v0.3.11-pre.007

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

@@ -1,5 +1,8 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs
-// version: 3
+// version: 4
+
+type PersistencePort = std::sync::Arc<dyn crate::RawTransactionIngestPersistencePort + 'static>;
+type PersistenceTasks = tokio::task::JoinSet<ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>>;
 
 /// Runtime-neutral boxed future resolving after one RAW transaction ingest Worker has fully reached a terminal lifecycle state.
 pub type RawTransactionIngestTerminalFuture<'a> =
@@ -85,31 +88,74 @@ fn current_runtime_handle() -> ksp_core_lib::Result<tokio::runtime::Handle> {
     };
 }
 
-async fn drain_admission(settings: &crate::RawTransactionIngestSettings, admission: &mut crate::RawTransactionAdmission) -> bool {
-    let mut clean = true;
+async fn drain_children(children: &mut tokio::task::JoinSet<()>) -> std::option::Option<ksp_core_lib::ErrorCode> {
+    let mut fault = std::option::Option::None;
+    while let std::option::Option::Some(joined) = children.join_next().await {
+        if joined.is_err() && fault.is_none() {
+            fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
+        }
+    }
+    return fault;
+}
+
+async fn drain_persistence(persistence: &mut PersistenceTasks) -> std::option::Option<ksp_core_lib::ErrorCode> {
+    let mut fault = std::option::Option::None;
+    while let std::option::Option::Some(joined) = persistence.join_next().await {
+        let current = persistence_fault(joined);
+        if fault.is_none() {
+            fault = current;
+        }
+    }
+    return fault;
+}
+
+async fn drain_admission_and_persistence(
+    settings: &crate::RawTransactionIngestSettings,
+    admission: &mut crate::RawTransactionAdmission,
+    persistence: &mut PersistenceTasks,
+    port: &std::option::Option<PersistencePort>,
+) -> std::option::Option<ksp_core_lib::ErrorCode> {
+    let mut fault = std::option::Option::None;
     admission.close();
     loop {
+        while persistence.len() >= settings.persistence_concurrency() {
+            let joined = persistence.join_next().await;
+            let current = match joined {
+                std::option::Option::Some(value) => persistence_fault(value),
+                std::option::Option::None => std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID),
+            };
+            if fault.is_none() {
+                fault = current;
+            }
+        }
         let received = admission.receive(settings.network()).await;
         match received {
-            std::result::Result::Ok(std::option::Option::Some(_acquisition)) => {},
-            std::result::Result::Ok(std::option::Option::None) => return clean,
-            std::result::Result::Err(_) => clean = false,
+            std::result::Result::Ok(std::option::Option::Some(acquisition)) => {
+                if !spawn_persistence(persistence, port, acquisition) && fault.is_none() {
+                    fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
+                }
+            },
+            std::result::Result::Ok(std::option::Option::None) => break,
+            std::result::Result::Err(_) => {
+                if fault.is_none() {
+                    fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
+                }
+            },
         }
     }
-}
-
-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;
-        }
+    let persistence_fault = drain_persistence(persistence).await;
+    if fault.is_none() {
+        fault = persistence_fault;
     }
-    return clean;
+    return fault;
 }
 
