diff --git a/Cargo.toml b/Cargo.toml index 41d0ec2..88af3a9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 578 +# version: 579 [workspace] resolver = "3" members = ["crates/ksp-app-backfill-desk", "crates/ksp-app-config-desk", "crates/ksp-app-solprices-desk", "crates/ksp-app-store-desk", "crates/ksp-app-wallet-desk", "crates/ksp-config-lib", "crates/ksp-core-lib", "crates/ksp-interface-lib", "crates/ksp-job-api", "crates/ksp-job-backfill-lib", "crates/ksp-logging-lib", "crates/ksp-offchain-transport-lib", "crates/ksp-onchain-transport-lib", "crates/ksp-program-api", "crates/ksp-raw-transaction-lib", "crates/ksp-store-api", "crates/ksp-store-lib", "crates/ksp-store-postgres-lib", "crates/ksp-wallet-lib", "crates/ksp-worker-api", "crates/ksp-worker-raw-transaction-ingest-lib"] [workspace.package] -version = "0.3.14-pre.8.fix.2" +version = "0.3.14-pre.9" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs index f8c9fc1..34d3ed8 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs +++ b/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, + ) -> ksp_core_lib::Result<(std::option::Option, 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); diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs index 81f24e1..92ce385 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs +++ b/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, + continuity_contracts: std::sync::Arc>, entry_index: usize, inventory: std::sync::Arc>, 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)> { + fn active_source_keys(&self) -> ksp_core_lib::Result> { 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> { + 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)> { + 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>, + continuity_contracts: &std::sync::Arc>, +) -> ksp_core_lib::Result { + 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>, inventory: std::sync::Arc>, + processing_frontier_sender: tokio::sync::watch::Sender, ) -> 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 => { diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs index 94899a8..34dacb0 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs +++ b/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, + 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, oldest_pending_slot: std::option::Option, @@ -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, ) -> 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, + 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 { + 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, + 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, diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs index edd3695..0ffd209 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs @@ -1,7 +1,7 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs -// version: 32 +// version: 33 -//! External public, security, redaction and release-boundary hardening canaries through `v0.3.14-pre.008`. +//! External public, security, redaction and release-boundary hardening canaries through `v0.3.14-pre.009`. fn network(value: &'static str) -> std::option::Option { let result = ksp_store_lib::RawNetworkId::new(value); @@ -1185,3 +1185,52 @@ fn v0_3_14_pre_008_reconciliation_and_source_loss_are_target_coverage_gated_with assert!(!root.contains("repair"), "pre.008 leaked lower-case repair responsibility through crate root"); return; } + +#[test] +fn v0_3_14_pre_009_health_policy_is_present_future_coverage_gated_and_source_neutral() { + let continuity = include_str!("../src/continuity.rs"); + let resources = include_str!("../src/runtime_resources.rs"); + let snapshot = include_str!("../src/snapshot.rs"); + let root = include_str!("../src/lib.rs"); + for required in [ + "health_projection", + "future_target_coverage", + "continuity_frontier", + "has_open_gaps", + "is_covered_by_active_sources", + "source_failures_reconciled", + "failed_source_losses_reconciled", + ] { + assert!(continuity.contains(required) || snapshot.contains(required), "required pre.009 coverage-health guard missing: {required}"); + } + for required in [ + "continuity_policy_observed", + "continuity_has_open_gaps", + "snapshot.continuity_frontier_slot != snapshot.processing_frontier_slot", + "!snapshot.future_target_coverage", + "snapshot.source_reconnecting > 0", + "snapshot.source_active == snapshot.source_total", + "ksp_worker_api::WorkerHealth::Healthy", + "ksp_worker_api::WorkerHealth::Degraded", + "ksp_worker_api::WorkerHealth::Unhealthy", + "ksp_worker_api::WorkerState::Faulted", + ] { + assert!(snapshot.contains(required), "required pre.009 health projection guard missing: {required}"); + } + for required in [ + "continuity_contracts", + "contracts.health_projection", + "inventory.active_source_keys()", + "inventory.failed_source_keys()", + "aggregate.with_continuity_health", + "with_failed_source_losses_reconciled", + ] { + assert!(resources.contains(required), "required pre.009 inventory-health bridge missing: {required}"); + } + assert!(!root.contains("SourceHealthByKey")); + assert!(!root.contains("provider_health")); + assert!(!root.contains("endpoint_health")); + assert!(!root.contains("pub use self::continuity::RawTransactionIngestTargetCoverage")); + assert!(!snapshot.contains("WorkerHealth::Faulted"), "Faulted must remain a Worker lifecycle state rather than a new health enum variant"); + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs index 024c674..0871acd 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs @@ -1,7 +1,7 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs -// version: 27 +// version: 28 -//! Release-completeness canaries through the `v0.3.14-pre.008` continuity reconciliation tranche. +//! Release-completeness canaries through the `v0.3.14-pre.009` multi-source health-policy tranche. #[test] fn pre_010_production_module_inventory_is_exact() -> std::io::Result<()> { @@ -343,3 +343,27 @@ fn v0_3_14_pre_008_reconciliation_canaries_are_present_without_public_surface_or assert!(!root.contains("pub use self::continuity::RawTransactionIngestSourceLossDecision")); return; } + +#[test] +fn v0_3_14_pre_009_health_policy_canaries_are_present_without_public_source_identity_growth() { + let continuity_tests = include_str!("../unit_tests/continuity.rs"); + let hardening = include_str!("hardening.rs"); + let resource_tests = include_str!("../unit_tests/runtime_resources.rs"); + let snapshot_tests = include_str!("../unit_tests/snapshot.rs"); + let root = include_str!("../src/lib.rs"); + for required in [ + "pre_009_health_projection_distinguishes_future_coverage_from_open_gap_reconciliation", + "pre_009_health_projection_rejects_duplicate_and_unknown_active_sources", + "v0_3_14_pre_009_inventory_health_projection_tracks_reconciled_coverage", + "v0_3_14_pre_009_health_requires_present_and_future_coverage_before_healthy", + ] { + assert!( + continuity_tests.contains(required) || resource_tests.contains(required) || snapshot_tests.contains(required), + "required pre.009 health canary missing: {required}" + ); + } + assert!(hardening.contains("v0_3_14_pre_009_health_policy_is_present_future_coverage_gated_and_source_neutral")); + assert!(!root.contains("SourceHealthByKey")); + assert!(!root.contains("pub use self::continuity::RawTransactionIngestTargetCoverage")); + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/continuity.rs b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/continuity.rs index 8579acb..134c719 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/continuity.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/continuity.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/continuity.rs -// version: 6 +// version: 7 fn network() -> std::option::Option { return match ksp_store_lib::RawNetworkId::new("mainnet") { @@ -566,3 +566,84 @@ fn pre_008_known_reference_missing_moves_to_continuity_ledger_and_reconciles_fro assert_eq!(contracts.continuity_frontier(std::option::Option::Some(120)), std::option::Option::Some(120)); return; } + +#[test] +fn pre_009_health_projection_distinguishes_future_coverage_from_open_gap_reconciliation() { + let first = match capability(1, ksp_onchain_transport_lib::SolanaCommitment::Confirmed, exact_scope("standard_logs", 7)) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("first capability fixture unavailable"), + }; + let second = match capability(2, ksp_onchain_transport_lib::SolanaCommitment::Confirmed, exact_scope("standard_logs", 7)) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("second capability fixture unavailable"), + }; + let mut contracts = match crate::RawTransactionIngestContinuityContracts::new(std::vec![first, second]) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("continuity contracts fixture failed: {error}"), + }; + let healthy = match contracts.health_projection(&[[1_u8; 32], [2_u8; 32]], &[], std::option::Option::Some(120)) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("health projection failed: {error}"), + }; + assert_eq!(healthy, (std::option::Option::Some(120), false, true, true)); + let redundant_future = match contracts.health_projection(&[[2_u8; 32]], &[[1_u8; 32]], std::option::Option::Some(120)) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("redundant health projection failed: {error}"), + }; + assert_eq!(redundant_future, (std::option::Option::Some(120), false, true, false)); + assert!(contracts.record_source_loss_gap([1_u8; 32], 100, 110).is_ok()); + let pending = match contracts.health_projection(&[[2_u8; 32]], &[[1_u8; 32]], std::option::Option::Some(120)) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("pending health projection failed: {error}"), + }; + assert_eq!(pending, (std::option::Option::Some(99), true, true, false)); + assert!(contracts.record_coverage_epoch([2_u8; 32], 90, 120).is_ok()); + let reconciled = match contracts.health_projection(&[[2_u8; 32]], &[[1_u8; 32]], std::option::Option::Some(120)) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("reconciled health projection failed: {error}"), + }; + assert_eq!(reconciled, (std::option::Option::Some(120), false, true, true)); + let uncovered_future = match contracts.health_projection(&[], &[], std::option::Option::Some(120)) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("uncovered health projection failed: {error}"), + }; + assert_eq!(uncovered_future, (std::option::Option::Some(120), false, false, true)); + return; +} + +#[test] +fn pre_009_health_projection_rejects_duplicate_and_unknown_active_sources() { + let capability = match capability(1, ksp_onchain_transport_lib::SolanaCommitment::Confirmed, exact_scope("standard_logs", 7)) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("capability fixture unavailable"), + }; + let mut contracts = match crate::RawTransactionIngestContinuityContracts::new(std::vec![capability]) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("continuity contracts fixture failed: {error}"), + }; + let duplicate = match contracts.health_projection(&[[1_u8; 32], [1_u8; 32]], &[], std::option::Option::Some(20)) { + std::result::Result::Ok(_) => return, + std::result::Result::Err(value) => value, + }; + assert_eq!(duplicate.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); + assert!(duplicate.context().iter().any(|context| return context.value() == "continuity.health_active_set_invalid")); + let unknown = match contracts.health_projection(&[[9_u8; 32]], &[], std::option::Option::Some(20)) { + std::result::Result::Ok(_) => return, + std::result::Result::Err(value) => value, + }; + assert_eq!(unknown.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); + assert!(unknown.context().iter().any(|context| return context.value() == "continuity.health_active_source_unknown")); + let overlap = match contracts.health_projection(&[[1_u8; 32]], &[[1_u8; 32]], std::option::Option::Some(20)) { + std::result::Result::Ok(_) => return, + std::result::Result::Err(value) => value, + }; + assert_eq!(overlap.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); + assert!(overlap.context().iter().any(|context| return context.value() == "continuity.health_failed_set_invalid")); + let unknown_failed = match contracts.health_projection(&[], &[[9_u8; 32]], std::option::Option::Some(20)) { + std::result::Result::Ok(_) => return, + std::result::Result::Err(value) => value, + }; + assert_eq!(unknown_failed.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); + assert!(unknown_failed.context().iter().any(|context| return context.value() == "continuity.health_failed_source_unknown")); + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs index 4ad2ffd..33617a8 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs -// version: 30 +// version: 31 fn grpc_endpoint(cluster: &str) -> std::option::Option { return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider"); @@ -1051,15 +1051,21 @@ async fn v0_3_13_pre_002_runtime_resources_reject_duplicate_and_cross_network_so #[test] fn v0_3_13_pre_007_source_inventory_aggregates_frontier_and_lifecycle_conservatively() { let source_keys = std::vec![[1_u8; 32], [2_u8; 32]]; + let continuity_contracts = match supervisor_contracts(&[(1, "fixture-a", 1), (2, "fixture-b", 2)]) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; let inventory = std::sync::Arc::new(std::sync::Mutex::new(super::RawTransactionIngestSourceInventory::new(source_keys))); let (aggregate_sender, aggregate_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); let first = super::RawTransactionIngestSourceInventoryPublisher { aggregate_sender: aggregate_sender.clone(), + continuity_contracts: std::sync::Arc::clone(&continuity_contracts), entry_index: 0, inventory: std::sync::Arc::clone(&inventory), source_key: [1_u8; 32], }; - let second = super::RawTransactionIngestSourceInventoryPublisher { aggregate_sender, entry_index: 1, inventory, source_key: [2_u8; 32] }; + let second = + super::RawTransactionIngestSourceInventoryPublisher { aggregate_sender, continuity_contracts, entry_index: 1, inventory, source_key: [2_u8; 32] }; assert!( first .publish( @@ -1077,6 +1083,8 @@ fn v0_3_13_pre_007_source_inventory_aggregates_frontier_and_lifecycle_conservati assert_eq!(first_aggregate.source_active(), 1); assert_eq!(first_aggregate.source_reconnecting(), 0); assert_eq!(first_aggregate.source_failed(), 0); + assert!(first_aggregate.continuity_policy_observed()); + assert!(!first_aggregate.future_target_coverage()); assert!( second .publish( @@ -1097,6 +1105,9 @@ fn v0_3_13_pre_007_source_inventory_aggregates_frontier_and_lifecycle_conservati assert_eq!(aggregate.source_reconnect_total(), 4); assert_eq!(aggregate.source_replay_attempt_total(), 6); assert_eq!(aggregate.source_continuity_gap_total(), 1); + assert!(aggregate.continuity_policy_observed()); + assert!(!aggregate.continuity_has_open_gaps()); + assert!(!aggregate.future_target_coverage()); return; } @@ -1137,7 +1148,15 @@ async fn v0_3_13_pre_007_source_supervisor_joins_all_children_on_stop() { }); } let supervisor = tokio::spawn(async move { - return super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory).await; + return super::supervise_live_source_tasks( + stop_receiver, + source_stop_sender, + children, + contracts, + inventory, + tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0, + ) + .await; }); for _ in 0..64 { if active.load(std::sync::atomic::Ordering::Acquire) == 3 { @@ -1197,7 +1216,15 @@ async fn v0_3_13_pre_007_source_failure_stops_and_joins_sibling_sources() { } } }); - let result = super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory).await; + let result = super::supervise_live_source_tasks( + stop_receiver, + source_stop_sender, + children, + contracts, + inventory, + tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0, + ) + .await; assert!(result.is_err()); assert!(!sibling_active.load(std::sync::atomic::Ordering::Acquire)); return; @@ -3804,7 +3831,15 @@ async fn v0_3_13_pre_011_aborting_outer_source_supervisor_aborts_nested_source_t return std::future::pending::<([u8; 32], ksp_core_lib::Result<()>)>().await; }); let supervisor = tokio::spawn(async move { - return super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory).await; + return super::supervise_live_source_tasks( + stop_receiver, + source_stop_sender, + children, + contracts, + inventory, + tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0, + ) + .await; }); for _ in 0..64 { if active.load(std::sync::atomic::Ordering::Acquire) == 1 { @@ -3863,7 +3898,15 @@ async fn v0_3_13_pre_011_stop_racing_ready_source_fault_preserves_fault_and_join } }); let supervisor = tokio::spawn(async move { - return super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory).await; + return super::supervise_live_source_tasks( + stop_receiver, + source_stop_sender, + children, + contracts, + inventory, + tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0, + ) + .await; }); for _ in 0..64 { if sibling_active.load(std::sync::atomic::Ordering::Acquire) == 1 { @@ -3929,18 +3972,22 @@ async fn v0_3_14_pre_008_proven_full_ledger_epoch_survives_filtered_peer_loss_wi } } }); + let (health_sender, health_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); let supervisor = tokio::spawn(async move { - return super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory).await; + return super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory, health_sender).await; }); for _ in 0..64 { - if sibling_active.load(std::sync::atomic::Ordering::Acquire) { + if sibling_active.load(std::sync::atomic::Ordering::Acquire) && health_receiver.borrow().failed_source_losses_reconciled() { break; } tokio::task::yield_now().await; } assert!(sibling_active.load(std::sync::atomic::Ordering::Acquire)); - tokio::task::yield_now().await; - assert!(sibling_active.load(std::sync::atomic::Ordering::Acquire)); + let health = *health_receiver.borrow(); + assert!(health.continuity_policy_observed()); + assert!(!health.continuity_has_open_gaps()); + assert!(health.future_target_coverage()); + assert!(health.failed_source_losses_reconciled()); stop_sender.send_replace(true); let result = match supervisor.await { std::result::Result::Ok(value) => value, @@ -4036,3 +4083,62 @@ fn v0_3_14_pre_003_websocket_incident_without_observed_slot_is_unbounded_and_cou assert!(reporter.observe_websocket_continuity(crate::RawTransactionIngestSourceState::Active, 0, 0).is_err()); return; } + +#[test] +fn v0_3_14_pre_009_inventory_health_projection_tracks_reconciled_coverage() { + let source_keys = std::vec![[1_u8; 32], [2_u8; 32]]; + let continuity_contracts = match supervisor_contracts(&[(1, "standard_logs", 7), (2, "standard_logs", 7)]) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let inventory = std::sync::Arc::new(std::sync::Mutex::new(super::RawTransactionIngestSourceInventory::new(source_keys))); + let (aggregate_sender, aggregate_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); + let first = super::RawTransactionIngestSourceInventoryPublisher { + aggregate_sender: aggregate_sender.clone(), + continuity_contracts: std::sync::Arc::clone(&continuity_contracts), + entry_index: 0, + inventory: std::sync::Arc::clone(&inventory), + source_key: [1_u8; 32], + }; + let second = super::RawTransactionIngestSourceInventoryPublisher { + aggregate_sender, + continuity_contracts: std::sync::Arc::clone(&continuity_contracts), + entry_index: 1, + inventory, + source_key: [2_u8; 32], + }; + let active = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(60), std::option::Option::None) + .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), 0, 0, 0); + assert!(first.publish(active).is_ok()); + assert!(second.publish(active).is_ok()); + let healthy = *aggregate_receiver.borrow(); + assert!(healthy.continuity_policy_observed()); + assert_eq!(healthy.continuity_frontier_slot(), std::option::Option::Some(60)); + assert!(!healthy.continuity_has_open_gaps()); + assert!(healthy.future_target_coverage()); + { + let mut contracts = match continuity_contracts.lock() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + assert!(contracts.record_source_loss_gap([2_u8; 32], 50, 55).is_ok()); + } + assert!(first.publish(active).is_ok()); + let pending = *aggregate_receiver.borrow(); + assert_eq!(pending.continuity_frontier_slot(), std::option::Option::Some(49)); + assert!(pending.continuity_has_open_gaps()); + assert!(pending.future_target_coverage()); + { + let mut contracts = match continuity_contracts.lock() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + assert!(contracts.record_coverage_epoch([1_u8; 32], 40, 60).is_ok()); + } + assert!(first.publish(active).is_ok()); + let reconciled = *aggregate_receiver.borrow(); + assert_eq!(reconciled.continuity_frontier_slot(), std::option::Option::Some(60)); + assert!(!reconciled.continuity_has_open_gaps()); + assert!(reconciled.future_target_coverage()); + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs index a10e9c9..fe0642a 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs -// version: 5 +// version: 6 fn snapshot_foundation_with_source_total( source_total: usize, @@ -249,3 +249,58 @@ fn v0_3_13_pre_010_multi_source_counts_and_health_are_conservative_and_source_ne assert_eq!(unhealthy.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy); return; } + +#[test] +fn v0_3_14_pre_009_health_requires_present_and_future_coverage_before_healthy() { + let (mut publisher, source) = match snapshot_foundation_with_source_total(2) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let reconnecting = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(50), std::option::Option::None) + .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting), 1, 1, 1) + .with_source_counts(2, 1, 1, 0) + .with_continuity_health(std::option::Option::Some(50), false, true) + .with_failed_source_losses_reconciled(true); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, reconnecting).is_ok()); + assert_eq!(source.current().worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy); + let gap_pending = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(50), std::option::Option::None) + .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), 1, 1, 1) + .with_source_counts(2, 2, 0, 0) + .with_continuity_health(std::option::Option::Some(49), true, true) + .with_failed_source_losses_reconciled(true); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, gap_pending).is_ok()); + assert_eq!(source.current().worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy); + let healthy = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(50), std::option::Option::None) + .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), 1, 1, 1) + .with_source_counts(2, 2, 0, 0) + .with_continuity_health(std::option::Option::Some(50), false, true) + .with_failed_source_losses_reconciled(true); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, healthy).is_ok()); + assert_eq!(source.current().worker_snapshot().health(), ksp_worker_api::WorkerHealth::Healthy); + let failed_unreconciled = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(50), std::option::Option::None) + .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Failed), 1, 1, 1) + .with_source_counts(2, 1, 0, 1) + .with_continuity_health(std::option::Option::Some(50), false, true) + .with_failed_source_losses_reconciled(false); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, failed_unreconciled).is_ok()); + assert_eq!(source.current().worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy); + let degraded = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(50), std::option::Option::None) + .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Failed), 1, 1, 1) + .with_source_counts(2, 1, 0, 1) + .with_continuity_health(std::option::Option::Some(50), false, true) + .with_failed_source_losses_reconciled(true); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, degraded).is_ok()); + assert_eq!(source.current().worker_snapshot().health(), ksp_worker_api::WorkerHealth::Degraded); + let future_uncovered = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(50), std::option::Option::None) + .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Failed), 1, 1, 1) + .with_source_counts(2, 1, 0, 1) + .with_continuity_health(std::option::Option::Some(50), false, false) + .with_failed_source_losses_reconciled(true); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, future_uncovered).is_ok()); + assert_eq!(source.current().worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy); + assert!(publisher.publish_state(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED), 0, 0).is_ok()); + let faulted = source.current(); + assert!(matches!(faulted.worker_snapshot().state(), ksp_worker_api::WorkerState::Faulted(_))); + assert_eq!(faulted.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy); + return; +} diff --git a/deltas/0.3.14/pre.009.md b/deltas/0.3.14/pre.009.md new file mode 100644 index 0000000..533c27e --- /dev/null +++ b/deltas/0.3.14/pre.009.md @@ -0,0 +1,270 @@ + + + +# Delta `0.3.14-pre.009` — health policy multi-source + +## Base requise + +```text +0.3.14-pre.008-fix.002 +workspace.package.version = 0.3.14-pre.8.fix.2 +deltas/0.3.14/pre.008-fix.002.md présent +``` + +## Gate de la base + +Le gate opérateur de `0.3.14-pre.008-fix.002` est validé avant ouverture de cette tranche : + +```text +cargo fmt --all : PASS +cargo fmt --all -- --check : PASS +audit Rust workspace rules : PASS +audit Markdown tables : PASS +cargo check --workspace : PASS +cargo clippy --workspace --all-targets --all-features -- -D warnings : PASS +cargo test -p ksp-worker-raw-transaction-ingest-lib --all-targets --all-features : PASS +``` + +Le gate Worker comprend notamment : + +```text +139 unit tests : PASS +cross_layer_completeness : 8 PASS +dependency_boundary : 19 PASS +hardening : 33 PASS +public_api : 20 PASS +release_completeness : 10 PASS +``` + +## Objectif + +Implémenter strictement la tranche `pre.009` du plan `035` : + +```text +projeter Healthy / Degraded / Unhealthy depuis des preuves de coverage présentes et futures +conserver Faulted comme état lifecycle terminal du Worker +ne jamais rendre Healthy sur le seul fait qu'un Transport ait reconnecté +rendre Healthy uniquement lorsque toutes les sources attendues sont Active et que les gaps sont fermés +rendre Degraded lorsqu'une source reste perdue mais que le présent est réconcilié et TargetCoverage reste couvert pour la suite +rendre Unhealthy pendant une réconciliation non prouvée, avec gap ouvert ou coverage future insuffisante +ne créer aucune health source/provider spécifique publique +``` + +## Projection de health fondée sur la continuité + +Le ledger de `pre.008` expose maintenant une projection privée et source-neutral : + +```text +continuity_frontier +has_open_gaps +future_target_coverage +failed_source_losses_reconciled +``` + +Pour chaque source projetée `Failed`, `failed_source_losses_reconciled` exige un gap `SourceFailure` effectivement enregistré puis fermé par une preuve run-local. Une source simplement marquée `Failed` dans l'inventaire ne suffit donc jamais à autoriser `Degraded`. + +`future_target_coverage` est calculé uniquement depuis les sources réellement `Active` et les relations `Exact` / `Superset` déjà validées par `TargetCoverage`. + +La configuration d'une source ne constitue toujours pas une preuve historique. Le présent reste gouverné par : + +```text +aucun gap ouvert +continuity frontier == processing frontier +``` + +Les deux dimensions sont volontairement distinctes : + +```text +coverage future disponible != gap historique réparé +gap historique réparé != coverage future encore disponible +``` + +## Pont Inventory -> Continuity -> Snapshot + +Chaque publication agrégée de l'inventaire multi-source calcule désormais, après libération du lock Inventory : + +```text +ensemble exact des sources Active +processing frontier agrégé +projection continuity/TargetCoverage sous le lock continuity séparé +projection health privée ajoutée au message latest-value +``` + +Aucun `MutexGuard` n'est conservé à travers un `await` et aucun client Transport/Store n'est exposé dans la snapshot publique. + +La source HTTP full-ledger republie sa projection après l'enregistrement d'un coverage epoch afin qu'une fermeture de gap puisse mettre à jour la health sans attendre une transaction ou un slot produit supplémentaire. + +## Politique Running + +Lorsque la projection de continuité est disponible, la health suit la politique conservative suivante : + +```text +source Reconnecting + -> Unhealthy tant que la continuité de l'incident n'est pas explicitement réconciliée + +gap ouvert +ou continuity frontier != processing frontier +ou TargetCoverage future non couverte par les sources Active + -> Unhealthy + +toutes les sources attendues Active ++ aucun gap ouvert ++ continuity frontier rattrapé ++ TargetCoverage future couverte + -> Healthy + +une source reste perdue/Failed ++ son gap SourceFailure a été enregistré et réconcilié ++ aucune source Reconnecting ++ aucun gap ouvert ++ continuity frontier rattrapé ++ TargetCoverage future couverte + -> Degraded +``` + +Une source perdue n'est donc jamais transformée en source optionnelle. `Degraded` indique uniquement que le service du run reste couvert malgré son absence ; les observations spécifiques que cette source aurait produites ne sont pas inventées. + +## Faulted reste un état lifecycle + +`ksp-worker-api::WorkerHealth` reste inchangé et conserve : + +```text +Unknown +Healthy +Degraded +Unhealthy +``` + +La quatrième politique terminale du plan est représentée par : + +```text +WorkerState::Faulted(error_code) +WorkerHealth::Unhealthy +``` + +Aucune nouvelle variante `WorkerHealth::Faulted` ni extension de l'API générique Worker n'est introduite. + +## Retour à Healthy + +Le retour à `Healthy` exige simultanément : + +```text +toutes les sources attendues projetées Active +aucune source projetée Reconnecting +aucun gap run-local ouvert +continuity frontier == processing frontier +TargetCoverage future couverte par les sources Active +``` + +Le simple passage Transport `Reconnecting -> Active` ne constitue pas une preuve suffisante si le continuity ledger n'est pas réconcilié. + +## Tests ajoutés + +Les unit tests couvrent notamment : + +```text +projection health séparant future TargetCoverage et gap historique ouvert +rejet des ensembles Active dupliqués ou inconnus +pont Inventory -> continuity health avec gap puis coverage epoch de réconciliation +Reconnecting -> Unhealthy tant que la preuve de continuité manque +sources Active + gap ouvert -> Unhealthy +sources Active + gaps fermés + coverage future -> Healthy +source Failed sans gap SourceFailure réconcilié -> Unhealthy +source Failed avec gap SourceFailure réconcilié et présent/futur couverts -> Degraded +publication supervisor immédiate après réconciliation réussie de la perte de source +coverage future insuffisante -> Unhealthy +WorkerState::Faulted -> WorkerHealth::Unhealthy +``` + +Les canaris `hardening` et `release_completeness` vérifient en plus : + +```text +health fondée sur continuity frontier + gaps + TargetCoverage +aucune health provider/source-key spécifique publique +aucune variante WorkerHealth::Faulted ajoutée +aucune croissance de la surface publique TargetCoverage +``` + +## Hors périmètre inchangé + +```text +aucune fairness spécifique nominal/repair de pre.010 +aucune observabilité publique détaillée des gaps/repair de pre.011 +aucun nouveau mécanisme de replay/reconnect Worker +aucun respawn de source +aucun EARLY/shred +aucun Job Backfill depuis Worker +aucun nouveau provider ou SDK +``` + +## Fichiers ajoutés + +```text +deltas/0.3.14/pre.009.md +``` + +## Fichiers modifiés + +```text +Cargo.toml +crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs +crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs +crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/continuity.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs +``` + +## Fichiers supprimés + +```text +aucun +``` + +## Version Cargo + +Conformément à `VER-ID-009` : + +```text +header Cargo.toml : 578 -> 579 +workspace.package.version : 0.3.14-pre.8.fix.2 -> 0.3.14-pre.9 +``` + +Versions des fichiers modifiés : + +```text +continuity.rs : 8 -> 9 +runtime_resources.rs : 36 -> 37 +snapshot.rs : 5 -> 6 +unit_tests/continuity.rs : 6 -> 7 +unit_tests/runtime_resources.rs : 30 -> 31 +unit_tests/snapshot.rs : 5 -> 6 +tests/hardening.rs : 32 -> 33 +tests/release_completeness.rs : 27 -> 28 +``` + +## Validation exécutée dans l'environnement de préparation + +```text +python3 scripts/audit_rust_workspace_rules.py : PASS +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas : PASS +scan des frontières Worker/Transport/Backfill/Store : PASS +scan de la crate-root historique : PASS +comparaison exacte pre.008-fix.002 -> pre.009 : PASS +``` + +Les gates Cargo ne sont pas déclarés PASS dans l'environnement de préparation lorsqu'ils ne peuvent pas y être exécutés. Ils restent obligatoires côté opérateur avant `pre.010`. + +## Décisions prises + +```text +la health est une projection de preuves run-local, pas un statut provider +le futur est prouvé par TargetCoverage sur les sources Active +le présent est prouvé par continuity frontier + absence de gap ouvert +Reconnecting reste Unhealthy tant que l'incident n'est pas réconcilié +Failed ne devient Degraded qu'après enregistrement et fermeture effective de son gap SourceFailure +une source perdue mais réellement redondante maintient Degraded, jamais Healthy +Faulted reste un WorkerState terminal avec WorkerHealth::Unhealthy +```