diff --git a/Cargo.toml b/Cargo.toml index 339ab74..f827029 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 583 +# version: 584 [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.10.fix.1" +version = "0.3.14-pre.11" 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 34d3ed8..022efe2 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: 9 +// version: 10 /// 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; @@ -105,6 +105,17 @@ impl RawTransactionIngestGapReason { Self::WebSocketReconnect, Self::YellowstoneRetention, ]; + + const fn public(self) -> crate::RawTransactionIngestGapReason { + return match self { + Self::HttpProducedBlockUnavailable => crate::RawTransactionIngestGapReason::HttpProducedBlockUnavailable, + Self::KnownReferenceMissing => crate::RawTransactionIngestGapReason::KnownReferenceMissing, + Self::SourceFailure => crate::RawTransactionIngestGapReason::SourceFailure, + Self::TransportOverflow => crate::RawTransactionIngestGapReason::TransportOverflow, + Self::WebSocketReconnect => crate::RawTransactionIngestGapReason::WebSocketReconnect, + Self::YellowstoneRetention => crate::RawTransactionIngestGapReason::YellowstoneRetention, + }; + } } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -121,6 +132,15 @@ impl RawTransactionIngestGapState { const fn is_open(self) -> bool { return !matches!(self, Self::Repaired); } + + const fn public(self) -> crate::RawTransactionIngestGapState { + return match self { + Self::Pending => crate::RawTransactionIngestGapState::Pending, + Self::Repairing => crate::RawTransactionIngestGapState::Repairing, + Self::Repaired => crate::RawTransactionIngestGapState::Repaired, + Self::Unresolved => crate::RawTransactionIngestGapState::Unresolved, + }; + } } /// Private supervisor decision for one terminal live-source loss after conservative continuity reconciliation. @@ -317,6 +337,20 @@ struct RawTransactionIngestGap { reason: RawTransactionIngestGapReason, source_key: [u8; 32], state: RawTransactionIngestGapState, + last_method: std::option::Option, +} + +impl RawTransactionIngestGap { + fn snapshot(&self) -> crate::RawTransactionIngestGapSnapshot { + return crate::RawTransactionIngestGapSnapshot::new( + crate::RawTransactionIngestGapId::new(self.gap_id.0), + self.range.start_slot(), + self.range.end_slot(), + self.state.public(), + self.reason.public(), + self.last_method, + ); + } } struct RawTransactionIngestGapLedger { @@ -410,23 +444,134 @@ impl RawTransactionIngestGapLedger { reason, source_key, state: RawTransactionIngestGapState::Pending, + last_method: std::option::Option::None, }); return self.validate_invariants(); } fn reconcile_with_coverage_epochs(&mut self, coverage_epochs: &RawTransactionIngestCoverageEpochLedger) -> ksp_core_lib::Result<()> { for gap in &mut self.gaps { - if !gap.state.is_open() { + if !gap.state.is_open() || gap.state == RawTransactionIngestGapState::Unresolved { continue; } let requirement = RawTransactionIngestCoverageRequirement { commitment: gap.commitment, scope: gap.coverage_requirement.clone() }; if coverage_epochs.redundant_relation_for_gap(gap.source_key, &requirement, gap.range).is_some() { gap.state = RawTransactionIngestGapState::Repaired; + gap.last_method = std::option::Option::Some(crate::RawTransactionIngestRepairMethod::RedundantCoverage); } } return self.validate_invariants(); } + fn observability_projection(&self) -> ksp_core_lib::Result { + let mut open_gap_count = 0_usize; + let mut repairing_gap_count = 0_usize; + let mut repaired_gap_total = 0_u64; + let mut unresolved_gap_total = 0_u64; + let mut replay_repair_total = 0_u64; + let mut redundant_coverage_repair_total = 0_u64; + let mut http_scan_repair_total = 0_u64; + let mut repair_block_fetch_total = 0_u64; + let mut repair_transaction_hydration_total = 0_u64; + let mut oldest_open_gap_start_slot = std::option::Option::None; + let mut gaps = std::vec::Vec::with_capacity(MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS); + for gap in &self.gaps { + if gap.state.is_open() { + open_gap_count = match open_gap_count.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::counter_exhausted_error("continuity.open_gap_count")), + }; + if gap.state == RawTransactionIngestGapState::Repairing { + repairing_gap_count = match repairing_gap_count.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(crate::counter_exhausted_error("continuity.repairing_gap_count")); + }, + }; + } + if gap.state == RawTransactionIngestGapState::Unresolved { + unresolved_gap_total = match unresolved_gap_total.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(crate::counter_exhausted_error("continuity.unresolved_gap_total")); + }, + }; + } + oldest_open_gap_start_slot = match oldest_open_gap_start_slot { + std::option::Option::Some(value) => std::option::Option::Some(value.min(gap.range.start_slot())), + std::option::Option::None => std::option::Option::Some(gap.range.start_slot()), + }; + gaps.push(gap.snapshot()); + continue; + } + repaired_gap_total = match repaired_gap_total.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::counter_exhausted_error("continuity.repaired_gap_total")), + }; + match gap.last_method { + std::option::Option::Some(crate::RawTransactionIngestRepairMethod::Replay) => { + replay_repair_total = match replay_repair_total.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(crate::counter_exhausted_error("continuity.replay_repair_total")); + }, + }; + }, + std::option::Option::Some(crate::RawTransactionIngestRepairMethod::RedundantCoverage) => { + redundant_coverage_repair_total = match redundant_coverage_repair_total.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(crate::counter_exhausted_error("continuity.redundant_coverage_repair_total")); + }, + }; + }, + std::option::Option::Some(crate::RawTransactionIngestRepairMethod::HttpScan) => { + http_scan_repair_total = match http_scan_repair_total.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(crate::counter_exhausted_error("continuity.http_scan_repair_total")); + }, + }; + }, + std::option::Option::Some(crate::RawTransactionIngestRepairMethod::BlockFetch) => { + repair_block_fetch_total = match repair_block_fetch_total.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(crate::counter_exhausted_error("continuity.repair_block_fetch_total")); + }, + }; + }, + std::option::Option::Some(crate::RawTransactionIngestRepairMethod::TransactionHydration) => { + repair_transaction_hydration_total = match repair_transaction_hydration_total.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(crate::counter_exhausted_error("continuity.repair_transaction_hydration_total")); + }, + }; + }, + std::option::Option::None => {}, + } + } + if gaps.len() < MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS { + for gap in self.gaps.iter().rev() { + if gap.state != RawTransactionIngestGapState::Repaired { + continue; + } + gaps.push(gap.snapshot()); + if gaps.len() == MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS { + break; + } + } + } + gaps.sort_unstable_by_key(|gap| return gap.gap_id().value()); + return std::result::Result::Ok( + crate::RawTransactionIngestContinuitySnapshotProjection::empty() + .with_gap_state(gaps, open_gap_count, repairing_gap_count, oldest_open_gap_start_slot) + .with_recovery_totals(repaired_gap_total, unresolved_gap_total, replay_repair_total, redundant_coverage_repair_total, http_scan_repair_total) + .with_material_totals(repair_block_fetch_total, repair_transaction_hydration_total), + ); + } + fn validate_invariants(&self) -> ksp_core_lib::Result<()> { if self.next_gap_id == 0 { return std::result::Result::Err(crate::runtime_error("continuity.gap_id_exhausted")); @@ -923,6 +1068,11 @@ impl crate::RawTransactionIngestContinuityContracts { return self.gap_ledger.continuity_frontier(processing_frontier_slot); } + /// Returns the bounded source-neutral continuity observability projection for the latest Worker snapshot. + pub(crate) fn observability_projection(&self) -> ksp_core_lib::Result { + return self.gap_ledger.observability_projection(); + } + /// 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 diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs index d9cd5a7..e4d3b69 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs +++ b/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: 32 +// version: 33 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -99,6 +99,16 @@ 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; +/// Stable source-neutral identifier of one run-local continuity gap. +pub use self::snapshot::RawTransactionIngestGapId; +/// Source-neutral reason why one run-local continuity gap was opened. +pub use self::snapshot::RawTransactionIngestGapReason; +/// Safe source-neutral snapshot of one recent or still-open run-local continuity gap. +pub use self::snapshot::RawTransactionIngestGapSnapshot; +/// Source-neutral lifecycle state of one run-local continuity gap. +pub use self::snapshot::RawTransactionIngestGapState; +/// Source-neutral mechanism last used while attempting to close one continuity gap. +pub use self::snapshot::RawTransactionIngestRepairMethod; /// 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. @@ -150,6 +160,8 @@ pub(crate) use self::persistence::RawTransactionIngestPersistencePort; pub(crate) use self::persistence::persist_raw_transaction_ingest_acquisition; /// Persists one canonical acquisition through the bounded cross-source convergence cache. pub(crate) use self::persistence::persist_raw_transaction_ingest_converged_acquisition; +/// Private bounded continuity-observability projection carried into the concrete Worker snapshot. +pub(crate) use self::snapshot::RawTransactionIngestContinuitySnapshotProjection; /// Private latest-value processing-frontier projection emitted by the productive source task. pub(crate) use self::snapshot::RawTransactionIngestProcessingFrontierProjection; /// Private latest-value publisher and checked counter owner shared by the Worker supervisor. diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs index f67a4ff..a98e751 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs +++ b/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: 17 +// version: 18 type PersistencePort = std::sync::Arc; type PersistenceTasks = tokio::task::JoinSet>; @@ -647,7 +647,7 @@ async fn wait_processing_frontier( if receiver.changed().await.is_err() { return std::option::Option::None; } - return std::option::Option::Some(*receiver.borrow_and_update()); + return std::option::Option::Some(receiver.borrow_and_update().clone()); } fn validate_store_network(settings: &crate::RawTransactionIngestSettings, store_network: &ksp_store_lib::RawNetworkId) -> ksp_core_lib::Result<()> { 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 64d90a3..aae5262 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: 40 +// version: 41 use sha2::Digest; // rust-rules: trait-import @@ -432,7 +432,7 @@ impl RawTransactionIngestLiveSource { tokio::select! { biased; result = &mut source_future => { - let latest = *source_frontier_receiver.borrow_and_update(); + let latest = source_frontier_receiver.borrow_and_update().clone(); let terminal_state = if result.is_ok() { crate::RawTransactionIngestSourceState::Closed } else { @@ -447,7 +447,7 @@ impl RawTransactionIngestLiveSource { if changed.is_err() { return std::result::Result::Err(crate::runtime_error("source.frontier_channel_closed")); } - if let std::result::Result::Err(error) = inventory_publisher.publish(*source_frontier_receiver.borrow_and_update()) { + if let std::result::Result::Err(error) = inventory_publisher.publish(source_frontier_receiver.borrow_and_update().clone()) { return std::result::Result::Err(error); } } @@ -637,17 +637,24 @@ fn source_inventory_health_projection( }; (aggregate, active_source_keys, failed_source_keys) }; - let (continuity_frontier_slot, continuity_has_open_gaps, future_target_coverage, failed_source_losses_reconciled) = { + let (continuity_frontier_slot, continuity_has_open_gaps, future_target_coverage, failed_source_losses_reconciled, continuity_snapshot) = { 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()) { + let health = 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), - } + }; + let continuity_snapshot = match contracts.observability_projection() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + (health.0, health.1, health.2, health.3, continuity_snapshot) }; - let aggregate = aggregate.with_continuity_health(continuity_frontier_slot, continuity_has_open_gaps, future_target_coverage); + let aggregate = aggregate + .with_continuity_health(continuity_frontier_slot, continuity_has_open_gaps, future_target_coverage) + .with_continuity_snapshot(continuity_snapshot); return std::result::Result::Ok(aggregate.with_failed_source_losses_reconciled(failed_source_losses_reconciled)); } 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 be8d404..d4fbc65 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: 8 +// version: 9 /// Runtime-neutral boxed future resolving to one newer concrete RAW transaction ingest Worker snapshot. pub type RawTransactionIngestSnapshotFuture<'a> = @@ -20,14 +20,267 @@ pub enum RawTransactionIngestSourceState { Failed, } -/// Private latest-value source-processing projection emitted by the productive source task. +/// Stable source-neutral identifier of one run-local continuity gap. +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub struct RawTransactionIngestGapId(u64); + +impl crate::RawTransactionIngestGapId { + /// Creates one internal gap identifier after the continuity ledger validated monotonicity. + pub(crate) const fn new(value: u64) -> Self { + return Self(value); + } + + /// Returns the run-local numeric identifier. + #[must_use] + pub const fn value(self) -> u64 { + return self.0; + } +} + +/// Source-neutral lifecycle state of one run-local continuity gap. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum RawTransactionIngestGapState { + /// The gap is known and waiting for an admissible recovery proof. + Pending, + /// One bounded recovery mechanism is actively processing the gap. + Repairing, + /// The complete required interval has been proven covered. + Repaired, + /// The gap remains unresolved after the admissible bounded mechanisms are exhausted. + Unresolved, +} + +/// Source-neutral reason why one run-local continuity gap was opened. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum RawTransactionIngestGapReason { + /// A slot was proven produced but its block material remained unavailable. + HttpProducedBlockUnavailable, + /// A previously observed transaction reference could not yet be materialized. + KnownReferenceMissing, + /// A logical live source became terminal before continuity was proven. + SourceFailure, + /// Transport reported bounded notification loss or overflow. + TransportOverflow, + /// A WebSocket reconnect opened a bounded continuity incident. + WebSocketReconnect, + /// Yellowstone replay retention could not cover the requested boundary. + YellowstoneRetention, +} + +/// Source-neutral mechanism last used while attempting to close one continuity gap. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum RawTransactionIngestRepairMethod { + /// Transport-owned replay supplied material later proven sufficient. + Replay, + /// A distinct live source supplied explicit interval coverage. + RedundantCoverage, + /// A bounded HTTP slot/block discovery scan supplied coverage evidence. + HttpScan, + /// A produced slot required direct block material retrieval. + BlockFetch, + /// A known transaction reference required observed transaction hydration. + TransactionHydration, +} + +/// Safe source-neutral snapshot of one recent or still-open run-local continuity gap. #[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct RawTransactionIngestGapSnapshot { + gap_id: crate::RawTransactionIngestGapId, + start_slot: u64, + end_slot: u64, + state: crate::RawTransactionIngestGapState, + reason: crate::RawTransactionIngestGapReason, + last_method: std::option::Option, +} + +impl crate::RawTransactionIngestGapSnapshot { + /// Creates one safe gap projection from already validated continuity-ledger state. + pub(crate) const fn new( + gap_id: crate::RawTransactionIngestGapId, + start_slot: u64, + end_slot: u64, + state: crate::RawTransactionIngestGapState, + reason: crate::RawTransactionIngestGapReason, + last_method: std::option::Option, + ) -> Self { + return Self { gap_id, start_slot, end_slot, state, reason, last_method }; + } + + /// Returns the run-local gap identifier. + #[must_use] + pub const fn gap_id(&self) -> crate::RawTransactionIngestGapId { + return self.gap_id; + } + + /// Returns the inclusive first slot of the gap. + #[must_use] + pub const fn start_slot(&self) -> u64 { + return self.start_slot; + } + + /// Returns the inclusive last slot of the gap. + #[must_use] + pub const fn end_slot(&self) -> u64 { + return self.end_slot; + } + + /// Returns the source-neutral gap lifecycle state. + #[must_use] + pub const fn state(&self) -> crate::RawTransactionIngestGapState { + return self.state; + } + + /// Returns the source-neutral reason that opened the gap. + #[must_use] + pub const fn reason(&self) -> crate::RawTransactionIngestGapReason { + return self.reason; + } + + /// Returns the last bounded recovery mechanism recorded for this gap, when any. + #[must_use] + pub const fn last_method(&self) -> std::option::Option { + return self.last_method; + } +} + +/// Private bounded continuity-observability projection carried into the public latest-value Worker snapshot. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct RawTransactionIngestContinuitySnapshotProjection { + gaps: std::vec::Vec, + open_gap_count: usize, + repairing_gap_count: usize, + repaired_gap_total: u64, + unresolved_gap_total: u64, + replay_repair_total: u64, + redundant_coverage_repair_total: u64, + http_scan_repair_total: u64, + repair_block_fetch_total: u64, + repair_transaction_hydration_total: u64, + oldest_open_gap_start_slot: std::option::Option, +} + +impl crate::RawTransactionIngestContinuitySnapshotProjection { + /// Returns the empty continuity-observability projection before any gap exists. + pub(crate) fn empty() -> Self { + return Self { + gaps: std::vec::Vec::new(), + open_gap_count: 0, + repairing_gap_count: 0, + repaired_gap_total: 0, + unresolved_gap_total: 0, + replay_repair_total: 0, + redundant_coverage_repair_total: 0, + http_scan_repair_total: 0, + repair_block_fetch_total: 0, + repair_transaction_hydration_total: 0, + oldest_open_gap_start_slot: std::option::Option::None, + }; + } + + /// Returns a copy carrying bounded gap details and current open-gap aggregates. + pub(crate) fn with_gap_state( + mut self, + gaps: std::vec::Vec, + open_gap_count: usize, + repairing_gap_count: usize, + oldest_open_gap_start_slot: std::option::Option, + ) -> Self { + self.gaps = gaps; + self.open_gap_count = open_gap_count; + self.repairing_gap_count = repairing_gap_count; + self.oldest_open_gap_start_slot = oldest_open_gap_start_slot; + return self; + } + + /// Returns a copy carrying cumulative source-neutral gap recovery totals. + pub(crate) fn with_recovery_totals( + mut self, + repaired_gap_total: u64, + unresolved_gap_total: u64, + replay_repair_total: u64, + redundant_coverage_repair_total: u64, + http_scan_repair_total: u64, + ) -> Self { + self.repaired_gap_total = repaired_gap_total; + self.unresolved_gap_total = unresolved_gap_total; + self.replay_repair_total = replay_repair_total; + self.redundant_coverage_repair_total = redundant_coverage_repair_total; + self.http_scan_repair_total = http_scan_repair_total; + return self; + } + + /// Returns a copy carrying cumulative bounded material-recovery totals. + pub(crate) fn with_material_totals(mut self, repair_block_fetch_total: u64, repair_transaction_hydration_total: u64) -> Self { + self.repair_block_fetch_total = repair_block_fetch_total; + self.repair_transaction_hydration_total = repair_transaction_hydration_total; + return self; + } + + /// Returns the bounded recent/open gap projections. + pub(crate) fn gaps(&self) -> &[crate::RawTransactionIngestGapSnapshot] { + return self.gaps.as_slice(); + } + + /// Returns the current number of open gaps. + pub(crate) const fn open_gap_count(&self) -> usize { + return self.open_gap_count; + } + + /// Returns the current number of actively recovering gaps. + pub(crate) const fn repairing_gap_count(&self) -> usize { + return self.repairing_gap_count; + } + + /// Returns the cumulative repaired-gap total. + pub(crate) const fn repaired_gap_total(&self) -> u64 { + return self.repaired_gap_total; + } + + /// Returns the cumulative unresolved-gap total. + pub(crate) const fn unresolved_gap_total(&self) -> u64 { + return self.unresolved_gap_total; + } + + /// Returns the cumulative replay recovery total. + pub(crate) const fn replay_repair_total(&self) -> u64 { + return self.replay_repair_total; + } + + /// Returns the cumulative distinct-source coverage recovery total. + pub(crate) const fn redundant_coverage_repair_total(&self) -> u64 { + return self.redundant_coverage_repair_total; + } + + /// Returns the cumulative bounded HTTP scan recovery total. + pub(crate) const fn http_scan_repair_total(&self) -> u64 { + return self.http_scan_repair_total; + } + + /// Returns the cumulative direct block material recovery total. + pub(crate) const fn repair_block_fetch_total(&self) -> u64 { + return self.repair_block_fetch_total; + } + + /// Returns the cumulative transaction hydration recovery total. + pub(crate) const fn repair_transaction_hydration_total(&self) -> u64 { + return self.repair_transaction_hydration_total; + } + + /// Returns the oldest first slot among currently open gaps. + pub(crate) const fn oldest_open_gap_start_slot(&self) -> std::option::Option { + return self.oldest_open_gap_start_slot; + } +} + +/// Private latest-value source-processing projection emitted by the productive source task. +#[derive(Clone, 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, + continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection, hydration_pending: usize, processing_frontier_slot: std::option::Option, oldest_pending_slot: std::option::Option, @@ -43,13 +296,14 @@ pub(crate) struct RawTransactionIngestProcessingFrontierProjection { impl crate::RawTransactionIngestProcessingFrontierProjection { /// Returns the empty run-local processing projection used before the source observes work. - pub(crate) const fn empty() -> Self { + pub(crate) 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, + continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection::empty(), hydration_pending: 0, processing_frontier_slot: std::option::Option::None, oldest_pending_slot: std::option::Option::None, @@ -65,17 +319,14 @@ impl crate::RawTransactionIngestProcessingFrontierProjection { } /// 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, - oldest_pending_slot: std::option::Option, - ) -> Self { + pub(crate) fn new(hydration_pending: usize, processing_frontier_slot: std::option::Option, 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, + continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection::empty(), hydration_pending, processing_frontier_slot, oldest_pending_slot, @@ -106,7 +357,7 @@ impl crate::RawTransactionIngestProcessingFrontierProjection { } /// Returns a copy carrying the latest source reconnect/replay projection. - pub(crate) const fn with_source_continuity( + pub(crate) fn with_source_continuity( mut self, source_state: std::option::Option, source_reconnect_total: u64, @@ -121,7 +372,7 @@ impl crate::RawTransactionIngestProcessingFrontierProjection { } /// Returns a copy carrying source-neutral multi-source lifecycle counts. - pub(crate) const fn with_source_counts(mut self, source_total: usize, source_active: usize, source_reconnecting: usize, source_failed: usize) -> Self { + pub(crate) fn with_source_counts(mut self, source_total: usize, source_active: usize, source_reconnecting: usize, source_failed: usize) -> Self { self.source_total = source_total; self.source_active = source_active; self.source_reconnecting = source_reconnecting; @@ -130,7 +381,7 @@ impl crate::RawTransactionIngestProcessingFrontierProjection { } /// Returns a copy carrying source-neutral run-local continuity evidence used only for Worker health classification. - pub(crate) const fn with_continuity_health( + pub(crate) fn with_continuity_health( mut self, continuity_frontier_slot: std::option::Option, continuity_has_open_gaps: bool, @@ -143,6 +394,17 @@ impl crate::RawTransactionIngestProcessingFrontierProjection { return self; } + /// Returns a copy carrying bounded source-neutral continuity observability. + pub(crate) fn with_continuity_snapshot(mut self, continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection) -> Self { + self.continuity_snapshot = continuity_snapshot; + return self; + } + + /// Returns the bounded source-neutral continuity observability carried by this projection. + pub(crate) const fn continuity_snapshot(&self) -> &crate::RawTransactionIngestContinuitySnapshotProjection { + return &self.continuity_snapshot; + } + /// 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; @@ -164,7 +426,7 @@ impl crate::RawTransactionIngestProcessingFrontierProjection { } /// 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 { + pub(crate) 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; } @@ -224,6 +486,17 @@ pub struct RawTransactionIngestSnapshot { continuity_policy_observed: bool, failed_source_losses_reconciled: bool, future_target_coverage: bool, + gaps: std::vec::Vec, + open_gap_count: usize, + repairing_gap_count: usize, + repaired_gap_total: u64, + unresolved_gap_total: u64, + replay_repair_total: u64, + redundant_coverage_repair_total: u64, + http_scan_repair_total: u64, + repair_block_fetch_total: u64, + repair_transaction_hydration_total: u64, + oldest_open_gap_start_slot: std::option::Option, admission_queue_capacity: usize, admission_queue_depth: usize, persistence_concurrency: usize, @@ -374,6 +647,74 @@ impl crate::RawTransactionIngestSnapshot { return self.oldest_pending_slot; } + /// Returns the bounded source-neutral gap projections retained in this latest-value snapshot. + /// + /// Every open gap is retained while capacity remains bounded; recent repaired entries may fill unused projection capacity. + #[must_use] + pub fn gaps(&self) -> &[crate::RawTransactionIngestGapSnapshot] { + return self.gaps.as_slice(); + } + + /// Returns the number of currently open run-local continuity gaps. + #[must_use] + pub const fn open_gap_count(&self) -> usize { + return self.open_gap_count; + } + + /// Returns the number of currently active gap recoveries. + #[must_use] + pub const fn repairing_gap_count(&self) -> usize { + return self.repairing_gap_count; + } + + /// Returns the number of gaps cumulatively proven repaired during this run. + #[must_use] + pub const fn repaired_gap_total(&self) -> u64 { + return self.repaired_gap_total; + } + + /// Returns the number of gaps cumulatively classified unresolved during this run. + #[must_use] + pub const fn unresolved_gap_total(&self) -> u64 { + return self.unresolved_gap_total; + } + + /// Returns the number of gaps whose latest successful recovery proof used replay. + #[must_use] + pub const fn replay_repair_total(&self) -> u64 { + return self.replay_repair_total; + } + + /// Returns the number of gaps whose latest successful recovery proof used distinct-source coverage. + #[must_use] + pub const fn redundant_coverage_repair_total(&self) -> u64 { + return self.redundant_coverage_repair_total; + } + + /// Returns the number of gaps whose latest successful recovery proof used a bounded HTTP scan. + #[must_use] + pub const fn http_scan_repair_total(&self) -> u64 { + return self.http_scan_repair_total; + } + + /// Returns the number of repaired gaps whose last recorded mechanism was direct block material retrieval. + #[must_use] + pub const fn repair_block_fetch_total(&self) -> u64 { + return self.repair_block_fetch_total; + } + + /// Returns the number of repaired gaps whose last recorded mechanism was transaction hydration. + #[must_use] + pub const fn repair_transaction_hydration_total(&self) -> u64 { + return self.repair_transaction_hydration_total; + } + + /// Returns the oldest first slot among currently open continuity gaps. + #[must_use] + pub const fn oldest_open_gap_start_slot(&self) -> std::option::Option { + return self.oldest_open_gap_start_slot; + } + /// Returns the configured number of logical live sources for this Worker run. #[must_use] pub const fn source_total(&self) -> usize { @@ -428,6 +769,17 @@ impl std::fmt::Debug for crate::RawTransactionIngestSnapshot { return formatter .debug_struct("RawTransactionIngestSnapshot") .field("worker", &self.worker) + .field("gaps", &self.gaps) + .field("open_gap_count", &self.open_gap_count) + .field("repairing_gap_count", &self.repairing_gap_count) + .field("repaired_gap_total", &self.repaired_gap_total) + .field("unresolved_gap_total", &self.unresolved_gap_total) + .field("replay_repair_total", &self.replay_repair_total) + .field("redundant_coverage_repair_total", &self.redundant_coverage_repair_total) + .field("http_scan_repair_total", &self.http_scan_repair_total) + .field("repair_block_fetch_total", &self.repair_block_fetch_total) + .field("repair_transaction_hydration_total", &self.repair_transaction_hydration_total) + .field("oldest_open_gap_start_slot", &self.oldest_open_gap_start_slot) .field("admission_queue_capacity", &self.admission_queue_capacity) .field("admission_queue_depth", &self.admission_queue_depth) .field("persistence_concurrency", &self.persistence_concurrency) @@ -558,6 +910,17 @@ impl crate::RawTransactionIngestSnapshotPublisher { continuity_policy_observed: false, failed_source_losses_reconciled: false, future_target_coverage: false, + gaps: std::vec::Vec::new(), + open_gap_count: 0, + repairing_gap_count: 0, + repaired_gap_total: 0, + unresolved_gap_total: 0, + replay_repair_total: 0, + redundant_coverage_repair_total: 0, + http_scan_repair_total: 0, + repair_block_fetch_total: 0, + repair_transaction_hydration_total: 0, + oldest_open_gap_start_slot: std::option::Option::None, admission_queue_capacity: settings.admission_queue_capacity(), admission_queue_depth: 0, persistence_concurrency: settings.persistence_concurrency(), @@ -679,6 +1042,18 @@ impl crate::RawTransactionIngestSnapshotPublisher { 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(); + let continuity_snapshot = projection.continuity_snapshot(); + self.snapshot.gaps = continuity_snapshot.gaps().to_vec(); + self.snapshot.open_gap_count = continuity_snapshot.open_gap_count(); + self.snapshot.repairing_gap_count = continuity_snapshot.repairing_gap_count(); + self.snapshot.repaired_gap_total = continuity_snapshot.repaired_gap_total(); + self.snapshot.unresolved_gap_total = continuity_snapshot.unresolved_gap_total(); + self.snapshot.replay_repair_total = continuity_snapshot.replay_repair_total(); + self.snapshot.redundant_coverage_repair_total = continuity_snapshot.redundant_coverage_repair_total(); + self.snapshot.http_scan_repair_total = continuity_snapshot.http_scan_repair_total(); + self.snapshot.repair_block_fetch_total = continuity_snapshot.repair_block_fetch_total(); + self.snapshot.repair_transaction_hydration_total = continuity_snapshot.repair_transaction_hydration_total(); + self.snapshot.oldest_open_gap_start_slot = continuity_snapshot.oldest_open_gap_start_slot(); 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(); 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 a74e0c6..98e8df0 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: 34 +// version: 35 -//! External public, security, redaction and release-boundary hardening canaries through `v0.3.14-pre.010`. +//! External public, security, redaction and release-boundary hardening canaries through `v0.3.14-pre.011`. fn network(value: &'static str) -> std::option::Option { let result = ksp_store_lib::RawNetworkId::new(value); @@ -1263,3 +1263,53 @@ fn v0_3_14_pre_010_repair_fairness_shares_existing_bounds_without_second_pipelin assert!(!root.contains("RawTransactionIngestFairTurnGate")); return; } + +#[test] +fn v0_3_14_pre_011_gap_observability_is_bounded_checked_and_redacted() { + let continuity = include_str!("../src/continuity.rs"); + let snapshot = include_str!("../src/snapshot.rs"); + let root = include_str!("../src/lib.rs"); + for required in [ + "pub struct RawTransactionIngestGapId", + "pub enum RawTransactionIngestGapState", + "pub enum RawTransactionIngestGapReason", + "pub enum RawTransactionIngestRepairMethod", + "pub struct RawTransactionIngestGapSnapshot", + "open_gap_count", + "repairing_gap_count", + "repaired_gap_total", + "unresolved_gap_total", + "replay_repair_total", + "redundant_coverage_repair_total", + "http_scan_repair_total", + "repair_block_fetch_total", + "repair_transaction_hydration_total", + "oldest_open_gap_start_slot", + ] { + assert!(snapshot.contains(required), "required pre.011 snapshot observability guard missing: {required}"); + } + for required in [ + "MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS: usize = 64", + "std::vec::Vec::with_capacity(MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS)", + "gaps.len() == MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS", + "counter_exhausted_error(\"continuity.repaired_gap_total\")", + "counter_exhausted_error(\"continuity.unresolved_gap_total\")", + "gap.last_method = std::option::Option::Some(crate::RawTransactionIngestRepairMethod::RedundantCoverage)", + ] { + assert!(continuity.contains(required), "required pre.011 bounded/checked continuity guard missing: {required}"); + } + for required in [ + "RawTransactionIngestGapId", + "RawTransactionIngestGapReason", + "RawTransactionIngestGapSnapshot", + "RawTransactionIngestGapState", + "RawTransactionIngestRepairMethod", + ] { + assert!(root.contains(required), "required pre.011 public source-neutral type missing: {required}"); + } + for forbidden in ["source_key:", "endpoint_url:", "signature:", "payload:", "provider_error:"] { + assert!(!snapshot.contains(forbidden), "pre.011 snapshot leaked sensitive/source-specific field: {forbidden}"); + } + assert!(!root.contains("source_key")); + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs index 3004037..8aef08f 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs -// version: 22 +// version: 23 //! External public-surface proofs for the RAW transaction ingest Worker foundation. @@ -428,3 +428,54 @@ fn v0_3_13_pre_012_completeness_closure_adds_no_public_implementation_surface() assert!(cross_layer.contains("v0_3_13_pre_012_legacy_v0_v1_is_proven_from_transport_through_worker_common_raw_to_store")); return; } + +#[test] +fn v0_3_14_pre_011_gap_observability_is_public_bounded_and_source_neutral() { + let _gap_id_value: fn(ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapId) -> u64 = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapId::value; + let _gap_id: fn( + &ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot, + ) -> ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapId = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot::gap_id; + let _start_slot: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot) -> u64 = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot::start_slot; + let _end_slot: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot) -> u64 = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot::end_slot; + let _state: fn( + &ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot, + ) -> ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapState = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot::state; + let _reason: fn( + &ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot, + ) -> ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapReason = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot::reason; + let _last_method: fn( + &ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot, + ) -> std::option::Option = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot::last_method; + let _gaps: fn( + &ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot, + ) -> &[ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestGapSnapshot] = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::gaps; + let _open_gap_count: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> usize = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::open_gap_count; + let _repairing_gap_count: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> usize = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::repairing_gap_count; + let _repaired_gap_total: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> u64 = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::repaired_gap_total; + let _unresolved_gap_total: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> u64 = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::unresolved_gap_total; + let _replay_repair_total: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> u64 = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::replay_repair_total; + let _redundant_coverage_repair_total: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> u64 = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::redundant_coverage_repair_total; + let _http_scan_repair_total: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> u64 = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::http_scan_repair_total; + let _repair_block_fetch_total: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> u64 = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::repair_block_fetch_total; + let _repair_transaction_hydration_total: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> u64 = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::repair_transaction_hydration_total; + let _oldest_open_gap_start_slot: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> std::option::Option = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::oldest_open_gap_start_slot; + let root = include_str!("../src/lib.rs"); + for forbidden in ["SourceGapByKey", "provider_gap", "endpoint_gap", "signature_gap", "payload_gap", "source_key"] { + assert!(!root.contains(forbidden), "pre.011 public root leaked source-specific gap material: {forbidden}"); + } + 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 b935af1..e3d862e 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: 29 +// version: 30 -//! Release-completeness canaries through the `v0.3.14-pre.010` repair backpressure/fairness tranche. +//! Release-completeness canaries through the `v0.3.14-pre.011` gap observability tranche. #[test] fn pre_010_production_module_inventory_is_exact() -> std::io::Result<()> { @@ -51,7 +51,7 @@ fn pre_010_production_module_inventory_is_exact() -> std::io::Result<()> { } #[test] -fn pre_010_public_root_export_inventory_is_exact() { +fn v0_3_14_pre_011_public_root_export_inventory_is_exact() { let root = include_str!("../src/lib.rs"); let mut exports = std::vec::Vec::new(); for line in root.lines() { @@ -94,9 +94,14 @@ fn pre_010_public_root_export_inventory_is_exact() { "MIN_RAW_TRANSACTION_INGEST_PERSISTENCE_CONCURRENCY", "MIN_RAW_TRANSACTION_INGEST_SHUTDOWN_DRAIN_TIMEOUT", "RAW_TRANSACTION_INGEST_WORKER_KIND_CODE", + "RawTransactionIngestGapId", + "RawTransactionIngestGapReason", + "RawTransactionIngestGapSnapshot", + "RawTransactionIngestGapState", "RawTransactionIngestHandle", "RawTransactionIngestHeliusTransactionSource", "RawTransactionIngestHttpBlockPollingSource", + "RawTransactionIngestRepairMethod", "RawTransactionIngestRuntimeResources", "RawTransactionIngestSettings", "RawTransactionIngestSnapshot", @@ -386,3 +391,34 @@ fn v0_3_14_pre_010_repair_fairness_canaries_are_present_without_public_or_pipeli assert!(!root.contains("RawTransactionIngestFairTurnGate")); return; } + +#[test] +fn v0_3_14_pre_011_gap_observability_canaries_are_present_without_source_identity_growth() { + let continuity_tests = include_str!("../unit_tests/continuity.rs"); + let hardening = include_str!("hardening.rs"); + let public_api = include_str!("public_api.rs"); + let snapshot_tests = include_str!("../unit_tests/snapshot.rs"); + let root = include_str!("../src/lib.rs"); + for required in [ + "pre_011_gap_observability_projects_open_and_recent_repaired_entries", + "v0_3_14_pre_011_snapshot_carries_checked_gap_observability", + "v0_3_14_pre_011_gap_observability_is_public_bounded_and_source_neutral", + ] { + assert!( + continuity_tests.contains(required) || snapshot_tests.contains(required) || public_api.contains(required), + "required pre.011 observability canary missing: {required}" + ); + } + assert!(hardening.contains("v0_3_14_pre_011_gap_observability_is_bounded_checked_and_redacted")); + for required in [ + "RawTransactionIngestGapId", + "RawTransactionIngestGapReason", + "RawTransactionIngestGapSnapshot", + "RawTransactionIngestGapState", + "RawTransactionIngestRepairMethod", + ] { + assert!(root.contains(required), "required pre.011 public type missing: {required}"); + } + assert!(!root.contains("source_key")); + 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 134c719..5cb4a91 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: 7 +// version: 8 fn network() -> std::option::Option { return match ksp_store_lib::RawNetworkId::new("mainnet") { @@ -62,6 +62,7 @@ fn gap( reason: super::RawTransactionIngestGapReason::WebSocketReconnect, source_key: [source_key_byte; 32], state, + last_method: std::option::Option::None, }); } @@ -647,3 +648,80 @@ fn pre_009_health_projection_rejects_duplicate_and_unknown_active_sources() { assert!(unknown_failed.context().iter().any(|context| return context.value() == "continuity.health_failed_source_unknown")); return; } + +#[test] +fn pre_011_gap_observability_projects_open_and_recent_repaired_entries() { + 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}"), + }; + assert!(contracts.record_source_loss_gap([1_u8; 32], 100, 110).is_ok()); + let pending = match contracts.observability_projection() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("pending observability projection failed: {error}"), + }; + assert_eq!(pending.open_gap_count(), 1); + assert_eq!(pending.repairing_gap_count(), 0); + assert_eq!(pending.repaired_gap_total(), 0); + assert_eq!(pending.unresolved_gap_total(), 0); + assert_eq!(pending.oldest_open_gap_start_slot(), std::option::Option::Some(100)); + assert_eq!(pending.gaps().len(), 1); + assert_eq!(pending.gaps()[0].start_slot(), 100); + assert_eq!(pending.gaps()[0].end_slot(), 110); + assert_eq!(pending.gaps()[0].state(), crate::RawTransactionIngestGapState::Pending); + assert_eq!(pending.gaps()[0].reason(), crate::RawTransactionIngestGapReason::SourceFailure); + assert_eq!(pending.gaps()[0].last_method(), std::option::Option::None); + assert!(contracts.record_coverage_epoch([2_u8; 32], 90, 120).is_ok()); + let repaired = match contracts.observability_projection() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("repaired observability projection failed: {error}"), + }; + assert_eq!(repaired.open_gap_count(), 0); + assert_eq!(repaired.repaired_gap_total(), 1); + assert_eq!(repaired.redundant_coverage_repair_total(), 1); + assert_eq!(repaired.replay_repair_total(), 0); + assert_eq!(repaired.http_scan_repair_total(), 0); + assert_eq!(repaired.repair_block_fetch_total(), 0); + assert_eq!(repaired.repair_transaction_hydration_total(), 0); + assert_eq!(repaired.oldest_open_gap_start_slot(), std::option::Option::None); + assert_eq!(repaired.gaps().len(), 1); + assert_eq!(repaired.gaps()[0].state(), crate::RawTransactionIngestGapState::Repaired); + assert_eq!(repaired.gaps()[0].last_method(), std::option::Option::Some(crate::RawTransactionIngestRepairMethod::RedundantCoverage)); + return; +} + +#[test] +fn pre_011_unresolved_gap_remains_visible_and_bounded() { + 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 unresolved = match gap(1, 1, 200, 205, super::RawTransactionIngestGapState::Unresolved) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("unresolved gap fixture unavailable"), + }; + contracts.gap_ledger.gaps.push(unresolved); + contracts.gap_ledger.next_gap_id = 2; + let projection = match contracts.observability_projection() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("unresolved observability projection failed: {error}"), + }; + assert_eq!(projection.open_gap_count(), 1); + assert_eq!(projection.unresolved_gap_total(), 1); + assert_eq!(projection.oldest_open_gap_start_slot(), std::option::Option::Some(200)); + assert!(projection.gaps().len() <= super::MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS); + assert_eq!(projection.gaps()[0].state(), crate::RawTransactionIngestGapState::Unresolved); + 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 dc84e32..6e4b8ca 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: 32 +// version: 33 fn grpc_endpoint(cluster: &str) -> std::option::Option { return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider"); @@ -1074,7 +1074,7 @@ fn v0_3_13_pre_007_source_inventory_aggregates_frontier_and_lifecycle_conservati ) .is_ok() ); - let first_aggregate = *aggregate_receiver.borrow(); + let first_aggregate = aggregate_receiver.borrow().clone(); assert_eq!(first_aggregate.hydration_pending(), 1); assert_eq!(first_aggregate.processing_frontier_slot(), std::option::Option::None); assert_eq!(first_aggregate.oldest_pending_slot(), std::option::Option::Some(45)); @@ -1093,7 +1093,7 @@ fn v0_3_13_pre_007_source_inventory_aggregates_frontier_and_lifecycle_conservati ) .is_ok() ); - let aggregate = *aggregate_receiver.borrow(); + let aggregate = aggregate_receiver.borrow().clone(); assert_eq!(aggregate.hydration_pending(), 3); assert_eq!(aggregate.processing_frontier_slot(), std::option::Option::Some(42)); assert_eq!(aggregate.oldest_pending_slot(), std::option::Option::Some(40)); @@ -2808,7 +2808,7 @@ fn pre_008_reconnect_replay_and_proven_gap_are_distinct_monotone_and_frontier_pr let mut reporter = super::RawTransactionIngestProcessingFrontierReporter::new(sender); assert!(reporter.observe_pending(42).is_ok()); assert!(reporter.observe_source_continuity(crate::RawTransactionIngestSourceState::Reconnecting, 0, 1, 0, 0, 0).is_ok()); - let replaying = *receiver.borrow(); + let replaying = receiver.borrow().clone(); assert_eq!(replaying.hydration_pending(), 1); assert_eq!(replaying.oldest_pending_slot(), std::option::Option::Some(42)); assert_eq!(replaying.source_state(), std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting)); @@ -2826,7 +2826,7 @@ fn pre_008_reconnect_replay_and_proven_gap_are_distinct_monotone_and_frontier_pr assert!(gap.context().iter().any(|context| { return context.value() == "source.continuity_gap_proven"; })); - let proven = *receiver.borrow(); + let proven = receiver.borrow().clone(); assert_eq!(proven.hydration_pending(), 1); assert_eq!(proven.oldest_pending_slot(), std::option::Option::Some(42)); assert_eq!(proven.source_reconnect_total(), 1); @@ -2849,7 +2849,7 @@ fn v0_3_14_pre_004_replay_delivery_is_not_coverage_and_unproven_coverage_fails_c let (sender, receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); let mut reporter = super::RawTransactionIngestProcessingFrontierReporter::new(sender); assert!(reporter.observe_source_continuity(crate::RawTransactionIngestSourceState::Active, 1, 1, 0, 0, 0).is_ok()); - let attempted = *receiver.borrow(); + let attempted = receiver.borrow().clone(); assert_eq!(attempted.source_reconnect_total(), 1); assert_eq!(attempted.source_replay_attempt_total(), 1); assert_eq!(attempted.source_continuity_gap_total(), 0); @@ -2914,7 +2914,7 @@ async fn pre_009_duplicate_storm_is_bounded_coalesced_and_abort_leaves_no_orphan assert_eq!(coordinator.pending.len(), 1); assert_eq!(coordinator.pending_signal_count, 3); assert!(coordinator.tasks.is_empty()); - let pending = *frontier_receiver.borrow(); + let pending = frontier_receiver.borrow().clone(); assert_eq!(pending.hydration_pending(), 3); assert_eq!(pending.oldest_pending_slot(), std::option::Option::Some(10)); assert!(pending.processing_frontier_slot() != std::option::Option::Some(10)); @@ -2923,7 +2923,7 @@ async fn pre_009_duplicate_storm_is_bounded_coalesced_and_abort_leaves_no_orphan assert!(coordinator.pending.is_empty()); assert_eq!(coordinator.pending_signal_count, 0); assert!(coordinator.tasks.is_empty()); - let cleaned = *frontier_receiver.borrow(); + let cleaned = frontier_receiver.borrow().clone(); assert_eq!(cleaned.hydration_pending(), 0); assert_eq!(cleaned.oldest_pending_slot(), std::option::Option::None); assert_eq!(cleaned.processing_frontier_slot(), safe_frontier_before_abort); @@ -2992,7 +2992,7 @@ async fn pre_009_abort_joins_in_flight_hydration_and_clears_pending_projection() assert_eq!(coordinator.pending_signal_count, 0); assert!(coordinator.pending.is_empty()); assert!(coordinator.tasks.is_empty()); - let cleaned = *frontier_receiver.borrow(); + let cleaned = frontier_receiver.borrow().clone(); assert_eq!(cleaned.hydration_pending(), 0); assert_eq!(cleaned.oldest_pending_slot(), std::option::Option::None); assert_eq!(cleaned.processing_frontier_slot(), std::option::Option::None); @@ -3983,7 +3983,7 @@ async fn v0_3_14_pre_008_proven_full_ledger_epoch_survives_filtered_peer_loss_wi tokio::task::yield_now().await; } assert!(sibling_active.load(std::sync::atomic::Ordering::Acquire)); - let health = *health_receiver.borrow(); + let health = health_receiver.borrow().clone(); assert!(health.continuity_policy_observed()); assert!(!health.continuity_has_open_gaps()); assert!(health.future_target_coverage()); @@ -4037,7 +4037,7 @@ fn v0_3_14_pre_003_websocket_reconnect_is_anchored_before_terminal_gap_projectio std::option::Option::None => panic!("bounded reconnect incident anchor missing"), }; assert_eq!(anchor.end_slot(), std::option::Option::Some(105)); - let projection = *receiver.borrow(); + let projection = receiver.borrow().clone(); assert_eq!(projection.source_reconnect_total(), 1); assert_eq!(projection.source_replay_attempt_total(), 0); assert_eq!(projection.source_continuity_gap_total(), 1); @@ -4059,7 +4059,7 @@ fn v0_3_14_pre_003_websocket_overflow_and_reconnect_storm_share_earliest_anchor( assert_eq!(anchor.end_slot(), std::option::Option::None); assert!(anchor.saw_reconnect()); assert!(anchor.saw_overflow()); - let projection = *receiver.borrow(); + let projection = receiver.borrow().clone(); assert_eq!(projection.source_reconnect_total(), 1); assert_eq!(projection.source_continuity_gap_total(), 3); let bounded = reporter.observe_websocket_post_incident_slot(201); @@ -4109,9 +4109,9 @@ fn v0_3_14_pre_009_inventory_health_projection_tracks_reconciled_coverage() { }; 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!(first.publish(active.clone()).is_ok()); + assert!(second.publish(active.clone()).is_ok()); + let healthy = aggregate_receiver.borrow().clone(); assert!(healthy.continuity_policy_observed()); assert_eq!(healthy.continuity_frontier_slot(), std::option::Option::Some(60)); assert!(!healthy.continuity_has_open_gaps()); @@ -4123,11 +4123,14 @@ fn v0_3_14_pre_009_inventory_health_projection_tracks_reconciled_coverage() { }; 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!(first.publish(active.clone()).is_ok()); + let pending = aggregate_receiver.borrow().clone(); assert_eq!(pending.continuity_frontier_slot(), std::option::Option::Some(49)); assert!(pending.continuity_has_open_gaps()); assert!(pending.future_target_coverage()); + assert_eq!(pending.continuity_snapshot().open_gap_count(), 1); + assert_eq!(pending.continuity_snapshot().oldest_open_gap_start_slot(), std::option::Option::Some(50)); + assert_eq!(pending.continuity_snapshot().gaps().len(), 1); { let mut contracts = match continuity_contracts.lock() { std::result::Result::Ok(value) => value, @@ -4135,11 +4138,18 @@ fn v0_3_14_pre_009_inventory_health_projection_tracks_reconciled_coverage() { }; assert!(contracts.record_coverage_epoch([1_u8; 32], 40, 60).is_ok()); } - assert!(first.publish(active).is_ok()); - let reconciled = *aggregate_receiver.borrow(); + assert!(first.publish(active.clone()).is_ok()); + let reconciled = aggregate_receiver.borrow().clone(); assert_eq!(reconciled.continuity_frontier_slot(), std::option::Option::Some(60)); assert!(!reconciled.continuity_has_open_gaps()); assert!(reconciled.future_target_coverage()); + assert_eq!(reconciled.continuity_snapshot().open_gap_count(), 0); + assert_eq!(reconciled.continuity_snapshot().repaired_gap_total(), 1); + assert_eq!(reconciled.continuity_snapshot().redundant_coverage_repair_total(), 1); + assert_eq!( + reconciled.continuity_snapshot().gaps()[0].last_method(), + std::option::Option::Some(crate::RawTransactionIngestRepairMethod::RedundantCoverage) + ); 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 fe0642a..36d2b1a 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: 6 +// version: 7 fn snapshot_foundation_with_source_total( source_total: usize, @@ -304,3 +304,45 @@ fn v0_3_14_pre_009_health_requires_present_and_future_coverage_before_healthy() assert_eq!(faulted.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy); return; } + +#[test] +fn v0_3_14_pre_011_snapshot_carries_checked_gap_observability() { + let (mut publisher, source) = match snapshot_foundation_with_source_total(2) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let gap = crate::RawTransactionIngestGapSnapshot::new( + crate::RawTransactionIngestGapId::new(7), + 100, + 110, + crate::RawTransactionIngestGapState::Pending, + crate::RawTransactionIngestGapReason::SourceFailure, + std::option::Option::None, + ); + let continuity_snapshot = crate::RawTransactionIngestContinuitySnapshotProjection::empty() + .with_gap_state(std::vec![gap], 1, 0, std::option::Option::Some(100)) + .with_recovery_totals(3, 1, 2, 1, 0) + .with_material_totals(4, 5); + let projection = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(120), std::option::Option::None) + .with_continuity_snapshot(continuity_snapshot); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, projection).is_ok()); + let snapshot = source.current(); + assert_eq!(snapshot.gaps().len(), 1); + assert_eq!(snapshot.gaps()[0].gap_id().value(), 7); + assert_eq!(snapshot.gaps()[0].start_slot(), 100); + assert_eq!(snapshot.gaps()[0].end_slot(), 110); + assert_eq!(snapshot.gaps()[0].state(), crate::RawTransactionIngestGapState::Pending); + assert_eq!(snapshot.gaps()[0].reason(), crate::RawTransactionIngestGapReason::SourceFailure); + assert_eq!(snapshot.gaps()[0].last_method(), std::option::Option::None); + assert_eq!(snapshot.open_gap_count(), 1); + assert_eq!(snapshot.repairing_gap_count(), 0); + assert_eq!(snapshot.repaired_gap_total(), 3); + assert_eq!(snapshot.unresolved_gap_total(), 1); + assert_eq!(snapshot.replay_repair_total(), 2); + assert_eq!(snapshot.redundant_coverage_repair_total(), 1); + assert_eq!(snapshot.http_scan_repair_total(), 0); + assert_eq!(snapshot.repair_block_fetch_total(), 4); + assert_eq!(snapshot.repair_transaction_hydration_total(), 5); + assert_eq!(snapshot.oldest_open_gap_start_slot(), std::option::Option::Some(100)); + return; +} diff --git a/deltas/0.3.14/pre.011.md b/deltas/0.3.14/pre.011.md new file mode 100644 index 0000000..358a93c --- /dev/null +++ b/deltas/0.3.14/pre.011.md @@ -0,0 +1,310 @@ + + + +# Delta `0.3.14-pre.011` — snapshots / observabilité gap et récupération + +## Base requise + +```text +0.3.14-pre.010-fix.001 +workspace.package.version = 0.3.14-pre.10.fix.1 +deltas/0.3.14/pre.010-fix.001.md présent +``` + +## Gate de la base + +Le gate opérateur de `0.3.14-pre.010-fix.001` 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 +147 unit tests : PASS +cross_layer_completeness : 8 PASS +dependency_boundary : 19 PASS +hardening : 35 PASS +public_api : 20 PASS +release_completeness : 12 PASS +``` + +## Objectif + +Implémenter strictement la tranche `pre.011` du plan `035` : + +```text +stabiliser une projection publique source-neutral des gaps run-local +exposer gap_id, plage inclusive, état, raison et dernière méthode éventuelle +exposer les agrégats open/repairing/repaired/unresolved et méthodes de récupération +conserver les compteurs checked +borner la liste détaillée publiée +ne jamais exposer source key, provider, endpoint, signature, payload ou erreur distante arbitraire +ne modifier ni ownership Transport ni pipeline Common RAW/Store +``` + +## Types publics stabilisés + +La surface `snapshot` ajoute : + +```text +RawTransactionIngestGapId +RawTransactionIngestGapState +RawTransactionIngestGapReason +RawTransactionIngestRepairMethod +RawTransactionIngestGapSnapshot +``` + +`RawTransactionIngestGapSnapshot` expose uniquement : + +```text +gap_id +start_slot +end_slot +state +reason +last_method éventuel +``` + +Aucune identité logique de source et aucune donnée provider n'est transportée. + +## Raisons et états + +Les raisons publiques reflètent le catalogue privé déjà validé : + +```text +HttpProducedBlockUnavailable +KnownReferenceMissing +SourceFailure +TransportOverflow +WebSocketReconnect +YellowstoneRetention +``` + +Les états publics sont : + +```text +Pending +Repairing +Repaired +Unresolved +``` + +Un gap `Unresolved` reste terminalement visible et n'est plus reclassé `Repaired` par l'arrivée tardive d'un coverage epoch. + +## Méthodes source-neutral + +Le catalogue public de méthode est : + +```text +Replay +RedundantCoverage +HttpScan +BlockFetch +TransactionHydration +``` + +Dans le comportement productif actuellement fermé, la réconciliation par coverage epoch distincte enregistre explicitement : + +```text +last_method = RedundantCoverage +``` + +Les autres méthodes restent à zéro tant qu'aucun mécanisme productif ne fournit une preuve correspondante ; la snapshot ne fabrique jamais une méthode depuis une simple configuration. + +## Projection détaillée bornée + +La liste publique `RawTransactionIngestSnapshot::gaps()` reste bornée par : + +```text +MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS = 64 +``` + +La projection conserve toujours tous les gaps ouverts, puis utilise uniquement la capacité restante pour les gaps `Repaired` les plus récents. Les entrées sont triées par `gap_id` avant publication. + +L'historique complet n'est donc jamais dupliqué dans la latest-value snapshot. + +## Agrégats publics + +`RawTransactionIngestSnapshot` expose désormais : + +```text +open_gap_count +repairing_gap_count +repaired_gap_total +unresolved_gap_total +replay_repair_total +redundant_coverage_repair_total +http_scan_repair_total +repair_block_fetch_total +repair_transaction_hydration_total +oldest_open_gap_start_slot +``` + +Les comptes sont calculés avec `checked_add`. Un overflow devient une erreur Worker `counter_exhausted` ; aucune saturation silencieuse n'est introduite. + +## Pont continuity -> snapshot + +La projection privée `RawTransactionIngestContinuitySnapshotProjection` reste interne à la crate. + +Le chemin de publication est : + +```text +RawTransactionIngestGapLedger + -> RawTransactionIngestContinuityContracts::observability_projection + -> source_inventory_health_projection + -> RawTransactionIngestProcessingFrontierProjection + -> RawTransactionIngestSnapshotPublisher + -> RawTransactionIngestSnapshot +``` + +Le même `watch` latest-value existant est réutilisé. Aucun second canal public ou pipeline de persistence n'est créé. + +## Non-Copy de la projection interne + +La projection processing interne transporte maintenant un `Vec` borné de snapshots de gaps ; elle devient donc `Clone` mais n'est plus `Copy`. + +Les points qui lisaient auparavant une valeur `watch` par déréférencement utilisent désormais `clone()` explicitement. Cela ne change ni les frontières d'ownership ni la sémantique latest-value. + +## Tests ajoutés ou étendus + +Les unit tests couvrent : + +```text +gap Pending visible avec plage/raison exactes +gap réconcilié visible comme Repaired avec RedundantCoverage +gap Unresolved conservé et oldest_open_gap_start_slot exact +agrégats repaired/unresolved/méthodes exacts +propagation continuity -> inventory aggregate -> concrete snapshot +surface publique complète des getters gap/agrégats +borne détaillée <= 64 +``` + +Les canaris `hardening`, `public_api` et `release_completeness` vérifient en plus : + +```text +catalogue public exact attendu +compteurs checked +projection détaillée bornée +absence de source_key/endpoint/signature/payload dans snapshot.rs +absence de source_key dans la crate root +aucune croissance du graphe de dépendances ou du pipeline +``` + +## Hors périmètre inchangé + +```text +aucun nouveau mécanisme de replay/reconnect Worker +aucun nouveau scan HTTP ou retry réseau +aucun changement de health policy pre.009 +aucun changement de fairness pre.010 +aucun respawn de source +aucun EARLY/shred +aucun Job Backfill depuis Worker +aucun backend Store physique +aucune nouvelle dépendance +shutdown/races réservés à pre.012 +``` + +## Fichiers ajoutés + +```text +deltas/0.3.14/pre.011.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/lib.rs +crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.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/public_api.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 : 583 -> 584 +workspace.package.version : 0.3.14-pre.10.fix.1 -> 0.3.14-pre.11 +``` + +Versions des fichiers Rust modifiés : + +```text +src/continuity.rs : 10 +src/lib.rs : 33 +src/runtime.rs : 18 +src/runtime_resources.rs : 41 +src/snapshot.rs : 9 +unit_tests/continuity.rs : 8 +unit_tests/runtime_resources.rs : 33 +unit_tests/snapshot.rs : 7 +tests/hardening.rs : 35 +tests/public_api.rs : 23 +tests/release_completeness.rs : 30 +``` + +## 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 redaction de la surface snapshot : PASS +contrôle de l'inventaire public attendu : PASS +comparaison exacte pre.010-fix.001 -> pre.011 : 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.012`. + +## Décisions prises + +```text +surface détaillée bornée à 64 entrées +priorité de projection à tous les gaps ouverts +historique réparé représenté par compteurs + capacité détaillée restante +Unresolved reste visible et non réconciliable tardivement +aucune identité de source publique +aucun nouveau canal latest-value +``` + +## Questions ouvertes + +```text +aucune pour cette tranche +``` + +## Gate opérateur après application + +```bash +cargo fmt --all +cargo fmt --all -- --check +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas +cargo check --workspace +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test -p ksp-worker-raw-transaction-ingest-lib --all-targets --all-features +```