diff --git a/Cargo.toml b/Cargo.toml index 9673995..3948502 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 226 +# version: 227 [workspace] resolver = "3" members = ["crates/ksp-app-config-desk", "crates/ksp-app-wallet-desk", "crates/ksp-config-lib", "crates/ksp-core-lib", "crates/ksp-logging-lib", "crates/ksp-onchain-transport-lib", "crates/ksp-wallet-lib"] [workspace.package] -version = "0.2.8-pre.6" +version = "0.2.8-pre.7" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-onchain-transport-lib/Cargo.toml b/crates/ksp-onchain-transport-lib/Cargo.toml index a38ac9a..565e89b 100644 --- a/crates/ksp-onchain-transport-lib/Cargo.toml +++ b/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 diff --git a/crates/ksp-onchain-transport-lib/src/ws_session.rs b/crates/ksp-onchain-transport-lib/src/ws_session.rs index 3e6c06f..0164485 100644 --- a/crates/ksp-onchain-transport-lib/src/ws_session.rs +++ b/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>; /// 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::::new(); let mut subscriptions = std::collections::BTreeMap::::new(); let mut remote_to_local = std::collections::BTreeMap::::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( + id: crate::WsSessionId, + endpoint: &crate::WsEndpointSettings, + websocket: &mut tokio_tungstenite::WebSocketStream, + shutdown_rx: &mut tokio::sync::watch::Receiver>, +) -> 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) diff --git a/crates/ksp-onchain-transport-lib/tests/release_completeness.rs b/crates/ksp-onchain-transport-lib/tests/release_completeness.rs index 842e944..a2e0603 100644 --- a/crates/ksp-onchain-transport-lib/tests/release_completeness.rs +++ b/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")); +} diff --git a/crates/ksp-onchain-transport-lib/unit_tests/ws_session.rs b/crates/ksp-onchain-transport-lib/unit_tests/ws_session.rs index 424c9dd..c1b4f97 100644 --- a/crates/ksp-onchain-transport-lib/unit_tests/ws_session.rs +++ b/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::); + 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; diff --git a/deltas/0.2.8/pre.007.md b/deltas/0.2.8/pre.007.md new file mode 100644 index 0000000..a0eaa58 --- /dev/null +++ b/deltas/0.2.8/pre.007.md @@ -0,0 +1,211 @@ + + + +# Delta `0.2.8-pre.007` — heartbeat Helius WebSocket + idle lifecycle + +## 1. Base et objet + +Base appliquée : + +```text +0.2.8-pre.6 +``` + +Le checkpoint opérateur de `pre.006` est intégralement vert et sans warning : + +```text +cargo fmt --all OK +python3 scripts/audit_rust_workspace_rules.py clean +cargo check --workspace OK +cargo clippy --workspace --all-targets OK +Transport unit 325/325 +Transport public API 40/40 +Transport release completeness 28/28 +Transport doctests compile-fail 4/4 +cargo test --workspace OK +``` + +Cette tranche ajoute uniquement la policy de heartbeat Helius LaserStream WebSocket dans l'actor physique commun. Elle ne rouvre pas le lifecycle transaction validé par `pre.006`. + +## 2. Version technique + +```text +workspace.package.version = 0.2.8-pre.7 +commit attendu = v0.2.8-pre.007 +Git tag = aucun tag prerelease +``` + +Le header root `Cargo.toml` passe en version `227`. + +## 3. Policy heartbeat provider-owned + +La décision d'architecture de `pre.001` est matérialisée sans nouveau réglage public : + +```text +protocol WsProtocolKind::HeliusLaserStream uniquement +intervalle nominal 60 s +frame WebSocket Ping control frame vide +état actor Active uniquement +succès réarmement à now + 60 s +reconnect réussi réarmement depuis la connexion de remplacement +write timeout/failure WsActorIoOutcome::Failed +close/shutdown timer abandonné ; shutdown prioritaire +SolanaStandard aucun heartbeat provider +``` + +Le heartbeat n'est pas un appel JSON-RPC `ping`. KSP utilise la frame de contrôle WebSocket `Ping`, déjà compatible avec le traitement `Pong` de l'actor. + +Aucun champ `heartbeat_*` n'est ajouté à : + +```text +WsSessionSettings +WsEndpointSettings +Config +.env.example +schema Transport +``` + +La cadence est donc une policy Helius WebSocket interne, pas une option de configuration générique. + +## 4. Intégration dans l'actor unique + +`run_ws_session_actor` possède le deadline heartbeat avec les mêmes priorités de shutdown que le reste du moteur : + +```text +shutdown +command +socket input +Helius heartbeat deadline +pending JSON-RPC timeout +``` + +Le branch heartbeat est désactivé pour `SolanaStandard`. + +Sur succès du Ping, le prochain deadline est recalculé à partir de `Instant::now()`. Pendant reconnect, aucun heartbeat n'est émis ; lorsque `recover_websocket_session` rend une connexion de remplacement active, le deadline est réarmé à 60 s. + +Le helper d'écriture sélectionne de façon bornée entre : + +```text +shutdown signal +websocket.send(Ping) +command_timeout +``` + +Une erreur d'écriture ou un timeout retourne le même `WsActorIoOutcome::Failed` que les autres erreurs de connexion. Le chemin de reconnect, son budget, son backoff, les remaps et le shutdown restent donc uniques. + +## 5. Timers déterministes de test + +La cadence runtime ne doit pas être raccourcie pour rendre les tests rapides. La crate active donc uniquement côté dev/test la feature Tokio : + +```text +test-util +``` + +avec `io-util` nécessaire au canari de socket cassé en mémoire. + +Il ne s'agit pas d'une nouvelle dépendance ; aucune feature runtime de production n'est ajoutée au contrat KSP. + +Les tests utilisent `tokio::time::pause()` / `advance()` pour vérifier la vraie constante de 60 s. + +## 6. Canaris ajoutés + +Six tests unitaires Transport sont ajoutés : + +```text +Helius-only + constante 60 s +premier Ping à 60 s + second Ping après réarmement +absence totale de heartbeat provider sur SolanaStandard +close à 30 s sans Ping +échec d'écriture Ping -> WsActorIoOutcome::Failed / ERROR_CODE_WS_CONNECTION_FAILED +reconnect avant heartbeat -> nouveau deadline 60 s depuis la connexion de remplacement +``` + +Le test hérité `websocket_shutdown_interrupts_reconnect_backoff_without_new_connection` continue de couvrir l'interruption du backoff par shutdown, tandis que le nouveau branch heartbeat place également le signal shutdown en première priorité. + +Le canari release-completeness vérifie que : + +- la policy reste dans `ws_session.rs` ; +- le Ping est une frame WebSocket ; +- `WsSessionSettings` ne gagne aucun champ de heartbeat ; +- la façade Helius ne possède aucun timer séparé. + +Comptages attendus : + +```text +Transport unit 331 +Transport public API 40 +release completeness 29 +doctests compile-fail 4 +``` + +## 7. Non-régressions + +Cette tranche ne modifie pas : + +```text +surface Helius standard 6 familles +transactionSubscribe request/filter/options +transactionNotification Full/Signature/Unknown +WsSubscriptionKind::HeliusTransaction +remote/local ID remapping +late notification handling +backpressure logical subscription +standard Solana WebSocket 18/18 +HTTP 52 current / 14 historical +Config Helius mainnet/devnet +dependency firewall +``` + +Aucune promesse de replay WebSocket ou de livraison lossless n'est introduite. + +## 8. Documentation + +Le plan `015` : + +- ferme `pre.006` sur preuve opérateur ; +- marque `pre.007` PREPARED ; +- matérialise la policy Helius-only 60 s et ses canaris. + +La validation `011` : + +- transforme les critères lifecycle `pre.006` en preuves acquises ; +- ouvre la gate `pre.007` ; +- conserve les tests adversariaux élargis pour `pre.008`. + +## 9. Hors scope + +Restent hors de `pre.007` : + +```text +provider adversarial lifecycle élargi pre.008 +payload/security/redaction adversarial pre.008 +compliance Helius + standard + HTTP pre.009 +smoke Helius WebSocket live opt-in pre.010 +LaserStream gRPC future backend distinct +``` + +## 10. Fichiers modifiés + +```text +Cargo.toml +crates/ksp-onchain-transport-lib/Cargo.toml +crates/ksp-onchain-transport-lib/src/ws_session.rs +crates/ksp-onchain-transport-lib/unit_tests/ws_session.rs +crates/ksp-onchain-transport-lib/tests/release_completeness.rs +docs/plans/015-V0_2_8_HELIUS_LASERSTREAM_WEBSOCKET_PLAN.md +docs/validation/011-V0_2_8_HELIUS_LASERSTREAM_WEBSOCKET.md +deltas/0.2.8/pre.007.md +``` + +## 11. Gate opérateur + +```bash +cargo fmt --all +python3 scripts/audit_rust_workspace_rules.py +cargo check --workspace +cargo clippy --workspace --all-targets +cargo test -p ksp-onchain-transport-lib +cargo test --workspace +``` + +Critère de fermeture : zéro warning nouveau, audit Rust clean, **331 unit / 40 public API / 29 completeness / 4 doctests** et workspace vert. diff --git a/docs/plans/015-V0_2_8_HELIUS_LASERSTREAM_WEBSOCKET_PLAN.md b/docs/plans/015-V0_2_8_HELIUS_LASERSTREAM_WEBSOCKET_PLAN.md index 3ae7aeb..ae7a390 100644 --- a/docs/plans/015-V0_2_8_HELIUS_LASERSTREAM_WEBSOCKET_PLAN.md +++ b/docs/plans/015-V0_2_8_HELIUS_LASERSTREAM_WEBSOCKET_PLAN.md @@ -1,9 +1,9 @@ - + # Plan `0.2.8` — Helius LaserStream WebSocket -> **Statut : `pre.005` + `fix.001` + `fix.002` sont validés sans warning. `0.2.8-pre.006` prépare maintenant le handle live Helius `transactionSubscribe`, le décodage `transactionNotification` et son intégration au registry/reconnect/backpressure de l'actor WebSocket unique.** +> **Statut : `0.2.8-pre.006` est validé intégralement. `0.2.8-pre.007` prépare maintenant le heartbeat Helius LaserStream WebSocket provider-owned dans l'actor physique unique : Ping control frame à 60 s, timer uniquement en `Active`, réarmement après reconnect et arrêt immédiat au shutdown.** ## 1. Objet, base et état courant @@ -33,10 +33,11 @@ pre.004-fix.002 provenance composée validée pre.005 contrat typed transactionSubscribe/unsubscribe validé pre.005-fix.001 correction types de canaris + visibilité/tests/règles validée pre.005-fix.002 suppression warnings dead_code via cfg(test) validée -pre.006 transactionNotification + lifecycle actor préparé +pre.006 transactionNotification + lifecycle actor validé +pre.007 heartbeat Helius WebSocket / idle préparé -workspace.package.version courant = 0.2.8-pre.6 -commit attendu = v0.2.8-pre.006 +workspace.package.version courant = 0.2.8-pre.7 +commit attendu = v0.2.8-pre.007 aucun tag prerelease ``` @@ -64,9 +65,9 @@ pre.005 DONE — transactionSubscribe request typed + filters/options/tokenAcco + bounds 50k + maxSupportedTransactionVersion conditionnel ; live handle différé à pre.006 fix.001 DONE — assertions Vec/Value corrigées + helpers test-only privés + audit super/crate durci fix.002 DONE — helpers wire non consommés en production bornés à #[cfg(test)] ; zéro warning dead_code -pre.006 PREPARED — transactionNotification + handle live + actor registry/reconnect/resubscribe/unsubscribe races +pre.006 DONE — transactionNotification + handle live + actor registry/reconnect/resubscribe/unsubscribe races + late notifications + backpressure ciblée, sans second actor/socket -pre.007 heartbeat Helius WebSocket/idle + timers + interaction reconnect/control frames/shutdown +pre.007 PREPARED — heartbeat Helius WebSocket/idle + timers + interaction reconnect/control frames/shutdown pre.008 provider adversarial lifecycle + capability guards + payload/backpressure + security/redaction pre.009 compliance Helius WebSocket + non-régressions Solana standard 18/18 + HTTP 52/14 + Config/API/dependency-firewall canaries @@ -1079,3 +1080,61 @@ Les canaris locaux `pre.006` doivent prouver ensemble : ``` Hors scope inchangé : heartbeat/idle (`pre.007`), adversarial provider/security élargi (`pre.008`), smoke Helius live (`pre.010`) et LaserStream gRPC. + +## 15. Fermeture `pre.006` et préparation `pre.007` + +Checkpoint opérateur `0.2.8-pre.006` reçu le 2026-08-23 : + +```text +[x] cargo fmt --all +[x] python3 scripts/audit_rust_workspace_rules.py = clean +[x] cargo check --workspace = sans warning +[x] cargo clippy --workspace --all-targets = sans warning +[x] cargo test -p ksp-onchain-transport-lib = 325 unit + 40 public API + 28 completeness + 4 doctests +[x] cargo test --workspace = vert +``` + +Le lifecycle Helius transaction est donc fermé : `transactionNotification`, handle live, remap remote ID, resubscribe, unsubscribe tardif et backpressure ciblée sont acquis sans second actor/socket. + +`pre.007` n'ajoute aucune surface publique générique de heartbeat. La policy reste possédée par le protocole Helius LaserStream WebSocket dans `WsSession` : + +```text +protocole actif WsProtocolKind::HeliusLaserStream uniquement +intervalle nominal 60 s +wire WebSocket Ping control frame vide +état d'émission Active uniquement +succès prochain deadline = now + 60 s +reconnect réussi deadline réarmé depuis la nouvelle connexion +write timeout/failure WsActorIoOutcome::Failed -> reconnect borné existant +close/shutdown branche shutdown prioritaire, timer abandonné +SolanaStandard aucun heartbeat provider +``` + +Aucun champ `heartbeat_*` n'est ajouté à `WsSessionSettings` ou Config en `0.2.8`. La cadence Helius est une policy interne provider-owned, comme décidé à l'audit d'ouverture. + +Pour rendre les canaris de timer déterministes sans réduire la cadence de production, `tokio` active uniquement la feature de dev/test `test-util`; aucune nouvelle dépendance runtime n'est ajoutée. + +Canaris locaux `pre.007` préparés : + +```text +[ ] policy Helius-only + intervalle exact 60 s +[ ] premier Ping après 60 s, puis réarmement périodique +[ ] SolanaStandard n'émet aucun Ping provider même après plusieurs intervalles +[ ] close avant deadline n'émet aucun heartbeat +[ ] write failure du Ping produit le même outcome Failed utilisé par le reconnect +[ ] reconnect réussi réarme le heartbeat depuis la nouvelle connexion, sans Ping immédiat +[ ] shutdown/backoff existant reste interruptible +[ ] aucune modification du lifecycle transaction pre.006 +``` + +Comptages attendus après ajout des canaris : + +```text +Transport unit 331 +Transport public API 40 +release completeness 29 +doctests compile-fail 4 +``` + +Le heartbeat Helius reste un mécanisme de maintien de connexion ; il n'introduit aucune promesse de replay WebSocket ni de livraison lossless. + diff --git a/docs/validation/011-V0_2_8_HELIUS_LASERSTREAM_WEBSOCKET.md b/docs/validation/011-V0_2_8_HELIUS_LASERSTREAM_WEBSOCKET.md index 0ea30ec..99f1ce4 100644 --- a/docs/validation/011-V0_2_8_HELIUS_LASERSTREAM_WEBSOCKET.md +++ b/docs/validation/011-V0_2_8_HELIUS_LASERSTREAM_WEBSOCKET.md @@ -3,7 +3,7 @@ # Validation `0.2.8` — Helius LaserStream WebSocket -> **Statut : `pre.005` + ses deux fixes sont validés sans warning. `pre.006` est préparé avec `transactionNotification`, handle live, remap reconnect, late-notification handling et backpressure ciblée sur l'actor WebSocket unique.** +> **Statut : `pre.006` est validé intégralement. `pre.007` est préparé pour le heartbeat Helius LaserStream WebSocket provider-owned : Ping control frame à 60 s uniquement en `Active`, réarmé après reconnect et interrompu par close/shutdown.** ## 1. Références @@ -137,7 +137,7 @@ Verdict : **gate `0.2.8-pre.001` positif ; `pre.002` + `fix.001` et `pre.003` so ### 4.2 Invariants architecture -| Critère | Décision | Preuve cible | État | +| Critère | Décision | Preuve cible | État | |-------------------------------------|------------------------------------------------------------------|---------------------------------|--------------------| | actor physique | un seul `WsSession` actor partagé | source/runtime canary | pre.002 implémenté | | façade standard | `SolanaStandardWsSession` | public API canary | pre.002 implémenté | @@ -212,19 +212,19 @@ aucun payload brut dans logs/snapshots [x] 6 familles standard Helius utilisent le wire standard exact — fixture `pre.003` opérateur verte [x] transaction subscribe/ack exact — fixture `pre.005` validée puis consommée par le handle live `pre.006` [x] transaction unsubscribe/result exact — fixture `pre.005` validée puis registry actor `pre.006` -[ ] transaction notification dispatch exact — canari `pre.006` préparé +[x] transaction notification dispatch exact — canari `pre.006` opérateur vert [ ] notification decode failure isole la logical subscription [ ] provider RPC application error ne tue pas la session -[ ] late notification après unsubscribe ne réactive rien — canari `pre.006` préparé -[ ] reconnect invalide/remappe remote ids — canari `pre.006` 41 -> 99 préparé -[ ] resubscribe garde local id — canari `pre.006` préparé +[x] late notification après unsubscribe ne réactive rien — canari `pre.006` opérateur vert +[x] reconnect invalide/remappe remote ids — canari `pre.006` 41 -> 99 opérateur vert +[x] resubscribe garde local id — canari `pre.006` opérateur vert [ ] unsubscribe pendant reconnect gagne [ ] heartbeat n'écrit qu'en Active [ ] heartbeat est annulé au close [ ] heartbeat write failure suit reconnect borné [ ] shutdown interrompt heartbeat/backoff [ ] oversized inbound frame reste borné -[ ] queue overflow d'une transaction subscription n'affecte pas les autres — canari transaction + root `pre.006` préparé +[x] queue overflow d'une transaction subscription n'affecte pas les autres — canari transaction + root `pre.006` opérateur vert [ ] payload/filters/api-key absents de Debug/snapshots sûrs ``` @@ -571,3 +571,61 @@ Gate opérateur `pre.006` : Comptages attendus si les nouveaux canaris passent : **325 tests unitaires Transport**, **40 public API**, **28 release-completeness**, **4 doctests compile-fail**. Verdict courant : **`pre.006` PREPARED.** + +## 17. Gate `pre.006` fermé et préparation `pre.007` + +Checkpoint opérateur `pre.006` : + +```text +[x] cargo fmt --all +[x] python3 scripts/audit_rust_workspace_rules.py = clean +[x] cargo check --workspace = sans warning +[x] cargo clippy --workspace --all-targets = sans warning +[x] cargo test -p ksp-onchain-transport-lib = 325 unit + 40 public API + 28 completeness + 4 doctests +[x] cargo test --workspace = vert +``` + +Preuves lifecycle désormais acquises : + +```text +[x] WsSubscriptionKind::HeliusTransaction possède le triplet exact +[x] HeliusLaserStreamWsSession::transaction_subscribe retourne le handle live typed +[x] notification Full / Signature / Unknown décodées +[x] local WsSubscriptionId stable après reconnect +[x] remote id remappé +[x] late notification après unsubscribe ignorée +[x] overflow transaction isole uniquement le handle lent +[x] cleanup overflow utilise transactionUnsubscribe +[x] une souscription Helius root saine reste active +[x] aucun second actor/socket/registry +[x] quatre compile-fail Helius restent verts +``` + +Gate préparé pour `pre.007` : + +```text +[ ] heartbeat activé uniquement pour WsProtocolKind::HeliusLaserStream +[ ] intervalle provider-owned exactement 60 s +[ ] heartbeat utilise WebSocket Ping control frame, pas un JSON-RPC provider method +[ ] aucun heartbeat provider sur SolanaStandard +[ ] heartbeat n'écrit qu'en état Active +[ ] succès réarme le prochain deadline à now + 60 s +[ ] reconnect réussi réarme le deadline depuis la nouvelle connexion +[ ] write timeout/failure suit le chemin Failed -> reconnect existant +[ ] close avant deadline annule le heartbeat +[ ] shutdown reste prioritaire sur le heartbeat +[ ] aucun champ heartbeat public ajouté à WsSessionSettings +[ ] aucune configuration/env supplémentaire +[ ] aucune régression transactionSubscribe/transactionNotification de pre.006 +[ ] cargo fmt --all +[ ] python3 scripts/audit_rust_workspace_rules.py = clean +[ ] cargo check --workspace = sans warning +[ ] cargo clippy --workspace --all-targets = sans warning +[ ] cargo test -p ksp-onchain-transport-lib = 331 unit + 40 public API + 29 completeness + 4 doctests attendus +[ ] cargo test --workspace = vert +``` + +Le `test-util` Tokio ajouté côté dev sert uniquement à avancer l'horloge des canaris déterministes ; la cadence runtime reste 60 s et aucune nouvelle dépendance de production n'est introduite. + +Verdict courant : **`pre.006` DONE ; `pre.007` PREPARED.** +