v0.2.7-pre.010

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/src/lib.rs
-// version: 26
+// version: 27
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -44,6 +44,7 @@ mod rpc_tokens;
 mod rpc_transactions;
 mod settings;
 mod ws_accounts;
+mod ws_cluster;
 mod ws_lifecycle;
 mod ws_session;
 mod ws_settings;
@@ -320,6 +321,8 @@ pub use self::ws_accounts::SolanaAccountSubscribeConfig;
 pub use self::ws_accounts::SolanaProgramNotification;
 /// Configuration accepted by the standard Solana `programSubscribe` WebSocket method.
 pub use self::ws_accounts::SolanaProgramSubscribeConfig;
+/// Slot relationship reported by the standard Solana `slotNotification` WebSocket method.
+pub use self::ws_cluster::SolanaSlotNotification;
 /// Stable local identity assigned to one physical WebSocket session.
 pub use self::ws_lifecycle::WsSessionId;
 /// Safe runtime snapshot for one physical WebSocket session.
@@ -360,6 +363,10 @@ pub use self::ws_subscription::WsSubscription;
 pub use self::ws_transactions::SolanaLogsNotification;
 /// Filter accepted by the standard Solana `logsSubscribe` WebSocket method.
 pub use self::ws_transactions::SolanaLogsSubscribeFilter;
+/// Typed value carried by standard Solana `signatureNotification` messages.
+pub use self::ws_transactions::SolanaSignatureNotification;
+/// Optional configuration accepted by standard Solana `signatureSubscribe`.
+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;
@@ -387,3 +394,5 @@ pub(crate) use self::ws_subscription::WsSubscriptionRegistration;
 pub(crate) use self::ws_subscription::WsSubscriptionRuntime;
 /// Crate-internal constructor for bounded typed notification channels.
 pub(crate) use self::ws_subscription::typed_notification_channel;
+/// Crate-internal constructor for bounded typed notification channels with terminal-value classification.
+pub(crate) use self::ws_subscription::typed_notification_channel_with_completion;

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

