v0.2.9-pre.010

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: 1
+// version: 2
 
 #[derive(Clone, Copy)]
 enum FixtureMode {
@@ -11,12 +11,16 @@ enum FixtureMode {
     Malformed,
     Oversized,
     Idle,
+    ReconnectReplay,
+    ReplayGap,
+    ReconnectExhausted,
 }
 
 #[derive(Clone)]
 struct FixtureGeyser {
     mode: FixtureMode,
     half_close_seen: std::sync::Arc<std::sync::atomic::AtomicBool>,
+    subscribe_calls: std::sync::Arc<std::sync::atomic::AtomicUsize>,
 }
 
 #[allow(clippy::implicit_return)] // tonic::async_trait generates async wrapper tails outside the authored fixture bodies.
@@ -32,6 +36,10 @@ impl yellowstone_grpc_proto::geyser::geyser_server::Geyser for FixtureGeyser {
         if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) {
             return std::result::Result::Err(error);
         }
+        let subscribe_call = self.subscribe_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst).saturating_add(1);
+        if matches!(self.mode, FixtureMode::ReconnectExhausted) && subscribe_call > 1 {
+            return std::result::Result::Err(tonic::Status::unavailable("GRPC-RECONNECT-SECRET-CANARY"));
+        }
         let mut inbound = request.into_inner();
         let (outbound_tx, outbound_rx) = tokio::sync::mpsc::channel(64);
         let mode = self.mode;
