v0.2.8-pre.008

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/src/ws_helius_transactions.rs
-// version: 4
+// version: 5
 
 const MAX_HELIUS_TRANSACTION_FILTER_ACCOUNTS: usize = 50_000;
 
@@ -348,7 +348,7 @@ fn helius_transaction_subscribe_params(request: &crate::HeliusTransactionSubscri
 ///
 /// The nested transaction payload is deliberately retained as JSON because its exact Solana wire representation depends on the requested encoding and detail
 /// mode. KSP types the stable provider envelope while preserving the full nested payload without Program-specific decoding.
-#[derive(Clone, Debug, PartialEq)]
+#[derive(Clone, PartialEq)]
 pub struct HeliusFullTransactionNotification {
     transaction: serde_json::Value,
     signature: std::string::String,
@@ -382,8 +382,20 @@ impl HeliusFullTransactionNotification {
     }
 }
 
+impl std::fmt::Debug for HeliusFullTransactionNotification {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("HeliusFullTransactionNotification")
+            .field("transaction", &"<omitted>")
+            .field("signature", &"<omitted>")
+            .field("slot", &self.slot)
+            .field("transaction_index", &self.transaction_index)
+            .finish();
+    }
+}
+
 /// Signatures-mode notification delivered by Helius `transactionSubscribe`.
-#[derive(Clone, Debug, PartialEq)]
+#[derive(Clone, PartialEq)]
 pub struct HeliusTransactionSignatureNotification {
     signature: std::string::String,
     slot: u64,
@@ -438,11 +450,36 @@ impl HeliusTransactionSignatureNotification {
     }
 }
 
+impl std::fmt::Debug for HeliusTransactionSignatureNotification {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("HeliusTransactionSignatureNotification")
+            .field("signature", &"<omitted>")
+            .field("slot", &self.slot)
+            .field("transaction_index", &self.transaction_index)
+            .field("err", &wire_field_debug_state(&self.err))
+            .field("memo", &wire_field_debug_state(&self.memo))
+            .field("block_time", &wire_field_debug_state(&self.block_time))
+            .field("confirmation_status", &wire_field_debug_state(&self.confirmation_status))
+            .finish();
+    }
+}
+
+fn wire_field_debug_state<T>(field: &crate::SolanaWireField<T>) -> &'static str {
+    if field.is_omitted() {
+        return "omitted";
+    }
+    if field.is_null() {
+        return "null";
+    }
+    return "value";
+}
+
 /// Typed Helius `transactionNotification` payload union.
 ///
 /// `Full` also covers the provider `accounts` detail mode because both contain the nested `transaction` member. `Signature` covers the lightweight
 /// signatures mode. `Unknown` preserves `none` mode and forward-compatible provider shapes instead of failing the logical subscription.
