v0.3.10-pre.004

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/unit_tests/rpc_blocks.rs
-// version: 6
+// version: 7
 
 #[test]
 fn transaction_details_and_get_block_config_preserve_all_modern_options() {
@@ -188,6 +188,35 @@ fn serve_once(body: &'static str) -> (std::string::String, std::thread::JoinHand
     return (format!("http://{address}"), handle);
 }
 
+fn serve_status_and_count(status_line: &'static str) -> (std::string::String, std::thread::JoinHandle<(usize, std::string::String)>) {
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("fixture listener must bind");
+    let address = listener.local_addr().expect("fixture listener address must resolve");
+    let handle = std::thread::spawn(move || {
+        let (mut stream, _) = listener.accept().expect("fixture server must accept first request");
+        let first_request = read_request(&mut stream);
+        let response = format!("HTTP/1.1 {status_line}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
+        std::io::Write::write_all(&mut stream, response.as_bytes()).expect("fixture response must write");
+        let mut count = 1_usize;
+        listener.set_nonblocking(true).expect("fixture listener must become nonblocking");
+        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(120);
+        while std::time::Instant::now() < deadline {
+            match listener.accept() {
+                std::result::Result::Ok((mut retry_stream, _)) => {
+                    let _ = read_request(&mut retry_stream);
+                    std::io::Write::write_all(&mut retry_stream, response.as_bytes()).expect("fixture retry response must write");
+                    count = count.saturating_add(1);
+                },
+                std::result::Result::Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                    std::thread::sleep(std::time::Duration::from_millis(5));
+                },
+                std::result::Result::Err(error) => panic!("fixture listener failed while counting retries: {error}"),
+            }
+        }
+        return (count, first_request);
+    });
+    return (format!("http://{address}"), handle);
+}
+
 fn read_request(stream: &mut std::net::TcpStream) -> std::string::String {
     let mut bytes = std::vec::Vec::new();
     let mut buffer = [0_u8; 1024];
@@ -784,3 +813,92 @@ async fn typed_get_block_rejects_processed_commitment_before_io() {
     assert_eq!(error.context()[1].key(), "commitment");
     assert_eq!(error.context()[1].value(), "processed");
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn pre_004_get_block_observed_reports_actual_winner_after_retry_reroute() {
+    let (first_url, first_handle) = serve_status_and_count("429 Too Many Requests");
+    let (winner_url, winner_handle) = serve_once(include_str!("../fixtures/http/get_block.observed_material.success.json"));
+    let urls = [(first_url.as_str(), "first-endpoint", "first-provider"), (winner_url.as_str(), "winner-endpoint", "winner-provider")];
+    let mut endpoints = std::vec::Vec::with_capacity(urls.len());
+    for (url, endpoint_name, provider) in urls {
+        let role = crate::HttpEndpointRoleSettings::new(
+            crate::HttpRoleName::new("default"),
+            true,
+            std::vec![crate::HttpRequestKind::wildcard()],
+            10,
+            crate::HttpRoleLimits::new(std::option::Option::None, std::option::Option::None, std::option::Option::None, std::option::Option::None),
+        );
+        endpoints.push(crate::HttpEndpointSettings::new(
+            endpoint_name,
+            true,
+            crate::HttpProviderName::new(provider),
+            crate::HttpClusterName::new("local"),
+            crate::HttpEndpointUrl::parse(url).expect("fixture URL must parse"),
+            std::time::Duration::from_millis(100),
+            std::time::Duration::from_secs(1),
+            std::option::Option::Some(1),
+            std::vec![role],
+        ));
+    }
+    let pool = crate::HttpTransportPool::new(crate::HttpTransportSettings::new(
+        endpoints,
+        crate::HttpRetrySettings::new(1, std::time::Duration::from_millis(1), std::time::Duration::from_millis(2)),
+    ))
+    .expect("observed fixture pool must build");
+    let config = crate::SolanaGetBlockConfig::new(
+        std::option::Option::Some(crate::SolanaCommitment::Confirmed),
+        std::option::Option::Some(crate::SolanaTransactionEncoding::Base64),
+        std::option::Option::Some(crate::SolanaTransactionDetails::Full),
+        std::option::Option::Some(0),
+        std::option::Option::Some(false),
+    );
+    let observed = pool
+        .get_block_observed(&crate::HttpRoleName::new("default"), 430_000_123, std::option::Option::Some(&config))
+        .await
+        .expect("retry-safe observed getBlock must succeed on the second endpoint");
+    assert_eq!(observed.endpoint_name(), "winner-endpoint");
+    assert_eq!(observed.provider().as_str(), "winner-provider");
+    let block = observed.value().as_ref().expect("winning response must contain a block");
+    let transactions = block.transactions().value().expect("winning block must contain transactions");
+    assert_eq!(transactions.len(), 1);
+    assert!(matches!(
+        transactions[0].transaction(),
+        crate::SolanaEncodedTransaction::Binary { encoding: crate::SolanaTransactionBinaryEncoding::Base64, .. }
+    ));
+    let rendered = format!("{observed:?}");
+    assert!(rendered.contains("winner-endpoint"));
+    assert!(rendered.contains("winner-provider"));
+    assert!(rendered.contains("<available>"));
+    assert!(!rendered.contains("AQAAAAAAAA"));
+    let (first_count, first_request) = first_handle.join().expect("first fixture server must join");
+    assert_eq!(first_count, 1);
+    assert_eq!(request_body(first_request.as_str())["method"], serde_json::json!("getBlock"));
+    let winner_request = winner_handle.join().expect("winner fixture server must join");
+    assert_eq!(request_body(winner_request.as_str())["method"], serde_json::json!("getBlock"));
+    return;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn pre_004_get_block_observed_preserves_null_and_reuses_get_block_validation() {
+    let (url, handle) = serve_once(include_str!("../fixtures/http/get_block.null.json"));
+    let pool = pool_for_url(url.as_str());
+    let observed = pool
+        .get_block_observed(&crate::HttpRoleName::new("default"), 430_000_123, std::option::Option::None)
+        .await
+        .expect("observed null getBlock must succeed");
+    assert!(observed.value().is_none());
+    assert_eq!(observed.endpoint_name(), "fixture");
+    assert_eq!(observed.provider().as_str(), "fixture");
+    handle.join().expect("fixture server must join");
+    let processed = crate::SolanaGetBlockConfig::new(
+        std::option::Option::Some(crate::SolanaCommitment::Processed),
+        std::option::Option::Some(crate::SolanaTransactionEncoding::Base64),
+        std::option::Option::Some(crate::SolanaTransactionDetails::Full),
+        std::option::Option::Some(0),
+        std::option::Option::Some(false),
+    );
+    let result = pool.get_block_observed(&crate::HttpRoleName::new("default"), 1, std::option::Option::Some(&processed)).await;
+    let error = result.expect_err("processed observed getBlock commitment must reject before I/O");
+    assert_eq!(error.code(), crate::ERROR_CODE_INVALID_RPC_PARAMETERS);
+    return;
+}