v0.2.7-pre.006

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

@@ -41,6 +41,7 @@ mod settings;
 mod ws_lifecycle;
 mod ws_session;
 mod ws_settings;
+mod ws_subscription;
 
 /// Passive runtime availability reported for one logical HTTP endpoint.
 pub use self::client::HttpEndpointAvailability;
@@ -340,6 +341,8 @@ pub use self::ws_settings::WsResubscribePolicy;
 pub use self::ws_settings::WsSessionSettings;
 /// Complete runtime settings consumed by the KSP WebSocket transport foundation.
 pub use self::ws_settings::WsTransportSettings;
+/// Typed handle for one logical Solana WebSocket subscription.
+pub use self::ws_subscription::WsSubscription;
 
 /// Owning tracing target for events emitted by the on-chain transport crate.
 pub(crate) use self::constants::TRACING_TARGET;
@@ -355,3 +358,15 @@ pub(crate) use self::rpc_common::decode_wire_json;
 pub(crate) use self::rpc_common::parse_wire_pubkey;
 /// Validates endpoint settings.
 pub(crate) use self::settings::validate_endpoint_settings;
+/// Crate-internal command surface shared by the physical session and typed subscription handle.
+pub(crate) use self::ws_session::WsSessionCommand;
+/// Crate-internal notification dispatch result.
+pub(crate) use self::ws_subscription::WsNotificationDispatchOutcome;
+/// Crate-internal type-erased notification dispatcher.
+pub(crate) use self::ws_subscription::WsNotificationDispatcher;
+/// Crate-internal actor registration returned after subscribe acknowledgement.
+pub(crate) use self::ws_subscription::WsSubscriptionRegistration;
+/// Crate-internal actor-owned logical subscription runtime entry.
+pub(crate) use self::ws_subscription::WsSubscriptionRuntime;
+/// Crate-internal constructor for bounded typed notification channels.
+pub(crate) use self::ws_subscription::typed_notification_channel;

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/src/ws_lifecycle.rs
-// version: 3
+// version: 4
 
 /// Stable local identity assigned to one physical WebSocket session.
 #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
@@ -116,6 +116,51 @@ impl WsSubscriptionKind {
             Self::Vote => "vote",
         };
     }
+
+    /// Returns the exact standard Solana subscribe JSON-RPC method for this family.
+    pub(crate) const fn subscribe_method(self) -> &'static str {
+        return match self {
+            Self::Account => "accountSubscribe",
+            Self::Block => "blockSubscribe",
+            Self::Logs => "logsSubscribe",
+            Self::Program => "programSubscribe",
+            Self::Root => "rootSubscribe",
+            Self::Signature => "signatureSubscribe",
+            Self::Slot => "slotSubscribe",
+            Self::SlotsUpdates => "slotsUpdatesSubscribe",
+            Self::Vote => "voteSubscribe",
+        };
+    }
+
+    /// Returns the exact standard Solana unsubscribe JSON-RPC method for this family.
+    pub(crate) const fn unsubscribe_method(self) -> &'static str {
+        return match self {
+            Self::Account => "accountUnsubscribe",
+            Self::Block => "blockUnsubscribe",
+            Self::Logs => "logsUnsubscribe",
+            Self::Program => "programUnsubscribe",
+            Self::Root => "rootUnsubscribe",
+            Self::Signature => "signatureUnsubscribe",
+            Self::Slot => "slotUnsubscribe",
+            Self::SlotsUpdates => "slotsUpdatesUnsubscribe",
+            Self::Vote => "voteUnsubscribe",
+        };
+    }
+
+    /// Returns the exact standard Solana notification method emitted for this family.
+    pub(crate) const fn notification_method(self) -> &'static str {
+        return match self {
+            Self::Account => "accountNotification",
+            Self::Block => "blockNotification",
+            Self::Logs => "logsNotification",
+            Self::Program => "programNotification",
+            Self::Root => "rootNotification",
+            Self::Signature => "signatureNotification",
+            Self::Slot => "slotNotification",
+            Self::SlotsUpdates => "slotsUpdatesNotification",
+            Self::Vote => "voteNotification",
+        };
+    }
 }
 
 /// Safe lifecycle projection for one logical WebSocket subscription.
