v0.3.13-pre.007

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: 18
+// version: 19
 
 fn grpc_endpoint(cluster: &str) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings> {
     return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider");
@@ -959,24 +959,118 @@ async fn v0_3_13_pre_002_runtime_resources_reject_duplicate_and_cross_network_so
     return;
 }
 
+#[test]
+fn v0_3_13_pre_007_source_inventory_aggregates_frontier_and_lifecycle_conservatively() {
+    let source_keys = std::vec![[1_u8; 32], [2_u8; 32]];
+    let inventory = std::sync::Arc::new(std::sync::Mutex::new(super::RawTransactionIngestSourceInventory::new(source_keys)));
+    let (aggregate_sender, aggregate_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());
+    let first = super::RawTransactionIngestSourceInventoryPublisher {
+        aggregate_sender: aggregate_sender.clone(),
+        entry_index: 0,
+        inventory: std::sync::Arc::clone(&inventory),
+        source_key: [1_u8; 32],
+    };
+    let second = super::RawTransactionIngestSourceInventoryPublisher { aggregate_sender, entry_index: 1, inventory, source_key: [2_u8; 32] };
+    first.publish(
+        crate::RawTransactionIngestProcessingFrontierProjection::new(1, std::option::Option::Some(50), std::option::Option::Some(45)).with_source_continuity(
+            std::option::Option::Some(crate::RawTransactionIngestSourceState::Active),
+            1,
+            2,
+            0,
+        ),
+    );
+    let first_aggregate = *aggregate_receiver.borrow();
+    assert_eq!(first_aggregate.hydration_pending(), 1);
+    assert_eq!(first_aggregate.processing_frontier_slot(), std::option::Option::None);
+    assert_eq!(first_aggregate.oldest_pending_slot(), std::option::Option::Some(45));
+    assert_eq!(first_aggregate.source_state(), std::option::Option::Some(crate::RawTransactionIngestSourceState::Active));
+    second.publish(
+        crate::RawTransactionIngestProcessingFrontierProjection::new(2, std::option::Option::Some(42), std::option::Option::Some(40)).with_source_continuity(
+            std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting),
+            3,
+            4,
+            1,
+        ),
+    );
+    let aggregate = *aggregate_receiver.borrow();
+    assert_eq!(aggregate.hydration_pending(), 3);
+    assert_eq!(aggregate.processing_frontier_slot(), std::option::Option::Some(42));
+    assert_eq!(aggregate.oldest_pending_slot(), std::option::Option::Some(40));
+    assert_eq!(aggregate.source_state(), std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting));
+    assert_eq!(aggregate.source_reconnect_total(), 4);
+    assert_eq!(aggregate.source_replay_attempt_total(), 6);
+    assert_eq!(aggregate.source_continuity_gap_total(), 1);
+    return;
+}
+
 #[tokio::test(flavor = "current_thread")]
-async fn v0_3_13_pre_002_multi_source_activation_fails_closed_until_supervisor_tranche() {
-    let first = match source_with_identity("devnet", "source-a", "tx", std::option::Option::None) {
-        std::option::Option::Some(value) => value,
-        std::option::Option::None => return,
+async fn v0_3_13_pre_007_source_supervisor_joins_all_children_on_stop() {
+    let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    let (stop_sender, stop_receiver) = tokio::sync::watch::channel(false);
+    let (source_stop_sender, source_stop_receiver) = tokio::sync::watch::channel(false);
+    let mut children = tokio::task::JoinSet::new();
+    for _ in 0..3 {
+        let active = std::sync::Arc::clone(&active);
+        let mut source_stop_receiver = source_stop_receiver.clone();
+        let _abort_handle = children.spawn(async move {
+            active.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
+            loop {
+                let changed = source_stop_receiver.changed().await;
+                if changed.is_err() || *source_stop_receiver.borrow() {
+                    active.fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
+                    return std::result::Result::Ok(());
+                }
+            }
+        });
+    }
+    let supervisor = tokio::spawn(super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children));
+    for _ in 0..64 {
+        if active.load(std::sync::atomic::Ordering::Acquire) == 3 {
+            break;
+        }
+        tokio::task::yield_now().await;
+    }
+    assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 3);
+    stop_sender.send_replace(true);
+    let joined = supervisor.await;
+    let result = match joined {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
     };
-    let second = match source_with_identity("devnet", "source-b", "tx", std::option::Option::None) {
-        std::option::Option::Some(value) => value,
-        std::option::Option::None => return,
-    };
-    let mut resources = crate::RawTransactionIngestRuntimeResources::new(first);
-    assert!(resources.try_push_yellowstone_source(second).is_ok());
-    let error = match resources.validate_single_source_activation() {
-        std::result::Result::Ok(()) => return,
-        std::result::Result::Err(error) => error,
-    };
-    assert_eq!(error.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
-    assert!(error.context().iter().any(|context| return context.value() == "runtime_resources.multi_source_activation_pending"));
+    assert!(result.is_ok());
+    assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 0);
+    return;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn v0_3_13_pre_007_source_failure_stops_and_joins_sibling_sources() {
+    let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(2));
+    let sibling_active = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let (_stop_sender, stop_receiver) = tokio::sync::watch::channel(false);
+    let (source_stop_sender, source_stop_receiver) = tokio::sync::watch::channel(false);
+    let mut children = tokio::task::JoinSet::new();
+    let failing_barrier = std::sync::Arc::clone(&barrier);
+    let _failing_abort_handle = children.spawn(async move {
+        let _barrier_wait = failing_barrier.wait().await;
+        return std::result::Result::Err(crate::runtime_error("test.pre_007_source_failed"));
+    });
+    let sibling_barrier = std::sync::Arc::clone(&barrier);
+    let sibling_active_for_task = std::sync::Arc::clone(&sibling_active);
+    let mut sibling_stop_receiver = source_stop_receiver.clone();
+    let _sibling_abort_handle = children.spawn(async move {
+        sibling_active_for_task.store(true, std::sync::atomic::Ordering::Release);
+        let _barrier_wait = sibling_barrier.wait().await;
+        loop {
+            let changed = sibling_stop_receiver.changed().await;
+            if changed.is_err() || *sibling_stop_receiver.borrow() {
+                sibling_active_for_task.store(false, std::sync::atomic::Ordering::Release);
+                return std::result::Result::Ok(());
+            }
+        }
+    });
+    let result = super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children).await;
+    assert!(result.is_err());
+    assert!(!sibling_active.load(std::sync::atomic::Ordering::Acquire));
     return;
 }