-#[derive(Clone, Debug, PartialEq)]
+#[derive(Clone, PartialEq)]
 #[non_exhaustive]
 pub enum HeliusTransactionNotification {
     /// Full/accounts notification carrying the nested transaction payload.
@@ -453,6 +490,16 @@ pub enum HeliusTransactionNotification {
     Unknown(serde_json::Value),
 }
 
+impl std::fmt::Debug for HeliusTransactionNotification {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return match self {
+            Self::Full(notification) => formatter.debug_tuple("Full").field(notification).finish(),
+            Self::Signature(notification) => formatter.debug_tuple("Signature").field(notification).finish(),
+            Self::Unknown(_) => formatter.debug_tuple("Unknown").field(&"<omitted>").finish(),
+        };
+    }
+}
+
 impl crate::HeliusLaserStreamWsSession {
     /// Opens one Helius `transactionSubscribe` logical subscription through the shared physical actor.
     ///

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

@@ -1,11 +1,20 @@
 // file: crates/ksp-onchain-transport-lib/src/ws_protocol_session.rs
-// version: 4
+// version: 5
 
 /// Typed facade for one standard Solana WebSocket physical session.
 ///
 /// The facade delegates to the same [`crate::WsSession`] actor used by the compatibility API. It owns no socket, registry, reconnect loop or queue of its
 /// own and therefore does not duplicate the physical WebSocket runtime. Subscription wrappers are implemented beside their wire owners in the
 /// `ws_accounts`, `ws_blocks`, `ws_cluster` and `ws_transactions` modules.
+///
+/// ```compile_fail
+/// async fn unsupported_helius_transaction(
+///     session: &ksp_onchain_transport_lib::SolanaStandardWsSession,
+///     request: &ksp_onchain_transport_lib::HeliusTransactionSubscribeRequest,
+/// ) {
+///     let _ = session.transaction_subscribe(request).await;
+/// }
+/// ```
 #[derive(Clone)]
 pub struct SolanaStandardWsSession {
     inner: crate::WsSession,

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/tests/release_completeness.rs
-// version: 29
+// version: 30
 
 //! Release-level completeness canaries for staged HTTP and WebSocket Transport coverage.
 
@@ -808,7 +808,7 @@ fn release_v0_2_8_pre_003_helius_surface_is_exactly_six_standard_families_before
     assert!(source.contains("unsupported_block"));
     assert!(source.contains("unsupported_slots_updates"));
     assert!(source.contains("unsupported_vote"));
-    assert!(!source.contains("transaction_subscribe"));
+    assert!(!source.contains("pub async fn transaction_subscribe"));
     assert_eq!(ksp_onchain_transport_lib::WsProtocolKind::HeliusLaserStream.as_str(), "helius_laserstream");
 }
 
@@ -865,3 +865,24 @@ fn release_v0_2_8_pre_007_helius_heartbeat_is_provider_owned_by_shared_actor_onl
     let protocol_source = include_str!("../src/ws_protocol_session.rs");
     assert!(!protocol_source.contains("heartbeat_interval"));
 }
+
+#[test]
+fn release_v0_2_8_pre_008_adversarial_guards_preserve_provider_isolation_and_safe_diagnostics() {
+    let protocol_source = include_str!("../src/ws_protocol_session.rs");
+    assert!(protocol_source.contains("unsupported_helius_transaction"));
+    assert!(protocol_source.contains("unsupported_block"));
+    assert!(protocol_source.contains("unsupported_slots_updates"));
+    assert!(protocol_source.contains("unsupported_vote"));
+    let helius_source = include_str!("../src/ws_helius_transactions.rs");
+    assert!(helius_source.contains("impl std::fmt::Debug for HeliusFullTransactionNotification"));
+    assert!(helius_source.contains("impl std::fmt::Debug for HeliusTransactionSignatureNotification"));
+    assert!(helius_source.contains("impl std::fmt::Debug for HeliusTransactionNotification"));
+    assert!(helius_source.contains("Self::Unknown(_)"));
+    assert!(helius_source.contains("<omitted>"));
+    let actor_source = include_str!("../src/ws_session.rs");
+    assert!(actor_source.contains("max_message_size_bytes"));
+    assert!(actor_source.contains("max_frame_size_bytes"));
+    assert!(actor_source.contains("WsNotificationDispatchOutcome::QueueFull"));
+    assert!(actor_source.contains("ERROR_CODE_WS_BACKPRESSURE_OVERFLOW"));
+    assert!(actor_source.contains("remote_to_local.remove"));
+}

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/unit_tests/ws_helius_transactions.rs
-// version: 4
+// version: 5
 
 use futures_util::SinkExt; // rust-rules: trait-import
 use futures_util::StreamExt; // rust-rules: trait-import
@@ -78,6 +78,23 @@ fn backpressure_session_settings() -> crate::WsSessionSettings {
     );
 }
 
