v0.2.9-pre.003-fix.001

crates/ksp-onchain-transport-lib/src/grpc_unary.rs

@@ -0,0 +1,398 @@
+// file: crates/ksp-onchain-transport-lib/src/grpc_unary.rs
+// version: 1
+
+const MAX_YELLOWSTONE_BLOCKHASH_LENGTH_BYTES: usize = 256;
+const MAX_YELLOWSTONE_VERSION_LENGTH_BYTES: usize = 512;
+const PATH_GET_BLOCK_HEIGHT: &str = "/geyser.Geyser/GetBlockHeight";
+const PATH_GET_LATEST_BLOCKHASH: &str = "/geyser.Geyser/GetLatestBlockhash";
+const PATH_GET_SLOT: &str = "/geyser.Geyser/GetSlot";
+const PATH_GET_VERSION: &str = "/geyser.Geyser/GetVersion";
+const PATH_IS_BLOCKHASH_VALID: &str = "/geyser.Geyser/IsBlockhashValid";
+const PATH_PING: &str = "/geyser.Geyser/Ping";
+const PATH_SUBSCRIBE_REPLAY_INFO: &str = "/geyser.Geyser/SubscribeReplayInfo";
+
+/// Standard Solana Yellowstone unary client layered on one KSP-owned physical gRPC channel.
+///
+/// This N2 facade deliberately exposes no raw Tonic client and no upstream protobuf types. Streaming `Subscribe` is not part of `0.2.9-pre.003`.
+#[derive(Clone)]
+pub struct SolanaYellowstoneGrpcUnaryClient {
+    channel: tonic::transport::Channel,
+    metadata: std::vec::Vec<crate::YellowstoneGrpcMetadataEntry>,
+    unary_timeout: std::time::Duration,
+    max_inbound_message_size_bytes: usize,
+    max_outbound_message_size_bytes: usize,
+}
+
+impl SolanaYellowstoneGrpcUnaryClient {
+    /// Creates the internal unary facade state from a KSP-owned physical channel and validated endpoint settings.
+    pub(crate) fn new(
+        channel: tonic::transport::Channel,
+        metadata: std::vec::Vec<crate::YellowstoneGrpcMetadataEntry>,
+        settings: crate::YellowstoneGrpcSessionSettings,
+    ) -> Self {
+        return Self {
+            channel,
+            metadata,
+            unary_timeout: settings.unary_timeout(),
+            max_inbound_message_size_bytes: settings.max_inbound_message_size_bytes(),
+            max_outbound_message_size_bytes: settings.max_outbound_message_size_bytes(),
+        };
+    }
+
+    /// Returns the first slot retained by the endpoint for replay when advertised.
+    pub async fn subscribe_replay_info(&self) -> ksp_core_lib::Result<crate::YellowstoneReplayInfo> {
+        let response: yellowstone_grpc_proto::geyser::SubscribeReplayInfoResponse =
+            match self.unary("SubscribeReplayInfo", PATH_SUBSCRIBE_REPLAY_INFO, yellowstone_grpc_proto::geyser::SubscribeReplayInfoRequest {}).await {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+        return std::result::Result::Ok(crate::YellowstoneReplayInfo::new(response.first_available));
+    }
+
+    /// Executes the standard Yellowstone unary `Ping` RPC and verifies the echoed count.
+    pub async fn ping(&self, count: i32) -> ksp_core_lib::Result<crate::YellowstonePong> {
+        let response: yellowstone_grpc_proto::geyser::PongResponse =
+            match self.unary("Ping", PATH_PING, yellowstone_grpc_proto::geyser::PingRequest { count }).await {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+        if response.count != count {
+            return invalid_unary_response("Ping", "Yellowstone Ping response count does not match the request");
+        }
+        return std::result::Result::Ok(crate::YellowstonePong::new(response.count));
+    }
+
+    /// Returns the latest blockhash advertised by the Yellowstone endpoint at an optional commitment.
+    pub async fn get_latest_blockhash(
+        &self,
+        commitment: std::option::Option<crate::SolanaCommitment>,
+    ) -> ksp_core_lib::Result<crate::YellowstoneLatestBlockhash> {
+        let response: yellowstone_grpc_proto::geyser::GetLatestBlockhashResponse = match self
+            .unary(
+                "GetLatestBlockhash",
+                PATH_GET_LATEST_BLOCKHASH,
+                yellowstone_grpc_proto::geyser::GetLatestBlockhashRequest { commitment: commitment_to_wire(commitment) },
+            )
+            .await
+        {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        if response.blockhash.is_empty() || response.blockhash.len() > MAX_YELLOWSTONE_BLOCKHASH_LENGTH_BYTES {
+            return invalid_unary_response("GetLatestBlockhash", "Yellowstone latest blockhash response violates the KSP text bound");
+        }
+        return std::result::Result::Ok(crate::YellowstoneLatestBlockhash::new(response.slot, response.blockhash, response.last_valid_block_height));
+    }
+
+    /// Returns the current block height at an optional commitment.
+    pub async fn get_block_height(&self, commitment: std::option::Option<crate::SolanaCommitment>) -> ksp_core_lib::Result<crate::YellowstoneBlockHeight> {
+        let response: yellowstone_grpc_proto::geyser::GetBlockHeightResponse = match self
+            .unary(
+                "GetBlockHeight",
+                PATH_GET_BLOCK_HEIGHT,
+                yellowstone_grpc_proto::geyser::GetBlockHeightRequest { commitment: commitment_to_wire(commitment) },
+            )
+            .await
+        {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        return std::result::Result::Ok(crate::YellowstoneBlockHeight::new(response.block_height));
+    }
+
+    /// Returns the current slot at an optional commitment.
+    pub async fn get_slot(&self, commitment: std::option::Option<crate::SolanaCommitment>) -> ksp_core_lib::Result<crate::YellowstoneSlot> {
+        let response: yellowstone_grpc_proto::geyser::GetSlotResponse =
+            match self.unary("GetSlot", PATH_GET_SLOT, yellowstone_grpc_proto::geyser::GetSlotRequest { commitment: commitment_to_wire(commitment) }).await {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+        return std::result::Result::Ok(crate::YellowstoneSlot::new(response.slot));
+    }
+
+    /// Checks one opaque Solana blockhash at an optional commitment.
+    pub async fn is_blockhash_valid(
+        &self,
+        blockhash: impl std::convert::Into<std::string::String>,
+        commitment: std::option::Option<crate::SolanaCommitment>,
+    ) -> ksp_core_lib::Result<crate::YellowstoneBlockhashValidity> {
+        let blockhash = blockhash.into();
+        if blockhash.is_empty() || blockhash.len() > MAX_YELLOWSTONE_BLOCKHASH_LENGTH_BYTES || blockhash.trim() != blockhash {
+            return std::result::Result::Err(
+                ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RPC_PARAMETERS, "Yellowstone blockhash must be non-empty, bounded and untrimmed")
+                    .with_context("grpc_operation", "IsBlockhashValid")
+                    .with_context("field", "blockhash"),
+            );
+        }
+        let response: yellowstone_grpc_proto::geyser::IsBlockhashValidResponse = match self
+            .unary(
+                "IsBlockhashValid",
+                PATH_IS_BLOCKHASH_VALID,
+                yellowstone_grpc_proto::geyser::IsBlockhashValidRequest { blockhash, commitment: commitment_to_wire(commitment) },
+            )
+            .await
+        {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        return std::result::Result::Ok(crate::YellowstoneBlockhashValidity::new(response.slot, response.valid));
+    }
+
+    /// Returns the endpoint's Yellowstone/validator version string after applying a deterministic KSP text bound.
+    pub async fn get_version(&self) -> ksp_core_lib::Result<crate::YellowstoneVersionInfo> {
+        let response: yellowstone_grpc_proto::geyser::GetVersionResponse =
+            match self.unary("GetVersion", PATH_GET_VERSION, yellowstone_grpc_proto::geyser::GetVersionRequest {}).await {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+        if response.version.is_empty() || response.version.len() > MAX_YELLOWSTONE_VERSION_LENGTH_BYTES {
+            return invalid_unary_response("GetVersion", "Yellowstone version response violates the KSP text bound");
+        }
+        return std::result::Result::Ok(crate::YellowstoneVersionInfo::new(response.version));
+    }
+
+    async fn unary<Request, Response>(&self, operation: &'static str, path: &'static str, message: Request) -> ksp_core_lib::Result<Response>
+    where
+        Request: tonic_prost::prost::Message + Send + Sync + 'static,
+        Response: tonic_prost::prost::Message + std::default::Default + Send + Sync + 'static,
+    {
+        let path = match path.parse::<http::uri::PathAndQuery>() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => {
+                return std::result::Result::Err(
+                    ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, "internal Yellowstone gRPC method path is invalid")
+                        .with_context("grpc_operation", operation),
+                );
+            },
+        };
+        let mut grpc = tonic::client::Grpc::new(self.channel.clone())
+            .max_decoding_message_size(self.max_inbound_message_size_bytes)
+            .max_encoding_message_size(self.max_outbound_message_size_bytes);
+        let future = async {
+            if grpc.ready().await.is_err() {
+                return std::result::Result::Err(
+                    ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_CHANNEL_FAILED, "Yellowstone gRPC channel is not ready for unary dispatch")
+                        .with_context("grpc_operation", operation),
+                );
+            }
+            let mut request = tonic::Request::new(message);
+            for entry in &self.metadata {
+                if let std::result::Result::Err(error) = entry.append_to(request.metadata_mut()) {
+                    return std::result::Result::Err(error);
+                }
+            }
+            let response = match grpc.unary(request, path, tonic_prost::ProstCodec::<Request, Response>::default()).await {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(status) => return std::result::Result::Err(grpc_status_error(operation, status)),
+            };
+            return std::result::Result::Ok(response.into_inner());
+        };
+        return match tokio::time::timeout(self.unary_timeout, future).await {
+            std::result::Result::Ok(result) => result,
+            std::result::Result::Err(_) => std::result::Result::Err(
+                ksp_core_lib::Error::new(crate::ERROR_CODE_TIMEOUT, "Yellowstone gRPC unary operation exceeded the KSP deadline")
+                    .with_context("grpc_operation", operation),
+            ),
+        };
+    }
+}
+
+impl std::fmt::Debug for SolanaYellowstoneGrpcUnaryClient {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("SolanaYellowstoneGrpcUnaryClient")
+            .field("metadata_count", &self.metadata.len())
+            .field("unary_timeout", &self.unary_timeout)
+            .field("max_inbound_message_size_bytes", &self.max_inbound_message_size_bytes)
+            .field("max_outbound_message_size_bytes", &self.max_outbound_message_size_bytes)
+            .field("channel", &"<private>")
+            .finish();
+    }
+}
+
+/// Replay availability advertised by `SubscribeReplayInfo`.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct YellowstoneReplayInfo {
+    first_available: std::option::Option<u64>,
+}
+
+impl YellowstoneReplayInfo {
+    /// Creates a replay-info projection.
+    #[must_use]
+    pub const fn new(first_available: std::option::Option<u64>) -> Self {
+        return Self { first_available };
+    }
+
+    /// Returns the first replayable slot when the endpoint advertises one.
+    #[must_use]
+    pub const fn first_available(&self) -> std::option::Option<u64> {
+        return self.first_available;
+    }
+}
+
+/// Echo returned by the Yellowstone unary `Ping` RPC.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct YellowstonePong {
+    count: i32,
+}
+
+impl YellowstonePong {
+    /// Creates a ping response projection.
+    #[must_use]
+    pub const fn new(count: i32) -> Self {
+        return Self { count };
+    }
+
+    /// Returns the echoed ping count.
+    #[must_use]
+    pub const fn count(&self) -> i32 {
+        return self.count;
+    }
+}
+
+/// Latest blockhash response returned by Yellowstone.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct YellowstoneLatestBlockhash {
+    slot: u64,
+    blockhash: std::string::String,
+    last_valid_block_height: u64,
+}
+
+impl YellowstoneLatestBlockhash {
+    /// Creates a latest-blockhash projection.
+    #[must_use]
+    pub fn new(slot: u64, blockhash: std::string::String, last_valid_block_height: u64) -> Self {
+        return Self { slot, blockhash, last_valid_block_height };
+    }
+
+    /// Returns the response slot.
+    #[must_use]
+    pub const fn slot(&self) -> u64 {
+        return self.slot;
+    }
+
+    /// Returns the opaque Solana blockhash text.
+    #[must_use]
+    pub fn blockhash(&self) -> &str {
+        return self.blockhash.as_str();
+    }
+
+    /// Returns the last valid block height associated with the blockhash.
+    #[must_use]
+    pub const fn last_valid_block_height(&self) -> u64 {
+        return self.last_valid_block_height;
+    }
+}
+
+/// Current block height returned by Yellowstone.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct YellowstoneBlockHeight {
+    block_height: u64,
+}
+
+impl YellowstoneBlockHeight {
+    /// Creates a block-height projection.
+    #[must_use]
+    pub const fn new(block_height: u64) -> Self {
+        return Self { block_height };
+    }
+
+    /// Returns the block height.
+    #[must_use]
+    pub const fn block_height(&self) -> u64 {
+        return self.block_height;
+    }
+}
+
+/// Current slot returned by Yellowstone.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct YellowstoneSlot {
+    slot: u64,
+}
+
+impl YellowstoneSlot {
+    /// Creates a slot projection.
+    #[must_use]
+    pub const fn new(slot: u64) -> Self {
+        return Self { slot };
+    }
+
+    /// Returns the slot.
+    #[must_use]
+    pub const fn slot(&self) -> u64 {
+        return self.slot;
+    }
+}
+
+/// Result of checking one blockhash with Yellowstone.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct YellowstoneBlockhashValidity {
+    slot: u64,
+    valid: bool,
+}
+
+impl YellowstoneBlockhashValidity {
+    /// Creates a blockhash-validity projection.
+    #[must_use]
+    pub const fn new(slot: u64, valid: bool) -> Self {
+        return Self { slot, valid };
+    }
+
+    /// Returns the response slot.
+    #[must_use]
+    pub const fn slot(&self) -> u64 {
+        return self.slot;
+    }
+
+    /// Returns whether the blockhash is valid at the requested commitment.
+    #[must_use]
+    pub const fn valid(&self) -> bool {
+        return self.valid;
+    }
+}
+
+/// Bounded Yellowstone endpoint version information.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct YellowstoneVersionInfo {
+    version: std::string::String,
+}
+
+impl YellowstoneVersionInfo {
+    /// Creates a version projection.
+    #[must_use]
+    pub fn new(version: std::string::String) -> Self {
+        return Self { version };
+    }
+
+    /// Returns the bounded endpoint version string.
+    #[must_use]
+    pub fn version(&self) -> &str {
+        return self.version.as_str();
+    }
+}
+
+fn commitment_to_wire(commitment: std::option::Option<crate::SolanaCommitment>) -> std::option::Option<i32> {
+    return commitment.map(|value| {
+        return match value {
+            crate::SolanaCommitment::Processed => yellowstone_grpc_proto::geyser::CommitmentLevel::Processed as i32,
+            crate::SolanaCommitment::Confirmed => yellowstone_grpc_proto::geyser::CommitmentLevel::Confirmed as i32,
+            crate::SolanaCommitment::Finalized => yellowstone_grpc_proto::geyser::CommitmentLevel::Finalized as i32,
+        };
+    });
+}
+
+fn grpc_status_error(operation: &'static str, status: tonic::Status) -> ksp_core_lib::Error {
+    let code = status.code().to_string();
+    ksp_logging_lib::warn!(target: crate::TRACING_TARGET, grpc_operation = operation, grpc_code = code.as_str(), "Yellowstone gRPC unary endpoint returned a status");
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_GRPC_STATUS, "Yellowstone gRPC endpoint returned a gRPC status")
+        .with_context("grpc_operation", operation)
+        .with_context("grpc_code", code);
+}
+
+fn invalid_unary_response<T>(operation: &'static str, message: &str) -> ksp_core_lib::Result<T> {
+    return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, message).with_context("grpc_operation", operation));
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/grpc_unary.rs"]
+mod tests;