v0.3.12-pre.009

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: 9
+// version: 10
 
 fn grpc_endpoint(cluster: &str) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings> {
     return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider");
@@ -1289,7 +1289,7 @@ async fn pre_006_coalescence_key_merges_transaction_and_status_before_http_fanou
     assert_eq!(coordinator.pending.len(), 1);
     assert_eq!(coordinator.pending_signal_count, 2);
     assert_eq!(coordinator.tasks.len(), 1);
-    coordinator.abort_all().await;
+    coordinator.abort_all(&mut processing_frontier).await;
     assert_eq!(coordinator.pending_signal_count, 0);
     assert!(coordinator.pending.is_empty());
     assert!(coordinator.tasks.is_empty());
@@ -1412,3 +1412,125 @@ fn pre_008_reconnect_replay_and_proven_gap_are_distinct_monotone_and_frontier_pr
     }));
     return;
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn pre_009_duplicate_storm_is_bounded_coalesced_and_abort_leaves_no_orphan_frontier() {
+    let source = match signal_source() {
+        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 signal = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::Transaction, 10, 1, &["storm"], 0x55) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let received_at = match ksp_store_lib::RawTimestamp::from_unix_millis(1_760_000_300_000) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    let mut coordinator = super::RawTransactionIngestHydrationCoordinator::new(&settings);
+    coordinator.max_in_flight = 0;
+    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);
+    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());
+    }
+    let saturated = coordinator.queue_signal(&source, signal, received_at, &mut processing_frontier);
+    let saturated = match saturated {
+        std::result::Result::Ok(()) => return,
+        std::result::Result::Err(value) => value,
+    };
+    assert!(saturated.context().iter().any(|context| {
+        return context.value() == "source.hydration_pending_saturated";
+    }));
+    assert_eq!(coordinator.pending.len(), 1);
+    assert_eq!(coordinator.pending_signal_count, 3);
+    assert!(coordinator.tasks.is_empty());
+    let pending = *frontier_receiver.borrow();
+    assert_eq!(pending.hydration_pending(), 3);
+    assert_eq!(pending.oldest_pending_slot(), std::option::Option::Some(10));
+    assert_eq!(pending.processing_frontier_slot(), std::option::Option::Some(9));
+    coordinator.abort_all(&mut processing_frontier).await;
+    assert!(coordinator.pending.is_empty());
+    assert_eq!(coordinator.pending_signal_count, 0);
+    assert!(coordinator.tasks.is_empty());
+    let cleaned = *frontier_receiver.borrow();
+    assert_eq!(cleaned.hydration_pending(), 0);
+    assert_eq!(cleaned.oldest_pending_slot(), std::option::Option::None);
+    assert_eq!(cleaned.processing_frontier_slot(), std::option::Option::Some(9));
+    return;
+}
+
+#[test]
+fn pre_009_frontier_counter_exhaustion_is_transactional_before_abort_cleanup() {
+    let mut frontier = super::RawTransactionIngestProcessingFrontier::new();
+    frontier.pending_total = 1;
+    frontier.slots.insert(77, super::RawTransactionIngestProcessingSlotState { pending: 1, settled: u64::MAX });
+    let before = frontier.projection();
+    let exhausted = frontier.settle_pending(77);
+    let exhausted = match exhausted {
+        std::result::Result::Ok(()) => return,
+        std::result::Result::Err(value) => value,
+    };
+    assert!(exhausted.context().iter().any(|context| {
+        return context.value() == "source.frontier_slot_settled_counter_exhausted";
+    }));
+    let after = frontier.projection();
+    assert_eq!(after.hydration_pending(), before.hydration_pending());
+    assert_eq!(after.oldest_pending_slot(), before.oldest_pending_slot());
+    assert_eq!(after.processing_frontier_slot(), before.processing_frontier_slot());
+    let state = match frontier.slots.get(&77) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    assert_eq!(state.pending, 1);
+    assert_eq!(state.settled, u64::MAX);
+    frontier.discard_all_pending();
+    let cleaned = frontier.projection();
+    assert_eq!(cleaned.hydration_pending(), 0);
+    assert_eq!(cleaned.oldest_pending_slot(), std::option::Option::None);
+    assert_eq!(cleaned.processing_frontier_slot(), std::option::Option::None);
+    return;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn pre_009_abort_joins_in_flight_hydration_and_clears_pending_projection() {
+    let source = match signal_source() {
+        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 signal = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::TransactionStatus, 81, 2, &["stop-hydration"], 0x66) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let received_at = match ksp_store_lib::RawTimestamp::from_unix_millis(1_760_000_301_000) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    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());
+    assert_eq!(coordinator.pending_signal_count, 1);
+    assert_eq!(coordinator.tasks.len(), 1);
+    assert_eq!(frontier_receiver.borrow().hydration_pending(), 1);
+    coordinator.abort_all(&mut processing_frontier).await;
+    assert_eq!(coordinator.pending_signal_count, 0);
+    assert!(coordinator.pending.is_empty());
+    assert!(coordinator.tasks.is_empty());
+    let cleaned = *frontier_receiver.borrow();
+    assert_eq!(cleaned.hydration_pending(), 0);
+    assert_eq!(cleaned.oldest_pending_slot(), std::option::Option::None);
+    assert_eq!(cleaned.processing_frontier_slot(), std::option::Option::None);
+    return;
+}