+fn adversarial_payload_session_settings() -> crate::WsSessionSettings {
+    let defaults = crate::WsSessionSettings::default();
+    return crate::WsSessionSettings::new(
+        std::time::Duration::from_millis(250),
+        std::time::Duration::from_millis(200),
+        crate::WsReconnectSettings::new(2, std::time::Duration::from_millis(20), std::time::Duration::from_millis(20)),
+        crate::WsResubscribePolicy::ActiveSubscriptions,
+        defaults.command_queue_capacity(),
+        defaults.notification_queue_capacity(),
+        defaults.max_active_subscriptions(),
+        defaults.max_pending_requests(),
+        256,
+        128,
+        defaults.max_write_buffer_size_bytes(),
+    );
+}
+
 async fn bind_local_listener() -> (tokio::net::TcpListener, std::string::String) {
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("local listener must bind");
     let address = listener.local_addr().expect("local listener must expose address");
@@ -97,6 +114,19 @@ async fn send_result(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::n
     return;
 }
 
+async fn send_error(
+    websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>,
+    request: &serde_json::Value,
+    code: i64,
+    message: &str,
+    data: serde_json::Value,
+) {
+    let id = request.get("id").and_then(serde_json::Value::as_u64).expect("request id must be numeric");
+    let response = serde_json::json!({"jsonrpc":"2.0","id":id,"error":{"code":code,"message":message,"data":data}});
+    websocket.send(tokio_tungstenite::tungstenite::Message::Text(response.to_string().into())).await.expect("local error response must send");
+    return;
+}
+
 async fn send_notification(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>, subscription: u64, result: serde_json::Value) {
     let notification = serde_json::json!({"jsonrpc":"2.0","method":"transactionNotification","params":{"subscription":subscription,"result":result}});
     websocket.send(tokio_tungstenite::tungstenite::Message::Text(notification.to_string().into())).await.expect("local notification must send");
@@ -115,7 +145,7 @@ async fn send_root_notification(websocket: &mut tokio_tungstenite::WebSocketStre
 async fn wait_for_gap_count(session: &crate::HeliusLaserStreamWsSession, expected: u64) {
     let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
     loop {
-        if session.snapshot().continuity_gap_count() >= expected {
+        if session.snapshot().continuity_gap_count() >= expected && session.state() == crate::WsSessionState::Active {
             return;
         }
         assert!(tokio::time::Instant::now() < deadline, "Helius session continuity gap count must advance before timeout");
@@ -134,6 +164,17 @@ async fn wait_for_overflow_count(session: &crate::HeliusLaserStreamWsSession, ex
     }
 }
 
+async fn wait_for_session_subscription_count(session: &crate::HeliusLaserStreamWsSession, expected: usize) {
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
+    loop {
+        if session.snapshot().subscription_count() == expected {
+            return;
+        }
+        assert!(tokio::time::Instant::now() < deadline, "Helius session subscription count must settle before timeout");
+        tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+    }
+}
+
 async fn wait_for_subscription_state<T>(subscription: &crate::WsSubscription<T>, expected: crate::WsSubscriptionState) {
     let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
     loop {
@@ -380,6 +421,157 @@ fn helius_transaction_filter_debug_omits_signature_and_account_values() {
     assert!(!debug.contains("Vote111111111111111111111111111111111111111"));
 }
 
+#[test]
+fn helius_transaction_notification_debug_omits_raw_provider_payloads() {
+    let full_value = serde_json::json!({
+        "transaction":{"raw":"MASSIVE-RAW-PAYLOAD-CANARY"},
+        "signature":"FULL-SIGNATURE-CANARY",
+        "slot":77,
+        "transactionIndex":3
+    });
+    let full = super::decode_helius_transaction_notification(full_value).expect("full Helius notification must decode");
+    let signature_value = serde_json::json!({
+        "signature":"SIGNATURE-MODE-CANARY",
+        "slot":78,
+        "transactionIndex":4,
+        "err":{"secret":"ERROR-DATA-CANARY"},
+        "memo":"MEMO-CANARY",
+        "blockTime":123,
+        "confirmationStatus":"CONFIRMATION-CANARY"
+    });
+    let signature = super::decode_helius_transaction_notification(signature_value).expect("signature Helius notification must decode");
+    let unknown = super::decode_helius_transaction_notification(serde_json::json!({"provider":"UNKNOWN-PAYLOAD-CANARY"}))
+        .expect("unknown Helius notification must remain forward-compatible");
+    let rendered = format!("{full:?} {signature:?} {unknown:?}");
+    assert!(rendered.contains("transaction: \"<omitted>\""));
+    assert!(rendered.contains("signature: \"<omitted>\""));
+    assert!(rendered.contains("err: \"value\""));
+    assert!(rendered.contains("memo: \"value\""));
+    assert!(rendered.contains("Unknown(\"<omitted>\")"));
+    for forbidden in [
+        "MASSIVE-RAW-PAYLOAD-CANARY",
+        "FULL-SIGNATURE-CANARY",
+        "SIGNATURE-MODE-CANARY",
+        "ERROR-DATA-CANARY",
+        "MEMO-CANARY",
+        "CONFIRMATION-CANARY",
+        "UNKNOWN-PAYLOAD-CANARY",
+    ] {
+        assert!(!rendered.contains(forbidden));
+    }
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn helius_provider_rpc_application_error_is_safe_and_does_not_fail_session() {
+    let (listener, url) = bind_local_listener().await;
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("local server must accept Helius client");
+        let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local Helius handshake must succeed");
+        let transaction_subscribe = read_request(&mut websocket).await;
+        assert_eq!(transaction_subscribe["method"], serde_json::json!("transactionSubscribe"));
+        send_error(
+            &mut websocket,
+            &transaction_subscribe,
+            -32602,
+            "PROVIDER-MESSAGE-SECRET-CANARY",
+            serde_json::json!({"apiKey":"PROVIDER-ERROR-SECRET-CANARY","payload":"X".repeat(4096)}),
+        )
+        .await;
+        let root_subscribe = read_request(&mut websocket).await;
+        assert_eq!(root_subscribe["method"], serde_json::json!("rootSubscribe"));
+        send_result(&mut websocket, &root_subscribe, serde_json::json!(72)).await;
+        send_root_notification(&mut websocket, 72, 88).await;
+        let root_unsubscribe = read_request(&mut websocket).await;
+        assert_eq!(root_unsubscribe["method"], serde_json::json!("rootUnsubscribe"));
+        assert_eq!(root_unsubscribe["params"], serde_json::json!([72]));
+        send_result(&mut websocket, &root_unsubscribe, serde_json::json!(true)).await;
+        wait_for_close_frame(&mut websocket).await;
+    });
+    let endpoint_url = format!("{url}/?api-key=HELIUS-ENDPOINT-SECRET-CANARY");
+    let session = crate::HeliusLaserStreamWsSession::connect(helius_endpoint(endpoint_url.as_str())).await.expect("Helius facade must connect");
+    let request = crate::HeliusTransactionSubscribeRequest::new(base_filter(), std::option::Option::None);
+    let error = session.transaction_subscribe(&request).await.expect_err("provider application error must reject only the logical subscribe request");
+    assert_eq!(error.code(), crate::ERROR_CODE_RPC_APPLICATION_ERROR);
+    assert!(error.context().iter().any(|entry| return entry.key() == "rpc_code" && entry.value() == "-32602"));
+    assert!(error.context().iter().any(|entry| return entry.key() == "method" && entry.value() == "transactionSubscribe"));
+    let rendered = format!("{error:?} {error} {session:?} {:?}", session.snapshot());
+    for forbidden in ["PROVIDER-MESSAGE-SECRET-CANARY", "PROVIDER-ERROR-SECRET-CANARY", "HELIUS-ENDPOINT-SECRET-CANARY", "fixture-signature-secret-canary"] {
+        assert!(!rendered.contains(forbidden));
+    }
+    assert_eq!(session.state(), crate::WsSessionState::Active);
+    wait_for_session_subscription_count(&session, 0).await;
+    let mut root = session.root_subscribe().await.expect("session must accept a healthy subscription after provider application error");
+    assert_eq!(root.recv().await.expect("healthy root notification must arrive").expect("healthy root notification must decode"), 88);
+    assert!(root.unsubscribe().await.expect("healthy root unsubscribe must complete"));
+    session.close().await.expect("Helius fixture session must close");
+    server.await.expect("provider error fixture server must finish");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn helius_notification_method_mismatch_fails_only_transaction_subscription() {
+    let (listener, url) = bind_local_listener().await;
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("local server must accept Helius client");
+        let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local Helius handshake must succeed");
+        let transaction_subscribe = read_request(&mut websocket).await;
+        send_result(&mut websocket, &transaction_subscribe, serde_json::json!(41)).await;
+        let root_subscribe = read_request(&mut websocket).await;
+        send_result(&mut websocket, &root_subscribe, serde_json::json!(42)).await;
+        send_root_notification(&mut websocket, 41, 5).await;
+        let cleanup = read_request(&mut websocket).await;
+        assert_eq!(cleanup["method"], serde_json::json!("transactionUnsubscribe"));
+        assert_eq!(cleanup["params"], serde_json::json!([41]));
+        send_result(&mut websocket, &cleanup, serde_json::json!(true)).await;
+        send_root_notification(&mut websocket, 42, 99).await;
+        wait_for_close_frame(&mut websocket).await;
+    });
+    let session = crate::HeliusLaserStreamWsSession::connect(helius_endpoint(url.as_str())).await.expect("Helius facade must connect");
+    let request = crate::HeliusTransactionSubscribeRequest::new(crate::HeliusTransactionSubscribeFilter::default(), std::option::Option::None);
+    let mut transaction = session.transaction_subscribe(&request).await.expect("transaction subscription must register");
+    let mut root = session.root_subscribe().await.expect("root subscription must register");
+    wait_for_subscription_state(&transaction, crate::WsSubscriptionState::Failed).await;
+    assert_eq!(transaction.terminal_error_code(), std::option::Option::Some(crate::ERROR_CODE_WS_PROTOCOL_ERROR));
+    assert!(transaction.recv().await.is_none());
+    assert_eq!(session.state(), crate::WsSessionState::Active);
+    wait_for_session_subscription_count(&session, 1).await;
+    assert_eq!(root.recv().await.expect("healthy root notification must arrive").expect("healthy root notification must decode"), 99);
+    assert_eq!(root.state(), crate::WsSubscriptionState::Active);
+    session.close().await.expect("Helius fixture session must close");
+    server.await.expect("notification mismatch fixture server must finish");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn helius_oversized_inbound_payload_reconnects_before_provider_json_decode() {
+    let (listener, url) = bind_local_listener().await;
+    let server = tokio::spawn(async move {
+        let (first_stream, _) = listener.accept().await.expect("initial Helius client must connect");
+        let mut first = tokio_tungstenite::accept_async(first_stream).await.expect("initial Helius handshake must succeed");
+        first
+            .send(tokio_tungstenite::tungstenite::Message::Text("PROVIDER-PAYLOAD-CANARY".repeat(32).into()))
+            .await
+            .expect("oversized provider fixture payload must send");
+        let (replacement_stream, _) = listener.accept().await.expect("replacement Helius client must connect");
+        let mut replacement = tokio_tungstenite::accept_async(replacement_stream).await.expect("replacement Helius handshake must succeed");
+        let root_subscribe = read_request(&mut replacement).await;
+        assert_eq!(root_subscribe["method"], serde_json::json!("rootSubscribe"));
+        send_result(&mut replacement, &root_subscribe, serde_json::json!(91)).await;
+        let root_unsubscribe = read_request(&mut replacement).await;
+        assert_eq!(root_unsubscribe["method"], serde_json::json!("rootUnsubscribe"));
+        send_result(&mut replacement, &root_unsubscribe, serde_json::json!(true)).await;
+        wait_for_close_frame(&mut replacement).await;
+    });
+    let session = crate::HeliusLaserStreamWsSession::connect(helius_endpoint_with_session(url.as_str(), adversarial_payload_session_settings()))
+        .await
+        .expect("Helius facade must connect before adversarial payload");
+    wait_for_gap_count(&session, 1).await;
+    assert_eq!(session.state(), crate::WsSessionState::Active);
+    assert_eq!(session.snapshot().continuity_gap_count(), 1);
+    let mut root = session.root_subscribe().await.expect("recovered Helius session must remain usable");
+    assert!(root.unsubscribe().await.expect("recovered root subscription must unsubscribe"));
+    session.close().await.expect("recovered Helius session must close");
+    server.await.expect("oversized provider payload fixture server must finish");
+}
+
 #[tokio::test(flavor = "current_thread")]
 async fn helius_transaction_live_handle_decodes_notification_and_unsubscribes_through_shared_actor() {
     let (listener, url) = bind_local_listener().await;