-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() {
+fn finish_faulted(
+    lifecycle: &mut ksp_worker_api::WorkerLifecycle,
+    sender: &tokio::sync::watch::Sender<ksp_worker_api::WorkerState>,
+    code: ksp_core_lib::ErrorCode,
+) {
+    if lifecycle.fault(code).is_err() {
         sender.send_replace(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
         return;
     }
@@ -131,10 +177,34 @@ fn finish_stopped(lifecycle: &mut ksp_worker_api::WorkerLifecycle, sender: &toki
     return;
 }
 
+fn merge_fault(
+    first: std::option::Option<ksp_core_lib::ErrorCode>,
+    next: std::option::Option<ksp_core_lib::ErrorCode>,
+) -> std::option::Option<ksp_core_lib::ErrorCode> {
+    if first.is_some() {
+        return first;
+    }
+    return next;
+}
+
+fn persistence_fault(
+    joined: std::result::Result<ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>, tokio::task::JoinError>,
+) -> std::option::Option<ksp_core_lib::ErrorCode> {
+    return match joined {
+        std::result::Result::Ok(std::result::Result::Ok(outcome)) => {
+            let _entity = outcome.entity();
+            let _observation = outcome.observation();
+            std::option::Option::None
+        },
+        std::result::Result::Ok(std::result::Result::Err(error)) => std::option::Option::Some(error.code()),
+        std::result::Result::Err(_) => std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID),
+    };
+}
+
 async fn run_supervisor<Spawner>(
     settings: crate::RawTransactionIngestSettings,
     mut lifecycle: ksp_worker_api::WorkerLifecycle,
-    _store_guard: std::option::Option<std::sync::Arc<ksp_store_lib::Store>>,
+    port: std::option::Option<PersistencePort>,
     mut stop_receiver: tokio::sync::watch::Receiver<bool>,
     terminal_sender: tokio::sync::watch::Sender<ksp_worker_api::WorkerState>,
     source_spawner: Spawner,
@@ -153,19 +223,40 @@ async fn run_supervisor<Spawner>(
     }
     terminal_sender.send_replace(lifecycle.state());
     let mut children = tokio::task::JoinSet::new();
+    let mut persistence = PersistenceTasks::new();
+    let (source_stop_sender, source_stop_receiver) = tokio::sync::watch::channel(false);
     let (mut admission, admission_sender) = crate::RawTransactionAdmission::new(settings.admission_queue_capacity());
-    source_spawner(&mut children, stop_receiver.clone(), admission_sender);
-    let supervised_clean = supervise_until_stop(&settings, &mut stop_receiver, &mut children, &mut admission).await;
-    let admission_clean = drain_admission(&settings, &mut admission).await;
-    let children_clean = drain_children(&mut children).await;
-    if !supervised_clean || !admission_clean || !children_clean {
-        finish_faulted(&mut lifecycle, &terminal_sender);
-        return;
+    source_spawner(&mut children, source_stop_receiver, admission_sender);
+    let mut fault = supervise_until_stop(&settings, &source_stop_sender, &mut stop_receiver, &mut children, &mut admission, &mut persistence, &port).await;
+    if fault.is_some() {
+        source_stop_sender.send_replace(true);
+    }
+    let drain_fault = drain_admission_and_persistence(&settings, &mut admission, &mut persistence, &port).await;
+    fault = merge_fault(fault, drain_fault);
+    let child_fault = drain_children(&mut children).await;
+    fault = merge_fault(fault, child_fault);
+    match fault {
+        std::option::Option::Some(code) => finish_faulted(&mut lifecycle, &terminal_sender, code),
+        std::option::Option::None => finish_stopped(&mut lifecycle, &terminal_sender),
     }
-    finish_stopped(&mut lifecycle, &terminal_sender);
     return;
 }
 
+fn spawn_persistence(
+    persistence: &mut PersistenceTasks,
+    port: &std::option::Option<PersistencePort>,
+    acquisition: ksp_raw_transaction_lib::RawTransactionAcquisition,
+) -> bool {
+    let port = match port {
+        std::option::Option::Some(value) => std::sync::Arc::clone(value),
+        std::option::Option::None => return false,
+    };
+    let _abort_handle = persistence.spawn(async move {
+        return crate::persist_raw_transaction_ingest_acquisition(port.as_ref(), acquisition).await;
+    });
+    return true;
+}
+
 fn start_foundation(
     settings: crate::RawTransactionIngestSettings,
     runtime: tokio::runtime::Handle,
@@ -174,10 +265,10 @@ fn start_foundation(
     return start_foundation_with_source_spawner(settings, runtime, store_guard, |_children, _stop_receiver, _admission_sender| {});
 }
 
-fn start_foundation_with_source_spawner<Spawner>(
+fn start_foundation_with_port_and_source_spawner<Spawner>(
     settings: crate::RawTransactionIngestSettings,
     runtime: tokio::runtime::Handle,
-    store_guard: std::option::Option<std::sync::Arc<ksp_store_lib::Store>>,
+    port: std::option::Option<PersistencePort>,
     source_spawner: Spawner,
 ) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle>
 where
@@ -197,25 +288,50 @@ where
     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_supervisor(settings, lifecycle, store_guard, stop_receiver, terminal_sender, source_spawner)));
+    std::mem::drop(runtime.spawn(run_supervisor(settings, lifecycle, port, stop_receiver, terminal_sender, source_spawner)));
     return std::result::Result::Ok(handle);
 }
 
+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>, tokio::sync::mpsc::Sender<crate::RawTransactionIngress>)
+        + std::marker::Send
+        + 'static,
+{
+    let port = match store_guard {
+        std::option::Option::Some(store) => {
+            let port: PersistencePort = store;
+            std::option::Option::Some(port)
+        },
+        std::option::Option::None => std::option::Option::None,
+    };
+    return start_foundation_with_port_and_source_spawner(settings, runtime, port, source_spawner);
+}
+
 async fn supervise_until_stop(
     settings: &crate::RawTransactionIngestSettings,
+    source_stop_sender: &tokio::sync::watch::Sender<bool>,
     stop_receiver: &mut tokio::sync::watch::Receiver<bool>,
     children: &mut tokio::task::JoinSet<()>,
     admission: &mut crate::RawTransactionAdmission,
-) -> bool {
+    persistence: &mut PersistenceTasks,
+    port: &std::option::Option<PersistencePort>,
+) -> std::option::Option<ksp_core_lib::ErrorCode> {
     let mut admission_open = true;
-    let mut clean = true;
     loop {
         if *stop_receiver.borrow() {
+            source_stop_sender.send_replace(true);
             break;
         }
-        if children.is_empty() && !admission_open {
+        if children.is_empty() && !admission_open && persistence.is_empty() {
             let changed = stop_receiver.changed().await;
             if changed.is_err() || *stop_receiver.borrow() {
+                source_stop_sender.send_replace(true);
                 break;
             }
             continue;
@@ -224,24 +340,43 @@ async fn supervise_until_stop(
             biased;
             changed = stop_receiver.changed() => {
                 if changed.is_err() || *stop_receiver.borrow() {
+                    source_stop_sender.send_replace(true);
                     break;
                 }
             }
             joined = children.join_next(), if !children.is_empty() => {
                 if let std::option::Option::Some(std::result::Result::Err(_)) = joined {
-                    clean = false;
+                    source_stop_sender.send_replace(true);
+                    return std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
                 }
             }
-            received = admission.receive(settings.network()), if admission_open => {
+            joined = persistence.join_next(), if !persistence.is_empty() => {
+                if let std::option::Option::Some(value) = joined {
+                    let fault = persistence_fault(value);
+                    if fault.is_some() {
+                        source_stop_sender.send_replace(true);
+                        return fault;
+                    }
+                }
+            }
+            received = admission.receive(settings.network()), if admission_open && persistence.len() < settings.persistence_concurrency() => {
                 match received {
-                    std::result::Result::Ok(std::option::Option::Some(_acquisition)) => {},
+                    std::result::Result::Ok(std::option::Option::Some(acquisition)) => {
+                        if !spawn_persistence(persistence, port, acquisition) {
+                            source_stop_sender.send_replace(true);
+                            return std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
+                        }
+                    },
                     std::result::Result::Ok(std::option::Option::None) => admission_open = false,
-                    std::result::Result::Err(_) => clean = false,
+                    std::result::Result::Err(_) => {
+                        source_stop_sender.send_replace(true);
+                        return std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
+                    },
                 }
             }
         }
     }
-    return clean;
+    return std::option::Option::None;
 }
 
 fn validate_store_network(settings: &crate::RawTransactionIngestSettings, store_network: &ksp_store_lib::RawNetworkId) -> ksp_core_lib::Result<()> {