@@ -111,6 +119,45 @@ impl yellowstone_grpc_proto::geyser::geyser_server::Geyser for FixtureGeyser {
                 FixtureMode::Idle => {
                     let _ = inbound.message().await;
                 },
+                FixtureMode::ReconnectReplay => {
+                    if subscribe_call == 1 {
+                        assert_eq!(initial.from_slot, std::option::Option::None);
+                        let _ = outbound_tx.send(std::result::Result::Ok(slot_update(500))).await;
+                    } else {
+                        assert_eq!(subscribe_call, 2);
+                        assert_eq!(initial.from_slot, std::option::Option::Some(500));
+                        if outbound_tx.send(std::result::Result::Ok(slot_update(500))).await.is_err() {
+                            return;
+                        }
+                        if outbound_tx.send(std::result::Result::Ok(slot_update(501))).await.is_err() {
+                            return;
+                        }
+                        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::ReplayGap => {
+                    if subscribe_call == 1 {
+                        assert_eq!(initial.from_slot, std::option::Option::None);
+                        let _ = outbound_tx.send(std::result::Result::Ok(slot_update(500))).await;
+                    } else {
+                        assert_eq!(subscribe_call, 2);
+                        assert_eq!(initial.from_slot, std::option::Option::Some(505));
+                        if outbound_tx.send(std::result::Result::Ok(slot_update(505))).await.is_err() {
+                            return;
+                        }
+                        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::ReconnectExhausted => {
+                    assert_eq!(subscribe_call, 1);
+                    let _ = outbound_tx.send(std::result::Result::Ok(slot_update(700))).await;
+                },
             }
         });
         return std::result::Result::Ok(tonic::Response::new(super::MpscStream::new(outbound_rx)));
@@ -125,9 +172,17 @@ impl yellowstone_grpc_proto::geyser::geyser_server::Geyser for FixtureGeyser {
 
     async fn subscribe_replay_info(
         &self,
-        _request: tonic::Request<yellowstone_grpc_proto::geyser::SubscribeReplayInfoRequest>,
+        request: tonic::Request<yellowstone_grpc_proto::geyser::SubscribeReplayInfoRequest>,
     ) -> std::result::Result<tonic::Response<yellowstone_grpc_proto::geyser::SubscribeReplayInfoResponse>, tonic::Status> {
-        return std::result::Result::Err(tonic::Status::unimplemented("unary is outside this fixture"));
+        if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) {
+            return std::result::Result::Err(error);
+        }
+        let first_available = match self.mode {
+            FixtureMode::ReconnectReplay | FixtureMode::ReconnectExhausted => std::option::Option::Some(400),
+            FixtureMode::ReplayGap => std::option::Option::Some(505),
+            _ => return std::result::Result::Err(tonic::Status::unimplemented("replay info is outside this fixture mode")),
+        };
+        return std::result::Result::Ok(tonic::Response::new(yellowstone_grpc_proto::geyser::SubscribeReplayInfoResponse { first_available }));
     }
 
     async fn ping(
@@ -176,6 +231,7 @@ impl yellowstone_grpc_proto::geyser::geyser_server::Geyser for FixtureGeyser {
 struct FixtureServer {
     endpoint_url: std::string::String,
     half_close_seen: std::sync::Arc<std::sync::atomic::AtomicBool>,
+    subscribe_calls: std::sync::Arc<std::sync::atomic::AtomicUsize>,
     shutdown: std::option::Option<tokio::sync::oneshot::Sender<()>>,
     task: tokio::task::JoinHandle<()>,
 }
@@ -187,9 +243,15 @@ impl FixtureServer {
         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 subscribe_calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
         let service_half_close_seen = half_close_seen.clone();
+        let service_subscribe_calls = subscribe_calls.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 service = yellowstone_grpc_proto::geyser::geyser_server::GeyserServer::new(FixtureGeyser {
+                mode,
+                half_close_seen: service_half_close_seen,
+                subscribe_calls: service_subscribe_calls,
+            });
             let result = tonic::transport::Server::builder()
                 .serve_with_incoming_shutdown(service, incoming, async move {
                     let _ = shutdown_receiver.await;
@@ -200,6 +262,7 @@ impl FixtureServer {
         return Self {
             endpoint_url: format!("http://{local_address}"),
             half_close_seen,
+            subscribe_calls,
             shutdown: std::option::Option::Some(shutdown),
             task,
         };
@@ -229,13 +292,32 @@ fn fixture_settings(
     update_capacity: usize,
     max_inbound_message_size_bytes: usize,
     max_outbound_message_size_bytes: usize,
+) -> crate::YellowstoneGrpcEndpointSettings {
+    let defaults = crate::YellowstoneGrpcSessionSettings::default();
+    return fixture_settings_with_reconnect(
+        url,
+        request_capacity,
+        update_capacity,
+        max_inbound_message_size_bytes,
+        max_outbound_message_size_bytes,
+        crate::YellowstoneGrpcReconnectSettings::new(0, defaults.reconnect().initial_backoff(), defaults.reconnect().max_backoff()),
+    );
+}
+
+fn fixture_settings_with_reconnect(
+    url: &str,
+    request_capacity: usize,
+    update_capacity: usize,
+    max_inbound_message_size_bytes: usize,
+    max_outbound_message_size_bytes: usize,
+    reconnect: crate::YellowstoneGrpcReconnectSettings,
 ) -> 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(),
+        reconnect,
         request_capacity,
         update_capacity,
         max_inbound_message_size_bytes,
@@ -466,3 +548,138 @@ async fn yellowstone_outbound_mutation_size_is_rejected_before_queue_dispatch()
     session.close().await.expect("idle fixture must observe client half-close");
     server.stop().await;
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn yellowstone_reconnect_replays_from_last_observed_slot_and_counts_duplicate_identity() {
+    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 first = session.next_update().await.expect("first slot must decode").expect("first slot must be present");
+    match first {
+        crate::YellowstoneSubscribeUpdate::Slot(value) => assert_eq!(value.slot(), 500),
+        _ => panic!("fixture must return first Slot"),
+    }
+    let duplicate = session.next_update().await.expect("replayed duplicate must decode").expect("replayed duplicate must be present");
+    match duplicate {
+        crate::YellowstoneSubscribeUpdate::Slot(value) => assert_eq!(value.slot(), 500),
+        _ => panic!("fixture must return replayed Slot"),
+    }
+    let resumed = session.next_update().await.expect("resumed slot must decode").expect("resumed slot must be present");
+    match resumed {
+        crate::YellowstoneSubscribeUpdate::Slot(value) => assert_eq!(value.slot(), 501),
+        _ => panic!("fixture must return resumed Slot"),
+    }
+    let snapshot = session.snapshot();
+    assert_eq!(snapshot.state(), crate::YellowstoneGrpcSubscribeState::Active);
+    assert_eq!(snapshot.reconnect_count(), 1);
+    assert_eq!(snapshot.replay_attempt_count(), 1);
+    assert_eq!(snapshot.continuity_gap_count(), 0);
+    assert_eq!(snapshot.duplicate_update_count(), 1);
+    assert_eq!(snapshot.last_requested_from_slot(), std::option::Option::Some(500));
+    assert_eq!(snapshot.last_observed_slot(), std::option::Option::Some(501));
+    session.close().await.expect("reconnected stream must close cleanly");
+    assert_eq!(server.subscribe_calls.load(std::sync::atomic::Ordering::SeqCst), 2);
+    server.stop().await;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn yellowstone_replay_info_proves_and_clamps_retention_gap_without_lossless_claim() {
+    let server = FixtureServer::start(FixtureMode::ReplayGap).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 _ = session.next_update().await.expect("first slot must decode").expect("first slot must be present");
+    let resumed = session.next_update().await.expect("clamped replay must decode").expect("clamped replay must be present");
+    match resumed {
+        crate::YellowstoneSubscribeUpdate::Slot(value) => assert_eq!(value.slot(), 505),
+        _ => panic!("fixture must return clamped Slot"),
+    }
+    let snapshot = session.snapshot();
+    assert_eq!(snapshot.reconnect_count(), 1);
+    assert_eq!(snapshot.replay_attempt_count(), 1);
+    assert_eq!(snapshot.continuity_gap_count(), 1);
+    assert_eq!(snapshot.duplicate_update_count(), 0);
+    assert_eq!(snapshot.last_requested_from_slot(), std::option::Option::Some(505));
+    assert_eq!(snapshot.last_observed_slot(), std::option::Option::Some(505));
+    session.close().await.expect("gap fixture must close cleanly");
+    server.stop().await;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn yellowstone_reconnect_budget_exhaustion_is_terminal_and_safe() {
+    let server = FixtureServer::start(FixtureMode::ReconnectExhausted).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(2, std::time::Duration::from_millis(5), std::time::Duration::from_millis(10)),
+    );
+    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 _ = session.next_update().await.expect("first slot must decode").expect("first slot must be present");
+    let error = session.next_update().await.expect_err("failed reconnect attempts must exhaust the bounded budget");
+    assert_eq!(error.code(), crate::ERROR_CODE_GRPC_CHANNEL_FAILED);
+    let snapshot = session.snapshot();
+    assert_eq!(snapshot.state(), crate::YellowstoneGrpcSubscribeState::Failed);
+    assert_eq!(snapshot.reconnect_count(), 0);
+    assert_eq!(snapshot.replay_attempt_count(), 2);
+    assert_eq!(snapshot.terminal_error_code(), std::option::Option::Some(crate::ERROR_CODE_GRPC_CHANNEL_FAILED));
+    let rendered = format!("{error:?} {session:?}");
+    assert!(!rendered.contains("GRPC-RECONNECT-SECRET-CANARY"));
+    assert!(!rendered.contains(server.endpoint_url.as_str()));
+    assert_eq!(server.subscribe_calls.load(std::sync::atomic::Ordering::SeqCst), 3);
+    let _ = session.close().await;
+    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;
+    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(500), std::time::Duration::from_millis(500)),
+    );
+    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 _ = session.next_update().await.expect("first slot must decode").expect("first slot must be present");
+    tokio::time::timeout(std::time::Duration::from_secs(1), async {
+        loop {
+            if session.state() == crate::YellowstoneGrpcSubscribeState::Reconnecting {
+                return;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("fixture must enter reconnecting state");
+    let mutation_error = session.try_update(&mutated_request()).expect_err("request mutation must be rejected during reconnect");
+    assert_eq!(mutation_error.code(), crate::ERROR_CODE_GRPC_SESSION_CLOSED);
+    session.close().await.expect("shutdown must interrupt reconnect backoff without opening another stream");
+    assert_eq!(server.subscribe_calls.load(std::sync::atomic::Ordering::SeqCst), 1);
+    server.stop().await;
+}