v0.3.14-pre.009

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs
-// version: 8
+// version: 9
 
 /// Maximum number of slots admitted by one private continuity HTTP discovery window outside this module.
 pub(crate) const MAX_RAW_TRANSACTION_INGEST_CONTINUITY_DISCOVERY_WINDOW_SLOTS: u64 = MAX_RAW_TRANSACTION_INGEST_REPAIR_DISCOVERY_WINDOW_SLOTS;
@@ -347,6 +347,25 @@ impl RawTransactionIngestGapLedger {
         return self.gaps.iter().any(|gap| return gap.state.is_open());
     }
 
+    fn source_failures_reconciled(&self, source_keys: &std::collections::BTreeSet<[u8; 32]>) -> bool {
+        for source_key in source_keys {
+            let mut found = false;
+            for gap in &self.gaps {
+                if gap.source_key != *source_key || gap.reason != RawTransactionIngestGapReason::SourceFailure {
+                    continue;
+                }
+                found = true;
+                if gap.state.is_open() {
+                    return false;
+                }
+            }
+            if !found {
+                return false;
+            }
+        }
+        return true;
+    }
+
     fn record_gap(
         &mut self,
         source_key: [u8; 32],
@@ -904,6 +923,44 @@ impl crate::RawTransactionIngestContinuityContracts {
         return self.gap_ledger.continuity_frontier(processing_frontier_slot);
     }
 
+    /// Reconciles known gaps and projects source-neutral present/future coverage evidence for Worker health.
+    ///
+    /// The returned tuple is `(continuity_frontier, has_open_gaps, future_target_coverage, failed_source_losses_reconciled)`. Configuration alone may
+    /// satisfy only the future-coverage component; present continuity and terminal source-loss history still require proven run-local reconciliation.
+    pub(crate) fn health_projection(
+        &mut self,
+        active_source_keys: &[[u8; 32]],
+        failed_source_keys: &[[u8; 32]],
+        processing_frontier_slot: std::option::Option<u64>,
+    ) -> ksp_core_lib::Result<(std::option::Option<u64>, bool, bool, bool)> {
+        let mut active = std::collections::BTreeSet::new();
+        for source_key in active_source_keys {
+            if !active.insert(*source_key) {
+                return std::result::Result::Err(crate::runtime_error("continuity.health_active_set_invalid"));
+            }
+            if !self.capabilities.iter().any(|capability| return capability.source_key == *source_key) {
+                return std::result::Result::Err(crate::runtime_error("continuity.health_active_source_unknown"));
+            }
+        }
+        let mut failed = std::collections::BTreeSet::new();
+        for source_key in failed_source_keys {
+            if active.contains(source_key) || !failed.insert(*source_key) {
+                return std::result::Result::Err(crate::runtime_error("continuity.health_failed_set_invalid"));
+            }
+            if !self.capabilities.iter().any(|capability| return capability.source_key == *source_key) {
+                return std::result::Result::Err(crate::runtime_error("continuity.health_failed_source_unknown"));
+            }
+        }
+        if let std::result::Result::Err(error) = self.gap_ledger.reconcile_with_coverage_epochs(&self.coverage_epochs) {
+            return std::result::Result::Err(error);
+        }
+        let continuity_frontier = self.continuity_frontier(processing_frontier_slot);
+        let has_open_gaps = self.gap_ledger.has_open_gaps();
+        let future_target_coverage = self.target_coverage.is_covered_by_active_sources(self.capabilities.as_slice(), &active);
+        let failed_source_losses_reconciled = self.gap_ledger.source_failures_reconciled(&failed);
+        return std::result::Result::Ok((continuity_frontier, has_open_gaps, future_target_coverage, failed_source_losses_reconciled));
+    }
+
     /// Reconciles known gaps against already-proven coverage and decides whether one lost source may remain absent without stopping sibling sources.
     ///
     /// Continuation requires all configured `TargetCoverage` requirements to remain covered by distinct currently active sources and requires the known
@@ -926,14 +983,12 @@ impl crate::RawTransactionIngestContinuityContracts {
                 return std::result::Result::Err(crate::runtime_error("continuity.source_loss_active_source_unknown"));
             }
         }
