v0.3.11-pre.008

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/admission.rs
-// version: 1
+// version: 2
 
 use sha2::Digest; // rust-rules: trait-import
 
@@ -30,6 +30,12 @@ impl crate::RawTransactionAdmission {
         return (Self { receiver }, sender);
     }
 
+    /// Returns the latest receiver-side queue depth without consuming an ingress entry.
+    #[must_use]
+    pub(crate) fn queue_depth(&self) -> usize {
+        return self.receiver.len();
+    }
+
     /// Closes new admissions while preserving already queued ingress for deterministic drain.
     pub(crate) fn close(&mut self) {
         self.receiver.close();

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

@@ -1,9 +1,12 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/error.rs
-// version: 4
+// version: 5
 
 /// Error code used when durable Store content conflicts with one admitted canonical RAW transaction.
 pub const ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT: ksp_core_lib::ErrorCode =
     ksp_core_lib::ErrorCode::new("worker_raw_transaction_ingest", "content_conflict");
+/// Error code used when one monotone RAW transaction ingest Worker counter or snapshot sequence cannot advance without wrapping.
+pub const ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED: ksp_core_lib::ErrorCode =
+    ksp_core_lib::ErrorCode::new("worker_raw_transaction_ingest", "counter_exhausted");
 /// Error code used when the RAW transaction ingest Worker reaches an invalid runtime or lifecycle condition.
 pub const ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID: ksp_core_lib::ErrorCode =
     ksp_core_lib::ErrorCode::new("worker_raw_transaction_ingest", "runtime_invalid");
@@ -14,6 +17,12 @@ pub const ERROR_CODE_RAW_TRANSACTION_INGEST_SETTINGS_INVALID: ksp_core_lib::Erro
 pub const ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED: ksp_core_lib::ErrorCode =
     ksp_core_lib::ErrorCode::new("worker_raw_transaction_ingest", "store_failed");
 
+/// Creates one terminal counter-exhaustion error without exposing runtime material.
+pub(crate) fn counter_exhausted_error(field: &'static str) -> ksp_core_lib::Error {
+    return ksp_core_lib::Error::new(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED, "RAW transaction ingest Worker counter exhausted")
+        .with_context("field", field);
+}
+
 /// Creates one terminal content-conflict error without copying conflicting material into diagnostics.
 pub(crate) fn content_conflict_error() -> ksp_core_lib::Error {
     return ksp_core_lib::Error::new(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT, "RAW transaction ingest Store content conflict");

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: 6
+// version: 7
 
 #![warn(missing_docs)]
 #![deny(unreachable_pub)]
@@ -10,8 +10,8 @@
 //! This tranche owns the concrete Worker family identity, validated technical settings
 //! and the caller-runtime-owned lifecycle with private child-task supervision. This tranche also
 //! owns bounded source-neutral admission, common RAW canonicalization/assembly and backend-neutral
-//! Store persistence in normal mode; latest-value snapshots remain in a later prerelease, and no live
-//! source or Transport dependency exists.
+//! Store persistence in normal mode plus concrete latest-value snapshots projected onto Worker API;
+//! no live source or Transport dependency exists.
 
 mod admission;
 mod error;
@@ -19,9 +19,12 @@ mod identity;
 mod persistence;
 mod runtime;
 mod settings;
+mod snapshot;
 
 /// Error code used when durable Store content conflicts with one admitted canonical RAW transaction.
 pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT;
+/// Error code used when one monotone Worker counter or snapshot sequence cannot advance without wrapping.
+pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED;
 /// Error code used when the RAW transaction ingest Worker reaches an invalid runtime or lifecycle condition.
 pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID;
 /// Error code used when RAW transaction ingest Worker settings violate one bounded runtime invariant.
@@ -56,6 +59,12 @@ pub use self::settings::MIN_RAW_TRANSACTION_INGEST_PERSISTENCE_CONCURRENCY;
 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;
+/// Complete safe latest-value snapshot of one continuous RAW transaction ingest Worker.
+pub use self::snapshot::RawTransactionIngestSnapshot;
+/// Runtime-neutral boxed future resolving to one newer concrete RAW transaction ingest Worker snapshot.
+pub use self::snapshot::RawTransactionIngestSnapshotFuture;
+/// Cloneable latest-value source exposing concrete and common Worker snapshots from one shared watch state.
+pub use self::snapshot::RawTransactionIngestSnapshotSource;
 
 /// Receiver-side owner of the private bounded RAW transaction admission queue.
 pub(crate) use self::admission::RawTransactionAdmission;
@@ -63,6 +72,8 @@ pub(crate) use self::admission::RawTransactionAdmission;
 pub(crate) use self::admission::RawTransactionIngress;
 /// Creates one terminal content-conflict error without copying conflicting material into diagnostics.
 pub(crate) use self::error::content_conflict_error;
+/// Creates one terminal counter-exhaustion error without exposing runtime material.
+pub(crate) use self::error::counter_exhausted_error;
 /// 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.
@@ -79,3 +90,5 @@ pub(crate) use self::persistence::RawTransactionIngestPersistenceOutcome;
 pub(crate) use self::persistence::RawTransactionIngestPersistencePort;
 /// Persists one already-canonical Worker acquisition through the private Store port in `Normal` mode.
 pub(crate) use self::persistence::persist_raw_transaction_ingest_acquisition;
+/// Private latest-value publisher and checked counter owner shared by the Worker supervisor.
+pub(crate) use self::snapshot::RawTransactionIngestSnapshotPublisher;

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: 4
+// version: 5
 
 type PersistencePort = std::sync::Arc<dyn crate::RawTransactionIngestPersistencePort + 'static>;
 type PersistenceTasks = tokio::task::JoinSet<ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>>;
@@ -11,9 +11,9 @@ pub type RawTransactionIngestTerminalFuture<'a> =
 /// Cloneable external control handle for one continuous RAW transaction ingest Worker.
 #[derive(Clone)]
 pub struct RawTransactionIngestHandle {
+    snapshots: crate::RawTransactionIngestSnapshotSource,
     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 {
@@ -26,22 +26,36 @@ impl crate::RawTransactionIngestHandle {
         return self.stop_sender.send(true).is_ok();
     }
 
-    /// Waits until the private runtime task has published and closed one terminal lifecycle state.
+    /// Returns an independent latest-value source for concrete RAW transaction ingest Worker snapshots.
+    #[must_use]
+    pub fn snapshot_source(&self) -> crate::RawTransactionIngestSnapshotSource {
+        return self.snapshots.clone();
+    }
+
+    /// Returns an independent source implementing the common [`ksp_worker_api::WorkerSnapshotSource`] projection.
+    #[must_use]
+    pub fn worker_snapshot_source(&self) -> crate::RawTransactionIngestSnapshotSource {
+        return self.snapshots.clone();
+    }
+
+    /// Waits until the private runtime task has published one terminal lifecycle state after draining owned work.
     #[must_use]
     pub fn wait_terminal(&self) -> crate::RawTransactionIngestTerminalFuture<'_> {
-        let mut receiver = self.terminal_receiver.clone();
+        let source = self.snapshots.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 current = source.current();
+                let state = current.worker_snapshot().state();
+                if state.is_terminal() {
+                    return std::result::Result::Ok(state);
                 }
-                let changed = receiver.changed().await;
-                if changed.is_err() {
+                let observed = current.worker_snapshot().sequence();
+                let changed = source.wait_for_change(observed).await;
+                let state = changed.worker_snapshot().state();
+                if state.is_terminal() {
+                    return std::result::Result::Ok(state);
+                }
+                if changed.worker_snapshot().sequence() == observed && source.is_closed() {
                     return std::result::Result::Err(crate::runtime_error("terminal.closed_before_terminal"));
                 }
             }
@@ -51,11 +65,12 @@ impl crate::RawTransactionIngestHandle {
 
 impl std::fmt::Debug for crate::RawTransactionIngestHandle {
     fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let state = *self.terminal_receiver.borrow();
+        let snapshot = self.snapshots.current();
         return formatter
             .debug_struct("RawTransactionIngestHandle")
             .field("stop_requested", &self.stop_token.is_stop_requested())
-            .field("state", &state)
+            .field("sequence", &snapshot.worker_snapshot().sequence())
+            .field("state", &snapshot.worker_snapshot().state())
             .finish();
     }
 }
@@ -81,6 +96,21 @@ impl crate::RawTransactionIngestWorker {
     }
 }
 
+fn begin_stopping(
+    lifecycle: &mut ksp_worker_api::WorkerLifecycle,
+    snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
+    admission_queue_depth: usize,
+    in_flight_persistence: usize,
+) -> std::option::Option<ksp_core_lib::ErrorCode> {
+    if lifecycle.mark_stopping().is_err() {
+        return std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
+    }
+    if let std::result::Result::Err(error) = snapshots.publish_state(lifecycle.state(), admission_queue_depth, in_flight_persistence) {
+        return std::option::Option::Some(error.code());
+    }
+    return std::option::Option::None;
+}
+
 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),
@@ -98,10 +128,15 @@ async fn drain_children(children: &mut tokio::task::JoinSet<()>) -> std::option:
     return fault;
 }
 
-async fn drain_persistence(persistence: &mut PersistenceTasks) -> std::option::Option<ksp_core_lib::ErrorCode> {
+async fn drain_persistence(
+    lifecycle: &ksp_worker_api::WorkerLifecycle,
+    admission: &crate::RawTransactionAdmission,
+    persistence: &mut PersistenceTasks,
+    snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
+) -> 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);
+        let current = persistence_completion(joined, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots);
         if fault.is_none() {
             fault = current;
         }
@@ -111,9 +146,11 @@ async fn drain_persistence(persistence: &mut PersistenceTasks) -> std::option::O
 
 async fn drain_admission_and_persistence(
     settings: &crate::RawTransactionIngestSettings,
+    lifecycle: &ksp_worker_api::WorkerLifecycle,
     admission: &mut crate::RawTransactionAdmission,
     persistence: &mut PersistenceTasks,
     port: &std::option::Option<PersistencePort>,
+    snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
 ) -> std::option::Option<ksp_core_lib::ErrorCode> {
     let mut fault = std::option::Option::None;
     admission.close();
@@ -121,7 +158,7 @@ async fn drain_admission_and_persistence(
         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::Some(value) => persistence_completion(value, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots),
                 std::option::Option::None => std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID),
             };
             if fault.is_none() {
@@ -131,19 +168,31 @@ async fn drain_admission_and_persistence(
         let received = admission.receive(settings.network()).await;
         match received {
             std::result::Result::Ok(std::option::Option::Some(acquisition)) => {
-                if !spawn_persistence(persistence, port, acquisition) && fault.is_none() {
+                let spawned = spawn_persistence(persistence, port, acquisition);
+                let published = snapshots.record_admission_success(lifecycle.state(), admission.queue_depth(), persistence.len());
+                if !spawned && fault.is_none() {
                     fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
                 }
+                if let std::result::Result::Err(error) = published {
+                    if fault.is_none() {
+                        fault = std::option::Option::Some(error.code());
+                    }
+                }
             },
             std::result::Result::Ok(std::option::Option::None) => break,
             std::result::Result::Err(_) => {
-                if fault.is_none() {
+                let published = snapshots.record_admission_failure(lifecycle.state(), admission.queue_depth(), persistence.len());
+                if let std::result::Result::Err(error) = published {
+                    if fault.is_none() {
+                        fault = std::option::Option::Some(error.code());
+                    }
+                } else if fault.is_none() {
                     fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
                 }
             },
         }
     }
-    let persistence_fault = drain_persistence(persistence).await;
+    let persistence_fault = drain_persistence(lifecycle, admission, persistence, snapshots).await;
     if fault.is_none() {
         fault = persistence_fault;
     }
@@ -152,28 +201,27 @@ async fn drain_admission_and_persistence(
 
 fn finish_faulted(
     lifecycle: &mut ksp_worker_api::WorkerLifecycle,
-    sender: &tokio::sync::watch::Sender<ksp_worker_api::WorkerState>,
+    snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
     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));
+        snapshots.force_terminal(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
         return;
     }
-    sender.send_replace(lifecycle.state());
+    if snapshots.publish_state(lifecycle.state(), 0, 0).is_err() {
+        snapshots.force_terminal(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED));
+    }
     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));
-        return;
-    }
-    sender.send_replace(lifecycle.state());
+fn finish_stopped(lifecycle: &mut ksp_worker_api::WorkerLifecycle, snapshots: &mut crate::RawTransactionIngestSnapshotPublisher) {
     if lifecycle.mark_stopped().is_err() {
-        sender.send_replace(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
+        snapshots.force_terminal(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
         return;
     }
-    sender.send_replace(lifecycle.state());
+    if snapshots.publish_state(lifecycle.state(), 0, 0).is_err() {
+        snapshots.force_terminal(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED));
+    }
     return;
 }
 
@@ -187,16 +235,28 @@ fn merge_fault(
     return next;
 }
 
-fn persistence_fault(
+fn persistence_completion(
     joined: std::result::Result<ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>, tokio::task::JoinError>,
+    state: ksp_worker_api::WorkerState,
+    admission_queue_depth: usize,
+    in_flight_persistence: usize,
+    snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
 ) -> 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
+            match snapshots.record_persistence_success(state, admission_queue_depth, in_flight_persistence, outcome) {
+                std::result::Result::Ok(()) => std::option::Option::None,
+                std::result::Result::Err(error) => std::option::Option::Some(error.code()),
+            }
+        },
+        std::result::Result::Ok(std::result::Result::Err(error)) => {
+            let code = error.code();
+            let published = snapshots.record_persistence_fault(state, admission_queue_depth, in_flight_persistence, code);
+            match published {
+                std::result::Result::Ok(()) => std::option::Option::Some(code),
+                std::result::Result::Err(snapshot_error) => std::option::Option::Some(snapshot_error.code()),
+            }
         },
-        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),
     };
 }
