v0.3.11-pre.006

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

@@ -0,0 +1,93 @@
+// file: crates/ksp-worker-raw-transaction-ingest-lib/src/admission.rs
+// version: 1
+
+use sha2::Digest; // rust-rules: trait-import
+
+const RAW_TRANSACTION_INGEST_OBSERVATION_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.observation.v1\0";
+
+/// Crate-private source-neutral ingress admitted by the central bounded Worker queue.
+pub(crate) struct RawTransactionIngress {
+    /// Complete Common RAW material supplied by one private source task.
+    pub(crate) material: ksp_raw_transaction_lib::RawTransactionMaterial,
+    /// Logical network expected to match both Worker settings and canonical material.
+    pub(crate) network: ksp_store_lib::RawNetworkId,
+    /// Safe source-independent provenance attached to the acquisition observation.
+    pub(crate) provenance: ksp_store_lib::RawAcquisitionProvenance,
+    /// Opaque deterministic source-owned key material used only for Worker observation-key derivation.
+    pub(crate) source_key: [u8; 32],
+}
+
+/// Receiver side of the bounded central RAW transaction admission queue owned by the Worker supervisor.
+pub(crate) struct RawTransactionAdmission {
+    receiver: tokio::sync::mpsc::Receiver<crate::RawTransactionIngress>,
+}
+
+impl crate::RawTransactionAdmission {
+    /// Creates one bounded admission queue and returns its crate-private source sender.
+    #[must_use]
+    pub(crate) fn new(capacity: usize) -> (crate::RawTransactionAdmission, tokio::sync::mpsc::Sender<crate::RawTransactionIngress>) {
+        let (sender, receiver) = tokio::sync::mpsc::channel(capacity);
+        return (Self { receiver }, sender);
+    }
+
+    /// Closes new admissions while preserving already queued ingress for deterministic drain.
+    pub(crate) fn close(&mut self) {
+        self.receiver.close();
+        return;
+    }
+
+    /// Receives and converts one queued ingress into a common RAW transaction acquisition.
+    pub(crate) async fn receive(
+        &mut self,
+        expected_network: &ksp_store_lib::RawNetworkId,
+    ) -> ksp_core_lib::Result<std::option::Option<ksp_raw_transaction_lib::RawTransactionAcquisition>> {
+        let ingress = match self.receiver.recv().await {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
+        };
+        let acquisition = canonicalize_ingress(expected_network, ingress);
+        return match acquisition {
+            std::result::Result::Ok(value) => std::result::Result::Ok(std::option::Option::Some(value)),
+            std::result::Result::Err(error) => std::result::Result::Err(error),
+        };
+    }
+}
+
+fn canonicalize_ingress(
+    expected_network: &ksp_store_lib::RawNetworkId,
+    ingress: crate::RawTransactionIngress,
+) -> ksp_core_lib::Result<ksp_raw_transaction_lib::RawTransactionAcquisition> {
+    if &ingress.network != expected_network {
+        return std::result::Result::Err(crate::runtime_error("admission.network_mismatch"));
+    }
+    let transaction = ksp_raw_transaction_lib::canonicalize_raw_transaction(ingress.material);
+    let transaction = match transaction {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("admission.canonicalization_invalid")),
+    };
+    if transaction.reference().network() != expected_network {
+        return std::result::Result::Err(crate::runtime_error("admission.material_network_mismatch"));
+    }
+    let observation_key = observation_key(transaction.reference(), &ingress.source_key);
+    return std::result::Result::Ok(ksp_raw_transaction_lib::assemble_raw_transaction_acquisition(transaction, observation_key, ingress.provenance));
+}
+
+fn hash_bytes(hasher: &mut sha2::Sha256, value: &[u8]) {
+    hasher.update((value.len() as u64).to_be_bytes());
+    hasher.update(value);
+    return;
+}
+
+fn observation_key(reference: &ksp_store_lib::RawTransactionReference, source_key: &[u8; 32]) -> ksp_store_lib::RawObservationKey {
+    let mut hasher = sha2::Sha256::new();
+    hasher.update(RAW_TRANSACTION_INGEST_OBSERVATION_DOMAIN);
+    hash_bytes(&mut hasher, reference.network().as_str().as_bytes());
+    hash_bytes(&mut hasher, reference.signature().as_bytes());
+    hash_bytes(&mut hasher, source_key);
+    let bytes: [u8; 32] = hasher.finalize().into();
+    return ksp_store_lib::RawObservationKey::new(bytes);
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/admission.rs"]
+mod tests;

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: 4
+// version: 5
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -8,10 +8,11 @@
 //! Source-neutral runtime foundation for continuous KSP RAW transaction ingestion.
 //!
 //! This tranche owns the concrete Worker family identity, validated technical settings
-//! 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.
+//! and the caller-runtime-owned lifecycle with private child-task supervision. This tranche also
+//! owns bounded source-neutral admission plus common RAW canonicalization/assembly; persistence and
+//! latest-value snapshots remain in later prereleases, and no live source or Transport dependency exists.
 
+mod admission;
 mod error;
 mod identity;
 mod runtime;
@@ -50,6 +51,10 @@ pub use self::settings::MIN_RAW_TRANSACTION_INGEST_SHUTDOWN_DRAIN_TIMEOUT;
 /// Validated source-neutral runtime settings for one continuous RAW transaction ingest Worker.
 pub use self::settings::RawTransactionIngestSettings;
 
+/// Receiver-side owner of the private bounded RAW transaction admission queue.
+pub(crate) use self::admission::RawTransactionAdmission;
+/// Crate-private source-neutral ingress sent through the bounded central admission queue.
+pub(crate) use self::admission::RawTransactionIngress;
 /// Creates one runtime-domain error without copying runtime/provider/Store values into diagnostics.
 pub(crate) use self::error::runtime_error;
 /// Creates one settings-domain error without copying caller-supplied values into diagnostics.

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: 2
+// version: 3
 
 /// 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,19 @@ 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;
+    admission.close();
+    loop {
+        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,
+        }
+    }
+}
+
 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 {
@@ -119,13 +132,16 @@ fn finish_stopped(lifecycle: &mut ksp_worker_api::WorkerLifecycle, sender: &toki
 }
 
 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>>,
     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,
+    Spawner: FnOnce(&mut tokio::task::JoinSet<()>, tokio::sync::watch::Receiver<bool>, tokio::sync::mpsc::Sender<crate::RawTransactionIngress>)
+        + std::marker::Send
+        + 'static,
 {
     if *stop_receiver.borrow() {
         finish_stopped(&mut lifecycle, &terminal_sender);
@@ -137,9 +153,12 @@ async fn run_supervisor<Spawner>(
     }
     terminal_sender.send_replace(lifecycle.state());
     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 {
+    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;
     }
@@ -152,7 +171,7 @@ 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| {});
+    return start_foundation_with_source_spawner(settings, runtime, store_guard, |_children, _stop_receiver, _admission_sender| {});
 }
 
 fn start_foundation_with_source_spawner<Spawner>(
@@ -162,7 +181,9 @@ fn start_foundation_with_source_spawner<Spawner>(
     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,
+    Spawner: FnOnce(&mut tokio::task::JoinSet<()>, tokio::sync::watch::Receiver<bool>, tokio::sync::mpsc::Sender<crate::RawTransactionIngress>)
+        + 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,
@@ -176,17 +197,23 @@ 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(lifecycle, store_guard, stop_receiver, terminal_sender, source_spawner)));
+    std::mem::drop(runtime.spawn(run_supervisor(settings, 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 {
+async fn supervise_until_stop(
+    settings: &crate::RawTransactionIngestSettings,
+    stop_receiver: &mut tokio::sync::watch::Receiver<bool>,
+    children: &mut tokio::task::JoinSet<()>,
+    admission: &mut crate::RawTransactionAdmission,
+) -> bool {
+    let mut admission_open = true;
     let mut clean = true;
     loop {
         if *stop_receiver.borrow() {
             break;
         }
-        if children.is_empty() {
+        if children.is_empty() && !admission_open {
             let changed = stop_receiver.changed().await;
             if changed.is_err() || *stop_receiver.borrow() {
                 break;
@@ -200,15 +227,21 @@ async fn supervise_until_stop(stop_receiver: &mut tokio::sync::watch::Receiver<b
                     break;
                 }
             }
-            joined = children.join_next() => {
+            joined = children.join_next(), if !children.is_empty() => {
                 if let std::option::Option::Some(std::result::Result::Err(_)) = joined {
                     clean = false;
                 }
             }
+            received = admission.receive(settings.network()), if admission_open => {
+                match received {
+                    std::result::Result::Ok(std::option::Option::Some(_acquisition)) => {},
+                    std::result::Result::Ok(std::option::Option::None) => admission_open = false,
+                    std::result::Result::Err(_) => clean = false,
+                }
+            }
         }
     }
-    let drained_clean = drain_children(children).await;
-    return clean && drained_clean;
+    return clean;
 }
 
 fn validate_store_network(settings: &crate::RawTransactionIngestSettings, store_network: &ksp_store_lib::RawNetworkId) -> ksp_core_lib::Result<()> {