v0.3.13-pre.003

crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs
-// version: 12
+// version: 13
 
 fn grpc_endpoint(cluster: &str) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings> {
     return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider");
@@ -62,6 +62,69 @@ fn http_pool(cluster: &str, role_name: &str, request_kind: &str) -> std::option:
     };
 }
 
+fn ws_endpoint(
+    cluster: &str,
+    endpoint_name: &str,
+    provider: &str,
+    protocol: ksp_onchain_transport_lib::WsProtocolKind,
+) -> std::option::Option<ksp_onchain_transport_lib::WsEndpointSettings> {
+    let url = match ksp_onchain_transport_lib::WsEndpointUrl::parse("ws://127.0.0.1:10001/WS-SECRET-CANARY") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::option::Option::None,
+    };
+    return std::option::Option::Some(ksp_onchain_transport_lib::WsEndpointSettings::new(
+        endpoint_name,
+        true,
+        ksp_onchain_transport_lib::WsProviderName::new(provider),
+        ksp_onchain_transport_lib::WsClusterName::new(cluster),
+        protocol,
+        url,
+        ksp_onchain_transport_lib::WsSessionSettings::default(),
+    ));
+}
+
+fn standard_logs_source(
+    cluster: &str,
+    endpoint_name: &str,
+    filter: ksp_onchain_transport_lib::SolanaLogsSubscribeFilter,
+    commitment: ksp_onchain_transport_lib::SolanaCommitment,
+    hydration_role: &str,
+) -> std::option::Option<crate::RawTransactionIngestStandardLogsSource> {
+    let endpoint = match ws_endpoint(cluster, endpoint_name, "fixture-ws-provider", ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::option::Option::None,
+    };
+    let pool = match http_pool(cluster, hydration_role, "get_transaction") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::option::Option::None,
+    };
+    return match crate::RawTransactionIngestStandardLogsSource::new(
+        endpoint,
+        filter,
+        commitment,
+        pool,
+        ksp_onchain_transport_lib::HttpRoleName::new(hydration_role),
+    ) {
+        std::result::Result::Ok(value) => std::option::Option::Some(value),
+        std::result::Result::Err(_) => std::option::Option::None,
+    };
+}
+
+struct StandardLogsViewFixture {
+    signature: std::string::String,
+    slot: u64,
+}
+
+impl super::RawTransactionIngestStandardLogsView for StandardLogsViewFixture {
+    fn signature(&self) -> &str {
+        return self.signature.as_str();
+    }
+
+    fn slot(&self) -> u64 {
+        return self.slot;
+    }
+}
+
 fn transaction_request(
     commitment: std::option::Option<ksp_onchain_transport_lib::SolanaCommitment>,
 ) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneSubscribeRequest> {
@@ -331,6 +394,30 @@ fn signal_source_for_http_url_with_commitment(
     };
 }
 
