v0.3.14-pre.012

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
-// version: 42
+// version: 43
 
 use sha2::Digest; // rust-rules: trait-import
 
@@ -1726,6 +1726,9 @@ impl crate::RawTransactionIngestHttpBlockPollingSource {
                         break 'source;
                     }
                 }
+                if *stop_receiver.borrow() {
+                    break;
+                }
                 if !blocked_by_null && let std::option::Option::Some(proven_end_slot) = discovery.proven_end_slot {
                     let coverage_result = {
                         let mut contracts = match continuity_contracts.lock() {
@@ -2584,6 +2587,7 @@ impl crate::RawTransactionIngestRuntimeResources {
             std::sync::Arc::clone(&continuity_contracts),
             std::sync::Arc::clone(&inventory),
             processing_frontier_sender,
+            settings.shutdown_drain_timeout(),
         )
         .await;
         let validation = {
@@ -2682,18 +2686,19 @@ async fn supervise_live_source_tasks(
     continuity_contracts: std::sync::Arc<std::sync::Mutex<crate::RawTransactionIngestContinuityContracts>>,
     inventory: std::sync::Arc<std::sync::Mutex<RawTransactionIngestSourceInventory>>,
     processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
+    shutdown_drain_timeout: std::time::Duration,
 ) -> ksp_core_lib::Result<()> {
     loop {
         if *stop_receiver.borrow() {
             source_stop_sender.send_replace(true);
-            return drain_live_source_tasks(&mut children, std::option::Option::None).await;
+            return drain_live_source_tasks(&mut children, std::option::Option::None, shutdown_drain_timeout).await;
         }
         let joined = tokio::select! {
             biased;
             changed = stop_receiver.changed() => {
                 if changed.is_err() || *stop_receiver.borrow() {
                     source_stop_sender.send_replace(true);
-                    return drain_live_source_tasks(&mut children, std::option::Option::None).await;
+                    return drain_live_source_tasks(&mut children, std::option::Option::None, shutdown_drain_timeout).await;
                 }
                 continue;
             }
@@ -2703,11 +2708,21 @@ async fn supervise_live_source_tasks(
             std::option::Option::Some(std::result::Result::Ok(value)) => value,
             std::option::Option::Some(std::result::Result::Err(_)) => {
                 source_stop_sender.send_replace(true);
-                return drain_live_source_tasks(&mut children, std::option::Option::Some(crate::runtime_error("source.task_join_failed"))).await;
+                return drain_live_source_tasks(
+                    &mut children,
+                    std::option::Option::Some(crate::runtime_error("source.task_join_failed")),
+                    shutdown_drain_timeout,
+                )
+                .await;
             },
             std::option::Option::None => {
                 source_stop_sender.send_replace(true);
-                return drain_live_source_tasks(&mut children, std::option::Option::Some(crate::runtime_error("source.task_set_empty"))).await;
+                return drain_live_source_tasks(
+                    &mut children,
+                    std::option::Option::Some(crate::runtime_error("source.task_set_empty")),
+                    shutdown_drain_timeout,
+                )
+                .await;
             },
         };
         let source_fault = match source_result {
@@ -2716,7 +2731,7 @@ async fn supervise_live_source_tasks(
         };
         if !source_loss_is_reconcilable(&source_fault) {
             source_stop_sender.send_replace(true);
-            return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await;
+            return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault), shutdown_drain_timeout).await;
         }
         let supervisor_state = {
             let inventory = match inventory.lock() {
@@ -2729,21 +2744,21 @@ async fn supervise_live_source_tasks(
             std::result::Result::Ok(value) => value,
             std::result::Result::Err(error) => {
                 source_stop_sender.send_replace(true);
-                return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await;
+                return drain_live_source_tasks(&mut children, std::option::Option::Some(error), shutdown_drain_timeout).await;
             },
         };
         let continuity_range = match source_loss_continuity_range(&source_fault) {
             std::result::Result::Ok(value) => value,
             std::result::Result::Err(error) => {
                 source_stop_sender.send_replace(true);
-                return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await;
+                return drain_live_source_tasks(&mut children, std::option::Option::Some(error), shutdown_drain_timeout).await;
             },
         };
         let continuity_range = match continuity_range {
             std::option::Option::Some(value) => value,
             std::option::Option::None => {
                 source_stop_sender.send_replace(true);
-                return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await;
+                return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault), shutdown_drain_timeout).await;
             },
         };
         let decision = {
@@ -2761,7 +2776,7 @@ async fn supervise_live_source_tasks(
             std::result::Result::Ok(value) => value,
             std::result::Result::Err(error) => {
                 source_stop_sender.send_replace(true);
-                return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await;
+                return drain_live_source_tasks(&mut children, std::option::Option::Some(error), shutdown_drain_timeout).await;
             },
         };
         match decision {
@@ -2770,7 +2785,7 @@ async fn supervise_live_source_tasks(
                     std::result::Result::Ok(value) => value,
                     std::result::Result::Err(error) => {
                         source_stop_sender.send_replace(true);
-                        return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await;
+                        return drain_live_source_tasks(&mut children, std::option::Option::Some(error), shutdown_drain_timeout).await;
                     },
                 };
                 processing_frontier_sender.send_replace(aggregate);
@@ -2778,7 +2793,7 @@ async fn supervise_live_source_tasks(
             },
             crate::RawTransactionIngestSourceLossDecision::Fault => {
                 source_stop_sender.send_replace(true);
-                return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await;
+                return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault), shutdown_drain_timeout).await;
             },
         }
     }
@@ -2834,16 +2849,29 @@ fn source_loss_is_reconcilable(error: &ksp_core_lib::Error) -> bool {
 async fn drain_live_source_tasks(
     children: &mut tokio::task::JoinSet<([u8; 32], ksp_core_lib::Result<()>)>,
     mut first_fault: std::option::Option<ksp_core_lib::Error>,
+    shutdown_drain_timeout: std::time::Duration,
 ) -> ksp_core_lib::Result<()> {
-    while let std::option::Option::Some(joined) = children.join_next().await {
-        if first_fault.is_some() {
-            continue;
+    let drain = async {
+        while let std::option::Option::Some(joined) = children.join_next().await {
+            if first_fault.is_some() {
+                continue;
+            }
+            first_fault = match joined {
+                std::result::Result::Ok((_source_key, std::result::Result::Ok(()))) => std::option::Option::None,
+                std::result::Result::Ok((_source_key, std::result::Result::Err(error))) => std::option::Option::Some(error),
+                std::result::Result::Err(_) => std::option::Option::Some(crate::runtime_error("source.task_join_failed")),
+            };
+        }
+    };
+    if tokio::time::timeout(shutdown_drain_timeout, drain).await.is_err() {
+        children.abort_all();
+        while children.join_next().await.is_some() {}
+        if first_fault.is_none() {
+            first_fault = std::option::Option::Some(ksp_core_lib::Error::new(
+                crate::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT,
+                "RAW transaction ingest source drain timed out",
+            ));
         }
-        first_fault = match joined {
-            std::result::Result::Ok((_source_key, std::result::Result::Ok(()))) => std::option::Option::None,
-            std::result::Result::Ok((_source_key, std::result::Result::Err(error))) => std::option::Option::Some(error),
-            std::result::Result::Err(_) => std::option::Option::Some(crate::runtime_error("source.task_join_failed")),
-        };
     }
     return match first_fault {
         std::option::Option::Some(error) => std::result::Result::Err(error),
@@ -4656,6 +4684,9 @@ impl RawTransactionIngestHydrationCoordinator {
             std::result::Result::Ok(std::result::Result::Err(error)) => return std::result::Result::Err(error),
             std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.hydration_task_join_failed")),
         };
+        if *stop_receiver.borrow() {
+            return std::result::Result::Ok(false);
+        }
         let pending = match self.pending.remove(&fetched.key) {
             std::option::Option::Some(value) => value,
             std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.hydration_result_without_pending")),
@@ -4665,6 +4696,9 @@ impl RawTransactionIngestHydrationCoordinator {
         }
         self.pending_signal_count -= pending.signals.len();
         for pending_signal in pending.signals {
+            if *stop_receiver.borrow() {
+                return std::result::Result::Ok(false);
+            }
             let signal_slot = pending_signal.signal.slot;
             let resolution = resolve_known_reference_hydration(hydration, settings, pending_signal.signal, pending_signal.received_at, &fetched.observed);
             let resolution = match resolution {

crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs
-// version: 29
+// version: 30
 
 //! Dependency firewall canaries for the RAW transaction ingest Worker foundation.
 
@@ -732,3 +732,29 @@ fn v0_3_13_pre_012_cross_layer_closure_does_not_expand_worker_dependency_graph()
     }
     return;
 }
+
+#[test]
+fn v0_3_14_pre_012_shutdown_hardening_stays_worker_local_and_facade_only() {
+    let runtime = include_str!("../src/runtime.rs");
+    let resources = include_str!("../src/runtime_resources.rs");
+    for required in [
+        "tokio::time::timeout(settings.shutdown_drain_timeout(), drain).await",
+        "tokio::time::timeout(shutdown_drain_timeout, drain).await",
+        "children.abort_all()",
+        "persistence.abort_all()",
+    ] {
+        assert!(runtime.contains(required) || resources.contains(required), "required pre.012 bounded shutdown guard missing: {required}");
+    }
+    for forbidden in [
+        "ksp_store_postgres_lib::",
+        "ksp_config_lib::",
+        "ksp_job_backfill_lib::",
+        "reqwest::",
+        "tokio_tungstenite::",
+        "tonic::",
+        "yellowstone_grpc_proto::",
+    ] {
+        assert!(!runtime.contains(forbidden) && !resources.contains(forbidden), "pre.012 added forbidden Worker dependency edge: {forbidden}");
+    }
+    return;
+}

crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs
-// version: 35
+// version: 36
 
 //! External public, security, redaction and release-boundary hardening canaries through `v0.3.14-pre.011`.
 
@@ -1313,3 +1313,42 @@ fn v0_3_14_pre_011_gap_observability_is_bounded_checked_and_redacted() {
     assert!(!root.contains("source_key"));
     return;
 }
+
+#[test]
+fn v0_3_14_pre_012_shutdown_races_are_bounded_across_source_and_worker_drains() {
+    let runtime = include_str!("../src/runtime.rs");
+    let resources = include_str!("../src/runtime_resources.rs");
+    for required in [
+        "tokio::time::timeout(settings.shutdown_drain_timeout(), drain).await",
+        "persistence.abort_all()",
+        "while persistence.join_next().await.is_some() {}",
+        "while children.join_next().await.is_some() {}",
+    ] {
+        assert!(runtime.contains(required), "required pre.012 outer shutdown guard missing: {required}");
+    }
+    for required in [
+        "shutdown_drain_timeout: std::time::Duration",
+        "tokio::time::timeout(shutdown_drain_timeout, drain).await",
+        "children.abort_all()",
+        "ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT",
+        "coordinator.abort_all(&mut processing_frontier).await",
+        "discover_http_block_window(",
+        "admission_sender.send(ingress)",
+        "if *stop_receiver.borrow()",
+    ] {
+        assert!(resources.contains(required), "required pre.012 source shutdown guard missing: {required}");
+    }
+    for forbidden in [
+        "unbounded_channel",
+        "ksp_store_postgres_lib::",
+        "ksp_config_lib::",
+        "ksp_job_backfill_lib::",
+        "reqwest::",
+        "tokio_tungstenite::",
+        "tonic::",
+        "yellowstone_grpc_proto::",
+    ] {
+        assert!(!runtime.contains(forbidden) && !resources.contains(forbidden), "pre.012 crossed a Worker/facade boundary: {forbidden}");
+    }
+    return;
+}

crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs
-// version: 30
+// version: 31
 
 //! Release-completeness canaries through the `v0.3.14-pre.011` gap observability tranche.
 
@@ -422,3 +422,22 @@ fn v0_3_14_pre_011_gap_observability_canaries_are_present_without_source_identit
     assert!(!root.contains("source_key"));
     return;
 }
+
+#[test]
+fn v0_3_14_pre_012_shutdown_race_canaries_cover_bounded_source_and_worker_drains() {
+    let dependency_boundary = include_str!("dependency_boundary.rs");
+    let hardening = include_str!("hardening.rs");
+    let resource_tests = include_str!("../unit_tests/runtime_resources.rs");
+    let root = include_str!("../src/lib.rs");
+    for required in [
+        "v0_3_14_pre_012_source_fault_preserves_first_fault_and_aborts_non_cooperative_sibling",
+        "v0_3_14_pre_012_stop_aborts_non_cooperative_source_after_bounded_drain",
+    ] {
+        assert!(resource_tests.contains(required), "required pre.012 source shutdown canary missing: {required}");
+    }
+    assert!(hardening.contains("v0_3_14_pre_012_shutdown_races_are_bounded_across_source_and_worker_drains"));
+    assert!(dependency_boundary.contains("v0_3_14_pre_012_shutdown_hardening_stays_worker_local_and_facade_only"));
+    assert!(!root.contains("RawTransactionIngestShutdown"));
+    assert!(!root.contains("SourceDrain"));
+    return;
+}

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: 33
+// version: 34
 
 fn grpc_endpoint(cluster: &str) -> std::option::Option<ksp_onchain_transport_lib::YellowstoneGrpcEndpointSettings> {
     return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider");
@@ -1155,6 +1155,7 @@ async fn v0_3_13_pre_007_source_supervisor_joins_all_children_on_stop() {
             contracts,
             inventory,
             tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0,
+            std::time::Duration::from_secs(5),
         )
         .await;
     });
@@ -1223,6 +1224,7 @@ async fn v0_3_13_pre_007_source_failure_stops_and_joins_sibling_sources() {
         contracts,
         inventory,
         tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0,
+        std::time::Duration::from_secs(5),
     )
     .await;
     assert!(result.is_err());
@@ -3838,6 +3840,7 @@ async fn v0_3_13_pre_011_aborting_outer_source_supervisor_aborts_nested_source_t
             contracts,
             inventory,
             tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0,
+            std::time::Duration::from_secs(5),
         )
         .await;
     });
@@ -3905,6 +3908,7 @@ async fn v0_3_13_pre_011_stop_racing_ready_source_fault_preserves_fault_and_join
             contracts,
             inventory,
             tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0,
+            std::time::Duration::from_secs(5),
         )
         .await;
     });
@@ -3930,6 +3934,110 @@ async fn v0_3_13_pre_011_stop_racing_ready_source_fault_preserves_fault_and_join
     return;
 }
 
+#[tokio::test(flavor = "current_thread")]
+async fn v0_3_14_pre_012_source_fault_preserves_first_fault_and_aborts_non_cooperative_sibling() {
+    let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(2));
+    let (_stop_sender, stop_receiver) = tokio::sync::watch::channel(false);
+    let (source_stop_sender, _source_stop_receiver) = tokio::sync::watch::channel(false);
+    let contracts = match supervisor_contracts(&[(1, "fixture-a", 1), (2, "fixture-b", 2)]) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let inventory = match supervisor_inventory(
+        std::vec![[1_u8; 32], [2_u8; 32]],
+        &[
+            std::option::Option::Some(crate::RawTransactionIngestSourceState::Failed),
+            std::option::Option::Some(crate::RawTransactionIngestSourceState::Active),
+        ],
+    ) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    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 ([1_u8; 32], std::result::Result::Err(crate::runtime_error("test.pre_012_source_fault")));
+    });
+    let sibling_barrier = std::sync::Arc::clone(&barrier);
+    let sibling_active = std::sync::Arc::clone(&active);
+    let _sibling_abort_handle = children.spawn(async move {
+        let _guard = Pre011SourceActiveGuard::new(sibling_active);
+        let _barrier_wait = sibling_barrier.wait().await;
+        return std::future::pending::<([u8; 32], ksp_core_lib::Result<()>)>().await;
+    });
+    let result = super::supervise_live_source_tasks(
+        stop_receiver,
+        source_stop_sender,
+        children,
+        contracts,
+        inventory,
+        tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0,
+        std::time::Duration::from_millis(5),
+    )
+    .await;
+    let error = match result {
+        std::result::Result::Ok(()) => return,
+        std::result::Result::Err(value) => value,
+    };
+    assert_eq!(error.context()[0].value(), "test.pre_012_source_fault");
+    assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 0);
+    return;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn v0_3_14_pre_012_stop_aborts_non_cooperative_source_after_bounded_drain() {
+    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 contracts = match supervisor_contracts(&[(1, "fixture-a", 1)]) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let inventory = match supervisor_inventory(std::vec![[1_u8; 32]], &[std::option::Option::Some(crate::RawTransactionIngestSourceState::Active)]) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let mut children = tokio::task::JoinSet::new();
+    let source_active = std::sync::Arc::clone(&active);
+    let _abort_handle = children.spawn(async move {
+        let _guard = Pre011SourceActiveGuard::new(source_active);
+        return std::future::pending::<([u8; 32], ksp_core_lib::Result<()>)>().await;
+    });
+    let supervisor = tokio::spawn(async move {
+        return super::supervise_live_source_tasks(
+            stop_receiver,
+            source_stop_sender,
+            children,
+            contracts,
+            inventory,
+            tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0,
+            std::time::Duration::from_millis(5),
+        )
+        .await;
+    });
+    for _ in 0..64 {
+        if active.load(std::sync::atomic::Ordering::Acquire) == 1 {
+            break;
+        }
+        tokio::task::yield_now().await;
+    }
+    assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 1);
+    stop_sender.send_replace(true);
+    let result = match supervisor.await {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    let error = match result {
+        std::result::Result::Ok(()) => return,
+        std::result::Result::Err(value) => value,
+    };
+    assert_eq!(error.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT);
+    assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 0);
+    return;
+}
+
 #[tokio::test(flavor = "current_thread")]
 async fn v0_3_14_pre_008_proven_full_ledger_epoch_survives_filtered_peer_loss_without_worker_respawn() {
     let sibling_active = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
@@ -3974,7 +4082,16 @@ async fn v0_3_14_pre_008_proven_full_ledger_epoch_survives_filtered_peer_loss_wi
     });
     let (health_sender, health_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());
     let supervisor = tokio::spawn(async move {
-        return super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory, health_sender).await;
+        return super::supervise_live_source_tasks(
+            stop_receiver,
+            source_stop_sender,
+            children,
+            contracts,
+            inventory,
+            health_sender,
+            std::time::Duration::from_secs(5),
+        )
+        .await;
     });
     for _ in 0..64 {
         if sibling_active.load(std::sync::atomic::Ordering::Acquire) && health_receiver.borrow().failed_source_losses_reconciled() {