diff --git a/Cargo.toml b/Cargo.toml index b321431..ed14a57 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 248 +# version: 249 [workspace] resolver = "3" members = ["crates/ksp-app-config-desk", "crates/ksp-app-wallet-desk", "crates/ksp-config-lib", "crates/ksp-core-lib", "crates/ksp-logging-lib", "crates/ksp-onchain-transport-lib", "crates/ksp-wallet-lib"] [workspace.package] -version = "0.2.9-pre.8.fix.1" +version = "0.2.9-pre.9" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-onchain-transport-lib/src/error.rs b/crates/ksp-onchain-transport-lib/src/error.rs index 6c1f828..b46d193 100644 --- a/crates/ksp-onchain-transport-lib/src/error.rs +++ b/crates/ksp-onchain-transport-lib/src/error.rs @@ -1,10 +1,14 @@ // file: crates/ksp-onchain-transport-lib/src/error.rs -// version: 6 +// version: 7 /// Error code used when no logical endpoint can satisfy a request. pub const ERROR_CODE_ENDPOINT_SELECTION_FAILED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "endpoint_selection_failed"); +/// Error code used when a bounded Yellowstone gRPC request/update queue is exhausted. +pub const ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "grpc_backpressure_overflow"); /// Error code used when a Yellowstone gRPC channel cannot be prepared safely. pub const ERROR_CODE_GRPC_CHANNEL_FAILED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "grpc_channel_failed"); +/// Error code used when a Yellowstone gRPC subscribe session is no longer available to the caller. +pub const ERROR_CODE_GRPC_SESSION_CLOSED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "grpc_session_closed"); /// Error code used when a Yellowstone gRPC endpoint returns a remote gRPC status. pub const ERROR_CODE_GRPC_STATUS: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("onchain_transport", "grpc_status"); /// Error code used when an HTTP connection cannot be established. diff --git a/crates/ksp-onchain-transport-lib/src/grpc_channel.rs b/crates/ksp-onchain-transport-lib/src/grpc_channel.rs index e0e4f18..92e82ee 100644 --- a/crates/ksp-onchain-transport-lib/src/grpc_channel.rs +++ b/crates/ksp-onchain-transport-lib/src/grpc_channel.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/src/grpc_channel.rs -// version: 3 +// version: 4 /// Prepared or connected Yellowstone gRPC channel owned by KSP Transport. /// @@ -117,6 +117,23 @@ impl YellowstoneGrpcChannel { return crate::SolanaYellowstoneGrpcUnaryClient::new(self.channel.clone(), self.metadata.clone(), self.session.clone()); } + /// Opens one standard Solana Yellowstone bidirectional `Subscribe` session over this physical channel. + pub async fn open_standard_subscribe( + &self, + initial_request: crate::YellowstoneSubscribeRequest, + ) -> ksp_core_lib::Result { + return crate::open_yellowstone_subscribe_session( + self.channel.clone(), + self.metadata.clone(), + self.session.clone(), + self.endpoint_name.clone(), + self.provider.clone(), + self.cluster.clone(), + initial_request, + ) + .await; + } + fn from_parts(settings: &crate::YellowstoneGrpcEndpointSettings, channel: tonic::transport::Channel) -> Self { return Self { endpoint_name: settings.name().to_owned(), diff --git a/crates/ksp-onchain-transport-lib/src/grpc_stream.rs b/crates/ksp-onchain-transport-lib/src/grpc_stream.rs new file mode 100644 index 0000000..6ff6508 --- /dev/null +++ b/crates/ksp-onchain-transport-lib/src/grpc_stream.rs @@ -0,0 +1,584 @@ +// file: crates/ksp-onchain-transport-lib/src/grpc_stream.rs +// version: 1 + +use tonic_prost::prost::Message; // rust-rules: trait-import + +const AUTO_SUBSCRIBE_PING_ID: i32 = 1; +const PATH_SUBSCRIBE: &str = "/geyser.Geyser/Subscribe"; + +/// Safe lifecycle state of one standard Yellowstone bidirectional subscribe session. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum YellowstoneGrpcSubscribeState { + /// The bidirectional stream is active and accepts request mutations. + Active, + /// KSP has started a bounded graceful half-close. + Closing, + /// The stream ended normally, including a server half-close. + Closed, + /// The stream terminated because a transport, protocol, decoding or backpressure error occurred. + Failed, +} + +/// Standard Solana Yellowstone bidirectional `Subscribe` session layered on one KSP-owned physical gRPC channel. +/// +/// Request mutations and decoded updates use bounded Tokio channels. The raw Tonic stream and upstream protobuf messages remain private to Transport. +pub struct SolanaYellowstoneGrpcSubscribeSession { + endpoint_name: std::string::String, + provider: crate::YellowstoneGrpcProviderName, + cluster: crate::YellowstoneGrpcClusterName, + request_tx: std::option::Option>, + update_rx: tokio::sync::mpsc::Receiver>, + shutdown_tx: tokio::sync::watch::Sender>, + state_rx: tokio::sync::watch::Receiver, + task: tokio::task::JoinHandle<()>, + close_timeout: std::time::Duration, + max_outbound_message_size_bytes: usize, +} + +impl SolanaYellowstoneGrpcSubscribeSession { + /// Returns the safe logical endpoint name. + #[must_use] + pub fn endpoint_name(&self) -> &str { + return self.endpoint_name.as_str(); + } + + /// Returns the open provider descriptor without exposing endpoint credentials. + #[must_use] + pub const fn provider(&self) -> &crate::YellowstoneGrpcProviderName { + return &self.provider; + } + + /// Returns the open cluster descriptor. + #[must_use] + pub const fn cluster(&self) -> &crate::YellowstoneGrpcClusterName { + return &self.cluster; + } + + /// Returns the current safe stream lifecycle state. + #[must_use] + pub fn state(&self) -> crate::YellowstoneGrpcSubscribeState { + return self.state_rx.borrow().public_state(); + } + + /// Queues one complete standard Yellowstone request mutation without waiting for network dispatch. + /// + /// The mutation is rejected synchronously when local validation fails, the encoded request exceeds the configured outbound bound, the bounded request + /// queue is full, or the session is no longer active. Accepted mutations preserve bounded-channel admission order. + pub fn try_update(&self, request: &crate::YellowstoneSubscribeRequest) -> ksp_core_lib::Result<()> { + if self.state() != crate::YellowstoneGrpcSubscribeState::Active { + return std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_SESSION_CLOSED, + "Yellowstone subscribe session is not active", + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )); + } + let wire = match crate::yellowstone_subscribe_request_to_wire(request) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if wire.encoded_len() > self.max_outbound_message_size_bytes { + return std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW, + "Yellowstone subscribe request exceeds the configured outbound message bound", + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )); + } + let sender = match self.request_tx.as_ref() { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_SESSION_CLOSED, + "Yellowstone subscribe request channel is closed", + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )); + }, + }; + return match sender.try_send(wire) { + std::result::Result::Ok(()) => std::result::Result::Ok(()), + std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW, + "Yellowstone subscribe request queue is full", + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )), + std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_SESSION_CLOSED, + "Yellowstone subscribe request queue is closed", + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )), + }; + } + + /// Receives the next decoded standard Yellowstone update. + /// + /// A normal server half-close returns `Ok(None)`. Remote statuses, malformed updates and overflow terminate the session and surface a safe KSP error. + pub async fn next_update(&mut self) -> ksp_core_lib::Result> { + return match self.update_rx.recv().await { + std::option::Option::Some(std::result::Result::Ok(update)) => std::result::Result::Ok(std::option::Option::Some(update)), + std::option::Option::Some(std::result::Result::Err(error)) => { + self.request_tx.take(); + std::result::Result::Err(error) + }, + std::option::Option::None => { + self.request_tx.take(); + match *self.state_rx.borrow() { + SubscribeActorState::Closed => std::result::Result::Ok(std::option::Option::None), + SubscribeActorState::Failed(code) => std::result::Result::Err(subscribe_session_error( + code, + terminal_message(code), + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )), + SubscribeActorState::Active | SubscribeActorState::Closing => std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_SESSION_CLOSED, + "Yellowstone subscribe update channel ended before a terminal state was published", + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )), + } + }, + }; + } + + /// Gracefully half-closes the client request side and waits up to the configured close timeout for the server side to finish. + pub async fn close(mut self) -> ksp_core_lib::Result<()> { + let terminal_before_close = *self.state_rx.borrow(); + self.request_tx.take(); + if terminal_before_close == SubscribeActorState::Active { + let deadline = tokio::time::Instant::now() + self.close_timeout; + self.shutdown_tx.send_replace(std::option::Option::Some(deadline)); + } + let deadline = tokio::time::Instant::now() + self.close_timeout; + match tokio::time::timeout_at(deadline, &mut self.task).await { + std::result::Result::Ok(join_result) => { + if join_result.is_err() { + return std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_SESSION_CLOSED, + "Yellowstone subscribe actor terminated unexpectedly", + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )); + } + }, + std::result::Result::Err(_) => { + self.task.abort(); + let _ = (&mut self.task).await; + return std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_TIMEOUT, + "Yellowstone subscribe graceful shutdown exceeded the configured close timeout", + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )); + }, + } + return match *self.state_rx.borrow() { + SubscribeActorState::Closed => std::result::Result::Ok(()), + SubscribeActorState::Failed(code) => std::result::Result::Err(subscribe_session_error( + code, + terminal_message(code), + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )), + SubscribeActorState::Active | SubscribeActorState::Closing => std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_SESSION_CLOSED, + "Yellowstone subscribe actor ended without a terminal lifecycle state", + self.endpoint_name.as_str(), + self.provider.as_str(), + self.cluster.as_str(), + )), + }; + } +} + +impl std::fmt::Debug for SolanaYellowstoneGrpcSubscribeSession { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + return formatter + .debug_struct("SolanaYellowstoneGrpcSubscribeSession") + .field("endpoint_name", &self.endpoint_name) + .field("provider", &self.provider) + .field("cluster", &self.cluster) + .field("state", &self.state()) + .field("request_queue_capacity", &self.request_tx.as_ref().map(|sender| return sender.capacity())) + .field("update_queue_capacity", &self.update_rx.capacity()) + .finish(); + } +} + +impl Drop for SolanaYellowstoneGrpcSubscribeSession { + fn drop(&mut self) { + self.request_tx.take(); + if self.state() == crate::YellowstoneGrpcSubscribeState::Active { + self.shutdown_tx.send_replace(std::option::Option::Some(tokio::time::Instant::now() + self.close_timeout)); + } + } +} + +/// Opens one standard Yellowstone `Subscribe` stream from a validated KSP channel. +pub(crate) async fn open_yellowstone_subscribe_session( + channel: tonic::transport::Channel, + metadata: std::vec::Vec, + settings: crate::YellowstoneGrpcSessionSettings, + endpoint_name: std::string::String, + provider: crate::YellowstoneGrpcProviderName, + cluster: crate::YellowstoneGrpcClusterName, + initial_request: crate::YellowstoneSubscribeRequest, +) -> ksp_core_lib::Result { + let initial_wire = match crate::yellowstone_subscribe_request_to_wire(&initial_request) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if initial_wire.encoded_len() > settings.max_outbound_message_size_bytes() { + return std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW, + "initial Yellowstone subscribe request exceeds the configured outbound message bound", + endpoint_name.as_str(), + provider.as_str(), + cluster.as_str(), + )); + } + let (request_tx, request_rx) = tokio::sync::mpsc::channel(settings.request_channel_capacity()); + if request_tx.try_send(initial_wire).is_err() { + return std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW, + "initial Yellowstone subscribe request cannot enter the bounded request queue", + endpoint_name.as_str(), + provider.as_str(), + cluster.as_str(), + )); + } + let path = match PATH_SUBSCRIBE.parse::() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => { + return std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_CHANNEL_FAILED, + "internal Yellowstone Subscribe method path is invalid", + endpoint_name.as_str(), + provider.as_str(), + cluster.as_str(), + )); + }, + }; + let mut grpc = tonic::client::Grpc::new(channel) + .max_decoding_message_size(settings.max_inbound_message_size_bytes()) + .max_encoding_message_size(settings.max_outbound_message_size_bytes()); + let open_future = async { + if grpc.ready().await.is_err() { + return std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_GRPC_CHANNEL_FAILED, + "Yellowstone gRPC channel is not ready for Subscribe dispatch", + endpoint_name.as_str(), + provider.as_str(), + cluster.as_str(), + )); + } + let mut request = tonic::Request::new(MpscStream::new(request_rx)); + for entry in &metadata { + if let std::result::Result::Err(error) = entry.append_to(request.metadata_mut()) { + return std::result::Result::Err(error); + } + } + let response = match grpc + .streaming( + request, + path, + tonic_prost::ProstCodec::::default(), + ) + .await + { + std::result::Result::Ok(value) => value, + std::result::Result::Err(status) => { + return std::result::Result::Err(stream_status_error("SubscribeOpen", status, endpoint_name.as_str(), provider.as_str(), cluster.as_str())); + }, + }; + return std::result::Result::Ok(response.into_inner()); + }; + let incoming = match tokio::time::timeout(settings.connect_timeout(), open_future).await { + std::result::Result::Ok(std::result::Result::Ok(value)) => value, + std::result::Result::Ok(std::result::Result::Err(error)) => return std::result::Result::Err(error), + std::result::Result::Err(_) => { + return std::result::Result::Err(subscribe_session_error( + crate::ERROR_CODE_TIMEOUT, + "Yellowstone Subscribe stream opening exceeded the configured connection timeout", + endpoint_name.as_str(), + provider.as_str(), + cluster.as_str(), + )); + }, + }; + let (update_tx, update_rx) = tokio::sync::mpsc::channel(settings.update_channel_capacity()); + let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(std::option::Option::None::); + let (state_tx, state_rx) = tokio::sync::watch::channel(SubscribeActorState::Active); + let actor_request_tx = request_tx.clone(); + let actor_endpoint_name = endpoint_name.clone(); + let actor_provider = provider.clone(); + let actor_cluster = cluster.clone(); + let close_timeout = settings.close_timeout(); + let max_outbound_message_size_bytes = settings.max_outbound_message_size_bytes(); + let task = tokio::spawn(run_subscribe_actor( + actor_endpoint_name, + actor_provider, + actor_cluster, + incoming, + actor_request_tx, + update_tx, + shutdown_rx, + state_tx, + max_outbound_message_size_bytes, + )); + return std::result::Result::Ok(crate::SolanaYellowstoneGrpcSubscribeSession { + endpoint_name, + provider, + cluster, + request_tx: std::option::Option::Some(request_tx), + update_rx, + shutdown_tx, + state_rx, + task, + close_timeout, + max_outbound_message_size_bytes, + }); +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum SubscribeActorState { + Active, + Closing, + Closed, + Failed(ksp_core_lib::ErrorCode), +} + +impl SubscribeActorState { + const fn public_state(self) -> crate::YellowstoneGrpcSubscribeState { + return match self { + Self::Active => crate::YellowstoneGrpcSubscribeState::Active, + Self::Closing => crate::YellowstoneGrpcSubscribeState::Closing, + Self::Closed => crate::YellowstoneGrpcSubscribeState::Closed, + Self::Failed(_) => crate::YellowstoneGrpcSubscribeState::Failed, + }; + } +} + +struct MpscStream { + receiver: tokio::sync::mpsc::Receiver, +} + +impl MpscStream { + const fn new(receiver: tokio::sync::mpsc::Receiver) -> Self { + return Self { receiver }; + } +} + +impl futures_util::Stream for MpscStream { + type Item = T; + + fn poll_next(self: std::pin::Pin<&mut Self>, context: &mut std::task::Context<'_>) -> std::task::Poll> { + return self.get_mut().receiver.poll_recv(context); + } +} + +#[allow(clippy::too_many_arguments)] +async fn run_subscribe_actor( + endpoint_name: std::string::String, + provider: crate::YellowstoneGrpcProviderName, + cluster: crate::YellowstoneGrpcClusterName, + mut incoming: tonic::Streaming, + request_tx: tokio::sync::mpsc::Sender, + update_tx: tokio::sync::mpsc::Sender>, + mut shutdown_rx: tokio::sync::watch::Receiver>, + state_tx: tokio::sync::watch::Sender, + max_outbound_message_size_bytes: usize, +) { + let mut request_tx = std::option::Option::Some(request_tx); + loop { + tokio::select! { + shutdown_changed = shutdown_rx.changed() => { + let deadline = shutdown_deadline(&shutdown_rx, shutdown_changed); + state_tx.send_replace(SubscribeActorState::Closing); + request_tx.take(); + finish_client_half_close(&endpoint_name, &provider, &cluster, &mut incoming, deadline, &state_tx).await; + return; + } + message = incoming.message() => { + match message { + std::result::Result::Ok(std::option::Option::Some(wire)) => { + let update = match crate::yellowstone_subscribe_update_from_wire(wire) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + let _ = update_tx.try_send(std::result::Result::Err(error)); + state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_INVALID_RESPONSE)); + request_tx.take(); + return; + }, + }; + if matches!(&update, crate::YellowstoneSubscribeUpdate::Ping(_)) { + let ping_wire = ping_request_wire(); + if ping_wire.encoded_len() > max_outbound_message_size_bytes { + state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW)); + request_tx.take(); + return; + } + let sender = match request_tx.as_ref() { + std::option::Option::Some(value) => value, + std::option::Option::None => { + state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_SESSION_CLOSED)); + return; + }, + }; + if sender.try_send(ping_wire).is_err() { + state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW)); + request_tx.take(); + return; + } + } + match update_tx.try_send(std::result::Result::Ok(update)) { + std::result::Result::Ok(()) => {}, + std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + ksp_logging_lib::warn!( + target: crate::TRACING_TARGET, + endpoint_name = endpoint_name.as_str(), + provider = provider.as_str(), + cluster = cluster.as_str(), + "Yellowstone subscribe update queue overflowed" + ); + state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW)); + request_tx.take(); + return; + }, + std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { + state_tx.send_replace(SubscribeActorState::Closed); + request_tx.take(); + return; + }, + } + }, + std::result::Result::Ok(std::option::Option::None) => { + state_tx.send_replace(SubscribeActorState::Closed); + request_tx.take(); + return; + }, + std::result::Result::Err(status) => { + let error = stream_status_error("Subscribe", status, endpoint_name.as_str(), provider.as_str(), cluster.as_str()); + let _ = update_tx.try_send(std::result::Result::Err(error)); + state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_STATUS)); + request_tx.take(); + return; + }, + } + } + } + } +} + +async fn finish_client_half_close( + endpoint_name: &str, + provider: &crate::YellowstoneGrpcProviderName, + cluster: &crate::YellowstoneGrpcClusterName, + incoming: &mut tonic::Streaming, + deadline: tokio::time::Instant, + state_tx: &tokio::sync::watch::Sender, +) { + loop { + let message = tokio::time::timeout_at(deadline, incoming.message()).await; + match message { + std::result::Result::Ok(std::result::Result::Ok(std::option::Option::Some(_))) => {}, + std::result::Result::Ok(std::result::Result::Ok(std::option::Option::None)) => { + state_tx.send_replace(SubscribeActorState::Closed); + return; + }, + std::result::Result::Ok(std::result::Result::Err(status)) => { + let code = status.code().to_string(); + ksp_logging_lib::warn!( + target: crate::TRACING_TARGET, + endpoint_name, + provider = provider.as_str(), + cluster = cluster.as_str(), + grpc_code = code.as_str(), + "Yellowstone subscribe endpoint returned a status during graceful shutdown" + ); + state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_GRPC_STATUS)); + return; + }, + std::result::Result::Err(_) => { + state_tx.send_replace(SubscribeActorState::Failed(crate::ERROR_CODE_TIMEOUT)); + return; + }, + } + } +} + +fn shutdown_deadline( + shutdown_rx: &tokio::sync::watch::Receiver>, + shutdown_changed: std::result::Result<(), tokio::sync::watch::error::RecvError>, +) -> tokio::time::Instant { + return match shutdown_changed { + std::result::Result::Ok(()) => match *shutdown_rx.borrow() { + std::option::Option::Some(deadline) => deadline, + std::option::Option::None => tokio::time::Instant::now(), + }, + std::result::Result::Err(_) => tokio::time::Instant::now(), + }; +} + +fn ping_request_wire() -> yellowstone_grpc_proto::geyser::SubscribeRequest { + return yellowstone_grpc_proto::geyser::SubscribeRequest { + ping: std::option::Option::Some(yellowstone_grpc_proto::geyser::SubscribeRequestPing { id: AUTO_SUBSCRIBE_PING_ID }), + ..std::default::Default::default() + }; +} + +fn stream_status_error(operation: &'static str, status: tonic::Status, endpoint_name: &str, provider: &str, cluster: &str) -> ksp_core_lib::Error { + let code = status.code().to_string(); + ksp_logging_lib::warn!( + target: crate::TRACING_TARGET, + endpoint_name, + provider, + cluster, + grpc_operation = operation, + grpc_code = code.as_str(), + "Yellowstone subscribe endpoint returned a gRPC status" + ); + return subscribe_session_error(crate::ERROR_CODE_GRPC_STATUS, "Yellowstone subscribe endpoint returned a gRPC status", endpoint_name, provider, cluster) + .with_context("grpc_operation", operation) + .with_context("grpc_code", code); +} + +fn subscribe_session_error(code: ksp_core_lib::ErrorCode, message: &'static str, endpoint_name: &str, provider: &str, cluster: &str) -> ksp_core_lib::Error { + return ksp_core_lib::Error::new(code, message) + .with_context("endpoint_name", endpoint_name) + .with_context("provider", provider) + .with_context("cluster", cluster); +} + +fn terminal_message(code: ksp_core_lib::ErrorCode) -> &'static str { + if code == crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW { + return "Yellowstone subscribe session terminated because a bounded queue overflowed"; + } + if code == crate::ERROR_CODE_GRPC_STATUS { + return "Yellowstone subscribe session terminated after a remote gRPC status"; + } + if code == crate::ERROR_CODE_INVALID_RESPONSE { + return "Yellowstone subscribe session terminated after an invalid update"; + } + if code == crate::ERROR_CODE_TIMEOUT { + return "Yellowstone subscribe session exceeded a configured deadline"; + } + return "Yellowstone subscribe session is closed"; +} + +#[cfg(test)] +#[path = "../unit_tests/grpc_stream.rs"] +mod tests; diff --git a/crates/ksp-onchain-transport-lib/src/grpc_subscribe.rs b/crates/ksp-onchain-transport-lib/src/grpc_subscribe.rs index 1bdee46..110126b 100644 --- a/crates/ksp-onchain-transport-lib/src/grpc_subscribe.rs +++ b/crates/ksp-onchain-transport-lib/src/grpc_subscribe.rs @@ -1,11 +1,8 @@ // file: crates/ksp-onchain-transport-lib/src/grpc_subscribe.rs -// version: 6 +// version: 7 -#[cfg(test)] const MAX_GRPC_BLOCKHASH_TEXT_LENGTH_BYTES: usize = 128; -#[cfg(test)] const MAX_GRPC_BLOCK_VECTOR_COUNT: usize = 65_536; -#[cfg(test)] const MAX_GRPC_SUBSCRIBE_ACCOUNT_DATA_LENGTH_BYTES: usize = 512 * 1024 * 1024; const MAX_GRPC_SUBSCRIBE_ACCOUNT_PREDICATE_COUNT: usize = 256; const MAX_GRPC_SUBSCRIBE_ACCOUNT_SELECTOR_COUNT: usize = 50_000; @@ -16,26 +13,16 @@ const MAX_GRPC_SUBSCRIBE_FILTER_GROUP_COUNT: usize = 1_024; const MAX_GRPC_SUBSCRIBE_FILTER_NAME_LENGTH_BYTES: usize = 128; const MAX_GRPC_SUBSCRIBE_MEMCMP_BYTES: usize = 1024 * 1024; const MAX_GRPC_SUBSCRIBE_MEMCMP_TEXT_LENGTH_BYTES: usize = 2 * 1024 * 1024; -#[cfg(test)] const MAX_GRPC_SUBSCRIBE_SLOT_DEAD_ERROR_LENGTH_BYTES: usize = 16 * 1024; const MAX_GRPC_SUBSCRIBE_TRANSACTION_SIGNATURE_TEXT_LENGTH_BYTES: usize = 128; -#[cfg(test)] const MAX_GRPC_SUBSCRIBE_UPDATE_FILTER_COUNT: usize = 1_024; -#[cfg(test)] const MAX_GRPC_TRANSACTION_ERROR_BYTES: usize = 64 * 1024; -#[cfg(test)] const MAX_GRPC_TRANSACTION_INSTRUCTION_DATA_BYTES: usize = 1024 * 1024; -#[cfg(test)] const MAX_GRPC_TRANSACTION_LOG_COUNT: usize = 16_384; -#[cfg(test)] const MAX_GRPC_TRANSACTION_LOG_LENGTH_BYTES: usize = 64 * 1024; -#[cfg(test)] const MAX_GRPC_TRANSACTION_RETURN_DATA_BYTES: usize = 1024 * 1024; -#[cfg(test)] const MAX_GRPC_TRANSACTION_TEXT_LENGTH_BYTES: usize = 64 * 1024; -#[cfg(test)] const MAX_GRPC_TRANSACTION_VECTOR_COUNT: usize = 65_536; -#[cfg(test)] const YELLOWSTONE_HASH_LENGTH_BYTES: usize = 32; const YELLOWSTONE_TRANSACTION_SIGNATURE_WIRE_LENGTH_BYTES: usize = 64; @@ -105,7 +92,6 @@ impl YellowstoneAccountsDataSlice { return self.length; } - #[cfg(test)] fn to_wire(self) -> yellowstone_grpc_proto::geyser::SubscribeRequestAccountsDataSlice { return yellowstone_grpc_proto::geyser::SubscribeRequestAccountsDataSlice { offset: self.offset, length: self.length }; } @@ -130,7 +116,6 @@ impl YellowstoneSubscribePing { return self.id; } - #[cfg(test)] fn to_wire(self) -> yellowstone_grpc_proto::geyser::SubscribeRequestPing { return yellowstone_grpc_proto::geyser::SubscribeRequestPing { id: self.id }; } @@ -144,7 +129,6 @@ pub enum YellowstoneCuckooHashAlgorithm { } impl YellowstoneCuckooHashAlgorithm { - #[cfg(test)] const fn to_wire(self) -> i32 { return match self { Self::SipHash => yellowstone_grpc_proto::geyser::CuckooHashAlgorithm::SipHash as i32, @@ -222,7 +206,6 @@ impl YellowstoneCuckooFilter { return self.hash_algorithm; } - #[cfg(test)] fn to_wire(&self) -> yellowstone_grpc_proto::geyser::CuckooFilter { return yellowstone_grpc_proto::geyser::CuckooFilter { data: self.data.clone(), @@ -335,7 +318,6 @@ impl YellowstoneAccountMemcmp { }; } - #[cfg(test)] fn to_wire(&self) -> yellowstone_grpc_proto::geyser::SubscribeRequestFilterAccountsFilterMemcmp { let data = match &self.data { YellowstoneAccountMemcmpData::Bytes(value) => { @@ -394,7 +376,6 @@ pub enum YellowstoneAccountFilterPredicate { } impl YellowstoneAccountFilterPredicate { - #[cfg(test)] fn to_wire(&self) -> yellowstone_grpc_proto::geyser::SubscribeRequestFilterAccountsFilter { let filter = match self { Self::Memcmp(value) => yellowstone_grpc_proto::geyser::subscribe_request_filter_accounts_filter::Filter::Memcmp(value.to_wire()), @@ -514,7 +495,6 @@ impl YellowstoneSubscribeAccountFilter { return self.cuckoo_accounts_filter.as_ref(); } - #[cfg(test)] fn to_wire(&self) -> yellowstone_grpc_proto::geyser::SubscribeRequestFilterAccounts { return yellowstone_grpc_proto::geyser::SubscribeRequestFilterAccounts { account: self.accounts.iter().map(std::string::ToString::to_string).collect(), @@ -581,7 +561,6 @@ impl YellowstoneSubscribeSlotFilter { return self.interslot_updates; } - #[cfg(test)] fn to_wire(self) -> yellowstone_grpc_proto::geyser::SubscribeRequestFilterSlots { return yellowstone_grpc_proto::geyser::SubscribeRequestFilterSlots { filter_by_commitment: self.filter_by_commitment, @@ -872,7 +851,6 @@ pub enum YellowstoneTokenAccountExpansion { } impl YellowstoneTokenAccountExpansion { - #[cfg(test)] const fn to_wire(self) -> i32 { return match self { Self::All => yellowstone_grpc_proto::geyser::TokenAccountExpansionControlFlag::All as i32, @@ -1033,7 +1011,6 @@ impl YellowstoneSubscribeTransactionFilter { return self.token_accounts; } - #[cfg(test)] fn to_wire(&self) -> yellowstone_grpc_proto::geyser::SubscribeRequestFilterTransactions { return yellowstone_grpc_proto::geyser::SubscribeRequestFilterTransactions { vote: self.vote, @@ -2027,7 +2004,6 @@ impl YellowstoneSubscribeBlockFilter { return self.cuckoo_account_include.as_ref(); } - #[cfg(test)] fn to_wire(&self) -> yellowstone_grpc_proto::geyser::SubscribeRequestFilterBlocks { return yellowstone_grpc_proto::geyser::SubscribeRequestFilterBlocks { account_include: self.account_include.iter().map(std::string::ToString::to_string).collect(), @@ -2424,6 +2400,99 @@ impl std::fmt::Debug for YellowstoneEntryUpdate { } } +/// Server-side Yellowstone keepalive ping update. +#[derive(Clone, Eq, PartialEq)] +pub struct YellowstoneSubscribePingUpdate { + filters: std::vec::Vec, + created_at: std::option::Option, +} + +impl YellowstoneSubscribePingUpdate { + /// Returns echoed matching filter names in server order. + #[must_use] + pub fn filters(&self) -> &[crate::YellowstoneSubscribeFilterName] { + return self.filters.as_slice(); + } + + /// Returns the optional server creation timestamp. + #[must_use] + pub const fn created_at(&self) -> std::option::Option { + return self.created_at; + } +} + +impl std::fmt::Debug for YellowstoneSubscribePingUpdate { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + return formatter + .debug_struct("YellowstoneSubscribePingUpdate") + .field("filter_count", &self.filters.len()) + .field("created_at", &self.created_at) + .finish(); + } +} + +/// Server-side Yellowstone keepalive pong update. +#[derive(Clone, Eq, PartialEq)] +pub struct YellowstoneSubscribePongUpdate { + filters: std::vec::Vec, + created_at: std::option::Option, + id: i32, +} + +impl YellowstoneSubscribePongUpdate { + /// Returns echoed matching filter names in server order. + #[must_use] + pub fn filters(&self) -> &[crate::YellowstoneSubscribeFilterName] { + return self.filters.as_slice(); + } + + /// Returns the optional server creation timestamp. + #[must_use] + pub const fn created_at(&self) -> std::option::Option { + return self.created_at; + } + + /// Returns the exact ping identifier echoed by the server. + #[must_use] + pub const fn id(&self) -> i32 { + return self.id; + } +} + +impl std::fmt::Debug for YellowstoneSubscribePongUpdate { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + return formatter + .debug_struct("YellowstoneSubscribePongUpdate") + .field("filter_count", &self.filters.len()) + .field("created_at", &self.created_at) + .field("id", &self.id) + .finish(); + } +} + +/// Any standard Yellowstone update delivered by the bidirectional `Subscribe` stream. +#[derive(Clone, Debug, PartialEq)] +pub enum YellowstoneSubscribeUpdate { + /// Account update. + Account(crate::YellowstoneAccountUpdate), + /// Slot lifecycle update. + Slot(crate::YellowstoneSlotUpdate), + /// Full transaction update. + Transaction(crate::YellowstoneTransactionUpdate), + /// Lightweight transaction-status update. + TransactionStatus(crate::YellowstoneTransactionStatusUpdate), + /// Full block update. + Block(crate::YellowstoneBlockUpdate), + /// Server keepalive ping. KSP replies automatically with a ping request. + Ping(crate::YellowstoneSubscribePingUpdate), + /// Server keepalive pong. + Pong(crate::YellowstoneSubscribePongUpdate), + /// Block-metadata update. + BlockMeta(crate::YellowstoneBlockMetaUpdate), + /// Entry update. + Entry(crate::YellowstoneEntryUpdate), +} + /// Named empty filter activating the standard Yellowstone `blocks_meta` family. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct YellowstoneSubscribeBlocksMetaFilter { @@ -2437,7 +2506,6 @@ impl YellowstoneSubscribeBlocksMetaFilter { return Self { _private: () }; } - #[cfg(test)] fn to_wire(self) -> yellowstone_grpc_proto::geyser::SubscribeRequestFilterBlocksMeta { return yellowstone_grpc_proto::geyser::SubscribeRequestFilterBlocksMeta::default(); } @@ -2456,7 +2524,6 @@ impl YellowstoneSubscribeEntryFilter { return Self { _private: () }; } - #[cfg(test)] fn to_wire(self) -> yellowstone_grpc_proto::geyser::SubscribeRequestFilterEntry { return yellowstone_grpc_proto::geyser::SubscribeRequestFilterEntry::default(); } @@ -2712,7 +2779,6 @@ impl YellowstoneSubscribeRequest { return std::result::Result::Ok(()); } - #[cfg(test)] fn to_wire(&self) -> ksp_core_lib::Result { let validation = self.validate(); if let std::result::Result::Err(error) = validation { @@ -2770,7 +2836,6 @@ impl YellowstoneSubscribeRequest { } } -#[cfg(test)] fn decode_account_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> ksp_core_lib::Result { let (filters, created_at) = match decode_update_envelope(wire.filters, wire.created_at) { std::result::Result::Ok(value) => value, @@ -2790,7 +2855,6 @@ fn decode_account_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) return std::result::Result::Ok(crate::YellowstoneAccountUpdate { filters, created_at, account, slot: update.slot, is_startup: update.is_startup }); } -#[cfg(test)] fn decode_slot_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> ksp_core_lib::Result { let (filters, created_at) = match decode_update_envelope(wire.filters, wire.created_at) { std::result::Result::Ok(value) => value, @@ -2823,7 +2887,6 @@ fn decode_slot_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> }); } -#[cfg(test)] fn decode_transaction_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> ksp_core_lib::Result { let (filters, created_at) = match decode_update_envelope(wire.filters, wire.created_at) { std::result::Result::Ok(value) => value, @@ -2843,7 +2906,6 @@ fn decode_transaction_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpda return std::result::Result::Ok(crate::YellowstoneTransactionUpdate { filters, created_at, transaction, slot: update.slot }); } -#[cfg(test)] fn decode_transaction_status_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> ksp_core_lib::Result { let (filters, created_at) = match decode_update_envelope(wire.filters, wire.created_at) { std::result::Result::Ok(value) => value, @@ -2875,7 +2937,30 @@ fn decode_transaction_status_update(wire: yellowstone_grpc_proto::geyser::Subscr }); } -#[cfg(test)] +fn decode_ping_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> ksp_core_lib::Result { + let (filters, created_at) = match decode_update_envelope(wire.filters, wire.created_at) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + match wire.update_oneof { + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Ping(_)) => {}, + _ => return invalid_subscribe_response("ping", "Yellowstone update does not contain a ping payload"), + } + return std::result::Result::Ok(crate::YellowstoneSubscribePingUpdate { filters, created_at }); +} + +fn decode_pong_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> ksp_core_lib::Result { + let (filters, created_at) = match decode_update_envelope(wire.filters, wire.created_at) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let pong = match wire.update_oneof { + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Pong(value)) => value, + _ => return invalid_subscribe_response("pong", "Yellowstone update does not contain a pong payload"), + }; + return std::result::Result::Ok(crate::YellowstoneSubscribePongUpdate { filters, created_at, id: pong.id }); +} + fn decode_block_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> ksp_core_lib::Result { let (filters, created_at) = match decode_update_envelope(wire.filters, wire.created_at) { std::result::Result::Ok(value) => value, @@ -2951,7 +3036,6 @@ fn decode_block_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> }); } -#[cfg(test)] fn decode_block_meta_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> ksp_core_lib::Result { let (filters, created_at) = match decode_update_envelope(wire.filters, wire.created_at) { std::result::Result::Ok(value) => value, @@ -2991,7 +3075,6 @@ fn decode_block_meta_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdat }); } -#[cfg(test)] fn decode_entry_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> ksp_core_lib::Result { let (filters, created_at) = match decode_update_envelope(wire.filters, wire.created_at) { std::result::Result::Ok(value) => value, @@ -3008,7 +3091,6 @@ fn decode_entry_update(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate) -> return std::result::Result::Ok(crate::YellowstoneEntryUpdate { filters, created_at, entry }); } -#[cfg(test)] fn decode_block_rewards(wire: yellowstone_grpc_proto::solana::storage::confirmed_block::Rewards) -> ksp_core_lib::Result { if wire.rewards.len() > MAX_GRPC_BLOCK_VECTOR_COUNT { return invalid_subscribe_response("block.rewards", "Yellowstone block reward count exceeds the KSP bound"); @@ -3024,7 +3106,6 @@ fn decode_block_rewards(wire: yellowstone_grpc_proto::solana::storage::confirmed return std::result::Result::Ok(crate::YellowstoneBlockRewards { rewards, num_partitions: wire.num_partitions.map(|value| return value.num_partitions) }); } -#[cfg(test)] fn decode_entry_info(wire: yellowstone_grpc_proto::geyser::SubscribeUpdateEntry) -> ksp_core_lib::Result { let hash = match decode_entry_hash(wire.hash) { std::result::Result::Ok(value) => value, @@ -3040,7 +3121,6 @@ fn decode_entry_info(wire: yellowstone_grpc_proto::geyser::SubscribeUpdateEntry) }); } -#[cfg(test)] fn decode_entry_hash(bytes: std::vec::Vec) -> ksp_core_lib::Result { let bytes: [u8; YELLOWSTONE_HASH_LENGTH_BYTES] = match bytes.try_into() { std::result::Result::Ok(value) => value, @@ -3049,7 +3129,6 @@ fn decode_entry_hash(bytes: std::vec::Vec) -> ksp_core_lib::Result ksp_core_lib::Result { if value.is_empty() || value.len() > MAX_GRPC_BLOCKHASH_TEXT_LENGTH_BYTES @@ -3061,7 +3140,6 @@ fn decode_blockhash_text(field: &'static str, value: std::string::String) -> ksp return std::result::Result::Ok(value); } -#[cfg(test)] fn decode_transaction_info(wire: yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo) -> ksp_core_lib::Result { let signature = match decode_transaction_signature("transaction.signature", wire.signature) { std::result::Result::Ok(value) => value, @@ -3084,7 +3162,6 @@ fn decode_transaction_info(wire: yellowstone_grpc_proto::geyser::SubscribeUpdate return std::result::Result::Ok(crate::YellowstoneTransactionInfo { signature, is_vote: wire.is_vote, transaction, meta, index: wire.index }); } -#[cfg(test)] fn decode_stored_transaction( wire: yellowstone_grpc_proto::solana::storage::confirmed_block::Transaction, ) -> ksp_core_lib::Result { @@ -3109,7 +3186,6 @@ fn decode_stored_transaction( return std::result::Result::Ok(crate::YellowstoneStoredTransaction { signatures, message }); } -#[cfg(test)] fn decode_transaction_message( wire: yellowstone_grpc_proto::solana::storage::confirmed_block::Message, ) -> ksp_core_lib::Result { @@ -3174,7 +3250,6 @@ fn decode_transaction_message( }); } -#[cfg(test)] fn decode_compiled_instruction( wire: yellowstone_grpc_proto::solana::storage::confirmed_block::CompiledInstruction, ) -> ksp_core_lib::Result { @@ -3188,7 +3263,6 @@ fn decode_compiled_instruction( }); } -#[cfg(test)] fn decode_message_address_table_lookup( wire: yellowstone_grpc_proto::solana::storage::confirmed_block::MessageAddressTableLookup, ) -> ksp_core_lib::Result { @@ -3206,7 +3280,6 @@ fn decode_message_address_table_lookup( }); } -#[cfg(test)] fn decode_transaction_status_meta( wire: yellowstone_grpc_proto::solana::storage::confirmed_block::TransactionStatusMeta, ) -> ksp_core_lib::Result { @@ -3310,7 +3383,6 @@ fn decode_transaction_status_meta( }); } -#[cfg(test)] fn decode_inner_instructions( wire: yellowstone_grpc_proto::solana::storage::confirmed_block::InnerInstructions, ) -> ksp_core_lib::Result { @@ -3332,7 +3404,6 @@ fn decode_inner_instructions( return std::result::Result::Ok(crate::YellowstoneInnerInstructions { index: wire.index, instructions }); } -#[cfg(test)] fn decode_token_balance(wire: yellowstone_grpc_proto::solana::storage::confirmed_block::TokenBalance) -> ksp_core_lib::Result { if wire.mint.len() > MAX_GRPC_TRANSACTION_TEXT_LENGTH_BYTES || wire.owner.len() > MAX_GRPC_TRANSACTION_TEXT_LENGTH_BYTES @@ -3363,7 +3434,6 @@ fn decode_token_balance(wire: yellowstone_grpc_proto::solana::storage::confirmed }); } -#[cfg(test)] fn decode_return_data(wire: yellowstone_grpc_proto::solana::storage::confirmed_block::ReturnData) -> ksp_core_lib::Result { if wire.data.len() > MAX_GRPC_TRANSACTION_RETURN_DATA_BYTES { return invalid_subscribe_response("transaction.meta.return_data", "Yellowstone transaction return data exceeds the KSP bound"); @@ -3375,7 +3445,6 @@ fn decode_return_data(wire: yellowstone_grpc_proto::solana::storage::confirmed_b return std::result::Result::Ok(crate::YellowstoneReturnData { program_id, data: wire.data }); } -#[cfg(test)] fn decode_reward(wire: yellowstone_grpc_proto::solana::storage::confirmed_block::Reward) -> ksp_core_lib::Result { if wire.pubkey.len() > MAX_GRPC_TRANSACTION_TEXT_LENGTH_BYTES || wire.commission.len() > MAX_GRPC_TRANSACTION_TEXT_LENGTH_BYTES @@ -3410,7 +3479,6 @@ fn decode_reward(wire: yellowstone_grpc_proto::solana::storage::confirmed_block: }); } -#[cfg(test)] fn decode_transaction_error( field: &'static str, wire: yellowstone_grpc_proto::solana::storage::confirmed_block::TransactionError, @@ -3421,7 +3489,6 @@ fn decode_transaction_error( return std::result::Result::Ok(crate::YellowstoneTransactionError { bytes: wire.err }); } -#[cfg(test)] fn decode_transaction_signature(field: &'static str, bytes: std::vec::Vec) -> ksp_core_lib::Result { let bytes: [u8; YELLOWSTONE_TRANSACTION_SIGNATURE_WIRE_LENGTH_BYTES] = match bytes.try_into() { std::result::Result::Ok(value) => value, @@ -3430,7 +3497,6 @@ fn decode_transaction_signature(field: &'static str, bytes: std::vec::Vec) - return std::result::Result::Ok(crate::YellowstoneTransactionSignature::new(bytes)); } -#[cfg(test)] fn decode_hash_bytes(field: &'static str, bytes: std::vec::Vec) -> ksp_core_lib::Result { let bytes: [u8; YELLOWSTONE_HASH_LENGTH_BYTES] = match bytes.try_into() { std::result::Result::Ok(value) => value, @@ -3439,7 +3505,6 @@ fn decode_hash_bytes(field: &'static str, bytes: std::vec::Vec) -> ksp_core_ return std::result::Result::Ok(crate::YellowstoneHashBytes { bytes }); } -#[cfg(test)] fn decode_update_envelope( filters: std::vec::Vec, created_at: std::option::Option, @@ -3471,7 +3536,6 @@ fn decode_update_envelope( return std::result::Result::Ok((decoded_filters, decoded_created_at)); } -#[cfg(test)] fn decode_account_info(wire: yellowstone_grpc_proto::geyser::SubscribeUpdateAccountInfo) -> ksp_core_lib::Result { if wire.data.len() > MAX_GRPC_SUBSCRIBE_ACCOUNT_DATA_LENGTH_BYTES { return invalid_subscribe_response("account.data", "Yellowstone account data exceeds the KSP bound"); @@ -3508,7 +3572,6 @@ fn decode_account_info(wire: yellowstone_grpc_proto::geyser::SubscribeUpdateAcco }); } -#[cfg(test)] fn decode_pubkey_bytes(field: &'static str, bytes: std::vec::Vec) -> ksp_core_lib::Result { let array: [u8; 32] = match bytes.try_into() { std::result::Result::Ok(value) => value, @@ -3588,16 +3651,53 @@ fn validate_memcmp_text(value: &str) -> ksp_core_lib::Result<()> { return std::result::Result::Ok(()); } +/// Converts one validated KSP subscribe request to the internal Yellowstone protobuf wire. +pub(crate) fn yellowstone_subscribe_request_to_wire( + request: &crate::YellowstoneSubscribeRequest, +) -> ksp_core_lib::Result { + return request.to_wire(); +} + +/// Decodes one internal Yellowstone protobuf update into the complete KSP-owned update enum. +pub(crate) fn yellowstone_subscribe_update_from_wire( + wire: yellowstone_grpc_proto::geyser::SubscribeUpdate, +) -> ksp_core_lib::Result { + let kind = match wire.update_oneof.as_ref() { + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Account(_)) => 1_u8, + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Slot(_)) => 2_u8, + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Transaction(_)) => 3_u8, + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::TransactionStatus(_)) => 4_u8, + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Block(_)) => 5_u8, + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Ping(_)) => 6_u8, + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Pong(_)) => 7_u8, + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::BlockMeta(_)) => 8_u8, + std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Entry(_)) => 9_u8, + std::option::Option::None => { + return invalid_subscribe_response("update.oneof", "Yellowstone subscribe update is missing its oneof payload"); + }, + }; + return match kind { + 1 => decode_account_update(wire).map(crate::YellowstoneSubscribeUpdate::Account), + 2 => decode_slot_update(wire).map(crate::YellowstoneSubscribeUpdate::Slot), + 3 => decode_transaction_update(wire).map(crate::YellowstoneSubscribeUpdate::Transaction), + 4 => decode_transaction_status_update(wire).map(crate::YellowstoneSubscribeUpdate::TransactionStatus), + 5 => decode_block_update(wire).map(crate::YellowstoneSubscribeUpdate::Block), + 6 => decode_ping_update(wire).map(crate::YellowstoneSubscribeUpdate::Ping), + 7 => decode_pong_update(wire).map(crate::YellowstoneSubscribeUpdate::Pong), + 8 => decode_block_meta_update(wire).map(crate::YellowstoneSubscribeUpdate::BlockMeta), + 9 => decode_entry_update(wire).map(crate::YellowstoneSubscribeUpdate::Entry), + _ => invalid_subscribe_response("update.oneof", "Yellowstone subscribe update kind is unsupported"), + }; +} + fn invalid_subscribe_parameter(field: &'static str, message: &'static str) -> ksp_core_lib::Result { return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RPC_PARAMETERS, message).with_context("field", field)); } -#[cfg(test)] fn invalid_subscribe_response(field: &'static str, message: &'static str) -> ksp_core_lib::Result { return std::result::Result::Err(ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, message).with_context("grpc_update", field)); } -#[cfg(test)] fn commitment_to_wire(commitment: std::option::Option) -> std::option::Option { return commitment.map(|value| { return match value { diff --git a/crates/ksp-onchain-transport-lib/src/lib.rs b/crates/ksp-onchain-transport-lib/src/lib.rs index 5fed480..5bb88dc 100644 --- a/crates/ksp-onchain-transport-lib/src/lib.rs +++ b/crates/ksp-onchain-transport-lib/src/lib.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/src/lib.rs -// version: 41 +// version: 42 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -40,11 +40,16 @@ //! staged for `pre.007–008`. //! `0.2.9-pre.006` normalizes the five unambiguously HTTP-owned private implementation modules with an `http_` prefix while preserving shared `rpc_*`, //! JSON-RPC, error and constants modules. +//! `0.2.9-pre.007` completes the standard transaction/transaction-status filters and storage-wire projections; `pre.008` completes Blocks, block-meta and +//! entry projections. `0.2.9-pre.009` promotes those protobuf bridges into runtime and opens one KSP-owned bounded bidirectional `Subscribe` session with +//! request mutation, automatic server-Ping reply, observable Pong, normal server half-close, terminal backpressure and bounded graceful shutdown. Reconnect and +//! replay policy remain outside this tranche. mod constants; mod error; mod grpc_channel; mod grpc_settings; +mod grpc_stream; mod grpc_subscribe; mod grpc_unary; mod http_client; @@ -75,8 +80,12 @@ mod ws_transactions; /// Error code used when no logical endpoint can satisfy a request. pub use self::error::ERROR_CODE_ENDPOINT_SELECTION_FAILED; +/// Error code used when bounded Yellowstone gRPC runtime capacity is exhausted. +pub use self::error::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW; /// Error code used when a Yellowstone gRPC channel cannot be prepared safely. pub use self::error::ERROR_CODE_GRPC_CHANNEL_FAILED; +/// Error code used when a Yellowstone gRPC subscribe session is no longer available. +pub use self::error::ERROR_CODE_GRPC_SESSION_CLOSED; /// Error code used when a Yellowstone gRPC endpoint returns a remote status. pub use self::error::ERROR_CODE_GRPC_STATUS; /// Error code used when an HTTP connection cannot be established. @@ -129,6 +138,10 @@ pub use self::grpc_settings::YellowstoneGrpcReconnectSettings; pub use self::grpc_settings::YellowstoneGrpcSessionSettings; /// Complete runtime settings consumed by the KSP Yellowstone gRPC transport engine. pub use self::grpc_settings::YellowstoneGrpcTransportSettings; +/// Standard Yellowstone bidirectional Subscribe session. +pub use self::grpc_stream::SolanaYellowstoneGrpcSubscribeSession; +/// Safe Yellowstone bidirectional Subscribe lifecycle state. +pub use self::grpc_stream::YellowstoneGrpcSubscribeState; /// One validated standard Yellowstone account predicate. pub use self::grpc_subscribe::YellowstoneAccountFilterPredicate; /// Typed account payload carried by one standard Yellowstone account update. @@ -191,12 +204,18 @@ pub use self::grpc_subscribe::YellowstoneSubscribeEntryFilter; pub use self::grpc_subscribe::YellowstoneSubscribeFilterName; /// Optional ping mutation carried by the standard Yellowstone Subscribe request. pub use self::grpc_subscribe::YellowstoneSubscribePing; +/// Standard Yellowstone server Ping update. +pub use self::grpc_subscribe::YellowstoneSubscribePingUpdate; +/// Standard Yellowstone server Pong update. +pub use self::grpc_subscribe::YellowstoneSubscribePongUpdate; /// Provider-neutral standard Yellowstone Subscribe request. pub use self::grpc_subscribe::YellowstoneSubscribeRequest; /// Complete slot-family filter group for standard Yellowstone Subscribe. pub use self::grpc_subscribe::YellowstoneSubscribeSlotFilter; /// Complete transaction-family filter shared by transactions and transaction-status maps. pub use self::grpc_subscribe::YellowstoneSubscribeTransactionFilter; +/// Any standard Yellowstone Subscribe update. +pub use self::grpc_subscribe::YellowstoneSubscribeUpdate; /// Optional token-account owner expansion for current Yellowstone transaction filters. pub use self::grpc_subscribe::YellowstoneTokenAccountExpansion; /// One pre/post token balance from Yellowstone transaction status metadata. @@ -558,6 +577,12 @@ pub use self::ws_transactions::SolanaSignatureSubscribeConfig; /// Owning tracing target for events emitted by the on-chain transport crate. pub(crate) use self::constants::TRACING_TARGET; +/// Internal Yellowstone Subscribe session opener used by the physical channel. +pub(crate) use self::grpc_stream::open_yellowstone_subscribe_session; +/// Internal Yellowstone Subscribe request wire conversion shared with the stream engine. +pub(crate) use self::grpc_subscribe::yellowstone_subscribe_request_to_wire; +/// Internal Yellowstone Subscribe update decoder shared with the stream engine. +pub(crate) use self::grpc_subscribe::yellowstone_subscribe_update_from_wire; /// Crate-internal `HttpConcurrencyPermit` state shared across the owning crate. pub(crate) use self::http_resilience::HttpConcurrencyPermit; /// Crate-internal `HttpRoleRuntime` state shared across the owning crate. diff --git a/crates/ksp-onchain-transport-lib/tests/public_api.rs b/crates/ksp-onchain-transport-lib/tests/public_api.rs index 26df56d..f0ff5f4 100644 --- a/crates/ksp-onchain-transport-lib/tests/public_api.rs +++ b/crates/ksp-onchain-transport-lib/tests/public_api.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/tests/public_api.rs -// version: 45 +// version: 46 //! Integration tests for the public `ksp-onchain-transport-lib` consumer contract. @@ -969,3 +969,26 @@ fn public_v0_2_9_pre_008_yellowstone_blocks_contract_is_available_from_crate_roo let _entry_info = std::any::type_name::(); let _entry_update = std::any::type_name::(); } + +#[test] +fn public_v0_2_9_pre_009_yellowstone_bidi_session_contract_is_available_from_crate_root() { + fn assert_send() { + return; + } + assert_send::(); + assert_send::(); + let _open = ksp_onchain_transport_lib::YellowstoneGrpcChannel::open_standard_subscribe; + let _session = std::any::type_name::(); + let _ping = std::any::type_name::(); + let _pong = std::any::type_name::(); + let _states = [ + ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Active, + ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Closing, + ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Closed, + ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Failed, + ]; + assert_eq!(ksp_onchain_transport_lib::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW.domain(), "onchain_transport"); + assert_eq!(ksp_onchain_transport_lib::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW.code(), "grpc_backpressure_overflow"); + assert_eq!(ksp_onchain_transport_lib::ERROR_CODE_GRPC_SESSION_CLOSED.domain(), "onchain_transport"); + assert_eq!(ksp_onchain_transport_lib::ERROR_CODE_GRPC_SESSION_CLOSED.code(), "grpc_session_closed"); +} diff --git a/crates/ksp-onchain-transport-lib/tests/release_completeness.rs b/crates/ksp-onchain-transport-lib/tests/release_completeness.rs index 256ae3e..83779e4 100644 --- a/crates/ksp-onchain-transport-lib/tests/release_completeness.rs +++ b/crates/ksp-onchain-transport-lib/tests/release_completeness.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/tests/release_completeness.rs -// version: 39 +// version: 40 //! Release-level completeness canaries for staged HTTP and WebSocket Transport coverage. @@ -1144,8 +1144,8 @@ fn release_v0_2_9_pre_005_accounts_and_slots_contract_remains_complete() { assert!(source.contains("MAX_GRPC_SUBSCRIBE_CUCKOO_DATA_LENGTH_BYTES")); assert!(source.contains("MAX_GRPC_SUBSCRIBE_ACCOUNT_DATA_LENGTH_BYTES")); assert!(source.contains("MAX_GRPC_SUBSCRIBE_SLOT_DEAD_ERROR_LENGTH_BYTES")); - assert!(source.contains("#[cfg(test)]\nfn decode_account_update")); - assert!(source.contains("#[cfg(test)]\nfn decode_slot_update")); + assert!(source.contains("fn decode_account_update")); + assert!(source.contains("fn decode_slot_update")); assert!(!source.contains("SubscribeDeshred")); assert!(!source.contains("PublicNode")); assert!(!crate_root.contains("pub use tonic")); @@ -1180,7 +1180,7 @@ fn release_v0_2_9_pre_006_namespaces_unambiguously_http_owned_private_modules() } #[test] -fn release_v0_2_9_pre_007_transactions_contract_remains_complete_without_bidi() { +fn release_v0_2_9_pre_007_transactions_contract_remains_complete() { let source = include_str!("../src/grpc_subscribe.rs"); let crate_root = include_str!("../src/lib.rs"); for required in [ @@ -1221,7 +1221,7 @@ fn release_v0_2_9_pre_007_transactions_contract_remains_complete_without_bidi() } #[test] -fn release_v0_2_9_pre_008_completes_standard_blocks_without_advancing_bidi() { +fn release_v0_2_9_pre_008_standard_blocks_contract_remains_complete() { let source = include_str!("../src/grpc_subscribe.rs"); let crate_root = include_str!("../src/lib.rs"); for required in [ @@ -1252,7 +1252,7 @@ fn release_v0_2_9_pre_008_completes_standard_blocks_without_advancing_bidi() { assert!(source.contains("std::vec::Vec")); assert!(source.contains("std::vec::Vec")); assert!(source.contains("std::vec::Vec")); - assert!(source.contains("#[cfg(test)]\nfn decode_block_update")); + assert!(source.contains("fn decode_block_update")); assert!(!source.contains("pub async fn subscribe(")); assert!(!source.contains("SubscribeDeshred")); assert!(!source.contains("PublicNode")); @@ -1262,3 +1262,49 @@ fn release_v0_2_9_pre_008_completes_standard_blocks_without_advancing_bidi() { let _meta = std::any::type_name::(); let _entry = std::any::type_name::(); } + +#[test] +fn release_v0_2_9_pre_009_opens_one_bounded_standard_bidi_session_without_reconnect_or_provider_coupling() { + let stream_source = include_str!("../src/grpc_stream.rs"); + let subscribe_source = include_str!("../src/grpc_subscribe.rs"); + let channel_source = include_str!("../src/grpc_channel.rs"); + let crate_root = include_str!("../src/lib.rs"); + for required in [ + "const PATH_SUBSCRIBE: &str = \"/geyser.Geyser/Subscribe\"", + ".streaming(", + "tokio::sync::mpsc::channel", + "tokio::sync::watch::channel", + "try_send", + "AUTO_SUBSCRIBE_PING_ID", + "ping_request_wire", + "finish_client_half_close", + "timeout_at", + "max_decoding_message_size", + "max_encoding_message_size", + "YellowstoneGrpcSubscribeState", + "SolanaYellowstoneGrpcSubscribeSession", + ] { + assert!(stream_source.contains(required), "missing pre.009 bidi runtime token: {required}"); + } + for required in [ + "YellowstoneSubscribeUpdate", + "YellowstoneSubscribePingUpdate", + "YellowstoneSubscribePongUpdate", + "yellowstone_subscribe_request_to_wire", + "yellowstone_subscribe_update_from_wire", + ] { + assert!(subscribe_source.contains(required), "missing pre.009 runtime wire bridge token: {required}"); + } + assert!(channel_source.contains("pub async fn open_standard_subscribe")); + assert!(!stream_source.contains("unbounded_channel")); + assert!(!stream_source.contains("SubscribeDeshred")); + assert!(!stream_source.contains("PublicNode")); + assert!(!stream_source.contains("OrbitFlare")); + assert!(!stream_source.contains("Helius")); + assert!(!stream_source.contains("reconnect")); + assert!(!crate_root.contains("pub use tonic")); + assert!(!crate_root.contains("pub use yellowstone_grpc_proto")); + let _session = std::any::type_name::(); + let _state = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Active; + let _update = std::any::type_name::(); +} diff --git a/crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs b/crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs new file mode 100644 index 0000000..4685914 --- /dev/null +++ b/crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs @@ -0,0 +1,468 @@ +// file: crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs +// version: 1 + +#[derive(Clone, Copy)] +enum FixtureMode { + RoundTrip, + ClientHalfClose, + HostileClose, + Flood, + RemoteStatus, + Malformed, + Oversized, + Idle, +} + +#[derive(Clone)] +struct FixtureGeyser { + mode: FixtureMode, + half_close_seen: std::sync::Arc, +} + +#[allow(clippy::implicit_return)] // tonic::async_trait generates async wrapper tails outside the authored fixture bodies. +#[tonic::async_trait] +impl yellowstone_grpc_proto::geyser::geyser_server::Geyser for FixtureGeyser { + type SubscribeStream = super::MpscStream>; + type SubscribeDeshredStream = futures_util::stream::Empty>; + + async fn subscribe( + &self, + request: tonic::Request>, + ) -> std::result::Result, tonic::Status> { + if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) { + return std::result::Result::Err(error); + } + let mut inbound = request.into_inner(); + let (outbound_tx, outbound_rx) = tokio::sync::mpsc::channel(64); + let mode = self.mode; + let half_close_seen = self.half_close_seen.clone(); + tokio::spawn(async move { + let initial = inbound.message().await; + let initial = match initial { + std::result::Result::Ok(std::option::Option::Some(value)) => value, + _ => return, + }; + match mode { + FixtureMode::RoundTrip => { + assert!(initial.slots.contains_key("initial")); + if outbound_tx.send(std::result::Result::Ok(ping_update())).await.is_err() { + return; + } + let ping = inbound.message().await; + let ping = match ping { + std::result::Result::Ok(std::option::Option::Some(value)) => value, + _ => return, + }; + assert_eq!(ping.ping.map(|value| value.id), std::option::Option::Some(1)); + assert!(ping.accounts.is_empty()); + assert!(ping.slots.is_empty()); + assert!(ping.transactions.is_empty()); + assert!(ping.transactions_status.is_empty()); + assert!(ping.blocks.is_empty()); + assert!(ping.blocks_meta.is_empty()); + assert!(ping.entry.is_empty()); + if outbound_tx.send(std::result::Result::Ok(pong_update(1))).await.is_err() { + return; + } + let mutation = inbound.message().await; + let mutation = match mutation { + std::result::Result::Ok(std::option::Option::Some(value)) => value, + _ => return, + }; + assert!(mutation.accounts.contains_key("mutated")); + let _ = outbound_tx.send(std::result::Result::Ok(slot_update(901))).await; + }, + FixtureMode::ClientHalfClose => { + let half_close = inbound.message().await; + if matches!(half_close, std::result::Result::Ok(std::option::Option::None)) { + half_close_seen.store(true, std::sync::atomic::Ordering::SeqCst); + } + }, + FixtureMode::HostileClose => { + let half_close = inbound.message().await; + if matches!(half_close, std::result::Result::Ok(std::option::Option::None)) { + half_close_seen.store(true, std::sync::atomic::Ordering::SeqCst); + } + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + drop(outbound_tx); + }, + FixtureMode::Flood => { + for slot in 0_u64..32_u64 { + if outbound_tx.send(std::result::Result::Ok(slot_update(1_000 + slot))).await.is_err() { + return; + } + } + }, + FixtureMode::RemoteStatus => { + let _ = outbound_tx.send(std::result::Result::Err(tonic::Status::permission_denied("GRPC-STREAM-REMOTE-SECRET-CANARY"))).await; + }, + FixtureMode::Malformed => { + let _ = outbound_tx + .send(std::result::Result::Ok(yellowstone_grpc_proto::geyser::SubscribeUpdate { + filters: std::vec::Vec::new(), + update_oneof: std::option::Option::None, + created_at: std::option::Option::None, + })) + .await; + }, + FixtureMode::Oversized => { + let _ = outbound_tx.send(std::result::Result::Ok(slot_update(77))).await; + }, + FixtureMode::Idle => { + let _ = inbound.message().await; + }, + } + }); + return std::result::Result::Ok(tonic::Response::new(super::MpscStream::new(outbound_rx))); + } + + async fn subscribe_deshred( + &self, + _request: tonic::Request>, + ) -> std::result::Result, tonic::Status> { + return std::result::Result::Err(tonic::Status::unimplemented("deshred is outside KSP 0.2.9")); + } + + async fn subscribe_replay_info( + &self, + _request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture")); + } + + async fn ping( + &self, + _request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture")); + } + + async fn get_latest_blockhash( + &self, + _request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture")); + } + + async fn get_block_height( + &self, + _request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture")); + } + + async fn get_slot( + &self, + _request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture")); + } + + async fn is_blockhash_valid( + &self, + _request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture")); + } + + async fn get_version( + &self, + _request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture")); + } +} + +struct FixtureServer { + endpoint_url: std::string::String, + half_close_seen: std::sync::Arc, + shutdown: std::option::Option>, + task: tokio::task::JoinHandle<()>, +} + +impl FixtureServer { + async fn start(mode: FixtureMode) -> Self { + let bind_address: std::net::SocketAddr = "127.0.0.1:0".parse().expect("fixture bind address must parse"); + let incoming = tonic::transport::server::TcpIncoming::bind(bind_address).expect("fixture gRPC listener must bind"); + let local_address = incoming.local_addr().expect("fixture gRPC listener must expose local address"); + let (shutdown, shutdown_receiver) = tokio::sync::oneshot::channel(); + let half_close_seen = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let service_half_close_seen = half_close_seen.clone(); + let task = tokio::spawn(async move { + let service = yellowstone_grpc_proto::geyser::geyser_server::GeyserServer::new(FixtureGeyser { mode, half_close_seen: service_half_close_seen }); + let result = tonic::transport::Server::builder() + .serve_with_incoming_shutdown(service, incoming, async move { + let _ = shutdown_receiver.await; + }) + .await; + assert!(result.is_ok()); + }); + return Self { + endpoint_url: format!("http://{local_address}"), + half_close_seen, + shutdown: std::option::Option::Some(shutdown), + task, + }; + } + + async fn stop(mut self) { + if let std::option::Option::Some(shutdown) = self.shutdown.take() { + let _ = shutdown.send(()); + } + let result = self.task.await; + assert!(result.is_ok()); + } +} + +fn verify_fixture_metadata(metadata: &tonic::metadata::MetadataMap) -> std::result::Result<(), tonic::Status> { + let public = metadata.get("x-ksp-public").and_then(|value| return value.to_str().ok()); + let secret = metadata.get("x-ksp-token").and_then(|value| return value.to_str().ok()); + if public != std::option::Option::Some("fixture-public") || secret != std::option::Option::Some("GRPC-STREAM-SECRET-CANARY") { + return std::result::Result::Err(tonic::Status::unauthenticated("fixture metadata mismatch")); + } + return std::result::Result::Ok(()); +} + +fn fixture_settings( + url: &str, + request_capacity: usize, + update_capacity: usize, + max_inbound_message_size_bytes: usize, + max_outbound_message_size_bytes: usize, +) -> crate::YellowstoneGrpcEndpointSettings { + let defaults = crate::YellowstoneGrpcSessionSettings::default(); + let session = crate::YellowstoneGrpcSessionSettings::new( + defaults.connect_timeout(), + defaults.unary_timeout(), + std::time::Duration::from_millis(250), + defaults.reconnect().clone(), + request_capacity, + update_capacity, + max_inbound_message_size_bytes, + max_outbound_message_size_bytes, + ); + let metadata = std::vec![ + crate::YellowstoneGrpcMetadataEntry::public("x-ksp-public", "fixture-public").expect("fixture public metadata must be valid"), + crate::YellowstoneGrpcMetadataEntry::secret("x-ksp-token", "GRPC-STREAM-SECRET-CANARY").expect("fixture secret metadata must be valid"), + ]; + return crate::YellowstoneGrpcEndpointSettings::new( + "fixture-stream", + true, + crate::YellowstoneGrpcProviderName::new("fixture-provider"), + crate::YellowstoneGrpcClusterName::new("devnet"), + crate::YellowstoneGrpcEndpointUrl::parse(url).expect("fixture URL must parse"), + session, + ) + .with_metadata(metadata) + .expect("fixture metadata settings must validate"); +} + +fn initial_request() -> crate::YellowstoneSubscribeRequest { + let mut request = crate::YellowstoneSubscribeRequest::new(); + request + .insert_slot_filter( + crate::YellowstoneSubscribeFilterName::new("initial").expect("fixture filter name must validate"), + crate::YellowstoneSubscribeSlotFilter::new(), + ) + .expect("fixture slot filter must insert"); + return request; +} + +fn mutated_request() -> crate::YellowstoneSubscribeRequest { + let mut request = crate::YellowstoneSubscribeRequest::new(); + request + .insert_account_filter( + crate::YellowstoneSubscribeFilterName::new("mutated").expect("fixture filter name must validate"), + crate::YellowstoneSubscribeAccountFilter::new(), + ) + .expect("fixture account filter must insert"); + return request; +} + +fn ping_update() -> yellowstone_grpc_proto::geyser::SubscribeUpdate { + return yellowstone_grpc_proto::geyser::SubscribeUpdate { + filters: std::vec::Vec::new(), + update_oneof: std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Ping( + yellowstone_grpc_proto::geyser::SubscribeUpdatePing {}, + )), + created_at: std::option::Option::None, + }; +} + +fn pong_update(id: i32) -> yellowstone_grpc_proto::geyser::SubscribeUpdate { + return yellowstone_grpc_proto::geyser::SubscribeUpdate { + filters: std::vec::Vec::new(), + update_oneof: std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Pong( + yellowstone_grpc_proto::geyser::SubscribeUpdatePong { id }, + )), + created_at: std::option::Option::None, + }; +} + +fn slot_update(slot: u64) -> yellowstone_grpc_proto::geyser::SubscribeUpdate { + return yellowstone_grpc_proto::geyser::SubscribeUpdate { + filters: std::vec!["initial".to_owned()], + update_oneof: std::option::Option::Some(yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof::Slot( + yellowstone_grpc_proto::geyser::SubscribeUpdateSlot { + slot, + parent: std::option::Option::Some(slot.saturating_sub(1)), + status: yellowstone_grpc_proto::geyser::SlotStatus::SlotProcessed as i32, + dead_error: std::option::Option::None, + }, + )), + created_at: std::option::Option::None, + }; +} + +#[tokio::test(flavor = "current_thread")] +async fn yellowstone_bidi_round_trip_mutates_request_replies_to_ping_and_observes_server_half_close() { + let server = FixtureServer::start(FixtureMode::RoundTrip).await; + let defaults = crate::YellowstoneGrpcSessionSettings::default(); + let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes()); + let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); + let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open"); + assert_eq!(session.state(), crate::YellowstoneGrpcSubscribeState::Active); + let ping = session.next_update().await.expect("Ping update must decode").expect("Ping update must be present"); + assert!(matches!(ping, crate::YellowstoneSubscribeUpdate::Ping(_))); + let pong = session.next_update().await.expect("Pong update must decode").expect("Pong update must be present"); + match pong { + crate::YellowstoneSubscribeUpdate::Pong(value) => assert_eq!(value.id(), 1), + _ => panic!("fixture must return Pong"), + } + session.try_update(&mutated_request()).expect("request mutation must enter bounded queue"); + let slot = session.next_update().await.expect("slot update must decode").expect("slot update must be present"); + match slot { + crate::YellowstoneSubscribeUpdate::Slot(value) => assert_eq!(value.slot(), 901), + _ => panic!("fixture must return Slot"), + } + assert!(session.next_update().await.expect("server half-close must be normal").is_none()); + assert_eq!(session.state(), crate::YellowstoneGrpcSubscribeState::Closed); + session.close().await.expect("already half-closed fixture must close cleanly"); + server.stop().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn yellowstone_explicit_close_half_closes_request_stream_before_deadline() { + let server = FixtureServer::start(FixtureMode::ClientHalfClose).await; + let defaults = crate::YellowstoneGrpcSessionSettings::default(); + let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes()); + let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); + let session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open"); + session.close().await.expect("graceful half-close must complete"); + assert!(server.half_close_seen.load(std::sync::atomic::Ordering::SeqCst)); + server.stop().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn yellowstone_hostile_server_shutdown_is_bounded_by_close_timeout() { + let server = FixtureServer::start(FixtureMode::HostileClose).await; + let defaults = crate::YellowstoneGrpcSessionSettings::default(); + let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes()); + let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); + let session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open"); + let error = session.close().await.expect_err("hostile server must hit the bounded graceful close deadline"); + assert_eq!(error.code(), crate::ERROR_CODE_TIMEOUT); + assert!(server.half_close_seen.load(std::sync::atomic::Ordering::SeqCst)); + server.stop().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn yellowstone_session_drop_best_effort_half_closes_request_stream() { + let server = FixtureServer::start(FixtureMode::ClientHalfClose).await; + let defaults = crate::YellowstoneGrpcSessionSettings::default(); + let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes()); + let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); + let session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open"); + drop(session); + let observed = tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if server.half_close_seen.load(std::sync::atomic::Ordering::SeqCst) { + return true; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("fixture must observe dropped-session half-close before timeout"); + assert!(observed); + server.stop().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn yellowstone_slow_receiver_overflow_is_terminal_and_observable() { + let server = FixtureServer::start(FixtureMode::Flood).await; + let defaults = crate::YellowstoneGrpcSessionSettings::default(); + let settings = fixture_settings(server.endpoint_url.as_str(), 8, 2, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes()); + let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); + let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open"); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + assert!(session.next_update().await.expect("first queued update must remain readable").is_some()); + assert!(session.next_update().await.expect("second queued update must remain readable").is_some()); + let error = session.next_update().await.expect_err("overflow must become terminal after bounded queued updates drain"); + assert_eq!(error.code(), crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW); + let rendered = format!("{error:?} {session:?}"); + assert!(!rendered.contains("GRPC-STREAM-SECRET-CANARY")); + assert!(!rendered.contains(server.endpoint_url.as_str())); + let _ = session.close().await; + server.stop().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn yellowstone_stream_remote_status_is_safe_and_terminal() { + let server = FixtureServer::start(FixtureMode::RemoteStatus).await; + let defaults = crate::YellowstoneGrpcSessionSettings::default(); + let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes()); + let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); + let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open"); + let error = session.next_update().await.expect_err("remote Status must fail stream"); + assert_eq!(error.code(), crate::ERROR_CODE_GRPC_STATUS); + let rendered = format!("{error:?} {session:?}"); + assert!(!rendered.contains("GRPC-STREAM-REMOTE-SECRET-CANARY")); + assert!(!rendered.contains("GRPC-STREAM-SECRET-CANARY")); + assert!(!rendered.contains(server.endpoint_url.as_str())); + let _ = session.close().await; + server.stop().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn yellowstone_malformed_update_is_rejected_without_raw_wire_escape() { + let server = FixtureServer::start(FixtureMode::Malformed).await; + let defaults = crate::YellowstoneGrpcSessionSettings::default(); + let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes()); + let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); + let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open"); + let error = session.next_update().await.expect_err("missing update oneof must be rejected"); + assert_eq!(error.code(), crate::ERROR_CODE_INVALID_RESPONSE); + let _ = session.close().await; + server.stop().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn yellowstone_inbound_message_limit_is_enforced_by_tonic_stream_decoder() { + let server = FixtureServer::start(FixtureMode::Oversized).await; + let defaults = crate::YellowstoneGrpcSessionSettings::default(); + let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, 1, defaults.max_outbound_message_size_bytes()); + let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); + let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open"); + let error = session.next_update().await.expect_err("message above configured inbound bound must fail decoding"); + assert_eq!(error.code(), crate::ERROR_CODE_GRPC_STATUS); + let rendered = format!("{error:?} {session:?}"); + assert!(!rendered.contains(server.endpoint_url.as_str())); + let _ = session.close().await; + server.stop().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn yellowstone_outbound_mutation_size_is_rejected_before_queue_dispatch() { + let server = FixtureServer::start(FixtureMode::Idle).await; + let settings = fixture_settings(server.endpoint_url.as_str(), 8, 8, 64 * 1024 * 1024, 1); + let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); + let session = channel + .open_standard_subscribe(crate::YellowstoneSubscribeRequest::new()) + .await + .expect("empty initial request must fit one-byte fixture bound"); + let error = session.try_update(&mutated_request()).expect_err("encoded mutation above configured bound must be rejected before dispatch"); + assert_eq!(error.code(), crate::ERROR_CODE_GRPC_BACKPRESSURE_OVERFLOW); + session.close().await.expect("idle fixture must observe client half-close"); + server.stop().await; +} diff --git a/deltas/0.2.9/pre.009.md b/deltas/0.2.9/pre.009.md new file mode 100644 index 0000000..3544f63 --- /dev/null +++ b/deltas/0.2.9/pre.009.md @@ -0,0 +1,68 @@ + + + +# Delta `0.2.9-pre.009` — bidi standard + backpressure + half-close + shutdown + +## Base + +```text +0.2.9-pre.008-fix.001 +gate opérateur final : fmt/audit/check/Clippy/workspace PASS sans warning +Transport : 370 unit + 47 public API + 41 release-completeness + 4 doctests +``` + +## Changements + +- ouvre `/geyser.Geyser/Subscribe` avec le channel Tonic N1 existant ; +- ajoute `SolanaYellowstoneGrpcSubscribeSession` et son état public sûr ; +- branche directement une `mpsc` request bornée dans le stream client Tonic ; +- ajoute `try_update()` avec validation, borne protobuf et erreurs Full/Closed structurées ; +- promeut les conversions request et décodeurs `pre.004–008` de test-only vers runtime privé ; +- expose `YellowstoneSubscribeUpdate` couvrant les neuf variantes standard ; +- répond automatiquement au Ping serveur avec un request ping-only `id=1` et conserve Pong observable ; +- traite le server half-close comme terminaison normale de `pre.009` ; +- ajoute client half-close, shutdown borné, Drop best-effort et overflow terminal ; +- conserve les `Status` distants sous forme de diagnostics KSP sûrs sans recopier message/details/metadata ; +- ajoute les codes `grpc_backpressure_overflow` et `grpc_session_closed`. + +## Fixture locale + +| Cas | Preuve | +|------------------------|-------------| +| round-trip bidi | PASS source | +| mutation request | PASS source | +| Ping -> ping id=1 | PASS source | +| Pong observable | PASS source | +| server half-close | PASS source | +| client half-close | PASS source | +| shutdown hostile borné | PASS source | +| Drop session | PASS source | +| slow receiver overflow | PASS source | +| remote Status sûr | PASS source | +| update malformed | PASS source | +| inbound oversized | PASS source | +| outbound oversized | PASS source | + +## Frontières + +```text +OUT pre.009 : reconnect/resubscribe/from_slot replay policy/gaps/duplicates +OUT 0.2.9 : SubscribeDeshred +OUT pre.009 : PublicNode / Config V3 / provider facade +``` + +Aucune dépendance ou feature Cargo n'est ajoutée. Le runtime n'utilise toujours pas `yellowstone-grpc-client` et n'expose aucun type Tonic/protobuf brut. + +## Validation candidate + +```bash +cargo fmt --all +python3 scripts/audit_rust_workspace_rules.py +cargo check --workspace +cargo clippy --workspace --all-targets +cargo test -p ksp-onchain-transport-lib +cargo test -p ksp-core-lib --test workspace_dependencies +cargo test --workspace +``` + +Le `cargo tree` n'est pas requis : aucune dépendance ni feature n'a changé. diff --git a/docs/plans/016-V0_2_9_YELLOWSTONE_GRPC_PLAN.md b/docs/plans/016-V0_2_9_YELLOWSTONE_GRPC_PLAN.md index c7ceb26..2b97126 100644 --- a/docs/plans/016-V0_2_9_YELLOWSTONE_GRPC_PLAN.md +++ b/docs/plans/016-V0_2_9_YELLOWSTONE_GRPC_PLAN.md @@ -1,9 +1,9 @@ - + # Plan `0.2.9` — moteur Yellowstone gRPC + standard Solana + PublicNode -> **Statut : `0.2.9-pre.008` est fonctionnellement verte sur gate opérateur : fmt/audit/check/tests/workspace PASS, Transport 370 unit + 47 public API + 41 completeness + 4 doctests. Clippy passe mais émet un unique warning `field_reassign_with_default` dans la fixture `unit_tests/grpc_subscribe.rs`. `0.2.9-pre.008-fix.001` est candidate et corrige uniquement cette hygiène de fixture ; aucun changement runtime Blocks ni bidi.** +> **Statut : `0.2.9-pre.008-fix.001` est fermée sur gate opérateur sans warning : fmt/audit/check/Clippy/workspace PASS, Transport 370 unit + 47 public API + 41 completeness + 4 doctests. `0.2.9-pre.009` est candidate : stream bidi standard réel, mutation request, Ping/Pong, backpressure, half-close et shutdown borné ; reconnect/replay restent `pre.010`.** ## 1. Objet, base et état d'ouverture @@ -931,11 +931,11 @@ pre.006 DONE — structure Transport : namespace privé HTTP explicite pre.007 DONE — standard Solana : Transactions + transaction_status budget : 15–20 min ; preuve : include/exclude/required/Cuckoo/token expansion + tx/meta + TransactionConfig V1 -pre.008 FIX.001 CANDIDATE — standard Solana : Blocks + block_meta + entry - budget : 15–20 min ; gate fonctionnel PASS 370/47/41/4 + workspace ; fix fixture Clippy warning-only +pre.008 DONE — standard Solana : Blocks + block_meta + entry + budget : 15–20 min ; gate final fix.001 : fmt/audit/check/Clippy/workspace PASS sans warning + Transport 370/47/41/4 -pre.009 moteur partagé : bidi mutation + Ping/Pong + half-close + backpressure + shutdown - budget : 15–20 min ; preuve : actor/session local + bounded queues + cleanup déterministe +pre.009 CANDIDATE — moteur partagé : bidi mutation + Ping/Pong + half-close + backpressure + shutdown + budget : 15–20 min ; preuve : fixture locale bidi + bounded queues + half-close/drop/timeout déterministes pre.010 moteur partagé : reconnect/resubscribe + from_slot/ReplayInfo + gaps/duplicates budget : 15–20 min ; preuve : reconnect local déterministe + aucune promesse lossless @@ -1342,5 +1342,32 @@ Le premier gate opérateur de `pre.008` confirme l'intégralité du contrat Bloc Le fix remplace la construction `TransactionStatusMeta::default()` suivie de `meta.fee = 5_000` par un initialiseur struct avec `fee: 5_000` et `..Default::default()`. Aucun `allow`, aucune API, aucun DTO, aucun wire et aucune logique runtime ne changent. -**Gate attendu :** même gate opérateur, sans warning Clippy. Le bidi reste strictement `pre.009`. +**Gate final :** fmt/audit/check/Clippy/workspace PASS sans warning. `pre.008-fix.001` est fermée ; le bidi reste strictement `pre.009`. + +## 26. `pre.009` — stream bidi standard + lifecycle borné candidate + +`pre.009` ouvre pour la première fois le RPC `/geyser.Geyser/Subscribe` en runtime KSP, sans `yellowstone-grpc-client` et sans second moteur physique. `YellowstoneGrpcChannel::open_standard_subscribe()` construit une `SolanaYellowstoneGrpcSubscribeSession` sur le channel Tonic déjà possédé par N1. + +Contrat de la tranche : + +```text +request initial validé + taille protobuf bornée +file request mpsc bornée et directement consommée par Tonic +try_update() non bloquant : validation + encoded_len + Full/Closed explicites +file update mpsc bornée ; slow receiver => terminal backpressure overflow +SubscribeUpdate complet : Account/Slot/Transaction/TransactionStatus/Block/Ping/Pong/BlockMeta/Entry +server Ping => réponse automatique ping-only id=1 +Pong => update KSP observable +server half-close => terminal normal Ok(None) +client close => drop des senders request + half-close + attente close_timeout +Drop session => signal best-effort de half-close borné +Status/malformed/overflow => terminal Failed avec KspError sûr +inbound/outbound max message sizes => Tonic + validation locale outbound +``` + +Les conversions request et décodeurs update introduits sous `#[cfg(test)]` en `pre.004–008` deviennent ici des helpers runtime privés réellement consommés. Aucun type Tonic/protobuf n'est exposé dans l'API publique. + +Le reconnect est volontairement absent de `pre.009` : un server half-close est normal et terminal pour cette tranche ; un `Status` ou une erreur de protocole termine la session. `from_slot`, ReplayInfo, resubscribe, gaps/duplicates et changement de node restent `pre.010`. `SubscribeDeshred`, PublicNode et Config V3 restent hors scope. + +**Gate candidat :** audit statique clean ; fixture locale couvre round-trip, mutation, Ping/Pong, server half-close, client half-close, shutdown hostile borné, Drop best-effort, update overflow, Status distant sûr, update malformed et rejet outbound oversized. Compilation/Clippy/tests opérateur requis avant fermeture. diff --git a/docs/validation/012-V0_2_9_YELLOWSTONE_GRPC.md b/docs/validation/012-V0_2_9_YELLOWSTONE_GRPC.md index 1d79031..4e2f526 100644 --- a/docs/validation/012-V0_2_9_YELLOWSTONE_GRPC.md +++ b/docs/validation/012-V0_2_9_YELLOWSTONE_GRPC.md @@ -1,9 +1,9 @@ - + # Validation `0.2.9` — moteur Yellowstone + standard Solana + PublicNode -> **Statut : `pre.008` est fonctionnellement verte : fmt/audit/check/tests/workspace PASS, Transport 370 unit + 47 public API + 41 completeness + 4 doctests. Clippy passe avec un unique warning `field_reassign_with_default` dans la fixture Blocks. `pre.008-fix.001` corrige uniquement ce warning ; bidi reste `pre.009`.** +> **Statut : `pre.008-fix.001` est fermée sur gate opérateur sans warning : fmt/audit/check/Clippy/workspace PASS, Transport 370 unit + 47 public API + 41 completeness + 4 doctests. `pre.009` est candidate pour le bidi standard, mutation request, Ping/Pong, backpressure, half-close et shutdown borné ; reconnect/replay restent `pre.010`.** ## 1. Autorités du gate @@ -55,17 +55,17 @@ crate proto : 12.6.0 ## 3. Matrice service `Geyser` -| RPC | Forme | Classification | Scope | Preuve cible | État | -|-----------------------|-------|------------------------------------------------------|-------|-----------------------|-------------------------------------------------------| -| `Subscribe` | bidi | standard | IN | fixture locale + live | PARTIAL pre.008 filters/updates / stream TODO pre.009 | -| `SubscribeDeshred` | bidi | Triton extension/pré-exécution malgré présence proto | OUT | canari d'absence/API | OUT | -| `SubscribeReplayInfo` | unary | standard | IN | fixture unary | DONE pre.003 | -| `Ping` | unary | standard | IN | fixture unary | DONE pre.003 | -| `GetLatestBlockhash` | unary | standard | IN | fixture unary | DONE pre.003 | -| `GetBlockHeight` | unary | standard | IN | fixture unary | DONE pre.003 | -| `GetSlot` | unary | standard | IN | fixture unary | DONE pre.003 | -| `IsBlockhashValid` | unary | standard | IN | fixture unary | DONE pre.003 | -| `GetVersion` | unary | standard | IN | fixture unary | DONE pre.003 | +| RPC | Forme | Classification | Scope | Preuve cible | État | +|-----------------------|-------|------------------------------------------------------|-------|-----------------------|--------------------------------| +| `Subscribe` | bidi | standard | IN | fixture locale + live | CANDIDATE pre.009 runtime bidi | +| `SubscribeDeshred` | bidi | Triton extension/pré-exécution malgré présence proto | OUT | canari d'absence/API | OUT | +| `SubscribeReplayInfo` | unary | standard | IN | fixture unary | DONE pre.003 | +| `Ping` | unary | standard | IN | fixture unary | DONE pre.003 | +| `GetLatestBlockhash` | unary | standard | IN | fixture unary | DONE pre.003 | +| `GetBlockHeight` | unary | standard | IN | fixture unary | DONE pre.003 | +| `GetSlot` | unary | standard | IN | fixture unary | DONE pre.003 | +| `IsBlockhashValid` | unary | standard | IN | fixture unary | DONE pre.003 | +| `GetVersion` | unary | standard | IN | fixture unary | DONE pre.003 | ## 4. `SubscribeRequest` — coverage normative @@ -289,15 +289,15 @@ DONE pre.003 source/tests channel/client Debug sans URL, metadata value ni raw | Cas | Attendu | État | |--------------------------------------------|------------------------------------------|-----------------------------------| -| stream open | session bornée | TODO | -| request mutation | ordre déterministe | TODO | -| server Ping -> client request ping -> Pong | explicite | TODO | -| server half-close | terminal/reconnect selon policy | TODO | -| client close | cleanup borné | TODO | -| receiver drop | cleanup capacité | TODO | -| slow subscription | pas de queue infinie | TODO | -| inbound oversized | rejet avant allocation excessive | TODO | -| outbound oversized | rejet avant write | TODO | +| stream open | session bornée | CANDIDATE pre.009 | +| request mutation | ordre déterministe | CANDIDATE pre.009 | +| server Ping -> client request ping -> Pong | explicite | CANDIDATE pre.009 | +| server half-close | terminal normal avant reconnect | CANDIDATE pre.009 | +| client close | half-close + cleanup borné | CANDIDATE pre.009 | +| receiver drop | Drop session => cleanup best-effort | CANDIDATE pre.009 | +| slow subscription | pas de queue infinie | CANDIDATE pre.009 | +| inbound oversized | limite Tonic avant payload KSP | CANDIDATE pre.009 | +| outbound oversized | encoded_len avant queue/write | CANDIDATE pre.009 | | reconnect budget | borné | TODO | | resubscribe order | déterministe | TODO | | `from_slot` | utilisé sans promesse lossless | TODO | @@ -410,8 +410,8 @@ pre.004 DONE standard: Subscribe common/from_slot/bounds 15 pre.005 DONE standard: accounts + slots 15–20 min ; gate final fix.001 PASS pre.006 DONE structure: namespace privé HTTP `http_*` 15–20 min ; gate PASS pre.007 DONE standard: transactions + transaction_status 15–20 min ; gate PASS -pre.008 FIX.001 CANDIDATE standard: blocks + block_meta + entry 15–20 min ; gate fonctionnel PASS, 1 warning Clippy fixture -pre.009 TODO moteur: bidi/backpressure/half-close/shutdown 15–20 min +pre.008 DONE standard: blocks + block_meta + entry 15–20 min ; gate final fix.001 PASS sans warning +pre.009 CANDIDATE moteur: bidi/backpressure/half-close/shutdown 15–20 min ; fixture locale adversariale pre.010 TODO moteur: reconnect/replay/gap/duplicate 15–20 min pre.011 TODO Config V3 + protocol/provider + profils PublicNode 15–20 min pre.012 TODO PublicNode live + compliance + docs/prompt 0.2.10 15–20 min @@ -882,5 +882,40 @@ Les helpers wire/decode restent test-only jusqu’au premier consommateur runtim Le warning provient uniquement de `minimal_transaction_info()` dans `unit_tests/grpc_subscribe.rs`. `fix.001` initialise `TransactionStatusMeta.fee` directement dans le literal struct et conserve `..Default::default()`. Aucun `allow`, aucun changement runtime, aucune dépendance et aucun élargissement vers le bidi. -**Verdict `fix.001` : candidate warning-only ; fermeture de `pre.008` après gate opérateur sans warning.** +**Verdict `fix.001` : PASS sans warning ; `pre.008` fermée.** + +## 27. Gate `pre.009` — bidi standard + backpressure + half-close + shutdown candidate + +| Surface / invariant | État candidate | +|-------------------------------------------------|----------------| +| workspace version | `0.2.9-pre.9` | +| `/geyser.Geyser/Subscribe` bidi | SOURCE+TEST | +| request initial dans queue bornée | SOURCE+TEST | +| mutation `try_update()` ordonnée/non bloquante | SOURCE+TEST | +| aucune `unbounded_channel` | CANARY | +| decode des 9 variantes standard | SOURCE+TEST | +| server Ping -> ping-only id=1 | SOURCE+TEST | +| Pong observable | SOURCE+TEST | +| server half-close => `Ok(None)` | SOURCE+TEST | +| client close => request half-close | SOURCE+TEST | +| shutdown hostile borné par `close_timeout` | SOURCE+TEST | +| Drop session => half-close best-effort | SOURCE+TEST | +| slow receiver => terminal backpressure overflow | SOURCE+TEST | +| Status distant sans message/details secrets | SOURCE+TEST | +| update malformed => invalid response | SOURCE+TEST | +| max inbound Tonic | SOURCE+TEST | +| outbound encoded_len avant dispatch | SOURCE+TEST | +| reconnect/resubscribe/replay | OUT pre.009 | +| `SubscribeDeshred` | OUT 0.2.9 | +| PublicNode / Config V3 | OUT pre.009 | +| audit Rust workspace local | PASS / clean | +| fmt/check/Clippy/tests/workspace | opérateur TODO | + +Les helpers protobuf qui étaient test-only jusqu'à `pre.008` sont maintenant compilés en runtime parce que `grpc_stream` les consomme effectivement. Aucun helper raw n'est public : les frontières restent `YellowstoneSubscribeRequest` et `YellowstoneSubscribeUpdate`. + +Le flux sortant utilise la même `mpsc` bornée que celle fournie à `tonic::client::Grpc::streaming`; aucune queue cachée non bornée n'est interposée. Le Ping de keepalive serveur provoque une mutation ping-only actor-owned avec `id=1`, puis le Pong reste visible dans la file d'updates KSP. + +`pre.009` ne tente aucun reconnect. Un half-close serveur est terminal normal ; `Status`, decode invalide et overflow sont terminaux en erreur. Le budget de reconnect, l'ordre de resubscribe, `from_slot`/ReplayInfo, gaps, duplicates et node divergence restent exclusivement `pre.010`. + +**Verdict candidat :** source/fixture/audit statique prêts ; gate Cargo opérateur requis.