@@ -0,0 +1,70 @@
+// file: crates/ksp-onchain-transport-lib/src/ws_cluster.rs
+// version: 1
+
+/// Slot relationship reported by the standard Solana `slotNotification` WebSocket method.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct SolanaSlotNotification {
+    slot: u64,
+    parent: u64,
+    root: u64,
+}
+
+impl SolanaSlotNotification {
+    /// Returns the newly processed slot.
+    #[must_use]
+    pub const fn slot(&self) -> u64 {
+        return self.slot;
+    }
+
+    /// Returns the parent slot reported by the validator.
+    #[must_use]
+    pub const fn parent(&self) -> u64 {
+        return self.parent;
+    }
+
+    /// Returns the current root slot reported alongside this slot update.
+    #[must_use]
+    pub const fn root(&self) -> u64 {
+        return self.root;
+    }
+}
+
+impl crate::WsSession {
+    /// Subscribes to standard Solana slot-processing notifications through `slotSubscribe`.
+    pub async fn slot_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaSlotNotification>> {
+        return self
+            .subscribe_typed(crate::WsSubscriptionKind::Slot, std::vec::Vec::new(), |value| {
+                return decode_slot_notification("slotSubscribe", value);
+            })
+            .await;
+    }
+
+    /// Subscribes to standard Solana root-slot notifications through `rootSubscribe`.
+    pub async fn root_subscribe(&self) -> ksp_core_lib::Result<crate::WsSubscription<u64>> {
+        return self
+            .subscribe_typed(crate::WsSubscriptionKind::Root, std::vec::Vec::new(), |value| {
+                return crate::decode_wire_json::<u64>("rootSubscribe", value);
+            })
+            .await;
+    }
+}
+
+#[derive(serde::Deserialize)]
+struct WireSlotNotification {
+    slot: u64,
+    parent: u64,
+    root: u64,
+}
+
+fn decode_slot_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result<crate::SolanaSlotNotification> {
+    let decoded = crate::decode_wire_json::<WireSlotNotification>(method, value);
+    let wire = match decoded {
+        std::result::Result::Ok(wire) => wire,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    return std::result::Result::Ok(crate::SolanaSlotNotification { slot: wire.slot, parent: wire.parent, root: wire.root });
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/ws_cluster.rs"]
+mod tests;

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/src/ws_session.rs
-// version: 10
+// version: 11
 
 use futures_util::SinkExt; // rust-rules: trait-import
 use futures_util::StreamExt; // rust-rules: trait-import
@@ -167,11 +167,27 @@ impl WsSession {
     where
         T: std::marker::Send + 'static,
         F: Fn(serde_json::Value) -> ksp_core_lib::Result<T> + std::marker::Send + std::marker::Sync + 'static,
+    {
+        return self.subscribe_typed_with_completion(kind, params, decoder, |_| return false).await;
+    }
+
+    /// Creates one crate-internal typed subscription whose decoder can identify a delivered terminal notification.
+    pub(crate) async fn subscribe_typed_with_completion<T, F, C>(
+        &self,
+        kind: crate::WsSubscriptionKind,
+        params: std::vec::Vec<serde_json::Value>,
+        decoder: F,
+        is_terminal: C,
+    ) -> ksp_core_lib::Result<crate::WsSubscription<T>>
+    where
+        T: std::marker::Send + 'static,
+        F: Fn(serde_json::Value) -> ksp_core_lib::Result<T> + std::marker::Send + std::marker::Sync + 'static,
+        C: Fn(&T) -> bool + std::marker::Send + std::marker::Sync + 'static,
     {
         if self.state() != crate::WsSessionState::Active {
             return std::result::Result::Err(ws_session_closed_error(self.id, "WebSocket session is not active"));
         }
-        let (dispatcher, notification_rx) = crate::typed_notification_channel(self.notification_queue_capacity, decoder);
+        let (dispatcher, notification_rx) = crate::typed_notification_channel_with_completion(self.notification_queue_capacity, decoder, is_terminal);
         let (response_tx, response_rx) = tokio::sync::oneshot::channel();
         let command = WsSessionCommand::Subscribe { kind, params, dispatcher, response_tx };
         let send_wait = tokio::time::timeout(self.command_timeout, self.command_tx.send(command)).await;
@@ -1901,6 +1917,17 @@ fn handle_subscription_notification(
     let dispatch = (runtime.dispatcher)(result);
     return match dispatch {
         crate::WsNotificationDispatchOutcome::Delivered => WsActorIoOutcome::Continue,
+        crate::WsNotificationDispatchOutcome::DeliveredTerminal => {
+            close_local_subscription(subscriptions, remote_to_local, local_id, crate::WsSubscriptionState::Closed);
+            ksp_logging_lib::debug!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                subscription_id = local_id.get(),
+                subscription_kind = subscription_kind.as_str(),
+                "logical WebSocket subscription observed its server-terminal notification"
+            );
+            WsActorIoOutcome::Continue
+        },
         crate::WsNotificationDispatchOutcome::ReceiverClosed => {
             close_local_subscription(subscriptions, remote_to_local, local_id, crate::WsSubscriptionState::Closed);
             ksp_logging_lib::debug!(

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/src/ws_subscription.rs
-// version: 3
+// version: 4
 
 /// Typed handle for one logical Solana WebSocket subscription.
 ///
@@ -124,6 +124,7 @@ impl<T> std::fmt::Debug for WsSubscription<T> {
 /// Internal result of dispatching one decoded wire notification into a bounded typed channel.
 pub(crate) enum WsNotificationDispatchOutcome {
     Delivered,
+    DeliveredTerminal,
     ReceiverClosed,
     QueueFull,
     DecodeFailed { code: ksp_core_lib::ErrorCode },
@@ -137,6 +138,20 @@ pub(crate) fn typed_notification_channel<T, F>(capacity: usize, decoder: F) -> (
 where
     T: std::marker::Send + 'static,
     F: Fn(serde_json::Value) -> ksp_core_lib::Result<T> + std::marker::Send + std::marker::Sync + 'static,
+{
+    return typed_notification_channel_with_completion(capacity, decoder, |_| return false);
+}
+
+/// Creates one bounded typed notification receiver whose dispatcher can mark a successfully delivered value as terminal.
+pub(crate) fn typed_notification_channel_with_completion<T, F, C>(
+    capacity: usize,
+    decoder: F,
+    is_terminal: C,
+) -> (WsNotificationDispatcher, tokio::sync::mpsc::Receiver<ksp_core_lib::Result<T>>)
+where
+    T: std::marker::Send + 'static,
+    F: Fn(serde_json::Value) -> ksp_core_lib::Result<T> + std::marker::Send + std::marker::Sync + 'static,
+    C: Fn(&T) -> bool + std::marker::Send + std::marker::Sync + 'static,
 {
     let (notification_tx, notification_rx) = tokio::sync::mpsc::channel(capacity);
     let dispatcher = move |value: serde_json::Value| -> WsNotificationDispatchOutcome {
@@ -151,7 +166,11 @@ where
         };
         return match decoder(value) {
             std::result::Result::Ok(notification) => {
+                let terminal = is_terminal(&notification);
                 permit.send(std::result::Result::Ok(notification));
+                if terminal {
+                    return WsNotificationDispatchOutcome::DeliveredTerminal;
+                }
                 WsNotificationDispatchOutcome::Delivered
             },
             std::result::Result::Err(error) => {

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

@@ -1,5 +1,79 @@
 // file: crates/ksp-onchain-transport-lib/src/ws_transactions.rs
-// version: 2
+// version: 3
+
+/// Optional configuration accepted by standard Solana `signatureSubscribe`.
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+pub struct SolanaSignatureSubscribeConfig {
+    commitment: std::option::Option<crate::SolanaCommitment>,
+    enable_received_notification: std::option::Option<bool>,
+}
+
+impl SolanaSignatureSubscribeConfig {
+    /// Creates an explicit signature-subscription configuration.
+    #[must_use]
+    pub const fn new(commitment: std::option::Option<crate::SolanaCommitment>, enable_received_notification: std::option::Option<bool>) -> Self {
+        return Self { commitment, enable_received_notification };
+    }
+
+    /// Returns the optional commitment level.
+    #[must_use]
+    pub const fn commitment(&self) -> std::option::Option<crate::SolanaCommitment> {
+        return self.commitment;
+    }
+
+    /// Returns whether the server should emit the early `receivedSignature` notification when explicitly configured.
+    #[must_use]
+    pub const fn enable_received_notification(&self) -> std::option::Option<bool> {
+        return self.enable_received_notification;
+    }
+
+    fn is_empty(&self) -> bool {
+        return self.commitment.is_none() && self.enable_received_notification.is_none();
+    }
+
+    fn to_json_value(self) -> serde_json::Value {
+        let mut object = serde_json::Map::new();
+        if let std::option::Option::Some(commitment) = self.commitment {
+            object.insert("commitment".to_owned(), serde_json::Value::String(commitment.as_str().to_owned()));
+        }
+        if let std::option::Option::Some(enable_received_notification) = self.enable_received_notification {
+            object.insert("enableReceivedNotification".to_owned(), serde_json::Value::Bool(enable_received_notification));
+        }
+        return serde_json::Value::Object(object);
+    }
+}
+
+/// Typed value carried by standard Solana `signatureNotification` messages.
+#[derive(Clone, Debug, PartialEq)]
+pub enum SolanaSignatureNotification {
+    /// Early notification emitted when the RPC node first receives the signature and `enableReceivedNotification` is enabled.
+    ReceivedSignature,
+    /// Terminal processing notification emitted when the configured commitment is reached.
+    Processed {
+        /// Nullable transaction-error wire value; `None` means the transaction succeeded at the requested commitment.
+        err: std::option::Option<serde_json::Value>,
+    },
+}
+
+impl SolanaSignatureNotification {
+    /// Returns whether this notification terminates the server-side one-shot subscription.
+    #[must_use]
+    pub const fn is_terminal(&self) -> bool {
+        return match self {
+            Self::ReceivedSignature => false,
+            Self::Processed { .. } => true,
+        };
+    }
+
+    /// Returns the transaction-error wire value for a terminal processing notification when present.
+    #[must_use]
+    pub const fn err(&self) -> std::option::Option<&serde_json::Value> {
+        return match self {
+            Self::ReceivedSignature | Self::Processed { err: std::option::Option::None } => std::option::Option::None,
+            Self::Processed { err: std::option::Option::Some(err) } => std::option::Option::Some(err),
+        };
+    }
+}
 
 /// Filter accepted by the standard Solana `logsSubscribe` WebSocket method.
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -51,6 +125,31 @@ impl SolanaLogsNotification {
 }
 
 impl crate::WsSession {
+    /// Subscribes to one Solana transaction signature through standard `signatureSubscribe`.
+    ///
+    /// The server automatically terminates this subscription after the terminal processed notification. When
+    /// `enableReceivedNotification` is enabled, an earlier `ReceivedSignature` value may be delivered first without closing the logical handle.
+    pub async fn signature_subscribe(
+        &self,
+        signature: &str,
+        config: std::option::Option<&crate::SolanaSignatureSubscribeConfig>,
+    ) -> ksp_core_lib::Result<crate::WsSubscription<crate::SolanaRpcResponse<crate::SolanaSignatureNotification>>> {
+        let mut params = std::vec![serde_json::Value::String(signature.to_owned())];
+        if let std::option::Option::Some(config) = config
+            && !config.is_empty()
+        {
+            params.push((*config).to_json_value());
+        }
+        return self
+            .subscribe_typed_with_completion(
+                crate::WsSubscriptionKind::Signature,
+                params,
+                |value| return decode_signature_notification("signatureSubscribe", value),
+                |notification| return notification.value().is_terminal(),
+            )
+            .await;
+    }
+
     /// Subscribes to Solana transaction logs through standard `logsSubscribe`.
     pub async fn logs_subscribe(
         &self,
@@ -67,6 +166,24 @@ impl crate::WsSession {
     }
 }
 
+#[derive(serde::Deserialize)]
+#[serde(untagged)]
+enum WireSignatureNotification {
+    Received(std::string::String),
+    Processed(WireSignatureProcessed),
+}
+
+#[derive(serde::Deserialize)]
+struct WireSignatureProcessed {
+    err: serde_json::Value,
+}
+
+#[derive(serde::Deserialize)]
+struct WireRpcResponseSignature {
+    context: serde_json::Value,
+    value: WireSignatureNotification,
+}
+
 #[derive(serde::Deserialize)]
 struct WireRpcResponse {
     context: serde_json::Value,
@@ -80,6 +197,36 @@ struct WireLogsNotification {
     logs: std::vec::Vec<std::string::String>,
 }
 
+fn decode_signature_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result<crate::SolanaRpcResponse<crate::SolanaSignatureNotification>> {
+    let decoded = crate::decode_wire_json::<WireRpcResponseSignature>(method, value);
+    let wire = match decoded {
+        std::result::Result::Ok(wire) => wire,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let context = crate::SolanaRpcContext::decode_wire(method, wire.context);
+    let context = match context {
+        std::result::Result::Ok(context) => context,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let notification = match wire.value {
+        WireSignatureNotification::Received(value) if value == "receivedSignature" => crate::SolanaSignatureNotification::ReceivedSignature,
+        WireSignatureNotification::Received(_) => {
+            return std::result::Result::Err(
+                ksp_core_lib::Error::new(crate::ERROR_CODE_INVALID_RESPONSE, "signatureSubscribe notification contains an unknown string variant")
+                    .with_context("rpc_method", method),
+            );
+        },
+        WireSignatureNotification::Processed(processed) => {
+            let err = match processed.err {
+                serde_json::Value::Null => std::option::Option::None,
+                value => std::option::Option::Some(value),
+            };
+            crate::SolanaSignatureNotification::Processed { err }
+        },
+    };
+    return std::result::Result::Ok(crate::SolanaRpcResponse::new(context, notification));
+}
+
 fn decode_logs_notification(method: &str, value: serde_json::Value) -> ksp_core_lib::Result<crate::SolanaRpcResponse<crate::SolanaLogsNotification>> {
     let decoded = crate::decode_wire_json::<WireRpcResponse>(method, value);
     let wire = match decoded {