v0.3.11-pre.004

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

@@ -0,0 +1,157 @@
+// file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs
+// version: 1
+
+/// Runtime-neutral boxed future resolving after one RAW transaction ingest Worker has fully reached a terminal lifecycle state.
+pub type RawTransactionIngestTerminalFuture<'a> =
+    std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = ksp_core_lib::Result<ksp_worker_api::WorkerState>> + std::marker::Send + 'a>>;
+
+/// Cloneable external control handle for one continuous RAW transaction ingest Worker.
+#[derive(Clone)]
+pub struct RawTransactionIngestHandle {
+    stop_sender: tokio::sync::watch::Sender<bool>,
+    stop_token: ksp_worker_api::WorkerStopToken,
+    terminal_receiver: tokio::sync::watch::Receiver<ksp_worker_api::WorkerState>,
+}
+
+impl crate::RawTransactionIngestHandle {
+    /// Requests cooperative stop and returns `true` only for the first request accepted by the live runtime.
+    #[must_use]
+    pub fn request_stop(&self) -> bool {
+        if !self.stop_token.request_stop() {
+            return false;
+        }
+        return self.stop_sender.send(true).is_ok();
+    }
+
+    /// Waits until the private runtime task has published and closed one terminal lifecycle state.
+    #[must_use]
+    pub fn wait_terminal(&self) -> crate::RawTransactionIngestTerminalFuture<'_> {
+        let mut receiver = self.terminal_receiver.clone();
+        return std::boxed::Box::pin(async move {
+            loop {
+                let current = *receiver.borrow();
+                if current.is_terminal() {
+                    let changed = receiver.changed().await;
+                    if changed.is_err() {
+                        return std::result::Result::Ok(current);
+                    }
+                    return std::result::Result::Err(crate::runtime_error("terminal.changed_after_terminal"));
+                }
+                let changed = receiver.changed().await;
+                if changed.is_err() {
+                    return std::result::Result::Err(crate::runtime_error("terminal.closed_before_terminal"));
+                }
+            }
+        });
+    }
+}
+
+impl std::fmt::Debug for crate::RawTransactionIngestHandle {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let state = *self.terminal_receiver.borrow();
+        return formatter
+            .debug_struct("RawTransactionIngestHandle")
+            .field("stop_requested", &self.stop_token.is_stop_requested())
+            .field("state", &state)
+            .finish();
+    }
+}
+
+/// Entry point owning synchronous validation and task launch for one RAW transaction ingest Worker run.
+pub struct RawTransactionIngestWorker;
+
+impl crate::RawTransactionIngestWorker {
+    /// Starts one Worker on the caller-owned current Tokio runtime while retaining the caller-owned Store facade through an `Arc`.
+    pub fn start(
+        settings: crate::RawTransactionIngestSettings,
+        store: std::sync::Arc<ksp_store_lib::Store>,
+    ) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle> {
+        let runtime = match current_runtime_handle() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let store_snapshot = store.runtime_snapshot();
+        if let std::result::Result::Err(error) = validate_store_network(&settings, store_snapshot.network()) {
+            return std::result::Result::Err(error);
+        }
+        return start_foundation(settings, runtime, std::option::Option::Some(store));
+    }
+}
+
+fn current_runtime_handle() -> ksp_core_lib::Result<tokio::runtime::Handle> {
+    return match tokio::runtime::Handle::try_current() {
+        std::result::Result::Ok(value) => std::result::Result::Ok(value),
+        std::result::Result::Err(_) => std::result::Result::Err(crate::runtime_error("start.runtime_unavailable")),
+    };
+}
+
+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));
+        return;
+    }
+    sender.send_replace(lifecycle.state());
+    if lifecycle.mark_stopped().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;
+}
+
+async fn run_foundation(
+    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>,
+) {
+    if *stop_receiver.borrow() {
+        finish_stopped(&mut lifecycle, &terminal_sender);
+        return;
+    }
+    if lifecycle.mark_running().is_err() {
+        terminal_sender.send_replace(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
+        return;
+    }
+    terminal_sender.send_replace(lifecycle.state());
+    loop {
+        let changed = stop_receiver.changed().await;
+        if changed.is_err() || *stop_receiver.borrow() {
+            break;
+        }
+    }
+    finish_stopped(&mut lifecycle, &terminal_sender);
+    return;
+}
+
+fn start_foundation(
+    settings: crate::RawTransactionIngestSettings,
+    runtime: tokio::runtime::Handle,
+    store_guard: std::option::Option<std::sync::Arc<ksp_store_lib::Store>>,
+) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle> {
+    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")),
+    };
+    let mut lifecycle = ksp_worker_api::WorkerLifecycle::new(settings.worker_id().clone(), kind);
+    if lifecycle.start().is_err() {
+        return std::result::Result::Err(crate::runtime_error("start.lifecycle_invalid"));
+    }
+    let stop_token = ksp_worker_api::WorkerStopToken::new();
+    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)));
+    return std::result::Result::Ok(handle);
+}
+
+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"));
+    }
+    return std::result::Result::Ok(());
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/runtime.rs"]
+mod tests;