@@ -206,7 +266,7 @@ async fn run_supervisor<Spawner>(
     mut lifecycle: ksp_worker_api::WorkerLifecycle,
     port: std::option::Option<PersistencePort>,
     mut stop_receiver: tokio::sync::watch::Receiver<bool>,
-    terminal_sender: tokio::sync::watch::Sender<ksp_worker_api::WorkerState>,
+    mut snapshots: crate::RawTransactionIngestSnapshotPublisher,
     source_spawner: Spawner,
 ) where
     Spawner: FnOnce(&mut tokio::task::JoinSet<()>, tokio::sync::watch::Receiver<bool>, tokio::sync::mpsc::Sender<crate::RawTransactionIngress>)
@@ -214,30 +274,50 @@ async fn run_supervisor<Spawner>(
         + 'static,
 {
     if *stop_receiver.borrow() {
-        finish_stopped(&mut lifecycle, &terminal_sender);
+        let stopping_fault = begin_stopping(&mut lifecycle, &mut snapshots, 0, 0);
+        match stopping_fault {
+            std::option::Option::Some(code) => finish_faulted(&mut lifecycle, &mut snapshots, code),
+            std::option::Option::None => finish_stopped(&mut lifecycle, &mut snapshots),
+        }
         return;
     }
     if lifecycle.mark_running().is_err() {
-        terminal_sender.send_replace(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
+        snapshots.force_terminal(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
+        return;
+    }
+    if let std::result::Result::Err(error) = snapshots.publish_state(lifecycle.state(), 0, 0) {
+        finish_faulted(&mut lifecycle, &mut snapshots, error.code());
         return;
     }
-    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, 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;
+    let mut fault = supervise_until_stop(
+        &settings,
+        &lifecycle,
+        &source_stop_sender,
+        &mut stop_receiver,
+        &mut children,
+        &mut admission,
+        &mut persistence,
+        &port,
+        &mut snapshots,
+    )
+    .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;
+    let stopping_fault = begin_stopping(&mut lifecycle, &mut snapshots, admission.queue_depth(), persistence.len());
+    fault = merge_fault(fault, stopping_fault);
+    let drain_fault = drain_admission_and_persistence(&settings, &lifecycle, &mut admission, &mut persistence, &port, &mut snapshots).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),
+        std::option::Option::Some(code) => finish_faulted(&mut lifecycle, &mut snapshots, code),
+        std::option::Option::None => finish_stopped(&mut lifecycle, &mut snapshots),
     }
     return;
 }
@@ -286,9 +366,9 @@ where
     }
     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_supervisor(settings, lifecycle, port, stop_receiver, terminal_sender, source_spawner)));
