v0.3.11-pre.005

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs
-// version: 3
+// version: 4
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -8,9 +8,9 @@
 //! Source-neutral runtime foundation for continuous KSP RAW transaction ingestion.
 //!
 //! This tranche owns the concrete Worker family identity, validated technical settings
-//! and the minimal caller-runtime-owned start/stop lifecycle. Task supervision, bounded
-//! admission, persistence and latest-value snapshots remain in their dedicated prereleases;
-//! no live source or Transport dependency exists here.
+//! and the caller-runtime-owned lifecycle with private child-task supervision. Bounded
+//! admission, persistence and latest-value snapshots remain in their dedicated prereleases; no
+//! live source or Transport dependency exists here.
 
 mod error;
 mod identity;

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs
-// version: 1
+// version: 2
 
 /// Runtime-neutral boxed future resolving after one RAW transaction ingest Worker has fully reached a terminal lifecycle state.
 pub type RawTransactionIngestTerminalFuture<'a> =
@@ -85,6 +85,25 @@ fn current_runtime_handle() -> ksp_core_lib::Result<tokio::runtime::Handle> {
     };
 }
 
+async fn drain_children(children: &mut tokio::task::JoinSet<()>) -> bool {
+    let mut clean = true;
+    while let std::option::Option::Some(joined) = children.join_next().await {
+        if joined.is_err() {
+            clean = false;
+        }
+    }
+    return clean;
+}
+
+fn finish_faulted(lifecycle: &mut ksp_worker_api::WorkerLifecycle, sender: &tokio::sync::watch::Sender<ksp_worker_api::WorkerState>) {
+    if lifecycle.fault(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID).is_err() {
+        sender.send_replace(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
+        return;
+    }
+    sender.send_replace(lifecycle.state());
+    return;
+}
+
 fn finish_stopped(lifecycle: &mut ksp_worker_api::WorkerLifecycle, sender: &tokio::sync::watch::Sender<ksp_worker_api::WorkerState>) {
     if lifecycle.mark_stopping().is_err() {
         sender.send_replace(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
@@ -99,12 +118,15 @@ fn finish_stopped(lifecycle: &mut ksp_worker_api::WorkerLifecycle, sender: &toki
     return;
 }
 
-async fn run_foundation(
+async fn run_supervisor<Spawner>(
     mut lifecycle: ksp_worker_api::WorkerLifecycle,
     _store_guard: std::option::Option<std::sync::Arc<ksp_store_lib::Store>>,
     mut stop_receiver: tokio::sync::watch::Receiver<bool>,
     terminal_sender: tokio::sync::watch::Sender<ksp_worker_api::WorkerState>,
-) {
+    source_spawner: Spawner,
+) where
+    Spawner: FnOnce(&mut tokio::task::JoinSet<()>, tokio::sync::watch::Receiver<bool>) + std::marker::Send + 'static,
+{
     if *stop_receiver.borrow() {
         finish_stopped(&mut lifecycle, &terminal_sender);
         return;
@@ -114,11 +136,12 @@ async fn run_foundation(
         return;
     }
     terminal_sender.send_replace(lifecycle.state());
-    loop {
-        let changed = stop_receiver.changed().await;
-        if changed.is_err() || *stop_receiver.borrow() {
-            break;
-        }
+    let mut children = tokio::task::JoinSet::new();
+    source_spawner(&mut children, stop_receiver.clone());
+    let supervised_clean = supervise_until_stop(&mut stop_receiver, &mut children).await;
+    if !supervised_clean {
+        finish_faulted(&mut lifecycle, &terminal_sender);
+        return;
     }
     finish_stopped(&mut lifecycle, &terminal_sender);
     return;
@@ -129,6 +152,18 @@ fn start_foundation(
     runtime: tokio::runtime::Handle,
     store_guard: std::option::Option<std::sync::Arc<ksp_store_lib::Store>>,
 ) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle> {
+    return start_foundation_with_source_spawner(settings, runtime, store_guard, |_children, _stop_receiver| {});
+}
+
+fn start_foundation_with_source_spawner<Spawner>(
+    settings: crate::RawTransactionIngestSettings,
+    runtime: tokio::runtime::Handle,
+    store_guard: std::option::Option<std::sync::Arc<ksp_store_lib::Store>>,
+    source_spawner: Spawner,
+) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle>
+where
+    Spawner: FnOnce(&mut tokio::task::JoinSet<()>, tokio::sync::watch::Receiver<bool>) + std::marker::Send + 'static,
+{
     let kind = match ksp_worker_api::WorkerKindCode::new(crate::RAW_TRANSACTION_INGEST_WORKER_KIND_CODE) {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("start.worker_kind_invalid")),
@@ -141,10 +176,41 @@ fn start_foundation(
     let (stop_sender, stop_receiver) = tokio::sync::watch::channel(false);
     let (terminal_sender, terminal_receiver) = tokio::sync::watch::channel(lifecycle.state());
     let handle = crate::RawTransactionIngestHandle { stop_sender, stop_token, terminal_receiver };
-    std::mem::drop(runtime.spawn(run_foundation(lifecycle, store_guard, stop_receiver, terminal_sender)));
+    std::mem::drop(runtime.spawn(run_supervisor(lifecycle, store_guard, stop_receiver, terminal_sender, source_spawner)));
     return std::result::Result::Ok(handle);
 }
 
+async fn supervise_until_stop(stop_receiver: &mut tokio::sync::watch::Receiver<bool>, children: &mut tokio::task::JoinSet<()>) -> bool {
+    let mut clean = true;
+    loop {
+        if *stop_receiver.borrow() {
+            break;
+        }
+        if children.is_empty() {
+            let changed = stop_receiver.changed().await;
+            if changed.is_err() || *stop_receiver.borrow() {
+                break;
+            }
+            continue;
+        }
+        tokio::select! {
+            biased;
+            changed = stop_receiver.changed() => {
+                if changed.is_err() || *stop_receiver.borrow() {
+                    break;
+                }
+            }
+            joined = children.join_next() => {
+                if let std::option::Option::Some(std::result::Result::Err(_)) = joined {
+                    clean = false;
+                }
+            }
+        }
+    }
+    let drained_clean = drain_children(children).await;
+    return clean && drained_clean;
+}
+
 fn validate_store_network(settings: &crate::RawTransactionIngestSettings, store_network: &ksp_store_lib::RawNetworkId) -> ksp_core_lib::Result<()> {
     if settings.network() != store_network {
         return std::result::Result::Err(crate::runtime_error("start.store_network_mismatch"));

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: 3
+// version: 4
 
 //! Dependency firewall canaries for the RAW transaction ingest Worker foundation.
 
@@ -52,22 +52,14 @@ fn pre_002_manifest_keeps_forbidden_layers_and_live_sources_out() {
 }
 
 #[test]
-fn pre_004_source_surface_opens_lifecycle_without_supervisor_admission_or_live_sources() {
+fn pre_005_source_surface_owns_private_supervisor_without_admission_persistence_or_live_sources() {
     let root = include_str!("../src/lib.rs");
     let runtime = include_str!("../src/runtime.rs");
-    for required in [
-        "mod runtime;",
-        "pub use self::runtime::RawTransactionIngestHandle;",
-        "pub use self::runtime::RawTransactionIngestTerminalFuture;",
-        "pub use self::runtime::RawTransactionIngestWorker;",
-        "pub fn start(",
-        "pub fn request_stop(&self) -> bool",
-        "pub fn wait_terminal(&self)",
-    ] {
-        assert!(root.contains(required) || runtime.contains(required), "required pre.004 runtime contract missing: {required}");
+    for required in ["tokio::task::JoinSet", "run_supervisor", "supervise_until_stop", "drain_children", "start_foundation_with_source_spawner"] {
+        assert!(runtime.contains(required), "required pre.005 private supervisor contract missing: {required}");
     }
     for forbidden in [
-        "JoinSet",
+        "pub use self::runtime::JoinSet",
         "mpsc::",
         "canonicalize_raw_transaction",
         "persist_raw_transaction",
@@ -75,7 +67,7 @@ fn pre_004_source_surface_opens_lifecycle_without_supervisor_admission_or_live_s
         "RawTransactionIngestSnapshot",
         "WorkerSnapshotSource",
     ] {
-        assert!(!root.contains(forbidden) && !runtime.contains(forbidden), "pre.004 opened later runtime scope too early: {forbidden}");
+        assert!(!root.contains(forbidden) && !runtime.contains(forbidden), "pre.005 opened later runtime scope too early: {forbidden}");
     }
     return;
 }

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

@@ -1,5 +1,23 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs
-// version: 1
+// version: 2
+
+struct ActiveTaskGuard {
+    active: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+}
+
+impl ActiveTaskGuard {
+    fn new(active: std::sync::Arc<std::sync::atomic::AtomicUsize>) -> Self {
+        active.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
+        return Self { active };
+    }
+}
+
+impl std::ops::Drop for ActiveTaskGuard {
+    fn drop(&mut self) {
+        self.active.fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
+        return;
+    }
+}
 
 fn settings(network: &str) -> std::option::Option<crate::RawTransactionIngestSettings> {
     let network = match ksp_store_lib::RawNetworkId::new(network) {
@@ -13,6 +31,16 @@ fn settings(network: &str) -> std::option::Option<crate::RawTransactionIngestSet
     return std::option::Option::Some(crate::RawTransactionIngestSettings::with_defaults(network, worker_id));
 }
 
+async fn wait_for_active_count(active: &std::sync::Arc<std::sync::atomic::AtomicUsize>, expected: usize) -> bool {
+    for _ in 0..64 {
+        if active.load(std::sync::atomic::Ordering::Acquire) == expected {
+            return true;
+        }
+        tokio::task::yield_now().await;
+    }
+    return active.load(std::sync::atomic::Ordering::Acquire) == expected;
+}
+
 #[test]
 fn pre_004_current_runtime_is_required_before_spawn() {
     let result = super::current_runtime_handle();
@@ -125,3 +153,101 @@ async fn pre_004_dropping_last_control_handle_causes_private_runtime_exit() {
         }
     }
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn pre_005_supervisor_joins_all_cooperative_source_tasks_before_terminal() {
+    let settings = match settings("mainnet") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    let source_active = std::sync::Arc::clone(&active);
+    let handle = match super::start_foundation_with_source_spawner(
+        settings,
+        tokio::runtime::Handle::current(),
+        std::option::Option::None,
+        move |children, stop_receiver| {
+            for _ in 0..3 {
+                let active = std::sync::Arc::clone(&source_active);
+                let mut child_stop = stop_receiver.clone();
+                let _abort_handle = children.spawn(async move {
+                    let _guard = ActiveTaskGuard::new(active);
+                    loop {
+                        let changed = child_stop.changed().await;
+                        if changed.is_err() || *child_stop.borrow() {
+                            return;
+                        }
+                    }
+                });
+            }
+        },
+    ) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    assert!(wait_for_active_count(&active, 3).await);
+    assert!(handle.request_stop());
+    let terminal = handle.wait_terminal().await;
+    let terminal = match terminal {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    assert_eq!(terminal, ksp_worker_api::WorkerState::Stopped);
+    assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 0);
+    return;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn pre_005_supervisor_reaps_completed_source_task_and_still_joins_live_child_on_stop() {
+    let settings = match settings("mainnet") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => return,
+    };
+    let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    let completed = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    let source_active = std::sync::Arc::clone(&active);
+    let source_completed = std::sync::Arc::clone(&completed);
+    let handle = match super::start_foundation_with_source_spawner(
+        settings,
+        tokio::runtime::Handle::current(),
+        std::option::Option::None,
+        move |children, stop_receiver| {
+            let completed = std::sync::Arc::clone(&source_completed);
+            let _completed_abort_handle = children.spawn(async move {
+                completed.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
+                return;
+            });
+            let active = std::sync::Arc::clone(&source_active);
+            let mut child_stop = stop_receiver.clone();
+            let _active_abort_handle = children.spawn(async move {
+                let _guard = ActiveTaskGuard::new(active);
+                loop {
+                    let changed = child_stop.changed().await;
+                    if changed.is_err() || *child_stop.borrow() {
+                        return;
+                    }
+                }
+            });
+        },
+    ) {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    assert!(wait_for_active_count(&active, 1).await);
+    for _ in 0..64 {
+        if completed.load(std::sync::atomic::Ordering::Acquire) == 1 {
+            break;
+        }
+        tokio::task::yield_now().await;
+    }
+    assert_eq!(completed.load(std::sync::atomic::Ordering::Acquire), 1);
+    assert!(handle.request_stop());
+    let terminal = handle.wait_terminal().await;
+    let terminal = match terminal {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return,
+    };
+    assert_eq!(terminal, ksp_worker_api::WorkerState::Stopped);
+    assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 0);
+    return;
+}