@@ -130,7 +175,6 @@ pub struct WsSubscriptionSnapshot {
 impl WsSubscriptionSnapshot {
     /// Creates one safe subscription lifecycle projection for Transport runtime internals.
     #[must_use]
-    #[cfg(test)]
     pub(crate) const fn new(id: crate::WsSubscriptionId, kind: crate::WsSubscriptionKind, state: crate::WsSubscriptionState, remote_bound: bool) -> Self {
         return Self { id, kind, state, remote_bound };
     }

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

@@ -0,0 +1,205 @@
+// file: crates/ksp-onchain-transport-lib/src/ws_subscription.rs
+// version: 1
+
+/// Typed handle for one logical Solana WebSocket subscription.
+///
+/// The handle owns the bounded typed notification receiver while the physical session actor owns the remote subscription identity and socket. The remote
+/// numeric subscription identifier is intentionally never exposed because it is transient and will be remapped by reconnect support in later tranches.
+pub struct WsSubscription<T> {
+    session_id: crate::WsSessionId,
+    id: crate::WsSubscriptionId,
+    kind: crate::WsSubscriptionKind,
+    notification_rx: tokio::sync::mpsc::Receiver<ksp_core_lib::Result<T>>,
+    state_rx: tokio::sync::watch::Receiver<crate::WsSubscriptionState>,
+    command_tx: tokio::sync::mpsc::Sender<crate::WsSessionCommand>,
+    command_timeout: std::time::Duration,
+}
+
+impl<T> WsSubscription<T> {
+    /// Creates a typed subscription handle from one actor registration.
+    pub(crate) fn new(
+        session_id: crate::WsSessionId,
+        registration: WsSubscriptionRegistration,
+        notification_rx: tokio::sync::mpsc::Receiver<ksp_core_lib::Result<T>>,
+        command_tx: tokio::sync::mpsc::Sender<crate::WsSessionCommand>,
+        command_timeout: std::time::Duration,
+    ) -> Self {
+        return Self {
+            session_id,
+            id: registration.id,
+            kind: registration.kind,
+            notification_rx,
+            state_rx: registration.state_rx,
+            command_tx,
+            command_timeout,
+        };
+    }
+
+    /// Returns the stable local logical subscription identity.
+    #[must_use]
+    pub const fn id(&self) -> crate::WsSubscriptionId {
+        return self.id;
+    }
+
+    /// Returns the standard Solana subscription family.
+    #[must_use]
+    pub const fn kind(&self) -> crate::WsSubscriptionKind {
+        return self.kind;
+    }
+
+    /// Returns the latest observable lifecycle state for this logical subscription.
+    #[must_use]
+    pub fn state(&self) -> crate::WsSubscriptionState {
+        return *self.state_rx.borrow();
+    }
+
+    /// Receives the next typed notification or terminal typed-decoding error.
+    ///
+    /// The underlying queue is bounded by `WsSessionSettings::notification_queue_capacity`. `None` means the actor closed this logical subscription and no
+    /// further notifications can arrive.
+    pub async fn recv(&mut self) -> std::option::Option<ksp_core_lib::Result<T>> {
+        return self.notification_rx.recv().await;
+    }
+
+    /// Cancels this logical subscription and sends the matching Solana unsubscribe request when a remote binding still exists.
+    ///
+    /// The returned boolean preserves the standard Solana unsubscribe result. Local cancellation is terminal for this handle even when the remote endpoint
+    /// returns an application error; reconnect race hardening is completed in `0.2.7-pre.007`.
+    pub async fn unsubscribe(&mut self) -> ksp_core_lib::Result<bool> {
+        match self.state() {
+            crate::WsSubscriptionState::Closed => return std::result::Result::Ok(false),
+            crate::WsSubscriptionState::Failed => {
+                return std::result::Result::Err(subscription_closed_error(self.session_id, self.id, "WebSocket subscription is already failed"));
+            },
+            _ => {},
+        }
+        let (response_tx, response_rx) = tokio::sync::oneshot::channel();
+        let command = crate::WsSessionCommand::Unsubscribe { subscription_id: self.id, response_tx };
+        let send_wait = tokio::time::timeout(self.command_timeout, self.command_tx.send(command)).await;
+        match send_wait {
+            std::result::Result::Ok(std::result::Result::Ok(())) => {},
+            std::result::Result::Ok(std::result::Result::Err(_)) => {
+                return std::result::Result::Err(subscription_closed_error(self.session_id, self.id, "WebSocket session command channel is closed"));
+            },
+            std::result::Result::Err(_) => {
+                return std::result::Result::Err(subscription_timeout_error(
+                    self.session_id,
+                    self.id,
+                    "WebSocket unsubscribe command queue remained unavailable until timeout",
+                ));
+            },
+        }
+        return match response_rx.await {
+            std::result::Result::Ok(result) => result,
+            std::result::Result::Err(_) => {
+                std::result::Result::Err(subscription_closed_error(self.session_id, self.id, "WebSocket session ended before unsubscribe completion"))
+            },
+        };
+    }
+}
+
+impl<T> std::fmt::Debug for WsSubscription<T> {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("WsSubscription")
+            .field("session_id", &self.session_id)
+            .field("id", &self.id)
+            .field("kind", &self.kind)
+            .field("state", &self.state())
+            .finish();
+    }
+}
+
+/// Internal result of dispatching one decoded wire notification into a bounded typed channel.
+pub(crate) enum WsNotificationDispatchOutcome {
+    Delivered,
+    ReceiverClosed,
+    QueueFull,
+    DecodeFailed,
+}
+
+/// Type-erased actor-owned dispatcher for one heterogeneous typed notification channel.
+pub(crate) type WsNotificationDispatcher = std::boxed::Box<dyn Fn(serde_json::Value) -> WsNotificationDispatchOutcome + std::marker::Send + std::marker::Sync>;
+
+/// Creates one bounded typed notification receiver and its type-erased actor dispatcher.
+pub(crate) fn typed_notification_channel<T, F>(capacity: usize, decoder: F) -> (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,
+{
+    let (notification_tx, notification_rx) = tokio::sync::mpsc::channel(capacity);
+    let dispatcher = move |value: serde_json::Value| -> WsNotificationDispatchOutcome {
+        let decoded = decoder(value);
+        return match decoded {
+            std::result::Result::Ok(notification) => match notification_tx.try_send(std::result::Result::Ok(notification)) {
+                std::result::Result::Ok(()) => WsNotificationDispatchOutcome::Delivered,
+                std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => WsNotificationDispatchOutcome::ReceiverClosed,
+                std::result::Result::Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => WsNotificationDispatchOutcome::QueueFull,
+            },
+            std::result::Result::Err(error) => {
+                let _ = notification_tx.try_send(std::result::Result::Err(error));
+                WsNotificationDispatchOutcome::DecodeFailed
+            },
+        };
+    };
+    return (std::boxed::Box::new(dispatcher), notification_rx);
+}
+
+/// Internal registration returned after a remote subscribe acknowledgement becomes atomically bound.
+pub(crate) struct WsSubscriptionRegistration {
+    id: crate::WsSubscriptionId,
+    kind: crate::WsSubscriptionKind,
+    state_rx: tokio::sync::watch::Receiver<crate::WsSubscriptionState>,
+}
+
+impl WsSubscriptionRegistration {
+    /// Creates one successful actor registration without exposing the transient remote identifier.
+    pub(crate) fn new(
+        id: crate::WsSubscriptionId,
+        kind: crate::WsSubscriptionKind,
+        state_rx: tokio::sync::watch::Receiver<crate::WsSubscriptionState>,
+    ) -> Self {
+        return Self { id, kind, state_rx };
+    }
+}
+
+/// Actor-owned runtime entry for one local logical WebSocket subscription.
+pub(crate) struct WsSubscriptionRuntime {
+    /// Stable local identity.
+    pub(crate) id: crate::WsSubscriptionId,
+    /// Standard Solana subscription family.
+    pub(crate) kind: crate::WsSubscriptionKind,
+    /// Current logical lifecycle state.
+    pub(crate) state: crate::WsSubscriptionState,
+    /// Current transient remote subscription identity when bound.
+    pub(crate) remote_id: std::option::Option<u64>,
+    /// Lifecycle publisher observed by the public typed handle.
+    pub(crate) state_tx: tokio::sync::watch::Sender<crate::WsSubscriptionState>,
+    /// Type-erased dispatcher into the bounded typed notification channel.
+    pub(crate) dispatcher: WsNotificationDispatcher,
+}
+
+impl WsSubscriptionRuntime {
+    /// Builds the safe session-snapshot projection for this runtime entry.
+    pub(crate) fn snapshot(&self) -> crate::WsSubscriptionSnapshot {
+        return crate::WsSubscriptionSnapshot::new(self.id, self.kind, self.state, self.remote_id.is_some());
+    }
+
+    /// Updates the runtime state and publishes it to the typed handle.
+    pub(crate) fn set_state(&mut self, state: crate::WsSubscriptionState) {
+        self.state = state;
+        self.state_tx.send_replace(state);
+    }
+}
+
+fn subscription_closed_error(session_id: crate::WsSessionId, subscription_id: crate::WsSubscriptionId, message: &'static str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_WS_SESSION_CLOSED, message)
+        .with_context("session_id", session_id.get().to_string())
+        .with_context("subscription_id", subscription_id.get().to_string());
+}
+
+fn subscription_timeout_error(session_id: crate::WsSessionId, subscription_id: crate::WsSubscriptionId, message: &'static str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_TIMEOUT, message)
+        .with_context("session_id", session_id.get().to_string())
+        .with_context("subscription_id", subscription_id.get().to_string());
+}