v0.2.7-pre.011

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

@@ -0,0 +1,222 @@
+// file: crates/ksp-onchain-transport-lib/unit_tests/ws_blocks.rs
+// version: 1
+
+use futures_util::SinkExt; // rust-rules: trait-import
+use futures_util::StreamExt; // rust-rules: trait-import
+
+fn fixture_pubkey() -> ksp_core_lib::Pubkey {
+    return "11111111111111111111111111111111".parse::<ksp_core_lib::Pubkey>().expect("fixture pubkey must parse");
+}
+
+fn local_endpoint(url: &str) -> crate::WsEndpointSettings {
+    return crate::WsEndpointSettings::new(
+        "local_ws_blocks",
+        true,
+        crate::WsProviderName::new("local-fixture"),
+        crate::WsClusterName::new("local"),
+        crate::WsProtocolKind::SolanaStandard,
+        crate::WsEndpointUrl::parse(url).expect("local test WebSocket URL must parse"),
+        crate::WsSessionSettings::default(),
+    );
+}
+
+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");
+    return (listener, format!("ws://{address}"));
+}
+
+async fn read_request(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>) -> serde_json::Value {
+    let message = websocket.next().await.expect("request message must exist").expect("request message must decode");
+    let text = message.to_text().expect("request must be text");
+    return serde_json::from_str(text).expect("request must contain JSON");
+}
+
+async fn send_result(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>, request: &serde_json::Value, result: 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,"result":result});
+    websocket.send(tokio_tungstenite::tungstenite::Message::Text(response.to_string().into())).await.expect("local response must send");
+}
+
+async fn send_error(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>, request: &serde_json::Value, code: i64, message: &str) {
+    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}});
+    websocket.send(tokio_tungstenite::tungstenite::Message::Text(response.to_string().into())).await.expect("local error response must send");
+}
+
+async fn send_notification(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>, remote_id: u64, result: serde_json::Value) {
+    let notification = serde_json::json!({"jsonrpc":"2.0","method":"blockNotification","params":{"result":result,"subscription":remote_id}});
+    websocket.send(tokio_tungstenite::tungstenite::Message::Text(notification.to_string().into())).await.expect("local notification must send");
+}
+
+async fn wait_for_close_frame(websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>) {
+    loop {
+        let message = websocket.next().await;
+        match message {
+            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,
+        }
+    }
+}
+
+#[test]
+fn block_subscribe_config_preserves_all_unstable_options_and_rejects_processed_commitment() {
+    let config = crate::SolanaBlockSubscribeConfig::new(
+        std::option::Option::Some(crate::SolanaCommitment::Confirmed),
+        std::option::Option::Some(crate::SolanaTransactionEncoding::JsonParsed),
+        std::option::Option::Some(crate::SolanaTransactionDetails::Accounts),
+        std::option::Option::Some(7),
+        std::option::Option::Some(true),
+    );
+    assert_eq!(config.commitment(), std::option::Option::Some(crate::SolanaCommitment::Confirmed));
+    assert_eq!(config.encoding(), std::option::Option::Some(crate::SolanaTransactionEncoding::JsonParsed));
+    assert_eq!(config.transaction_details(), std::option::Option::Some(crate::SolanaTransactionDetails::Accounts));
+    assert_eq!(config.max_supported_transaction_version(), std::option::Option::Some(7));
+    assert_eq!(config.show_rewards(), std::option::Option::Some(true));
+    assert_eq!(
+        config.to_json_value(),
+        serde_json::json!({
+            "commitment": "confirmed",
+            "encoding": "jsonParsed",
+            "transactionDetails": "accounts",
+            "maxSupportedTransactionVersion": 7,
+            "showRewards": true
+        })
+    );
+    let invalid = crate::SolanaBlockSubscribeConfig::new(
+        std::option::Option::Some(crate::SolanaCommitment::Processed),
+        std::option::Option::None,
+        std::option::Option::None,
+        std::option::Option::None,
+        std::option::Option::None,
+    );
+    assert_eq!(invalid.validate().expect_err("processed commitment must be rejected").code(), crate::ERROR_CODE_INVALID_RPC_PARAMETERS);
+}
+
+#[test]
+fn block_notification_decoder_preserves_nulls_and_shared_confirmed_block_shape() {
+    let nulls = super::decode_block_notification(
+        "blockSubscribe",
+        serde_json::json!({"context":{"slot":51},"value":{"slot":51,"block":null,"err":{"reason":"missing"}}}),
+    )
+    .expect("nullable block notification must decode");
+    assert_eq!(nulls.context().slot(), 51);
+    assert_eq!(nulls.value().slot(), 51);
+    assert!(nulls.value().block().is_none());
+    assert_eq!(nulls.value().err(), std::option::Option::Some(&serde_json::json!({"reason":"missing"})));
+    let block = super::decode_block_notification(
+        "blockSubscribe",
+        serde_json::json!({
+            "context":{"slot":52},
+            "value":{
+                "slot":52,
+                "block":{
+                    "previousBlockhash":"prev",
+                    "blockhash":"current",
+                    "parentSlot":51,
+                    "signatures":["sig-a"],
+                    "rewards":null,
+                    "numRewardPartitions":4,
+                    "blockTime":123,
+                    "blockHeight":9
+                },
+                "err":null
+            }
+        }),
+    )
+    .expect("shared confirmed block shape must decode");
+    assert_eq!(block.value().block().expect("block must exist").blockhash(), "current");
+    assert!(block.value().err().is_none());
+    let large_transaction = "A".repeat(2_048);
+    let large_wire = serde_json::json!({
+        "context":{"slot":53},
+        "value":{
+            "slot":53,
+            "block":{
+                "previousBlockhash":"prev",
+                "blockhash":"large-current",
+                "parentSlot":52,
+                "transactions":[{"transaction":[large_transaction,"base64"],"meta":{"err":null,"fee":5000},"version":"legacy"}],
+                "rewards":[],
+                "blockTime":123,
+                "blockHeight":10
+            },
+            "err":null
+        }
+    });
+    assert!(large_wire.to_string().len() > 1_232);
+    assert!(super::decode_block_notification("blockSubscribe", large_wire).is_ok());
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn unstable_block_wrapper_uses_exact_request_notification_and_handle_unsubscribe() {
+    let (listener, url) = bind_local_listener().await;
+    let pubkey = fixture_pubkey();
+    let server_pubkey = pubkey;
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("local server must accept client");
+        let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
+        let subscribe = read_request(&mut websocket).await;
+        assert_eq!(subscribe["method"], serde_json::json!("blockSubscribe"));
+        assert_eq!(
+            subscribe["params"],
+            serde_json::json!([
+                {"mentionsAccountOrProgram":server_pubkey.to_string()},
+                {"commitment":"confirmed","encoding":"base64","transactionDetails":"signatures","maxSupportedTransactionVersion":3,"showRewards":false}
+            ])
+        );
+        send_result(&mut websocket, &subscribe, serde_json::json!(301)).await;
+        send_notification(&mut websocket, 301, serde_json::json!({"context":{"slot":77},"value":{"slot":77,"block":null,"err":null}})).await;
+        let unsubscribe = read_request(&mut websocket).await;
+        assert_eq!(unsubscribe["method"], serde_json::json!("blockUnsubscribe"));
+        assert_eq!(unsubscribe["params"], serde_json::json!([301]));
+        send_result(&mut websocket, &unsubscribe, serde_json::json!(true)).await;
+        wait_for_close_frame(&mut websocket).await;
+    });
+    let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
+    let config = crate::SolanaBlockSubscribeConfig::new(
+        std::option::Option::Some(crate::SolanaCommitment::Confirmed),
+        std::option::Option::Some(crate::SolanaTransactionEncoding::Base64),
+        std::option::Option::Some(crate::SolanaTransactionDetails::Signatures),
+        std::option::Option::Some(3),
+        std::option::Option::Some(false),
+    );
+    let mut subscription = session
+        .block_subscribe(&crate::SolanaBlockSubscribeFilter::MentionsAccountOrProgram(pubkey), std::option::Option::Some(&config))
+        .await
+        .expect("blockSubscribe must register");
+    let notification = subscription.recv().await.expect("block notification must arrive").expect("block notification must decode");
+    assert_eq!(notification.value().slot(), 77);
+    assert!(notification.value().block().is_none());
+    assert!(subscription.unsubscribe().await.expect("block unsubscribe must complete"));
+    session.close().await.expect("session close must complete");
+    server.await.expect("local server task must complete");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn unstable_block_validator_capability_rpc_error_does_not_fail_physical_session() {
+    let (listener, url) = bind_local_listener().await;
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("local server must accept client");
+        let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
+        let block_subscribe = read_request(&mut websocket).await;
+        assert_eq!(block_subscribe["method"], serde_json::json!("blockSubscribe"));
+        send_error(&mut websocket, &block_subscribe, -32601, "block subscription disabled").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!(302)).await;
+        wait_for_close_frame(&mut websocket).await;
+    });
+    let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
+    let error = session
+        .block_subscribe(&crate::SolanaBlockSubscribeFilter::All, std::option::Option::None)
+        .await
+        .expect_err("validator capability application error must surface to the caller");
+    assert_eq!(error.code(), crate::ERROR_CODE_RPC_APPLICATION_ERROR);
+    assert_eq!(session.state(), crate::WsSessionState::Active);
+    let root = session.root_subscribe().await.expect("session must remain usable after block application error");
+    assert_eq!(root.state(), crate::WsSubscriptionState::Active);
+    session.close().await.expect("session close must complete");
+    server.await.expect("local server task must complete");
+}

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/unit_tests/ws_cluster.rs
-// version: 1
+// version: 2
 
 use futures_util::SinkExt; // rust-rules: trait-import
 use futures_util::StreamExt; // rust-rules: trait-import
