// 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, 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, 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 { 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 { 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, ) -> ksp_core_lib::Result { 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) -> ksp_core_lib::Result { 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) -> ksp_core_lib::Result { 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, commitment: std::option::Option, ) -> ksp_core_lib::Result { 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 { 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(&self, operation: &'static str, path: &'static str, message: Request) -> ksp_core_lib::Result 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::() { 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::::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", &"") .finish(); } } /// Replay availability advertised by `SubscribeReplayInfo`. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct YellowstoneReplayInfo { first_available: std::option::Option, } impl YellowstoneReplayInfo { /// Creates a replay-info projection. #[must_use] pub const fn new(first_available: std::option::Option) -> 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 { 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) -> std::option::Option { 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(operation: &'static str, message: &str) -> ksp_core_lib::Result { 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;