+fn standard_logs_source_for_http_url(
+    url: &str,
+    filter: ksp_onchain_transport_lib::SolanaLogsSubscribeFilter,
+) -> std::option::Option<crate::RawTransactionIngestStandardLogsSource> {
+    let endpoint = match ws_endpoint("devnet", "standard-logs-fixture", "fixture-ws-provider", ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::option::Option::None,
+    };
+    let pool = match http_pool_for_url(url, "devnet", "http-hydration-fixture", "fixture-http-provider") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return std::option::Option::None,
+    };
+    return match crate::RawTransactionIngestStandardLogsSource::new(
+        endpoint,
+        filter,
+        ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        pool,
+        ksp_onchain_transport_lib::HttpRoleName::new("hydration"),
+    ) {
+        std::result::Result::Ok(value) => std::option::Option::Some(value),
+        std::result::Result::Err(_) => std::option::Option::None,
+    };
+}
+
 fn pre_004_settings() -> std::option::Option<crate::RawTransactionIngestSettings> {
     let network = match ksp_store_lib::RawNetworkId::new("devnet") {
         std::result::Result::Ok(value) => value,
@@ -1049,7 +1136,7 @@ fn pre_004_composite_provenance_rejects_overflow_without_truncation() {
         std::result::Result::Err(_) => return,
     };
     let route = super::RawTransactionIngestSourceRoute { endpoint_id, provider };
-    let result = super::composite_provenance_codes(&route, "h".repeat(120).as_str(), "http-endpoint");
+    let result = super::composite_provenance_codes(&route, "ys", "h".repeat(120).as_str(), "http-endpoint");
     let error = match result {
         std::result::Result::Ok(_) => return,
         std::result::Result::Err(error) => error,
@@ -1100,7 +1187,7 @@ async fn pre_004_observed_get_transaction_closes_signal_to_common_raw_ingress_wi
     assert!(request_body.contains(PRE_004_ZERO_SIGNATURE_TEXT));
     assert!(request_body.contains("\"commitment\":\"confirmed\""));
     assert!(request_body.contains("\"encoding\":\"base64\""));
-    assert!(request_body.contains("\"maxSupportedTransactionVersion\":0"));
+    assert!(request_body.contains("\"maxSupportedTransactionVersion\":1"));
     let ingress = match outcome {
         super::RawTransactionIngestHydrationOutcome::Available(value) => value,
         super::RawTransactionIngestHydrationOutcome::Missing(_) => return,
@@ -1172,7 +1259,7 @@ async fn pre_004_finalized_hydration_preserves_request_and_provenance_commitment
     let request_body = http_request_body(request.as_str());
     assert!(request_body.contains("\"commitment\":\"finalized\""));
     assert!(request_body.contains("\"encoding\":\"base64\""));
-    assert!(request_body.contains("\"maxSupportedTransactionVersion\":0"));
+    assert!(request_body.contains("\"maxSupportedTransactionVersion\":1"));
     let ingress = match outcome {
         super::RawTransactionIngestHydrationOutcome::Available(value) => value,
         super::RawTransactionIngestHydrationOutcome::Missing(_) => return,
@@ -1421,11 +1508,12 @@ async fn pre_006_coalescence_key_merges_transaction_and_status_before_http_fanou
         std::option::Option::Some(value) => value,
         std::option::Option::None => return,
     };
-    let transaction_key = match super::hydration_key(&source, &transaction) {
+    let hydration = source.hydration_context();
+    let transaction_key = match super::hydration_key(&hydration, &transaction) {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => return,
     };
-    let status_key = match super::hydration_key(&source, &status) {
+    let status_key = match super::hydration_key(&hydration, &status) {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => return,
     };
@@ -1437,10 +1525,10 @@ async fn pre_006_coalescence_key_merges_transaction_and_status_before_http_fanou
     let mut coordinator = super::RawTransactionIngestHydrationCoordinator::new(&settings);
     let (frontier_sender, _frontier_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());
     let mut processing_frontier = super::RawTransactionIngestProcessingFrontierReporter::new(frontier_sender);
-    if coordinator.queue_signal(&source, transaction, received_at, &mut processing_frontier).is_err() {
+    if coordinator.queue_signal(&hydration, transaction, received_at, &mut processing_frontier).is_err() {
         return;
     }
-    if coordinator.queue_signal(&source, status, received_at, &mut processing_frontier).is_err() {
+    if coordinator.queue_signal(&hydration, status, received_at, &mut processing_frontier).is_err() {
         return;
     }
     assert_eq!(coordinator.pending.len(), 1);
@@ -1467,11 +1555,12 @@ fn pre_006_coalescence_key_separates_network_signature_and_commitment() {
         std::option::Option::Some(value) => value,
         std::option::Option::None => return,
     };
-    let first_key = match super::hydration_key(&source, &first) {
+    let hydration = source.hydration_context();
+    let first_key = match super::hydration_key(&hydration, &first) {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => return,
     };
-    let second_key = match super::hydration_key(&source, &second) {
+    let second_key = match super::hydration_key(&hydration, &second) {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => return,
     };
@@ -1481,6 +1570,273 @@ fn pre_006_coalescence_key_separates_network_signature_and_commitment() {
     return;
 }
 
+#[test]
+fn v0_3_13_pre_003_standard_logs_source_validates_protocol_commitment_and_hydration_route() {
+    let accepted = standard_logs_source(
+        "devnet",
+        "standard-logs-a",
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All,
+        ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        "hydration",
+    );
+    let accepted = match accepted {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let resources = crate::RawTransactionIngestRuntimeResources::from_standard_logs_source(accepted);
+    assert_eq!(resources.source_count(), 1);
+    let helius_endpoint =
+        match ws_endpoint("devnet", "standard-logs-helius", "fixture-ws-provider", ksp_onchain_transport_lib::WsProtocolKind::HeliusLaserStream) {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => return,
+        };
+    let pool = match http_pool("devnet", "hydration", "get_transaction") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let protocol_error = crate::RawTransactionIngestStandardLogsSource::new(
+        helius_endpoint,
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All,
+        ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        pool.clone(),
+        ksp_onchain_transport_lib::HttpRoleName::new("hydration"),
+    );
+    let protocol_error = match protocol_error {
+        std::result::Result::Ok(_) => return,
+        std::result::Result::Err(value) => value,
+    };
+    assert!(protocol_error.context().iter().any(|context| return context.value() == "runtime_resources.standard_logs_protocol_invalid"));
+    let standard_endpoint =
+        match ws_endpoint("devnet", "standard-logs-processed", "fixture-ws-provider", ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard) {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => return,
+        };
+    let commitment_error = crate::RawTransactionIngestStandardLogsSource::new(
+        standard_endpoint,
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes,
+        ksp_onchain_transport_lib::SolanaCommitment::Processed,
+        pool,
+        ksp_onchain_transport_lib::HttpRoleName::new("hydration"),
+    );
+    let commitment_error = match commitment_error {
+        std::result::Result::Ok(_) => return,
+        std::result::Result::Err(value) => value,
+    };
+    assert!(commitment_error.context().iter().any(|context| return context.value() == "runtime_resources.hydration_commitment_invalid"));
+    let cross_network_endpoint =
+        match ws_endpoint("devnet", "standard-logs-cross-network", "fixture-ws-provider", ksp_onchain_transport_lib::WsProtocolKind::SolanaStandard) {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => return,
+        };
+    let cross_network_pool = match http_pool("testnet", "hydration", "get_transaction") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let cross_network = crate::RawTransactionIngestStandardLogsSource::new(
+        cross_network_endpoint,
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All,
+        ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        cross_network_pool,
+        ksp_onchain_transport_lib::HttpRoleName::new("hydration"),
+    );
+    let cross_network = match cross_network {
+        std::result::Result::Ok(_) => return,
+        std::result::Result::Err(value) => value,
+    };
+    assert!(cross_network.context().iter().any(|context| return context.value() == "runtime_resources.transport_network_mismatch"));
+    return;
+}
+
+#[test]
+fn v0_3_13_pre_003_standard_logs_source_key_is_filter_sensitive_hydration_role_neutral_and_redacted() {
+    let all_a = match standard_logs_source(
+        "devnet",
+        "standard-logs-key",
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All,
+        ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        "hydration",
+    ) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let all_b = match standard_logs_source(
+        "devnet",
+        "standard-logs-key",
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All,
+        ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        "secondary_hydration",
+    ) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let votes = match standard_logs_source(
+        "devnet",
+        "standard-logs-key",
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes,
+        ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        "hydration",
+    ) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    assert_eq!(all_a.source_key, all_b.source_key);
+    assert_ne!(all_a.source_key, votes.source_key);
+    let mention = ksp_core_lib::Pubkey::new_from_array([0x5a_u8; 32]);
+    let mention_text = mention.to_string();
+    let mentioned = match standard_logs_source(
+        "devnet",
+        "standard-logs-key",
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::Mentions(mention),
+        ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        "hydration",
+    ) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    assert_ne!(all_a.source_key, mentioned.source_key);
+    let debug = std::format!("{mentioned:?}");
+    assert!(debug.contains("filter_kind"));
+    assert!(debug.contains("mentions"));
+    assert!(!debug.contains(mention_text.as_str()));
+    assert!(!debug.contains("WS-SECRET-CANARY"));
+    assert!(!debug.contains("source_key: ["));
+    return;
+}
+
+#[test]
+fn v0_3_13_pre_003_standard_logs_projection_keeps_only_reference_material() {
+    let source = match standard_logs_source(
+        "devnet",
+        "standard-logs-projection",
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All,
+        ksp_onchain_transport_lib::SolanaCommitment::Finalized,
+        "hydration",
+    ) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let fixture = StandardLogsViewFixture { signature: PRE_004_ZERO_SIGNATURE_TEXT.to_owned(), slot: 73 };
+    let signal = match super::project_standard_logs_signal(&source, &fixture) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    assert_eq!(signal.family, super::RawTransactionIngestSourceFamily::Logs);
+    assert_eq!(signal.slot, 73);
+    assert_eq!(signal.transaction_index, std::option::Option::None);
+    assert_eq!(signal.created_at, std::option::Option::None);
+    assert_eq!(signal.matched_filter_count, 1);
+    assert_eq!(signal.matched_filter_id, std::option::Option::None);
+    assert_eq!(signal.signature, ksp_store_lib::RawTransactionSignature::new([0_u8; 64]));
+    let debug = std::format!("{signal:?}");
+    assert!(!debug.contains(PRE_004_ZERO_SIGNATURE_TEXT));
+    return;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn v0_3_13_pre_003_standard_logs_reference_hydrates_once_to_exact_common_raw() {
+    let body = std::format!(
+        "{{\"jsonrpc\":\"2.0\",\"result\":{{\"slot\":42,\"blockTime\":1760000120,\"transaction\":[\"{}\",\"base64\"],\"meta\":{{\"err\":null,\"fee\":5000}},\"version\":\"legacy\",\"transactionIndex\":7}},\"id\":1}}",
+        PRE_004_ZERO_TRANSACTION_BASE64,
+    );
+    let (url, server) = match serve_http_once(body) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    let source = match standard_logs_source_for_http_url(url.as_str(), ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::All) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let settings = match pre_004_settings() {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let fixture = StandardLogsViewFixture { signature: PRE_004_ZERO_SIGNATURE_TEXT.to_owned(), slot: 42 };
+    let signal = match super::project_standard_logs_signal(&source, &fixture) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    let received_at = match ksp_store_lib::RawTimestamp::from_unix_millis(1_760_000_200_000) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    let hydration = source.hydration_context();
+    let key = match super::hydration_key(&hydration, &signal) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    let fetched =
+        super::fetch_hydration(hydration.http_pool.clone(), hydration.hydration_role.clone(), hydration.network.clone(), key, hydration.commitment).await;
+    let fetched = match fetched {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    let ingress = super::finalize_hydration(&hydration, &settings, signal, received_at, &fetched.observed);
+    let ingress = match ingress {
+        std::result::Result::Ok(std::option::Option::Some(value)) => value,
+        std::result::Result::Ok(std::option::Option::None) | std::result::Result::Err(_) => return,
+    };
+    assert_eq!(ingress.provenance.protocol().as_str(), "solana_ws_http");
+    assert_eq!(ingress.provenance.acquisition_method().as_str(), "logs_get_transaction");
+    assert_eq!(ingress.material.slot(), 42);
+    assert_eq!(ingress.material.signature(), &ksp_store_lib::RawTransactionSignature::new([0_u8; 64]));
+    let (mut admission, sender) = crate::RawTransactionAdmission::new(1);
+    if sender.send(ingress).await.is_err() {
+        return;
+    }
+    std::mem::drop(sender);
+    let acquisition = match admission.receive(settings.network()).await {
+        std::result::Result::Ok(std::option::Option::Some(value)) => value,
+        _ => return,
+    };
+    assert_eq!(acquisition.transaction().reference().network().as_str(), "devnet");
+    assert_eq!(acquisition.transaction().reference().signature().as_bytes(), &[0_u8; 64]);
+    assert_eq!(acquisition.transaction().slot(), 42);
+    assert_eq!(acquisition.transaction().block_time().map(|value| return value.unix_millis()), std::option::Option::Some(1_760_000_120_000));
+    assert_eq!(acquisition.observation().provenance().protocol().as_str(), "solana_ws_http");
+    assert_eq!(acquisition.observation().provenance().acquisition_method().as_str(), "logs_get_transaction");
+    let request_body = match server.join() {
+        std::result::Result::Ok(std::result::Result::Ok(value)) => value,
+        std::result::Result::Ok(std::result::Result::Err(_)) | std::result::Result::Err(_) => return,
+    };
+    assert!(request_body.contains("\"method\":\"getTransaction\""));
+    assert!(request_body.contains("\"encoding\":\"base64\""));
+    assert!(request_body.contains("\"maxSupportedTransactionVersion\":1"));
+    return;
+}
+
+#[test]
+fn v0_3_13_pre_003_runtime_resources_reject_duplicate_standard_logs_identity() {
+    let first = match standard_logs_source(
+        "devnet",
+        "standard-logs-duplicate",
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes,
+        ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        "hydration",
+    ) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let duplicate = match standard_logs_source(
+        "devnet",
+        "standard-logs-duplicate",
+        ksp_onchain_transport_lib::SolanaLogsSubscribeFilter::AllWithVotes,
+        ksp_onchain_transport_lib::SolanaCommitment::Confirmed,
+        "secondary_hydration",
+    ) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let mut resources = crate::RawTransactionIngestRuntimeResources::from_standard_logs_source(first);
+    let result = resources.try_push_standard_logs_source(duplicate);
+    let error = match result {
+        std::result::Result::Ok(()) => return,
+        std::result::Result::Err(value) => value,
+    };
+    assert!(error.context().iter().any(|context| return context.value() == "runtime_resources.duplicate_source_identity"));
+    assert_eq!(resources.source_count(), 1);
+    return;
+}
+
 #[test]
 fn pre_007_processing_frontier_never_advances_through_pending_work() {
     let mut frontier = super::RawTransactionIngestProcessingFrontier::new();
@@ -1593,12 +1949,13 @@ async fn pre_009_duplicate_storm_is_bounded_coalesced_and_abort_leaves_no_orphan
     coordinator.max_pending_signals = 3;
     let (frontier_sender, frontier_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());
     let mut processing_frontier = super::RawTransactionIngestProcessingFrontierReporter::new(frontier_sender);
+    let hydration = source.hydration_context();
     assert!(processing_frontier.observe_settled(9).is_ok());
     assert!(processing_frontier.observe_settled(10).is_ok());
     for _ in 0..3 {
-        assert!(coordinator.queue_signal(&source, signal.clone(), received_at, &mut processing_frontier).is_ok());
+        assert!(coordinator.queue_signal(&hydration, signal.clone(), received_at, &mut processing_frontier).is_ok());
     }
-    let saturated = coordinator.queue_signal(&source, signal, received_at, &mut processing_frontier);
+    let saturated = coordinator.queue_signal(&hydration, signal, received_at, &mut processing_frontier);
     let saturated = match saturated {
         std::result::Result::Ok(()) => return,
         std::result::Result::Err(value) => value,
@@ -1678,7 +2035,8 @@ async fn pre_009_abort_joins_in_flight_hydration_and_clears_pending_projection()
     let mut coordinator = super::RawTransactionIngestHydrationCoordinator::new(&settings);
     let (frontier_sender, frontier_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());
     let mut processing_frontier = super::RawTransactionIngestProcessingFrontierReporter::new(frontier_sender);
-    assert!(coordinator.queue_signal(&source, signal, received_at, &mut processing_frontier).is_ok());
+    let hydration = source.hydration_context();
+    assert!(coordinator.queue_signal(&hydration, signal, received_at, &mut processing_frontier).is_ok());
     assert_eq!(coordinator.pending_signal_count, 1);
     assert_eq!(coordinator.tasks.len(), 1);
     assert_eq!(frontier_receiver.borrow().hydration_pending(), 1);