@@ -98,3 +98,136 @@ async fn stable_slot_and_root_wrappers_use_no_params_decode_exact_notifications_
     session.close().await.expect("session close must complete");
     server.await.expect("local server task must complete");
 }
+
+#[test]
+fn slots_update_decoder_preserves_all_known_variants_and_unknown_bounded_fallback() {
+    let cases = [
+        (serde_json::json!({"slot":1,"timestamp":10,"type":"firstShredReceived"}), "firstShredReceived"),
+        (serde_json::json!({"slot":2,"timestamp":20,"type":"completed"}), "completed"),
+        (serde_json::json!({"slot":3,"timestamp":30,"type":"createdBank","parent":2}), "createdBank"),
+        (
+            serde_json::json!({
+                "slot":4,
+                "timestamp":40,
+                "type":"frozen",
+                "stats":{"maxTransactionsPerEntry":64,"numFailedTransactions":1,"numSuccessfulTransactions":9,"numTransactionEntries":3}
+            }),
+            "frozen",
+        ),
+        (serde_json::json!({"slot":5,"timestamp":50,"type":"dead","err":"fixture dead"}), "dead"),
+        (serde_json::json!({"slot":6,"timestamp":60,"type":"optimisticConfirmation"}), "optimisticConfirmation"),
+        (serde_json::json!({"slot":7,"timestamp":70,"type":"root"}), "root"),
+    ];
+    for (wire, expected_type) in cases {
+        let update = super::decode_slots_update_notification("slotsUpdatesSubscribe", wire).expect("known slot update must decode");
+        assert_eq!(update.update_type(), expected_type);
+        assert!(update.slot().is_some());
+        assert!(update.timestamp().is_some());
+        assert!(update.unknown_raw().is_none());
+    }
+    let frozen = super::decode_slots_update_notification(
+        "slotsUpdatesSubscribe",
+        serde_json::json!({
+            "slot":4,
+            "timestamp":40,
+            "type":"frozen",
+            "stats":{"maxTransactionsPerEntry":64,"numFailedTransactions":1,"numSuccessfulTransactions":9,"numTransactionEntries":3}
+        }),
+    )
+    .expect("frozen update must decode");
+    match frozen {
+        crate::SolanaSlotUpdate::Frozen { stats, .. } => {
+            assert_eq!(stats.max_transactions_per_entry(), 64);
+            assert_eq!(stats.num_failed_transactions(), 1);
+            assert_eq!(stats.num_successful_transactions(), 9);
+            assert_eq!(stats.num_transaction_entries(), 3);
+        },
+        _ => panic!("fixture must decode as frozen"),
+    }
+    let unknown_wire = serde_json::json!({"slot":8,"timestamp":80,"type":"futureBankState","futureField":{"x":1}});
+    let unknown = super::decode_slots_update_notification("slotsUpdatesSubscribe", unknown_wire.clone()).expect("unknown update must remain consumable");
+    assert_eq!(unknown.update_type(), "futureBankState");
+    assert_eq!(unknown.slot(), std::option::Option::Some(8));
+    assert_eq!(unknown.timestamp(), std::option::Option::Some(80));
+    assert_eq!(unknown.unknown_raw(), std::option::Option::Some(&unknown_wire));
+}
+
+#[test]
+fn slots_update_known_variants_require_their_variant_specific_fields() {
+    assert!(super::decode_slots_update_notification("slotsUpdatesSubscribe", serde_json::json!({"slot":3,"timestamp":30,"type":"createdBank"})).is_err());
+    assert!(super::decode_slots_update_notification("slotsUpdatesSubscribe", serde_json::json!({"slot":4,"timestamp":40,"type":"frozen"})).is_err());
+    assert!(super::decode_slots_update_notification("slotsUpdatesSubscribe", serde_json::json!({"slot":5,"timestamp":50,"type":"dead"})).is_err());
+}
+
+#[test]
+fn vote_notification_decoder_preserves_timestamp_omitted_null_and_value() {
+    let pubkey = "11111111111111111111111111111111";
+    for (wire, expected_timestamp) in [
+        (serde_json::json!({"votePubkey":pubkey,"slots":[1,2],"hash":"hash-a","signature":"sig-a"}), std::option::Option::None),
+        (serde_json::json!({"votePubkey":pubkey,"slots":[1,2],"hash":"hash-b","timestamp":null,"signature":"sig-b"}), std::option::Option::None),
+        (serde_json::json!({"votePubkey":pubkey,"slots":[1,2],"hash":"hash-c","timestamp":123,"signature":"sig-c"}), std::option::Option::Some(123)),
+    ] {
+        let vote = super::decode_vote_notification("voteSubscribe", wire).expect("vote notification must decode");
+        assert_eq!(vote.vote_pubkey().to_string(), pubkey);
+        assert_eq!(vote.slots(), &[1, 2]);
+        assert_eq!(vote.timestamp(), expected_timestamp);
+    }
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn unstable_slots_updates_and_vote_wrappers_use_no_params_and_handle_unsubscribe() {
+    let (listener, url) = bind_local_listener().await;
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("local server must accept client");
+        let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
+        let slots_subscribe = read_request(&mut websocket).await;
+        assert_eq!(slots_subscribe["method"], serde_json::json!("slotsUpdatesSubscribe"));
+        assert_eq!(slots_subscribe["params"], serde_json::json!([]));
+        send_result(&mut websocket, &slots_subscribe, serde_json::json!(401)).await;
+        send_notification(
+            &mut websocket,
+            "slotsUpdatesNotification",
+            401,
+            serde_json::json!({"slot":76,"timestamp":1625081266243_i64,"type":"optimisticConfirmation"}),
+        )
+        .await;
+        let vote_subscribe = read_request(&mut websocket).await;
+        assert_eq!(vote_subscribe["method"], serde_json::json!("voteSubscribe"));
+        assert_eq!(vote_subscribe["params"], serde_json::json!([]));
+        send_result(&mut websocket, &vote_subscribe, serde_json::json!(402)).await;
+        send_notification(
+            &mut websocket,
+            "voteNotification",
+            402,
+            serde_json::json!({
+                "votePubkey":"11111111111111111111111111111111",
+                "slots":[75,76],
+                "hash":"fixture-hash",
+                "timestamp":null,
+                "signature":"fixture-signature"
+            }),
+        )
+        .await;
+        let slots_unsubscribe = read_request(&mut websocket).await;
+        assert_eq!(slots_unsubscribe["method"], serde_json::json!("slotsUpdatesUnsubscribe"));
+        assert_eq!(slots_unsubscribe["params"], serde_json::json!([401]));
+        send_result(&mut websocket, &slots_unsubscribe, serde_json::json!(true)).await;
+        let vote_unsubscribe = read_request(&mut websocket).await;
+        assert_eq!(vote_unsubscribe["method"], serde_json::json!("voteUnsubscribe"));
+        assert_eq!(vote_unsubscribe["params"], serde_json::json!([402]));
+        send_result(&mut websocket, &vote_unsubscribe, serde_json::json!(true)).await;
+        wait_for_close_frame(&mut websocket).await;
+    });
+    let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
+    let mut slots_subscription = session.slots_updates_subscribe().await.expect("slotsUpdatesSubscribe must register");
+    let update = slots_subscription.recv().await.expect("slots update must arrive").expect("slots update must decode");
+    assert_eq!(update.update_type(), "optimisticConfirmation");
+    let mut vote_subscription = session.vote_subscribe().await.expect("voteSubscribe must register");
+    let vote = vote_subscription.recv().await.expect("vote notification must arrive").expect("vote notification must decode");
+    assert_eq!(vote.slots(), &[75, 76]);
+    assert!(vote.timestamp().is_none());
+    assert!(slots_subscription.unsubscribe().await.expect("slots update unsubscribe must complete"));
+    assert!(vote_subscription.unsubscribe().await.expect("vote unsubscribe must complete"));
+    session.close().await.expect("session close must complete");
+    server.await.expect("local server task must complete");
+}

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/unit_tests/ws_lifecycle.rs
-// version: 3
+// version: 4
 
 fn non_zero(value: u64) -> std::num::NonZeroU64 {
     return std::num::NonZeroU64::new(value).expect("test ID must be non-zero");
@@ -105,3 +105,21 @@ fn websocket_subscription_kinds_map_exact_standard_method_triplets() {
         assert_eq!(kind.notification_method(), notification);
     }
 }
+
+#[test]
+fn websocket_unstable_subscription_partition_is_exact() {
+    let cases = [
+        (crate::WsSubscriptionKind::Account, false),
+        (crate::WsSubscriptionKind::Block, true),
+        (crate::WsSubscriptionKind::Logs, false),
+        (crate::WsSubscriptionKind::Program, false),
+        (crate::WsSubscriptionKind::Root, false),
+        (crate::WsSubscriptionKind::Signature, false),
+        (crate::WsSubscriptionKind::Slot, false),
+        (crate::WsSubscriptionKind::SlotsUpdates, true),
+        (crate::WsSubscriptionKind::Vote, true),
+    ];
+    for (kind, unstable) in cases {
+        assert_eq!(kind.is_unstable(), unstable);
+    }
+}