v0.3.12-pre.007

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

@@ -1,10 +1,49 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs
-// version: 2
+// version: 3
 
 /// 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>>;
 
+/// Private latest-value processing-frontier projection emitted by the productive source task.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) struct RawTransactionIngestProcessingFrontierProjection {
+    hydration_pending: usize,
+    processing_frontier_slot: std::option::Option<u64>,
+    oldest_pending_slot: std::option::Option<u64>,
+}
+
+impl crate::RawTransactionIngestProcessingFrontierProjection {
+    /// Returns the empty run-local processing projection used before the source observes work.
+    pub(crate) const fn empty() -> Self {
+        return Self { hydration_pending: 0, processing_frontier_slot: std::option::Option::None, oldest_pending_slot: std::option::Option::None };
+    }
+
+    /// Creates one run-local processing projection from bounded source-owned state.
+    pub(crate) const fn new(
+        hydration_pending: usize,
+        processing_frontier_slot: std::option::Option<u64>,
+        oldest_pending_slot: std::option::Option<u64>,
+    ) -> Self {
+        return Self { hydration_pending, processing_frontier_slot, oldest_pending_slot };
+    }
+
+    /// Returns the number of source signals still pending hydration/admission processing.
+    pub(crate) const fn hydration_pending(&self) -> usize {
+        return self.hydration_pending;
+    }
+
+    /// Returns the highest actually observed slot not blocked by older pending source work.
+    pub(crate) const fn processing_frontier_slot(&self) -> std::option::Option<u64> {
+        return self.processing_frontier_slot;
+    }
+
+    /// Returns the oldest actually observed slot that still owns pending source work.
+    pub(crate) const fn oldest_pending_slot(&self) -> std::option::Option<u64> {
+        return self.oldest_pending_slot;
+    }
+}
+
 /// Complete safe latest-value snapshot of one continuous RAW transaction ingest Worker.
 #[derive(Clone, Eq, PartialEq)]
 pub struct RawTransactionIngestSnapshot {
@@ -25,6 +64,9 @@ pub struct RawTransactionIngestSnapshot {
     store_failure_total: u64,
     source_failure_total: u64,
     backpressure_wait_total: u64,
+    hydration_pending: usize,
+    processing_frontier_slot: std::option::Option<u64>,
+    oldest_pending_slot: std::option::Option<u64>,
 }
 
 impl crate::RawTransactionIngestSnapshot {
@@ -129,6 +171,24 @@ impl crate::RawTransactionIngestSnapshot {
     pub const fn backpressure_wait_total(&self) -> u64 {
         return self.backpressure_wait_total;
     }
+
+    /// Returns the latest number of source signals pending hydration/admission processing.
+    #[must_use]
+    pub const fn hydration_pending(&self) -> usize {
+        return self.hydration_pending;
+    }
+
+    /// Returns the run-local processing frontier over actually observed source work.
+    #[must_use]
+    pub const fn processing_frontier_slot(&self) -> std::option::Option<u64> {
+        return self.processing_frontier_slot;
+    }
+
+    /// Returns the oldest actually observed slot that still owns pending source work.
+    #[must_use]
+    pub const fn oldest_pending_slot(&self) -> std::option::Option<u64> {
+        return self.oldest_pending_slot;
+    }
 }
 
 impl std::fmt::Debug for crate::RawTransactionIngestSnapshot {
@@ -152,6 +212,9 @@ impl std::fmt::Debug for crate::RawTransactionIngestSnapshot {
             .field("store_failure_total", &self.store_failure_total)
             .field("source_failure_total", &self.source_failure_total)
             .field("backpressure_wait_total", &self.backpressure_wait_total)
+            .field("hydration_pending", &self.hydration_pending)
+            .field("processing_frontier_slot", &self.processing_frontier_slot)
+            .field("oldest_pending_slot", &self.oldest_pending_slot)
             .finish();
     }
 }
@@ -265,6 +328,9 @@ impl crate::RawTransactionIngestSnapshotPublisher {
             store_failure_total: 0,
             source_failure_total: 0,
             backpressure_wait_total: 0,
+            hydration_pending: 0,
+            processing_frontier_slot: std::option::Option::None,
+            oldest_pending_slot: std::option::Option::None,
         };
         let (sender, receiver) = tokio::sync::watch::channel(snapshot.clone());
         return (Self { sender, snapshot }, crate::RawTransactionIngestSnapshotSource { receiver });
@@ -346,6 +412,20 @@ impl crate::RawTransactionIngestSnapshotPublisher {
         return self.publish(state, admission_queue_depth, in_flight_persistence);
     }
 
+    /// Publishes one latest run-local processing-frontier projection emitted by the source task.
+    pub(crate) fn record_processing_frontier(
+        &mut self,
+        state: ksp_worker_api::WorkerState,
+        admission_queue_depth: usize,
+        in_flight_persistence: usize,
+        projection: crate::RawTransactionIngestProcessingFrontierProjection,
+    ) -> ksp_core_lib::Result<()> {
+        self.snapshot.hydration_pending = projection.hydration_pending();
+        self.snapshot.processing_frontier_slot = projection.processing_frontier_slot();
+        self.snapshot.oldest_pending_slot = projection.oldest_pending_slot();
+        return self.publish(state, admission_queue_depth, in_flight_persistence);
+    }
+
     /// Records one failed private source task without retaining provider-specific error material.
     pub(crate) fn record_source_failure(
         &mut self,
@@ -457,7 +537,7 @@ impl crate::RawTransactionIngestSnapshotPublisher {
             sequence,
             state,
             health_for_state(state, self.snapshot.worker.health()),
-            activity_for_state(state, admission_queue_depth, in_flight_persistence),
+            activity_for_state(state, admission_queue_depth, in_flight_persistence, self.snapshot.hydration_pending),
         );
         self.snapshot.worker = worker;
         self.snapshot.admission_queue_depth = admission_queue_depth;
@@ -467,8 +547,13 @@ impl crate::RawTransactionIngestSnapshotPublisher {
     }
 }
 
-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 {
+fn activity_for_state(
+    state: ksp_worker_api::WorkerState,
+    admission_queue_depth: usize,
+    in_flight_persistence: usize,
+    hydration_pending: usize,
+) -> ksp_worker_api::WorkerActivity {
+    if admission_queue_depth > 0 || in_flight_persistence > 0 || hydration_pending > 0 {
         return ksp_worker_api::WorkerActivity::Active;
     }
     return match state {