v0.3.12-pre.008

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/src/grpc_stream.rs
-// version: 3
+// version: 4
 
 use tonic_prost::prost::Message; // rust-rules: trait-import
 
@@ -39,6 +39,43 @@ pub struct YellowstoneGrpcSubscribeSnapshot {
     terminal_error_code: std::option::Option<ksp_core_lib::ErrorCode>,
 }
 
+/// Cloneable latest-value observer for one standard Yellowstone Subscribe session snapshot.
+///
+/// The observer exposes only the already-safe transport snapshot and keeps the internal Tokio watch channel private. Cloning the observer does not duplicate
+/// the gRPC stream, request state or reconnect actor.
+#[derive(Clone)]
+pub struct YellowstoneGrpcSubscribeSnapshotSource {
+    receiver: tokio::sync::watch::Receiver<crate::YellowstoneGrpcSubscribeSnapshot>,
+}
+
+impl crate::YellowstoneGrpcSubscribeSnapshotSource {
+    fn new(receiver: tokio::sync::watch::Receiver<crate::YellowstoneGrpcSubscribeSnapshot>) -> Self {
+        return Self { receiver };
+    }
+
+    /// Returns the current safe reconnect/replay snapshot without waiting for another actor transition.
+    #[must_use]
+    pub fn current(&self) -> crate::YellowstoneGrpcSubscribeSnapshot {
+        return *self.receiver.borrow();
+    }
+
+    /// Waits for one newer safe reconnect/replay snapshot.
+    ///
+    /// `None` means the owning Subscribe actor dropped the latest-value publisher and no further snapshot can arrive.
+    pub async fn wait_for_change(&mut self) -> std::option::Option<crate::YellowstoneGrpcSubscribeSnapshot> {
+        return match self.receiver.changed().await {
+            std::result::Result::Ok(()) => std::option::Option::Some(*self.receiver.borrow_and_update()),
+            std::result::Result::Err(_) => std::option::Option::None,
+        };
+    }
+}
+
+impl std::fmt::Debug for crate::YellowstoneGrpcSubscribeSnapshotSource {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter.debug_struct("YellowstoneGrpcSubscribeSnapshotSource").field("current", &self.current()).finish();
+    }
+}
+
 impl YellowstoneGrpcSubscribeSnapshot {
     const fn new(initial_from_slot: std::option::Option<u64>) -> Self {
         return Self {
@@ -154,6 +191,12 @@ impl SolanaYellowstoneGrpcSubscribeSession {
         return *self.snapshot_rx.borrow();
     }
 
+    /// Returns one cloneable latest-value observer for reconnect/replay snapshot transitions.
+    #[must_use]
+    pub fn snapshot_source(&self) -> crate::YellowstoneGrpcSubscribeSnapshotSource {
+        return crate::YellowstoneGrpcSubscribeSnapshotSource::new(self.snapshot_rx.clone());
+    }
+
     /// 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

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/src/lib.rs
-// version: 45
+// version: 46
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -143,6 +143,8 @@ pub use self::grpc_settings::YellowstoneGrpcTransportSettings;
 pub use self::grpc_stream::SolanaYellowstoneGrpcSubscribeSession;
 /// Safe Yellowstone reconnect/replay continuity snapshot.
 pub use self::grpc_stream::YellowstoneGrpcSubscribeSnapshot;
+/// Cloneable latest-value observer for one standard Yellowstone Subscribe session snapshot.
+pub use self::grpc_stream::YellowstoneGrpcSubscribeSnapshotSource;
 /// Safe Yellowstone bidirectional Subscribe lifecycle state.
 pub use self::grpc_stream::YellowstoneGrpcSubscribeState;
 /// One validated standard Yellowstone account predicate.

crates/ksp-onchain-transport-lib/tests/public_api.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/tests/public_api.rs
-// version: 50
+// version: 51
 
 //! Integration tests for the public `ksp-onchain-transport-lib` consumer contract.
 
@@ -1013,6 +1013,10 @@ fn public_v0_2_9_pre_010_yellowstone_reconnect_snapshot_is_available_from_crate_
     let _snapshot = std::any::type_name::<ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshot>();
     let _state = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Reconnecting;
     let _session_snapshot = ksp_onchain_transport_lib::SolanaYellowstoneGrpcSubscribeSession::snapshot;
+    let _snapshot_source = ksp_onchain_transport_lib::SolanaYellowstoneGrpcSubscribeSession::snapshot_source;
+    let _source_type = std::any::type_name::<ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshotSource>();
+    let _source_current = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshotSource::current;
+    let _source_wait = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshotSource::wait_for_change;
     let _reconnect_count = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshot::reconnect_count;
     let _gap_count = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshot::continuity_gap_count;
     let _duplicate_count = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshot::duplicate_update_count;

crates/ksp-onchain-transport-lib/tests/release_completeness.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/tests/release_completeness.rs
-// version: 42
+// version: 43
 
 //! Release-level completeness canaries for staged HTTP and WebSocket Transport coverage.
 
@@ -1307,6 +1307,9 @@ fn release_v0_2_9_pre_010_adds_bounded_reconnect_replay_and_conservative_continu
     for required in [
         "YellowstoneGrpcSubscribeState::Reconnecting",
         "YellowstoneGrpcSubscribeSnapshot",
+        "YellowstoneGrpcSubscribeSnapshotSource",
+        "snapshot_source",
+        "wait_for_change",
         "reconnect_subscribe_stream",
         "replay_first_available",
         "SubscribeReplayInfo",

crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs
-// version: 4
+// version: 5
 
 #[derive(Clone, Copy)]
 enum FixtureMode {
@@ -159,6 +159,9 @@ impl yellowstone_grpc_proto::geyser::geyser_server::Geyser for FixtureGeyser {
                 },
             }
         });
+        if matches!(mode, FixtureMode::ReconnectReplay) && subscribe_call == 2 {
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        }
         return std::result::Result::Ok(tonic::Response::new(super::MpscStream::new(outbound_rx)));
     }
 
@@ -662,6 +665,45 @@ async fn yellowstone_reconnect_budget_exhaustion_is_terminal_and_safe() {
     server.stop().await;
 }
 
+#[tokio::test(flavor = "current_thread")]
+async fn yellowstone_snapshot_source_observes_replay_attempt_before_successful_reconnect() {
+    let server = FixtureServer::start(FixtureMode::ReconnectReplay).await;
+    let defaults = crate::YellowstoneGrpcSessionSettings::default();
+    let settings = fixture_settings_with_reconnect(
+        server.endpoint_url.as_str(),
+        8,
+        8,
+        defaults.max_inbound_message_size_bytes(),
+        defaults.max_outbound_message_size_bytes(),
+        crate::YellowstoneGrpcReconnectSettings::new(3, std::time::Duration::from_millis(5), std::time::Duration::from_millis(20)),
+    );
+    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 mut snapshots = session.snapshot_source();
+    assert_eq!(snapshots.current().state(), crate::YellowstoneGrpcSubscribeState::Active);
+    let _ = session.next_update().await.expect("first slot must decode").expect("first slot must be present");
+    let replaying = tokio::time::timeout(std::time::Duration::from_secs(1), async {
+        loop {
+            let snapshot = match snapshots.wait_for_change().await {
+                std::option::Option::Some(value) => value,
+                std::option::Option::None => return std::option::Option::None,
+            };
+            if snapshot.replay_attempt_count() > 0 && snapshot.reconnect_count() == 0 {
+                return std::option::Option::Some(snapshot);
+            }
+        }
+    })
+    .await
+    .expect("snapshot source must observe reconnect progress")
+    .expect("snapshot source must stay open during reconnect");
+    assert_eq!(replaying.state(), crate::YellowstoneGrpcSubscribeState::Reconnecting);
+    assert_eq!(replaying.replay_attempt_count(), 1);
+    assert_eq!(replaying.reconnect_count(), 0);
+    let _ = session.next_update().await.expect("replayed duplicate must decode").expect("replayed duplicate must be present");
+    session.close().await.expect("reconnected stream must close cleanly");
+    server.stop().await;
+}
+
 #[tokio::test(flavor = "current_thread")]
 async fn yellowstone_shutdown_interrupts_reconnect_backoff_and_mutation_is_rejected_during_reconnect() {
     let server = FixtureServer::start(FixtureMode::ReconnectReplay).await;