-        if let std::result::Result::Err(error) = self.gap_ledger.reconcile_with_coverage_epochs(&self.coverage_epochs) {
-            return std::result::Result::Err(error);
-        }
-        let continuity_frontier = self.continuity_frontier(processing_frontier_slot);
-        if self.gap_ledger.has_open_gaps() || continuity_frontier != processing_frontier_slot {
-            return std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Fault);
-        }
-        if !self.target_coverage.is_covered_by_active_sources(self.capabilities.as_slice(), &active) {
+        let (continuity_frontier, has_open_gaps, future_target_coverage, failed_source_losses_reconciled) =
+            match self.health_projection(active_source_keys, std::slice::from_ref(&lost_source_key), processing_frontier_slot) {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+        if has_open_gaps || continuity_frontier != processing_frontier_slot || !future_target_coverage || !failed_source_losses_reconciled {
             return std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Fault);
         }
         return std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Continue);

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs
-// version: 36
+// version: 37
 
 use sha2::Digest; // rust-rules: trait-import
 
@@ -104,6 +104,7 @@ struct RawTransactionIngestSourceInventory {
 #[derive(Clone)]
 struct RawTransactionIngestSourceInventoryPublisher {
     aggregate_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
+    continuity_contracts: std::sync::Arc<std::sync::Mutex<crate::RawTransactionIngestContinuityContracts>>,
     entry_index: usize,
     inventory: std::sync::Arc<std::sync::Mutex<RawTransactionIngestSourceInventory>>,
     source_key: [u8; 32],
@@ -361,7 +362,7 @@ impl RawTransactionIngestSourceInventory {
         return Self { source_keys, source_projections };
     }
 
-    fn supervisor_state(&self) -> ksp_core_lib::Result<(std::vec::Vec<[u8; 32]>, std::option::Option<u64>)> {
+    fn active_source_keys(&self) -> ksp_core_lib::Result<std::vec::Vec<[u8; 32]>> {
         if self.source_keys.len() != self.source_projections.len() {
             return std::result::Result::Err(crate::runtime_error("source.inventory_shape_mismatch"));
         }
@@ -371,6 +372,27 @@ impl RawTransactionIngestSourceInventory {
                 active_source_keys.push(*source_key);
             }
         }
+        return std::result::Result::Ok(active_source_keys);
+    }
+
+    fn failed_source_keys(&self) -> ksp_core_lib::Result<std::vec::Vec<[u8; 32]>> {
+        if self.source_keys.len() != self.source_projections.len() {
+            return std::result::Result::Err(crate::runtime_error("source.inventory_shape_mismatch"));
+        }
+        let mut failed_source_keys = std::vec::Vec::new();
+        for (source_key, projection) in self.source_keys.iter().zip(self.source_projections.iter()) {
+            if projection.source_state() == std::option::Option::Some(crate::RawTransactionIngestSourceState::Failed) {
+                failed_source_keys.push(*source_key);
+            }
+        }
+        return std::result::Result::Ok(failed_source_keys);
+    }
+
+    fn supervisor_state(&self) -> ksp_core_lib::Result<(std::vec::Vec<[u8; 32]>, std::option::Option<u64>)> {
+        let active_source_keys = match self.active_source_keys() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
         let aggregate = match self.aggregate() {
             std::result::Result::Ok(value) => value,
             std::result::Result::Err(error) => return std::result::Result::Err(error),
@@ -400,9 +422,49 @@ impl RawTransactionIngestSourceInventory {
     }
 }
 
+fn source_inventory_health_projection(
+    inventory: &std::sync::Arc<std::sync::Mutex<RawTransactionIngestSourceInventory>>,
+    continuity_contracts: &std::sync::Arc<std::sync::Mutex<crate::RawTransactionIngestContinuityContracts>>,
+) -> ksp_core_lib::Result<crate::RawTransactionIngestProcessingFrontierProjection> {
+    let (aggregate, active_source_keys, failed_source_keys) = {
+        let inventory = match inventory.lock() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(poisoned) => poisoned.into_inner(),
+        };
+        let aggregate = match inventory.aggregate() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let active_source_keys = match inventory.active_source_keys() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        let failed_source_keys = match inventory.failed_source_keys() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        (aggregate, active_source_keys, failed_source_keys)
+    };
+    let (continuity_frontier_slot, continuity_has_open_gaps, future_target_coverage, failed_source_losses_reconciled) = {
+        let mut contracts = match continuity_contracts.lock() {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(poisoned) => poisoned.into_inner(),
+        };
+        match contracts.health_projection(active_source_keys.as_slice(), failed_source_keys.as_slice(), aggregate.processing_frontier_slot()) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        }
+    };
+    return std::result::Result::Ok(
+        aggregate
+            .with_continuity_health(continuity_frontier_slot, continuity_has_open_gaps, future_target_coverage)
+            .with_failed_source_losses_reconciled(failed_source_losses_reconciled),
+    );
+}
+
 impl RawTransactionIngestSourceInventoryPublisher {
     fn publish(&self, projection: crate::RawTransactionIngestProcessingFrontierProjection) -> ksp_core_lib::Result<()> {
-        let aggregate = {
+        {
             let mut inventory = match self.inventory.lock() {
                 std::result::Result::Ok(value) => value,
                 std::result::Result::Err(poisoned) => poisoned.into_inner(),
@@ -410,10 +472,10 @@ impl RawTransactionIngestSourceInventoryPublisher {
             if let std::result::Result::Err(error) = inventory.update(self.entry_index, self.source_key, projection) {
                 return std::result::Result::Err(error);
             }
-            match inventory.aggregate() {
-                std::result::Result::Ok(value) => value,
-                std::result::Result::Err(error) => return std::result::Result::Err(error),
-            }
+        }
+        let aggregate = match source_inventory_health_projection(&self.inventory, &self.continuity_contracts) {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
         };
         self.aggregate_sender.send_replace(aggregate);
         return std::result::Result::Ok(());
@@ -1481,6 +1543,7 @@ impl crate::RawTransactionIngestHttpBlockPollingSource {
                         fault = std::option::Option::Some(error);
                         break;
                     }
+                    processing_frontier.publish();
                     next_scan_slot = match proven_end_slot.checked_add(1) {
                         std::option::Option::Some(value) => value,
                         std::option::Option::None => {
@@ -2295,6 +2358,7 @@ impl crate::RawTransactionIngestRuntimeResources {
             };
             let publisher = RawTransactionIngestSourceInventoryPublisher {
                 aggregate_sender: processing_frontier_sender.clone(),
+                continuity_contracts: std::sync::Arc::clone(&continuity_contracts),
                 entry_index,
                 inventory: std::sync::Arc::clone(&inventory),
                 source_key: source.source_key(),
@@ -2321,6 +2385,7 @@ impl crate::RawTransactionIngestRuntimeResources {
             children,
             std::sync::Arc::clone(&continuity_contracts),
             std::sync::Arc::clone(&inventory),
+            processing_frontier_sender,
         )
         .await;
         let validation = {
@@ -2418,6 +2483,7 @@ async fn supervise_live_source_tasks(
     mut children: tokio::task::JoinSet<([u8; 32], ksp_core_lib::Result<()>)>,
     continuity_contracts: std::sync::Arc<std::sync::Mutex<crate::RawTransactionIngestContinuityContracts>>,
     inventory: std::sync::Arc<std::sync::Mutex<RawTransactionIngestSourceInventory>>,
+    processing_frontier_sender: tokio::sync::watch::Sender<crate::RawTransactionIngestProcessingFrontierProjection>,
 ) -> ksp_core_lib::Result<()> {
     loop {
         if *stop_receiver.borrow() {
@@ -2502,6 +2568,14 @@ async fn supervise_live_source_tasks(
         };
         match decision {
             crate::RawTransactionIngestSourceLossDecision::Continue => {
+                let aggregate = match source_inventory_health_projection(&inventory, &continuity_contracts) {
+                    std::result::Result::Ok(value) => value,
+                    std::result::Result::Err(error) => {
+                        source_stop_sender.send_replace(true);
+                        return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await;
+                    },
+                };
+                processing_frontier_sender.send_replace(aggregate);
                 continue;
             },
             crate::RawTransactionIngestSourceLossDecision::Fault => {

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

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs
-// version: 5
+// version: 6
 
 /// Runtime-neutral boxed future resolving to one newer concrete RAW transaction ingest Worker snapshot.
 pub type RawTransactionIngestSnapshotFuture<'a> =
@@ -23,6 +23,11 @@ pub enum RawTransactionIngestSourceState {
 /// Private latest-value source-processing projection emitted by the productive source task.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(crate) struct RawTransactionIngestProcessingFrontierProjection {
+    continuity_frontier_slot: std::option::Option<u64>,
+    continuity_has_open_gaps: bool,
+    continuity_policy_observed: bool,
+    failed_source_losses_reconciled: bool,
+    future_target_coverage: bool,
     hydration_pending: usize,
     processing_frontier_slot: std::option::Option<u64>,
     oldest_pending_slot: std::option::Option<u64>,
@@ -40,6 +45,11 @@ impl crate::RawTransactionIngestProcessingFrontierProjection {
     /// Returns the empty run-local processing projection used before the source observes work.
     pub(crate) const fn empty() -> Self {
         return Self {
+            continuity_frontier_slot: std::option::Option::None,
+            continuity_has_open_gaps: false,
+            continuity_policy_observed: false,
+            failed_source_losses_reconciled: false,
+            future_target_coverage: false,
             hydration_pending: 0,
             processing_frontier_slot: std::option::Option::None,
             oldest_pending_slot: std::option::Option::None,
@@ -61,6 +71,11 @@ impl crate::RawTransactionIngestProcessingFrontierProjection {
         oldest_pending_slot: std::option::Option<u64>,
     ) -> Self {
         return Self {
+            continuity_frontier_slot: std::option::Option::None,
+            continuity_has_open_gaps: false,
+            continuity_policy_observed: false,
+            failed_source_losses_reconciled: false,
+            future_target_coverage: false,
             hydration_pending,
             processing_frontier_slot,
             oldest_pending_slot,
@@ -114,6 +129,51 @@ impl crate::RawTransactionIngestProcessingFrontierProjection {
         return self;
     }
 
+    /// Returns a copy carrying source-neutral run-local continuity evidence used only for Worker health classification.
+    pub(crate) const fn with_continuity_health(
+        mut self,
+        continuity_frontier_slot: std::option::Option<u64>,
+        continuity_has_open_gaps: bool,
+        future_target_coverage: bool,
+    ) -> Self {
+        self.continuity_frontier_slot = continuity_frontier_slot;
+        self.continuity_has_open_gaps = continuity_has_open_gaps;
+        self.continuity_policy_observed = true;
+        self.future_target_coverage = future_target_coverage;
+        return self;
+    }
+
+    /// Returns whether a run-local continuity policy projection has been observed.
+    pub(crate) const fn continuity_policy_observed(&self) -> bool {
+        return self.continuity_policy_observed;
+    }
+
+    /// Returns the gap-aware continuity frontier carried by this private projection.
+    pub(crate) const fn continuity_frontier_slot(&self) -> std::option::Option<u64> {
+        return self.continuity_frontier_slot;
+    }
+
+    /// Returns whether at least one run-local continuity gap remains unresolved.
+    pub(crate) const fn continuity_has_open_gaps(&self) -> bool {
+        return self.continuity_has_open_gaps;
+    }
+
+    /// Returns whether currently Active sources still cover the complete configured future `TargetCoverage`.
+    pub(crate) const fn future_target_coverage(&self) -> bool {
+        return self.future_target_coverage;
+    }
+
+    /// Returns a copy carrying whether every terminal Failed source has a reconciled source-loss gap.
+    pub(crate) const fn with_failed_source_losses_reconciled(mut self, failed_source_losses_reconciled: bool) -> Self {
+        self.failed_source_losses_reconciled = failed_source_losses_reconciled;
+        return self;
+    }
+
+    /// Returns whether every terminal Failed source has a reconciled source-loss gap.
+    pub(crate) const fn failed_source_losses_reconciled(&self) -> bool {
+        return self.failed_source_losses_reconciled;
+    }
+
     /// Returns the configured logical source count carried by this private projection.
     pub(crate) const fn source_total(&self) -> usize {
         return self.source_total;
@@ -159,6 +219,11 @@ impl crate::RawTransactionIngestProcessingFrontierProjection {
 #[derive(Clone, Eq, PartialEq)]
 pub struct RawTransactionIngestSnapshot {
     worker: ksp_worker_api::WorkerSnapshot,
+    continuity_frontier_slot: std::option::Option<u64>,
+    continuity_has_open_gaps: bool,
+    continuity_policy_observed: bool,
+    failed_source_losses_reconciled: bool,
+    future_target_coverage: bool,
     admission_queue_capacity: usize,
     admission_queue_depth: usize,
     persistence_concurrency: usize,
@@ -488,6 +553,11 @@ impl crate::RawTransactionIngestSnapshotPublisher {
         );
         let snapshot = crate::RawTransactionIngestSnapshot {
             worker,
+            continuity_frontier_slot: std::option::Option::None,
+            continuity_has_open_gaps: false,
+            continuity_policy_observed: false,
+            failed_source_losses_reconciled: false,
+            future_target_coverage: false,
             admission_queue_capacity: settings.admission_queue_capacity(),
             admission_queue_depth: 0,
             persistence_concurrency: settings.persistence_concurrency(),
@@ -527,14 +597,7 @@ impl crate::RawTransactionIngestSnapshotPublisher {
             self.snapshot.worker.kind().clone(),
             self.snapshot.worker.sequence(),
             state,
-            health_for_state(
-                state,
-                self.snapshot.worker.health(),
-                self.snapshot.source_total,
-                self.snapshot.source_active,
-                self.snapshot.source_reconnecting,
-                self.snapshot.source_failed,
-            ),
+            health_for_state(state, self.snapshot.worker.health(), &self.snapshot),
             ksp_worker_api::WorkerActivity::Idle,
         );
         self.snapshot.worker = worker;
@@ -611,6 +674,11 @@ impl crate::RawTransactionIngestSnapshotPublisher {
         in_flight_persistence: usize,
         projection: crate::RawTransactionIngestProcessingFrontierProjection,
     ) -> ksp_core_lib::Result<()> {
+        self.snapshot.continuity_frontier_slot = projection.continuity_frontier_slot();
+        self.snapshot.continuity_has_open_gaps = projection.continuity_has_open_gaps();
+        self.snapshot.continuity_policy_observed = projection.continuity_policy_observed();
+        self.snapshot.failed_source_losses_reconciled = projection.failed_source_losses_reconciled();
+        self.snapshot.future_target_coverage = projection.future_target_coverage();
         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();
@@ -737,14 +805,7 @@ impl crate::RawTransactionIngestSnapshotPublisher {
             self.snapshot.worker.kind().clone(),
             sequence,
             state,
-            health_for_state(
-                state,
-                self.snapshot.worker.health(),
-                self.snapshot.source_total,
-                self.snapshot.source_active,
-                self.snapshot.source_reconnecting,
-                self.snapshot.source_failed,
-            ),
+            health_for_state(state, self.snapshot.worker.health(), &self.snapshot),
             activity_for_state(state, admission_queue_depth, in_flight_persistence, self.snapshot.hydration_pending, self.snapshot.source_state),
         );
         self.snapshot.worker = worker;
@@ -799,15 +860,35 @@ fn checked_optional_counter(current: u64, increment: bool, field: &'static str)
 fn health_for_state(
     state: ksp_worker_api::WorkerState,
     previous: ksp_worker_api::WorkerHealth,
-    source_total: usize,
-    source_active: usize,
-    source_reconnecting: usize,
-    source_failed: usize,
+    snapshot: &crate::RawTransactionIngestSnapshot,
 ) -> ksp_worker_api::WorkerHealth {
+    if state == ksp_worker_api::WorkerState::Running && snapshot.continuity_policy_observed {
+        if snapshot.source_reconnecting > 0
+            || snapshot.continuity_has_open_gaps
+            || snapshot.continuity_frontier_slot != snapshot.processing_frontier_slot
+            || !snapshot.future_target_coverage
+        {
+            return ksp_worker_api::WorkerHealth::Unhealthy;
+        }
+        if snapshot.source_total > 0 && snapshot.source_active == snapshot.source_total {
+            return ksp_worker_api::WorkerHealth::Healthy;
+        }
+        if snapshot.source_total > 0
+            && snapshot.source_active < snapshot.source_total
+            && snapshot.source_failed > 0
+            && snapshot.failed_source_losses_reconciled
+            && snapshot.source_failed == snapshot.source_total - snapshot.source_active
+        {
+            return ksp_worker_api::WorkerHealth::Degraded;
+        }
+        return ksp_worker_api::WorkerHealth::Unhealthy;
+    }
     return match state {
-        ksp_worker_api::WorkerState::Running if source_failed > 0 => ksp_worker_api::WorkerHealth::Unhealthy,
-        ksp_worker_api::WorkerState::Running if source_reconnecting > 0 => ksp_worker_api::WorkerHealth::Degraded,
-        ksp_worker_api::WorkerState::Running if source_total > 0 && source_active < source_total => ksp_worker_api::WorkerHealth::Degraded,
+        ksp_worker_api::WorkerState::Running if snapshot.source_failed > 0 => ksp_worker_api::WorkerHealth::Unhealthy,
+        ksp_worker_api::WorkerState::Running if snapshot.source_reconnecting > 0 => ksp_worker_api::WorkerHealth::Degraded,
+        ksp_worker_api::WorkerState::Running if snapshot.source_total > 0 && snapshot.source_active < snapshot.source_total => {
+            ksp_worker_api::WorkerHealth::Degraded
+        },
         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,