+    let (snapshots, snapshot_source) = crate::RawTransactionIngestSnapshotPublisher::new(&settings, &lifecycle);
+    let handle = crate::RawTransactionIngestHandle { snapshots: snapshot_source, stop_sender, stop_token };
+    std::mem::drop(runtime.spawn(run_supervisor(settings, lifecycle, port, stop_receiver, snapshots, source_spawner)));
     return std::result::Result::Ok(handle);
 }
 
@@ -315,12 +395,14 @@ where
 
 async fn supervise_until_stop(
     settings: &crate::RawTransactionIngestSettings,
+    lifecycle: &ksp_worker_api::WorkerLifecycle,
     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,
     persistence: &mut PersistenceTasks,
     port: &std::option::Option<PersistencePort>,
+    snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
 ) -> std::option::Option<ksp_core_lib::ErrorCode> {
     let mut admission_open = true;
     loop {
@@ -352,7 +434,7 @@ async fn supervise_until_stop(
             }
             joined = persistence.join_next(), if !persistence.is_empty() => {
                 if let std::option::Option::Some(value) = joined {
-                    let fault = persistence_fault(value);
+                    let fault = persistence_completion(value, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots);
                     if fault.is_some() {
                         source_stop_sender.send_replace(true);
                         return fault;
@@ -362,15 +444,25 @@ async fn supervise_until_stop(
             received = admission.receive(settings.network()), if admission_open && persistence.len() < settings.persistence_concurrency() => {
                 match received {
                     std::result::Result::Ok(std::option::Option::Some(acquisition)) => {
-                        if !spawn_persistence(persistence, port, acquisition) {
+                        let spawned = spawn_persistence(persistence, port, acquisition);
+                        let published = snapshots.record_admission_success(lifecycle.state(), admission.queue_depth(), persistence.len());
+                        if !spawned {
                             source_stop_sender.send_replace(true);
                             return std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
                         }
+                        if let std::result::Result::Err(error) = published {
+                            source_stop_sender.send_replace(true);
+                            return std::option::Option::Some(error.code());
+                        }
                     },
                     std::result::Result::Ok(std::option::Option::None) => admission_open = false,
                     std::result::Result::Err(_) => {
+                        let published = snapshots.record_admission_failure(lifecycle.state(), admission.queue_depth(), persistence.len());
                         source_stop_sender.send_replace(true);
-                        return std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
+                        return match published {
+                            std::result::Result::Ok(()) => std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID),
+                            std::result::Result::Err(error) => std::option::Option::Some(error.code()),
+                        };
                     },
                 }
             }

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

@@ -0,0 +1,472 @@
+// file: crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs
+// version: 1
+
+/// Runtime-neutral boxed future resolving to one newer concrete RAW transaction ingest Worker snapshot.
+pub type RawTransactionIngestSnapshotFuture<'a> =
+    std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = crate::RawTransactionIngestSnapshot> + std::marker::Send + 'a>>;
+
+/// Complete safe latest-value snapshot of one continuous RAW transaction ingest Worker.
+#[derive(Clone, Eq, PartialEq)]
+pub struct RawTransactionIngestSnapshot {
+    worker: ksp_worker_api::WorkerSnapshot,
+    admission_queue_capacity: usize,
+    admission_queue_depth: usize,
+    persistence_concurrency: usize,
+    in_flight_persistence: usize,
+    admitted_total: u64,
+    canonicalized_total: u64,
+    persisted_total: u64,
+    entity_inserted_total: u64,
+    entity_already_present_total: u64,
+    entity_skipped_purged_total: u64,
+    observation_inserted_total: u64,
+    observation_already_present_total: u64,
+    content_conflict_total: u64,
+    store_failure_total: u64,
+    source_failure_total: u64,
+    backpressure_wait_total: u64,
+}
+
+impl crate::RawTransactionIngestSnapshot {
+    /// Returns the common Worker projection carried by this concrete snapshot.
+    #[must_use]
+    pub const fn worker_snapshot(&self) -> &ksp_worker_api::WorkerSnapshot {
+        return &self.worker;
+    }
+
+    /// Returns the configured bounded admission queue capacity.
+    #[must_use]
+    pub const fn admission_queue_capacity(&self) -> usize {
+        return self.admission_queue_capacity;
+    }
+
+    /// Returns the latest observed number of queued ingress entries awaiting supervisor admission.
+    #[must_use]
+    pub const fn admission_queue_depth(&self) -> usize {
+        return self.admission_queue_depth;
+    }
+
+    /// Returns the configured maximum number of concurrent Store persistence operations.
+    #[must_use]
+    pub const fn persistence_concurrency(&self) -> usize {
+        return self.persistence_concurrency;
+    }
+
+    /// Returns the latest observed number of in-flight Store persistence operations.
+    #[must_use]
+    pub const fn in_flight_persistence(&self) -> usize {
+        return self.in_flight_persistence;
+    }
+
+    /// Returns the number of ingress entries removed from the bounded admission queue.
+    #[must_use]
+    pub const fn admitted_total(&self) -> u64 {
+        return self.admitted_total;
+    }
+
+    /// Returns the number of admitted ingress entries successfully canonicalized through the Common RAW contract.
+    #[must_use]
+    pub const fn canonicalized_total(&self) -> u64 {
+        return self.canonicalized_total;
+    }
+
+    /// Returns the number of successful atomic Store persistence outcomes.
+    #[must_use]
+    pub const fn persisted_total(&self) -> u64 {
+        return self.persisted_total;
+    }
+
+    /// Returns the number of newly inserted canonical RAW transaction entities.
+    #[must_use]
+    pub const fn entity_inserted_total(&self) -> u64 {
+        return self.entity_inserted_total;
+    }
+
+    /// Returns the number of identical canonical RAW transaction entities already durable.
+    #[must_use]
+    pub const fn entity_already_present_total(&self) -> u64 {
+        return self.entity_already_present_total;
+    }
+
+    /// Returns the number of durable purge tombstones respected by normal persistence.
+    #[must_use]
+    pub const fn entity_skipped_purged_total(&self) -> u64 {
+        return self.entity_skipped_purged_total;
+    }
+
+    /// Returns the number of newly inserted deterministic acquisition observations.
+    #[must_use]
+    pub const fn observation_inserted_total(&self) -> u64 {
+        return self.observation_inserted_total;
+    }
+
+    /// Returns the number of deterministic acquisition observations already durable.
+    #[must_use]
+    pub const fn observation_already_present_total(&self) -> u64 {
+        return self.observation_already_present_total;
+    }
+
+    /// Returns the number of durable Store content conflicts observed by this run.
+    #[must_use]
+    pub const fn content_conflict_total(&self) -> u64 {
+        return self.content_conflict_total;
+    }
+
+    /// Returns the number of non-conflict Store persistence failures observed by this run.
+    #[must_use]
+    pub const fn store_failure_total(&self) -> u64 {
+        return self.store_failure_total;
+    }
+
+    /// Returns the number of source-task failures observed by this run.
+    #[must_use]
+    pub const fn source_failure_total(&self) -> u64 {
+        return self.source_failure_total;
+    }
+
+    /// Returns the number of source-side admission waits explicitly classified as backpressure waits.
+    #[must_use]
+    pub const fn backpressure_wait_total(&self) -> u64 {
+        return self.backpressure_wait_total;
+    }
+}
+
+impl std::fmt::Debug for crate::RawTransactionIngestSnapshot {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        return formatter
+            .debug_struct("RawTransactionIngestSnapshot")
+            .field("worker", &self.worker)
+            .field("admission_queue_capacity", &self.admission_queue_capacity)
+            .field("admission_queue_depth", &self.admission_queue_depth)
+            .field("persistence_concurrency", &self.persistence_concurrency)
+            .field("in_flight_persistence", &self.in_flight_persistence)
+            .field("admitted_total", &self.admitted_total)
+            .field("canonicalized_total", &self.canonicalized_total)
+            .field("persisted_total", &self.persisted_total)
+            .field("entity_inserted_total", &self.entity_inserted_total)
+            .field("entity_already_present_total", &self.entity_already_present_total)
+            .field("entity_skipped_purged_total", &self.entity_skipped_purged_total)
+            .field("observation_inserted_total", &self.observation_inserted_total)
+            .field("observation_already_present_total", &self.observation_already_present_total)
+            .field("content_conflict_total", &self.content_conflict_total)
+            .field("store_failure_total", &self.store_failure_total)
+            .field("source_failure_total", &self.source_failure_total)
+            .field("backpressure_wait_total", &self.backpressure_wait_total)
+            .finish();
+    }
+}
+
+/// Cloneable latest-value source exposing concrete and common Worker snapshots from one shared watch state.
+#[derive(Clone)]
+pub struct RawTransactionIngestSnapshotSource {
+    receiver: tokio::sync::watch::Receiver<crate::RawTransactionIngestSnapshot>,
+}
+
+impl crate::RawTransactionIngestSnapshotSource {
+    /// Returns the complete current concrete snapshot without replaying intermediate updates.
+    #[must_use]
+    pub fn current(&self) -> crate::RawTransactionIngestSnapshot {
+        return self.receiver.borrow().clone();
+    }
+
+    /// Reports whether the private publisher has closed after the runtime task returned.
+    #[must_use]
+    pub(crate) fn is_closed(&self) -> bool {
+        return self.receiver.has_changed().is_err();
+    }
+
+    /// Waits for one concrete snapshot newer than `observed`, coalescing intermediate updates to the latest value.
+    #[must_use]
+    pub fn wait_for_change(&self, observed: ksp_worker_api::WorkerSnapshotSequence) -> crate::RawTransactionIngestSnapshotFuture<'_> {
+        let mut receiver = self.receiver.clone();
+        return std::boxed::Box::pin(async move {
+            loop {
+                let current = receiver.borrow_and_update().clone();
+                if current.worker_snapshot().sequence().is_after(observed) {
+                    return current;
+                }
+                let changed = receiver.changed().await;
+                if changed.is_err() {
+                    return receiver.borrow_and_update().clone();
+                }
+            }
+        });
+    }
+}
+
+impl ksp_worker_api::WorkerSnapshotSource for crate::RawTransactionIngestSnapshotSource {
+    fn current(&self) -> ksp_worker_api::WorkerSnapshot {
+        return self.receiver.borrow().worker_snapshot().clone();
+    }
+
+    fn wait_for_change(&self, observed: ksp_worker_api::WorkerSnapshotSequence) -> ksp_worker_api::WorkerSnapshotFuture<'_> {
+        let mut receiver = self.receiver.clone();
+        return std::boxed::Box::pin(async move {
+            loop {
+                let current = receiver.borrow_and_update().worker_snapshot().clone();
+                if current.sequence().is_after(observed) {
+                    return current;
+                }
+                let changed = receiver.changed().await;
+                if changed.is_err() {
+                    return receiver.borrow_and_update().worker_snapshot().clone();
+                }
+            }
+        });
+    }
+}
+
+impl std::fmt::Debug for crate::RawTransactionIngestSnapshotSource {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let current = self.receiver.borrow();
+        return formatter
+            .debug_struct("RawTransactionIngestSnapshotSource")
+            .field("sequence", &current.worker_snapshot().sequence())
+            .field("state", &current.worker_snapshot().state())
+            .finish();
+    }
+}
+
+/// Private latest-value publisher and checked counter owner shared by the Worker supervisor.
+pub(crate) struct RawTransactionIngestSnapshotPublisher {
+    sender: tokio::sync::watch::Sender<crate::RawTransactionIngestSnapshot>,
+    snapshot: crate::RawTransactionIngestSnapshot,
+}
+
+impl crate::RawTransactionIngestSnapshotPublisher {
+    /// Creates the initial `Starting` snapshot stream for one validated Worker run.
+    pub(crate) fn new(
+        settings: &crate::RawTransactionIngestSettings,
+        lifecycle: &ksp_worker_api::WorkerLifecycle,
+    ) -> (crate::RawTransactionIngestSnapshotPublisher, crate::RawTransactionIngestSnapshotSource) {
+        let worker = ksp_worker_api::WorkerSnapshot::new(
+            lifecycle.id().clone(),
+            lifecycle.kind().clone(),
+            ksp_worker_api::WorkerSnapshotSequence::initial(),
+            lifecycle.state(),
+            ksp_worker_api::WorkerHealth::Unknown,
+            ksp_worker_api::WorkerActivity::Unknown,
+        );
+        let snapshot = crate::RawTransactionIngestSnapshot {
+            worker,
+            admission_queue_capacity: settings.admission_queue_capacity(),
+            admission_queue_depth: 0,
+            persistence_concurrency: settings.persistence_concurrency(),
+            in_flight_persistence: 0,
+            admitted_total: 0,
+            canonicalized_total: 0,
+            persisted_total: 0,
+            entity_inserted_total: 0,
+            entity_already_present_total: 0,
+            entity_skipped_purged_total: 0,
+            observation_inserted_total: 0,
+            observation_already_present_total: 0,
+            content_conflict_total: 0,
+            store_failure_total: 0,
+            source_failure_total: 0,
+            backpressure_wait_total: 0,
+        };
+        let (sender, receiver) = tokio::sync::watch::channel(snapshot.clone());
+        return (Self { sender, snapshot }, crate::RawTransactionIngestSnapshotSource { receiver });
+    }
+
+    /// Forces one terminal latest value without advancing the sequence, used only when sequence publication itself is exhausted or invalid.
+    pub(crate) fn force_terminal(&mut self, state: ksp_worker_api::WorkerState) {
+        let worker = ksp_worker_api::WorkerSnapshot::new(
+            self.snapshot.worker.id().clone(),
+            self.snapshot.worker.kind().clone(),
+            self.snapshot.worker.sequence(),
+            state,
+            health_for_state(state, self.snapshot.worker.health()),
+            ksp_worker_api::WorkerActivity::Idle,
+        );
+        self.snapshot.worker = worker;
+        self.snapshot.admission_queue_depth = 0;
+        self.snapshot.in_flight_persistence = 0;
+        self.sender.send_replace(self.snapshot.clone());
+        return;
+    }
+
+    /// Publishes one lifecycle/depth refresh with a strictly newer common sequence.
+    pub(crate) fn publish_state(
+        &mut self,
+        state: ksp_worker_api::WorkerState,
+        admission_queue_depth: usize,
+        in_flight_persistence: usize,
+    ) -> ksp_core_lib::Result<()> {
+        return self.publish(state, admission_queue_depth, in_flight_persistence);
+    }
+
+    /// Records one dequeued ingress that failed canonicalization and publishes the resulting latest value.
+    pub(crate) fn record_admission_failure(
+        &mut self,
+        state: ksp_worker_api::WorkerState,
+        admission_queue_depth: usize,
+        in_flight_persistence: usize,
+    ) -> ksp_core_lib::Result<()> {
+        let admitted_total = match checked_counter(self.snapshot.admitted_total, "admitted_total") {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        self.snapshot.admitted_total = admitted_total;
+        return self.publish(state, admission_queue_depth, in_flight_persistence);
+    }
+
+    /// Records one successfully canonicalized admitted ingress and publishes the resulting latest value.
+    pub(crate) fn record_admission_success(
+        &mut self,
+        state: ksp_worker_api::WorkerState,
+        admission_queue_depth: usize,
+        in_flight_persistence: usize,
+    ) -> ksp_core_lib::Result<()> {
+        let admitted_total = match checked_counter(self.snapshot.admitted_total, "admitted_total") {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let canonicalized_total = match checked_counter(self.snapshot.canonicalized_total, "canonicalized_total") {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        self.snapshot.admitted_total = admitted_total;
+        self.snapshot.canonicalized_total = canonicalized_total;
+        return self.publish(state, admission_queue_depth, in_flight_persistence);
+    }
+
+    /// Records one successful Store persistence outcome and publishes the resulting latest value.
+    pub(crate) fn record_persistence_success(
+        &mut self,
+        state: ksp_worker_api::WorkerState,
+        admission_queue_depth: usize,
+        in_flight_persistence: usize,
+        outcome: crate::RawTransactionIngestPersistenceOutcome,
+    ) -> ksp_core_lib::Result<()> {
+        let persisted_total = match checked_counter(self.snapshot.persisted_total, "persisted_total") {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let mut entity_inserted_total = self.snapshot.entity_inserted_total;
+        let mut entity_already_present_total = self.snapshot.entity_already_present_total;
+        let mut entity_skipped_purged_total = self.snapshot.entity_skipped_purged_total;
+        let mut observation_inserted_total = self.snapshot.observation_inserted_total;
+        let mut observation_already_present_total = self.snapshot.observation_already_present_total;
+        match outcome.entity() {
+            crate::RawTransactionIngestEntityPersistence::Inserted => {
+                entity_inserted_total = match checked_counter(entity_inserted_total, "entity_inserted_total") {
+                    std::result::Result::Ok(value) => value,
+                    std::result::Result::Err(error) => return std::result::Result::Err(error),
+                };
+            },
+            crate::RawTransactionIngestEntityPersistence::AlreadyPresent => {
+                entity_already_present_total = match checked_counter(entity_already_present_total, "entity_already_present_total") {
+                    std::result::Result::Ok(value) => value,
+                    std::result::Result::Err(error) => return std::result::Result::Err(error),
+                };
+            },
+            crate::RawTransactionIngestEntityPersistence::SkippedPurged => {
+                entity_skipped_purged_total = match checked_counter(entity_skipped_purged_total, "entity_skipped_purged_total") {
+                    std::result::Result::Ok(value) => value,
+                    std::result::Result::Err(error) => return std::result::Result::Err(error),
+                };
+            },
+        }
+        match outcome.observation() {
+            crate::RawTransactionIngestObservationPersistence::Inserted => {
+                observation_inserted_total = match checked_counter(observation_inserted_total, "observation_inserted_total") {
+                    std::result::Result::Ok(value) => value,
+                    std::result::Result::Err(error) => return std::result::Result::Err(error),
+                };
+            },
+            crate::RawTransactionIngestObservationPersistence::AlreadyPresent => {
+                observation_already_present_total = match checked_counter(observation_already_present_total, "observation_already_present_total") {
+                    std::result::Result::Ok(value) => value,
+                    std::result::Result::Err(error) => return std::result::Result::Err(error),
+                };
+            },
+            crate::RawTransactionIngestObservationPersistence::NotRecorded => {},
+        }
+        self.snapshot.persisted_total = persisted_total;
+        self.snapshot.entity_inserted_total = entity_inserted_total;
+        self.snapshot.entity_already_present_total = entity_already_present_total;
+        self.snapshot.entity_skipped_purged_total = entity_skipped_purged_total;
+        self.snapshot.observation_inserted_total = observation_inserted_total;
+        self.snapshot.observation_already_present_total = observation_already_present_total;
+        return self.publish(state, admission_queue_depth, in_flight_persistence);
+    }
+
+    /// Records one classified persistence fault counter and publishes the resulting latest value.
+    pub(crate) fn record_persistence_fault(
+        &mut self,
+        state: ksp_worker_api::WorkerState,
+        admission_queue_depth: usize,
+        in_flight_persistence: usize,
+        code: ksp_core_lib::ErrorCode,
+    ) -> ksp_core_lib::Result<()> {
+        if code == crate::ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT {
+            let next = match checked_counter(self.snapshot.content_conflict_total, "content_conflict_total") {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            self.snapshot.content_conflict_total = next;
+        } else if code == crate::ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED {
+            let next = match checked_counter(self.snapshot.store_failure_total, "store_failure_total") {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            self.snapshot.store_failure_total = next;
+        }
+        return self.publish(state, admission_queue_depth, in_flight_persistence);
+    }
+
+    fn publish(&mut self, state: ksp_worker_api::WorkerState, admission_queue_depth: usize, in_flight_persistence: usize) -> ksp_core_lib::Result<()> {
+        let sequence = match self.snapshot.worker.sequence().next() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => return std::result::Result::Err(crate::counter_exhausted_error("snapshot_sequence")),
+        };
+        let worker = ksp_worker_api::WorkerSnapshot::new(
+            self.snapshot.worker.id().clone(),
+            self.snapshot.worker.kind().clone(),
+            sequence,
+            state,
+            health_for_state(state, self.snapshot.worker.health()),
+            activity_for_state(state, admission_queue_depth, in_flight_persistence),
+        );
+        self.snapshot.worker = worker;
+        self.snapshot.admission_queue_depth = admission_queue_depth;
+        self.snapshot.in_flight_persistence = in_flight_persistence;
+        self.sender.send_replace(self.snapshot.clone());
+        return std::result::Result::Ok(());
+    }
+}
+
+fn activity_for_state(state: ksp_worker_api::WorkerState, admission_queue_depth: usize, in_flight_persistence: usize) -> ksp_worker_api::WorkerActivity {
+    if admission_queue_depth > 0 || in_flight_persistence > 0 {
+        return ksp_worker_api::WorkerActivity::Active;
+    }
+    return match state {
+        ksp_worker_api::WorkerState::Running
+        | ksp_worker_api::WorkerState::Stopping
+        | ksp_worker_api::WorkerState::Stopped
+        | ksp_worker_api::WorkerState::Faulted(_) => ksp_worker_api::WorkerActivity::Idle,
+        _ => ksp_worker_api::WorkerActivity::Unknown,
+    };
+}
+
+fn checked_counter(current: u64, field: &'static str) -> ksp_core_lib::Result<u64> {
+    return match current.checked_add(1) {
+        std::option::Option::Some(value) => std::result::Result::Ok(value),
+        std::option::Option::None => std::result::Result::Err(crate::counter_exhausted_error(field)),
+    };
+}
+
+fn health_for_state(state: ksp_worker_api::WorkerState, previous: ksp_worker_api::WorkerHealth) -> ksp_worker_api::WorkerHealth {
+    return match state {
+        ksp_worker_api::WorkerState::Running => ksp_worker_api::WorkerHealth::Healthy,
+        ksp_worker_api::WorkerState::Stopping | ksp_worker_api::WorkerState::Stopped => previous,
+        ksp_worker_api::WorkerState::Faulted(_) => ksp_worker_api::WorkerHealth::Unhealthy,
+        _ => ksp_worker_api::WorkerHealth::Unknown,
+    };
+}
+
+#[cfg(test)]
+#[path = "../unit_tests/snapshot.rs"]
+mod tests;