v0.2.8-pre.007

crates/ksp-onchain-transport-lib/Cargo.toml

@@ -1,5 +1,5 @@
 # file: crates/ksp-onchain-transport-lib/Cargo.toml
-# version: 5
+# version: 6
 
 [package]
 name = "ksp-onchain-transport-lib"
@@ -18,7 +18,7 @@ tokio = { workspace = true, features = ["macros", "net", "rt", "sync", "time"] }
 tokio-tungstenite = { workspace = true, features = ["connect", "rustls-tls-webpki-roots"] }
 
 [dev-dependencies]
-tokio = { workspace = true, features = ["net", "rt"] }
+tokio = { workspace = true, features = ["io-util", "net", "rt", "test-util"] }
 
 [lints]
 workspace = true

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

@@ -1,11 +1,13 @@
 // file: crates/ksp-onchain-transport-lib/src/ws_session.rs
-// version: 13
+// version: 14
 
 use futures_util::SinkExt; // rust-rules: trait-import
 use futures_util::StreamExt; // rust-rules: trait-import
 
 static NEXT_WS_SESSION_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
 
+const HELIUS_WS_HEARTBEAT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60);
+
 type WsPhysicalStream = tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
 
 /// Shareable handle for one explicitly created physical WebSocket session.
@@ -412,6 +414,8 @@ async fn run_ws_session_actor(
     let mut pending = std::collections::BTreeMap::<u64, PendingWsRequest>::new();
     let mut subscriptions = std::collections::BTreeMap::<u64, crate::WsSubscriptionRuntime>::new();
     let mut remote_to_local = std::collections::BTreeMap::<u64, crate::WsSubscriptionId>::new();
+    let heartbeat_enabled = helius_heartbeat_enabled(endpoint.protocol());
+    let mut heartbeat_deadline = next_helius_heartbeat_deadline();
     loop {
         prune_cancelled_pending(id, &mut pending, &mut subscriptions, &mut remote_to_local);
         let timeout_deadline = next_pending_deadline(&pending);
@@ -471,6 +475,13 @@ async fn run_ws_session_actor(
                 )
                 .await
             },
+            () = tokio::time::sleep_until(heartbeat_deadline), if heartbeat_enabled => {
+                let heartbeat = send_helius_heartbeat(id, &endpoint, &mut websocket, &mut shutdown_rx).await;
+                if matches!(&heartbeat, WsActorIoOutcome::Continue) {
+                    heartbeat_deadline = next_helius_heartbeat_deadline();
+                }
+                heartbeat
+            },
             () = tokio::time::sleep_until(timeout_deadline) => {
                 expire_pending_requests(id, &mut pending, &mut subscriptions, &mut remote_to_local);
                 WsActorIoOutcome::Continue
@@ -550,7 +561,12 @@ async fn run_ws_session_actor(
                 )
                 .await;
                 match recovery {
-                    WsReconnectOutcome::Connected { websocket: replacement } => websocket = *replacement,
+                    WsReconnectOutcome::Connected { websocket: replacement } => {
+                        websocket = *replacement;
+                        if heartbeat_enabled {
+                            heartbeat_deadline = next_helius_heartbeat_deadline();
+                        }
+                    },
                     WsReconnectOutcome::ShutdownRequested { deadline } => {
                         finish_disconnected_shutdown(
                             id,
@@ -611,7 +627,12 @@ async fn run_ws_session_actor(
                 )
                 .await;
                 match recovery {
-                    WsReconnectOutcome::Connected { websocket: replacement } => websocket = *replacement,
+                    WsReconnectOutcome::Connected { websocket: replacement } => {
+                        websocket = *replacement;
+                        if heartbeat_enabled {
+                            heartbeat_deadline = next_helius_heartbeat_deadline();
+                        }
+                    },
                     WsReconnectOutcome::ShutdownRequested { deadline } => {
                         finish_disconnected_shutdown(
                             id,
@@ -658,6 +679,69 @@ async fn run_ws_session_actor(
     }
 }
 
+const fn helius_heartbeat_enabled(protocol: crate::WsProtocolKind) -> bool {
+    return matches!(protocol, crate::WsProtocolKind::HeliusLaserStream);
+}
+
+fn next_helius_heartbeat_deadline() -> tokio::time::Instant {
+    return tokio::time::Instant::now() + HELIUS_WS_HEARTBEAT_INTERVAL;
+}
+
+async fn send_helius_heartbeat<S>(
+    id: crate::WsSessionId,
+    endpoint: &crate::WsEndpointSettings,
+    websocket: &mut tokio_tungstenite::WebSocketStream<S>,
+    shutdown_rx: &mut tokio::sync::watch::Receiver<std::option::Option<tokio::time::Instant>>,
+) -> WsActorIoOutcome
+where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + std::marker::Unpin,
+{
+    let message = tokio_tungstenite::tungstenite::Message::Ping(std::vec::Vec::new().into());
+    let send_result = tokio::select! {
+        biased;
+        shutdown_changed = shutdown_rx.changed() => {
+            let deadline = resolve_shutdown_deadline(shutdown_rx, shutdown_changed, endpoint.session().close_timeout());
+            return WsActorIoOutcome::ShutdownRequested { deadline };
+        },
+        send_result = websocket.send(message) => send_result,
+        () = tokio::time::sleep(endpoint.session().command_timeout()) => {
+            ksp_logging_lib::warn!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                endpoint_name = endpoint.name(),
+                "Helius WebSocket heartbeat Ping write timed out"
+            );
+            return WsActorIoOutcome::Failed {
+                code: crate::ERROR_CODE_WS_CONNECTION_FAILED,
+                pending_message: "WebSocket connection failed while writing Helius heartbeat Ping",
+            };
+        },
+    };
+    return match send_result {
+        std::result::Result::Ok(()) => {
+            ksp_logging_lib::trace!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                endpoint_name = endpoint.name(),
+                "sent Helius WebSocket heartbeat Ping control frame"
+            );
+            WsActorIoOutcome::Continue
+        },
+        std::result::Result::Err(_) => {
+            ksp_logging_lib::warn!(
+                target: crate::TRACING_TARGET,
+                session_id = id.get(),
+                endpoint_name = endpoint.name(),
+                "Helius WebSocket heartbeat Ping write failed"
+            );
+            WsActorIoOutcome::Failed {
+                code: crate::ERROR_CODE_WS_CONNECTION_FAILED,
+                pending_message: "WebSocket connection failed while writing Helius heartbeat Ping",
+            }
+        },
+    };
+}
+
 fn websocket_config(endpoint: &crate::WsEndpointSettings) -> tokio_tungstenite::tungstenite::protocol::WebSocketConfig {
     return tokio_tungstenite::tungstenite::protocol::WebSocketConfig::default()
         .write_buffer_size(0)

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/tests/release_completeness.rs
-// version: 28
+// version: 29
 
 //! Release-level completeness canaries for staged HTTP and WebSocket Transport coverage.
 
@@ -850,3 +850,18 @@ fn release_v0_2_8_pre_006_helius_transaction_lifecycle_is_actor_integrated_witho
     assert!(protocol_source.contains("unsupported_slots_updates"));
     assert!(protocol_source.contains("unsupported_vote"));
 }
+
+#[test]
+fn release_v0_2_8_pre_007_helius_heartbeat_is_provider_owned_by_shared_actor_only() {
+    let actor_source = include_str!("../src/ws_session.rs");
+    assert!(actor_source.contains("HELIUS_WS_HEARTBEAT_INTERVAL"));
+    assert!(actor_source.contains("std::time::Duration::from_secs(60)"));
+    assert!(actor_source.contains("WsProtocolKind::HeliusLaserStream"));
+    assert!(actor_source.contains("tungstenite::Message::Ping"));
+    assert!(actor_source.contains("send_helius_heartbeat"));
+    let settings_source = include_str!("../src/ws_settings.rs");
+    assert!(!settings_source.contains("heartbeat_interval"));
+    assert!(!settings_source.contains("heartbeat_enabled"));
+    let protocol_source = include_str!("../src/ws_protocol_session.rs");
+    assert!(!protocol_source.contains("heartbeat_interval"));
+}

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/unit_tests/ws_session.rs
-// version: 7
+// version: 8
 
 use futures_util::SinkExt; // rust-rules: trait-import
 use futures_util::StreamExt; // rust-rules: trait-import
@@ -20,6 +20,22 @@ fn local_endpoint_with_session(url: &str, session: crate::WsSessionSettings) ->
     );
 }
 
+fn helius_local_endpoint(url: &str) -> crate::WsEndpointSettings {
+    return helius_local_endpoint_with_session(url, crate::WsSessionSettings::default());
+}
+
+fn helius_local_endpoint_with_session(url: &str, session: crate::WsSessionSettings) -> crate::WsEndpointSettings {
+    return crate::WsEndpointSettings::new(
+        "local_helius_ws",
+        true,
+        crate::WsProviderName::new("helius-fixture"),
+        crate::WsClusterName::new("local"),
+        crate::WsProtocolKind::HeliusLaserStream,
+        crate::WsEndpointUrl::parse(url).expect("local Helius test WebSocket URL must parse"),
+        session,
+    );
+}
+
 fn session_settings(
     command_timeout: std::time::Duration,
     close_timeout: std::time::Duration,
@@ -166,6 +182,12 @@ async fn wait_for_close_frame(websocket: &mut tokio_tungstenite::WebSocketStream
     }
 }
 
+async fn yield_runtime_steps() {
+    for _ in 0..16 {
+        tokio::task::yield_now().await;
+    }
+}
+
 #[tokio::test(flavor = "current_thread")]
 async fn websocket_session_connects_and_round_trips_internal_json_rpc() {
     let (listener, url) = bind_local_listener().await;
@@ -400,6 +422,188 @@ async fn websocket_ping_flushes_automatic_pong_and_keeps_session_active() {
     server.await.expect("local server task must complete");
 }
 
+#[test]
+fn helius_heartbeat_policy_is_provider_owned_and_fixed_to_sixty_seconds() {
+    assert!(super::helius_heartbeat_enabled(crate::WsProtocolKind::HeliusLaserStream));
+    assert!(!super::helius_heartbeat_enabled(crate::WsProtocolKind::SolanaStandard));
+    assert_eq!(super::HELIUS_WS_HEARTBEAT_INTERVAL, std::time::Duration::from_secs(60));
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn helius_heartbeat_sends_ping_at_sixty_seconds_and_rearms() {
+    let (listener, url) = bind_local_listener().await;
+    let (ping_tx, mut ping_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("local Helius server must accept client");
+        let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local Helius handshake must succeed");
+        loop {
+            let message = websocket.next().await;
+            match message {
+                std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Ping(_))) => {
+                    ping_tx.send(()).expect("heartbeat observation channel must remain open");
+                    websocket.flush().await.expect("automatic heartbeat Pong must flush");
+                },
+                std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => return,
+                std::option::Option::Some(std::result::Result::Ok(_)) => {},
+                std::option::Option::Some(std::result::Result::Err(_)) | std::option::Option::None => return,
+            }
+        }
+    });
+    let session = crate::HeliusLaserStreamWsSession::connect(helius_local_endpoint(url.as_str())).await.expect("Helius fixture session must connect");
+    tokio::time::pause();
+    tokio::time::advance(std::time::Duration::from_secs(59)).await;
+    yield_runtime_steps().await;
+    assert!(matches!(ping_rx.try_recv(), std::result::Result::Err(tokio::sync::mpsc::error::TryRecvError::Empty)));
+    tokio::time::advance(std::time::Duration::from_secs(1)).await;
+    yield_runtime_steps().await;
+    assert!(matches!(ping_rx.try_recv(), std::result::Result::Ok(())));
+    tokio::time::advance(std::time::Duration::from_secs(59)).await;
+    yield_runtime_steps().await;
+    assert!(matches!(ping_rx.try_recv(), std::result::Result::Err(tokio::sync::mpsc::error::TryRecvError::Empty)));
+    tokio::time::advance(std::time::Duration::from_secs(1)).await;
+    yield_runtime_steps().await;
+    assert!(matches!(ping_rx.try_recv(), std::result::Result::Ok(())));
+    session.close().await.expect("Helius heartbeat fixture session must close");
+    tokio::time::resume();
+    server.await.expect("Helius heartbeat fixture server must complete");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn standard_session_never_emits_helius_provider_heartbeat() {
+    let (listener, url) = bind_local_listener().await;
+    let (ping_tx, mut ping_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("local standard server must accept client");
+        let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local standard handshake must succeed");
+        loop {
+            let message = websocket.next().await;
+            match message {
+                std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Ping(_))) => {
+                    ping_tx.send(()).expect("standard heartbeat observation channel must remain open");
+                    websocket.flush().await.expect("automatic Pong must flush");
+                },
+                std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => return,
+                std::option::Option::Some(std::result::Result::Ok(_)) => {},
+                std::option::Option::Some(std::result::Result::Err(_)) | std::option::Option::None => return,
+            }
+        }
+    });
+    let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("standard fixture session must connect");
+    tokio::time::pause();
+    tokio::time::advance(std::time::Duration::from_secs(180)).await;
+    yield_runtime_steps().await;
+    assert!(matches!(ping_rx.try_recv(), std::result::Result::Err(tokio::sync::mpsc::error::TryRecvError::Empty)));
+    session.close().await.expect("standard fixture session must close");
+    tokio::time::resume();
+    server.await.expect("standard heartbeat absence fixture server must complete");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn helius_explicit_close_cancels_heartbeat_before_deadline() {
+    let (listener, url) = bind_local_listener().await;
+    let (ping_tx, mut ping_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("local Helius close server must accept client");
+        let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local Helius close handshake must succeed");
+        loop {
+            let message = websocket.next().await;
+            match message {
+                std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Ping(_))) => {
+                    ping_tx.send(()).expect("close heartbeat observation channel must remain open");
+                    websocket.flush().await.expect("automatic heartbeat Pong must flush");
+                },
+                std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => return,
+                std::option::Option::Some(std::result::Result::Ok(_)) => {},
+                std::option::Option::Some(std::result::Result::Err(_)) | std::option::Option::None => return,
+            }
+        }
+    });
+    let session = crate::HeliusLaserStreamWsSession::connect(helius_local_endpoint(url.as_str())).await.expect("Helius close fixture session must connect");
+    tokio::time::pause();
+    tokio::time::advance(std::time::Duration::from_secs(30)).await;
+    yield_runtime_steps().await;
+    session.close().await.expect("Helius close fixture session must close before heartbeat deadline");
+    yield_runtime_steps().await;
+    assert!(matches!(ping_rx.try_recv(), std::result::Result::Err(tokio::sync::mpsc::error::TryRecvError::Empty)));
+    tokio::time::resume();
+    server.await.expect("Helius close fixture server must complete");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn helius_heartbeat_write_failure_maps_to_existing_reconnect_failure_outcome() {
+    let (client_io, peer_io) = tokio::io::duplex(64);
+    let mut websocket =
+        tokio_tungstenite::WebSocketStream::from_raw_socket(client_io, tokio_tungstenite::tungstenite::protocol::Role::Client, std::option::Option::None).await;
+    drop(peer_io);
+    let endpoint = helius_local_endpoint("ws://127.0.0.1:65535");
+    let (_shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(std::option::Option::None::<tokio::time::Instant>);
+    let session_id = crate::WsSessionId::new(std::num::NonZeroU64::new(1).expect("fixture session id must be non-zero"));
+    let outcome = super::send_helius_heartbeat(session_id, &endpoint, &mut websocket, &mut shutdown_rx).await;
+    match outcome {
+        super::WsActorIoOutcome::Failed { code, pending_message } => {
+            assert_eq!(code, crate::ERROR_CODE_WS_CONNECTION_FAILED);
+            assert_eq!(pending_message, "WebSocket connection failed while writing Helius heartbeat Ping");
+        },
+        _ => panic!("heartbeat write failure must use the existing reconnect failure outcome"),
+    }
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn helius_heartbeat_is_rearmed_from_successful_reconnect() {
+    let (listener, url) = bind_local_listener().await;
+    let (disconnect_tx, disconnect_rx) = tokio::sync::oneshot::channel::<()>();
+    let (replacement_tx, replacement_rx) = tokio::sync::oneshot::channel::<()>();
+    let (ping_tx, mut ping_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
+    let server = tokio::spawn(async move {
+        let (first_stream, _) = listener.accept().await.expect("first Helius connection must be accepted");
+        let mut first = tokio_tungstenite::accept_async(first_stream).await.expect("first Helius handshake must succeed");
+        disconnect_rx.await.expect("disconnect trigger must arrive");
+        first.send(tokio_tungstenite::tungstenite::Message::Close(std::option::Option::None)).await.expect("first Helius close must send");
+        let (replacement_stream, _) = listener.accept().await.expect("replacement Helius connection must be accepted");
+        let mut replacement = tokio_tungstenite::accept_async(replacement_stream).await.expect("replacement Helius handshake must succeed");
+        replacement_tx.send(()).expect("replacement observation must be delivered");
+        loop {
+            let message = replacement.next().await;
+            match message {
+                std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Ping(_))) => {
+                    ping_tx.send(()).expect("replacement heartbeat observation channel must remain open");
+                    replacement.flush().await.expect("replacement automatic Pong must flush");
+                },
+                std::option::Option::Some(std::result::Result::Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => return,
+                std::option::Option::Some(std::result::Result::Ok(_)) => {},
+                std::option::Option::Some(std::result::Result::Err(_)) | std::option::Option::None => return,
+            }
+        }
+    });
+    let settings = reconnect_session_settings(1, std::time::Duration::from_millis(20), crate::WsResubscribePolicy::ActiveSubscriptions);
+    let session = crate::HeliusLaserStreamWsSession::connect(helius_local_endpoint_with_session(url.as_str(), settings))
+        .await
+        .expect("Helius reconnect fixture session must connect");
+    disconnect_tx.send(()).expect("disconnect trigger must send");
+    tokio::time::timeout(std::time::Duration::from_secs(1), replacement_rx)
+        .await
+        .expect("replacement connection must remain bounded")
+        .expect("replacement connection must be observed");
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
+    loop {
+        if session.state() == crate::WsSessionState::Active && session.snapshot().continuity_gap_count() == 1 {
+            break;
+        }
+        assert!(tokio::time::Instant::now() < deadline, "Helius fixture must recover before heartbeat rearm check");
+        tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+    }
+    tokio::time::pause();
+    tokio::time::advance(std::time::Duration::from_secs(59)).await;
+    yield_runtime_steps().await;
+    assert!(matches!(ping_rx.try_recv(), std::result::Result::Err(tokio::sync::mpsc::error::TryRecvError::Empty)));
+    tokio::time::advance(std::time::Duration::from_secs(1)).await;
+    yield_runtime_steps().await;
+    assert!(matches!(ping_rx.try_recv(), std::result::Result::Ok(())));
+    session.close().await.expect("reconnected Helius fixture session must close");
+    tokio::time::resume();
+    server.await.expect("Helius reconnect heartbeat fixture server must complete");
+}
+
 #[tokio::test(flavor = "current_thread")]
 async fn websocket_remote_close_consumes_bounded_reconnect_budget_before_terminal_failure() {
     let (listener, url) = bind_local_listener().await;