From 505cc8c3f624eecfb5ef88e144bcff5a0b56f792 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Sat, 12 Sep 2026 09:00:36 +0200 Subject: [PATCH] v0.3.14-pre.008 --- Cargo.toml | 4 +- .../src/continuity.rs | 274 +++++++++++++- .../src/lib.rs | 7 +- .../src/runtime_resources.rs | 353 +++++++++++++----- .../tests/hardening.rs | 46 ++- .../tests/release_completeness.rs | 25 +- .../unit_tests/continuity.rs | 95 ++++- .../unit_tests/runtime_resources.rs | 222 ++++++++++- deltas/0.3.14/pre.008.md | 285 ++++++++++++++ 9 files changed, 1184 insertions(+), 127 deletions(-) create mode 100644 deltas/0.3.14/pre.008.md diff --git a/Cargo.toml b/Cargo.toml index 49bf349..c2cba82 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 575 +# version: 576 [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.7.fix.1" +version = "0.3.14-pre.8" 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 b3209d6..f8c9fc1 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: 7 +// version: 8 /// 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; @@ -89,6 +89,7 @@ struct RawTransactionIngestCoverageRequirement { #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum RawTransactionIngestGapReason { HttpProducedBlockUnavailable, + KnownReferenceMissing, SourceFailure, TransportOverflow, WebSocketReconnect, @@ -96,8 +97,14 @@ enum RawTransactionIngestGapReason { } impl RawTransactionIngestGapReason { - const ALL: [Self; 5] = - [Self::HttpProducedBlockUnavailable, Self::SourceFailure, Self::TransportOverflow, Self::WebSocketReconnect, Self::YellowstoneRetention]; + const ALL: [Self; 6] = [ + Self::HttpProducedBlockUnavailable, + Self::KnownReferenceMissing, + Self::SourceFailure, + Self::TransportOverflow, + Self::WebSocketReconnect, + Self::YellowstoneRetention, + ]; } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -116,6 +123,15 @@ impl RawTransactionIngestGapState { } } +/// Private supervisor decision for one terminal live-source loss after conservative continuity reconciliation. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum RawTransactionIngestSourceLossDecision { + /// Remaining active sources explicitly preserve all configured target coverage and no known gap blocks continuity. + Continue, + /// Coverage is incomplete, a known gap remains open, or the source-loss proof is otherwise insufficient. + Fault, +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] struct RawTransactionIngestGapRange { end_slot: u64, @@ -267,7 +283,6 @@ impl crate::RawTransactionIngestWebSocketIncidentAnchor { } /// Returns the first source slot that safely anchors the incident, inclusively. - #[cfg(test)] pub(crate) const fn start_slot(self) -> u64 { return self.start_slot; } @@ -315,6 +330,84 @@ impl RawTransactionIngestGapLedger { return Self { gaps: std::vec::Vec::new(), network, next_gap_id: 1 }; } + fn continuity_frontier(&self, processing_frontier_slot: std::option::Option) -> std::option::Option { + let processing_frontier_slot = match processing_frontier_slot { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + let earliest_open_gap_start = self.gaps.iter().filter(|gap| return gap.state.is_open()).map(|gap| return gap.range.start_slot()).min(); + let earliest_open_gap_start = match earliest_open_gap_start { + std::option::Option::Some(value) if value <= processing_frontier_slot => value, + std::option::Option::Some(_) | std::option::Option::None => return std::option::Option::Some(processing_frontier_slot), + }; + return earliest_open_gap_start.checked_sub(1); + } + + fn has_open_gaps(&self) -> bool { + return self.gaps.iter().any(|gap| return gap.state.is_open()); + } + + fn record_gap( + &mut self, + source_key: [u8; 32], + commitment: ksp_onchain_transport_lib::SolanaCommitment, + coverage_requirement: crate::RawTransactionIngestCoverageScope, + reason: RawTransactionIngestGapReason, + range: RawTransactionIngestGapRange, + ) -> ksp_core_lib::Result<()> { + for gap in &mut self.gaps { + if !gap.state.is_open() + || gap.source_key != source_key + || gap.commitment != commitment + || gap.coverage_requirement != coverage_requirement + || gap.reason != reason + { + continue; + } + let merged = match gap.range.try_merge(range) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if let std::option::Option::Some(merged) = merged { + gap.range = merged; + return self.validate_invariants(); + } + } + let open_gap_count = self.gaps.iter().filter(|gap| return gap.state.is_open()).count(); + if open_gap_count >= MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS { + return std::result::Result::Err(crate::runtime_error("continuity.open_gap_limit_exceeded")); + } + let gap_id = self.next_gap_id; + self.next_gap_id = match self.next_gap_id.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.gap_id_exhausted")), + }; + self.gaps.push(RawTransactionIngestGap { + commitment, + coverage_requirement, + gap_id: RawTransactionIngestGapId(gap_id), + network: self.network.clone(), + range, + reason, + source_key, + state: RawTransactionIngestGapState::Pending, + }); + 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() { + 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; + } + } + return self.validate_invariants(); + } + 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")); @@ -382,7 +475,7 @@ impl RawTransactionIngestGapLedger { } } } - if RawTransactionIngestGapReason::ALL.len() != 5 || RawTransactionIngestGapState::ALL.len() != 4 { + if RawTransactionIngestGapReason::ALL.len() != 6 || RawTransactionIngestGapState::ALL.len() != 4 { return std::result::Result::Err(crate::runtime_error("continuity.gap_catalog_invalid")); } return std::result::Result::Ok(()); @@ -477,6 +570,21 @@ impl RawTransactionIngestTargetCoverage { return std::result::Result::Ok(value); } + fn is_covered_by_active_sources( + &self, + capabilities: &[crate::RawTransactionIngestContinuityCapabilityDescriptor], + active_source_keys: &std::collections::BTreeSet<[u8; 32]>, + ) -> bool { + return self.requirements.iter().all(|requirement| { + return capabilities.iter().any(|capability| { + if !active_source_keys.contains(&capability.source_key) || capability.commitment != requirement.commitment { + return false; + } + return capability.source_scope.relation_to(&requirement.scope).is_some(); + }); + }); + } + fn include( &mut self, commitment: ksp_onchain_transport_lib::SolanaCommitment, @@ -518,6 +626,16 @@ impl RawTransactionIngestCoverageRange { const fn contains_gap(self, gap: RawTransactionIngestGapRange) -> bool { return self.start_slot <= gap.start_slot() && self.end_slot >= gap.end_slot(); } + + fn try_merge(self, other: Self) -> std::option::Option { + let overlaps = self.start_slot <= other.end_slot && other.start_slot <= self.end_slot; + let adjacent = self.end_slot.checked_add(1) == std::option::Option::Some(other.start_slot) + || other.end_slot.checked_add(1) == std::option::Option::Some(self.start_slot); + if !overlaps && !adjacent { + return std::option::Option::None; + } + return std::option::Option::Some(Self { end_slot: self.end_slot.max(other.end_slot), start_slot: self.start_slot.min(other.start_slot) }); + } } #[derive(Clone, Debug, Eq, PartialEq)] @@ -569,6 +687,47 @@ impl RawTransactionIngestCoverageEpochLedger { return Self { epochs: std::vec::Vec::new(), next_epoch_id: 1 }; } + fn record( + &mut self, + source_key: [u8; 32], + commitment: ksp_onchain_transport_lib::SolanaCommitment, + scope: crate::RawTransactionIngestCoverageScope, + range: RawTransactionIngestCoverageRange, + capabilities: &[crate::RawTransactionIngestContinuityCapabilityDescriptor], + ) -> ksp_core_lib::Result<()> { + let matching_capability = capabilities.iter().find(|capability| return capability.source_key == source_key); + let matching_capability = match matching_capability { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_source_unknown")), + }; + if matching_capability.commitment != commitment || matching_capability.source_scope != scope { + return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_capability_mismatch")); + } + for epoch in &mut self.epochs { + if epoch.source_key != source_key || epoch.commitment != commitment || epoch.scope != scope { + continue; + } + if let std::option::Option::Some(merged) = epoch.range.try_merge(range) { + epoch.range = merged; + return self.validate_invariants(capabilities); + } + } + if self.epochs.len() >= MAX_RAW_TRANSACTION_INGEST_COVERAGE_EPOCHS { + return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_inventory_invalid")); + } + let epoch_id = self.next_epoch_id; + self.next_epoch_id = match self.next_epoch_id.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_inventory_invalid")), + }; + let epoch = match RawTransactionIngestCoverageEpoch::new(epoch_id, source_key, commitment, scope, range) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + self.epochs.push(epoch); + return self.validate_invariants(capabilities); + } + fn redundant_relation_for_gap( &self, target_source_key: [u8; 32], @@ -675,6 +834,111 @@ impl crate::RawTransactionIngestContinuityContracts { return std::result::Result::Ok(Self { capabilities, coverage_epochs, gap_ledger, target_coverage }); } + /// Records one interval that a configured source has actually proven covered during this run. + pub(crate) fn record_coverage_epoch(&mut self, source_key: [u8; 32], start_slot: u64, end_slot: u64) -> ksp_core_lib::Result<()> { + let capability = self.capabilities.iter().find(|capability| return capability.source_key == source_key); + let capability = match capability { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.coverage_epoch_source_unknown")), + }; + let range = match RawTransactionIngestCoverageRange::new(start_slot, end_slot) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let commitment = capability.commitment; + let scope = capability.source_scope.clone(); + if let std::result::Result::Err(error) = self.coverage_epochs.record(source_key, commitment, scope, range, self.capabilities.as_slice()) { + return std::result::Result::Err(error); + } + return self.gap_ledger.reconcile_with_coverage_epochs(&self.coverage_epochs); + } + + /// Records one unresolved known-reference obligation as a single-slot continuity gap without blocking the processing frontier. + pub(crate) fn record_known_reference_gap(&mut self, source_key: [u8; 32], slot: u64) -> ksp_core_lib::Result<()> { + let capability = self.capabilities.iter().find(|capability| return capability.source_key == source_key); + let capability = match capability { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.known_reference_source_unknown")), + }; + let range = match RawTransactionIngestGapRange::new(slot, slot) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if let std::result::Result::Err(error) = self.gap_ledger.record_gap( + source_key, + capability.commitment, + capability.source_scope.clone(), + RawTransactionIngestGapReason::KnownReferenceMissing, + range, + ) { + return std::result::Result::Err(error); + } + return self.gap_ledger.reconcile_with_coverage_epochs(&self.coverage_epochs); + } + + /// Records one bounded run-local source-loss gap using the exact configured capability as its required coverage. + pub(crate) fn record_source_loss_gap(&mut self, source_key: [u8; 32], start_slot: u64, end_slot: u64) -> ksp_core_lib::Result<()> { + let capability = self.capabilities.iter().find(|capability| return capability.source_key == source_key); + let capability = match capability { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.source_loss_unknown_source")), + }; + let range = match RawTransactionIngestGapRange::new(start_slot, end_slot) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return self.gap_ledger.record_gap( + source_key, + capability.commitment, + capability.source_scope.clone(), + RawTransactionIngestGapReason::SourceFailure, + range, + ); + } + + /// Returns the gap-aware continuity frontier without conflating it with the processing frontier. + /// + /// The processing frontier is only an upper bound: any known open continuity gap at or below it clamps the returned frontier to the slot immediately + /// before that gap. A gap beginning at slot zero yields no continuity frontier. + pub(crate) fn continuity_frontier(&self, processing_frontier_slot: std::option::Option) -> std::option::Option { + return self.gap_ledger.continuity_frontier(processing_frontier_slot); + } + + /// 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 + /// gap ledger to be fully reconciled. The method never treats configuration alone as historical coverage evidence and never respawns a source. + pub(crate) fn source_loss_decision( + &mut self, + lost_source_key: [u8; 32], + active_source_keys: &[[u8; 32]], + processing_frontier_slot: std::option::Option, + ) -> ksp_core_lib::Result { + if !self.capabilities.iter().any(|capability| return capability.source_key == lost_source_key) { + return std::result::Result::Err(crate::runtime_error("continuity.source_loss_unknown_source")); + } + let mut active = std::collections::BTreeSet::new(); + for source_key in active_source_keys { + if *source_key == lost_source_key || !active.insert(*source_key) { + return std::result::Result::Err(crate::runtime_error("continuity.source_loss_active_set_invalid")); + } + if !self.capabilities.iter().any(|capability| return capability.source_key == *source_key) { + 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) { + return std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Fault); + } + return std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Continue); + } + /// Revalidates that the run-local contracts still correspond exactly to the caller-composed source count before tasks are spawned. pub(crate) fn validate_for_source_count(&self, expected_source_count: usize) -> ksp_core_lib::Result<()> { if self.capabilities.len() != expected_source_count || self.target_coverage.requirements.is_empty() { 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 7a03466..d9cd5a7 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: 31 +// version: 32 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -18,7 +18,8 @@ //! Transaction/TransactionStatus/Block and standard logs notifications become signature/slot references; BlockMeta/Slot remain continuity-only signals. //! Hydration is coalesced by network/signature/commitment under bounded in-flight and pending budgets. A bounded run-local processing frontier projects //! hydration pending, oldest pending slot and highest unblocked actually observed slot. The productive source also projects safe Transport reconnect/replay -//! state and faults conservatively when Transport proves a replay-retention continuity gap; history remediation remains outside this crate. +//! state. Run-local continuity contracts reconcile bounded source-loss and known-reference gaps only from explicit coverage epochs; a lost source may remain +//! absent only while currently active siblings preserve the configured TargetCoverage. Transport source respawn remains forbidden. mod admission; mod continuity; @@ -121,6 +122,8 @@ pub(crate) use self::continuity::RawTransactionIngestContinuityContracts; pub(crate) use self::continuity::RawTransactionIngestCoverageScope; /// Private run-local obligation for one transaction reference already observed by a live source. pub(crate) use self::continuity::RawTransactionIngestKnownReferenceObligation; +/// Private supervisor decision after one live source becomes terminal. +pub(crate) use self::continuity::RawTransactionIngestSourceLossDecision; /// Private run-local WebSocket incident anchor built only from observed source slots and safe Transport counters. pub(crate) use self::continuity::RawTransactionIngestWebSocketIncidentAnchor; /// Creates one terminal content-conflict error without copying conflicting material into diagnostics. 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 2f52dfb..2671f67 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: 34 +// version: 35 use sha2::Digest; // rust-rules: trait-import @@ -88,6 +88,14 @@ enum RawTransactionIngestLiveSource { Yellowstone(crate::RawTransactionIngestYellowstoneSource), } +#[derive(Clone)] +struct RawTransactionIngestSourceRuntimeShared { + continuity_contracts: std::sync::Arc>, + global_hydration_registry: std::sync::Arc, + hydration_in_flight_limit: usize, + hydration_pending_limit: usize, +} + struct RawTransactionIngestSourceInventory { source_keys: std::vec::Vec<[u8; 32]>, source_projections: std::vec::Vec, @@ -211,56 +219,20 @@ impl RawTransactionIngestLiveSource { settings: crate::RawTransactionIngestSettings, stop_receiver: tokio::sync::watch::Receiver, admission_sender: tokio::sync::mpsc::Sender, - global_hydration_registry: std::sync::Arc, - hydration_pending_limit: usize, - hydration_in_flight_limit: usize, inventory_publisher: RawTransactionIngestSourceInventoryPublisher, + shared: RawTransactionIngestSourceRuntimeShared, ) -> ksp_core_lib::Result<()> { let (source_frontier_sender, mut source_frontier_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); let mut source_future = std::boxed::Box::pin(async move { return match self { - Self::HeliusTransaction(source) => { - source - .run( - settings, - stop_receiver, - admission_sender, - source_frontier_sender, - global_hydration_registry, - hydration_pending_limit, - hydration_in_flight_limit, - ) - .await + Self::HeliusTransaction(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await, + Self::HttpBlockPolling(source) => { + source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared.continuity_contracts).await }, - Self::HttpBlockPolling(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender).await, Self::StandardBlock(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender).await, - Self::StandardLogs(source) => { - source - .run( - settings, - stop_receiver, - admission_sender, - source_frontier_sender, - global_hydration_registry, - hydration_pending_limit, - hydration_in_flight_limit, - ) - .await - }, - Self::Yellowstone(source) => { - source - .run( - settings, - stop_receiver, - admission_sender, - source_frontier_sender, - global_hydration_registry, - hydration_pending_limit, - hydration_in_flight_limit, - ) - .await - }, + Self::StandardLogs(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await, + Self::Yellowstone(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await, }; }); loop { @@ -389,6 +361,23 @@ impl RawTransactionIngestSourceInventory { return Self { source_keys, source_projections }; } + fn supervisor_state(&self) -> ksp_core_lib::Result<(std::vec::Vec<[u8; 32]>, std::option::Option)> { + if self.source_keys.len() != self.source_projections.len() { + return std::result::Result::Err(crate::runtime_error("source.inventory_shape_mismatch")); + } + let mut active_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::Active) { + active_source_keys.push(*source_key); + } + } + let aggregate = match self.aggregate() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok((active_source_keys, aggregate.processing_frontier_slot())); + } + fn update( &mut self, entry_index: usize, @@ -725,6 +714,7 @@ struct RawTransactionIngestHydrationContext { protocol: &'static str, route: RawTransactionIngestSourceRoute, route_prefix: &'static str, + source_key: [u8; 32], source_key_domain: &'static [u8], } @@ -809,6 +799,7 @@ impl crate::RawTransactionIngestYellowstoneSource { protocol: RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL, route: self.route.clone(), route_prefix: "ys", + source_key: self.source_key, source_key_domain: RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN, }; } @@ -820,10 +811,10 @@ impl crate::RawTransactionIngestYellowstoneSource { mut stop_receiver: tokio::sync::watch::Receiver, admission_sender: tokio::sync::mpsc::Sender, processing_frontier_sender: tokio::sync::watch::Sender, - global_hydration_registry: std::sync::Arc, - hydration_pending_limit: usize, - hydration_in_flight_limit: usize, + shared: RawTransactionIngestSourceRuntimeShared, ) -> ksp_core_lib::Result<()> { + let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } = + shared; let opened = tokio::select! { biased; _ = stop_receiver.changed() => { @@ -894,6 +885,7 @@ impl crate::RawTransactionIngestYellowstoneSource { &admission_sender, &mut stop_receiver, &mut processing_frontier, + &continuity_contracts, ) .await; match handled { @@ -1049,6 +1041,7 @@ impl crate::RawTransactionIngestHeliusTransactionSource { protocol: RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_HTTP_PROTOCOL, route: self.route.clone(), route_prefix: "hx", + source_key: self.source_key, source_key_domain: RAW_TRANSACTION_INGEST_HELIUS_TRANSACTION_HTTP_SOURCE_KEY_DOMAIN, }; } @@ -1060,10 +1053,10 @@ impl crate::RawTransactionIngestHeliusTransactionSource { mut stop_receiver: tokio::sync::watch::Receiver, admission_sender: tokio::sync::mpsc::Sender, processing_frontier_sender: tokio::sync::watch::Sender, - global_hydration_registry: std::sync::Arc, - hydration_pending_limit: usize, - hydration_in_flight_limit: usize, + shared: RawTransactionIngestSourceRuntimeShared, ) -> ksp_core_lib::Result<()> { + let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } = + shared; let connected = tokio::select! { biased; _ = stop_receiver.changed() => { @@ -1101,7 +1094,7 @@ impl crate::RawTransactionIngestHeliusTransactionSource { return std::result::Result::Err(error); } let mut fault = std::option::Option::None; - let mut bounded_websocket_incident = false; + let mut bounded_websocket_incident = std::option::Option::None; loop { if *stop_receiver.borrow() { break; @@ -1110,11 +1103,14 @@ impl crate::RawTransactionIngestHeliusTransactionSource { fault = std::option::Option::Some(error); break; } - if bounded_websocket_incident && coordinator.pending_signal_count == 0 && coordinator.tasks.is_empty() { - fault = std::option::Option::Some(crate::runtime_error("source.continuity_gap_proven")); + if let std::option::Option::Some((start_slot, end_slot)) = bounded_websocket_incident + && coordinator.pending_signal_count == 0 + && coordinator.tasks.is_empty() + { + fault = std::option::Option::Some(continuity_range_error("source.continuity_gap_proven", start_slot, end_slot)); break; } - let can_receive = coordinator.can_receive() && !bounded_websocket_incident; + let can_receive = coordinator.can_receive() && bounded_websocket_incident.is_none(); if !can_receive && coordinator.tasks.is_empty() { fault = std::option::Option::Some(crate::runtime_error("source.hydration_stalled")); break; @@ -1153,6 +1149,7 @@ impl crate::RawTransactionIngestHeliusTransactionSource { &admission_sender, &mut stop_receiver, &mut processing_frontier, + &continuity_contracts, ) .await; match handled { @@ -1212,7 +1209,9 @@ impl crate::RawTransactionIngestHeliusTransactionSource { fault = std::option::Option::Some(error); break; } - bounded_websocket_incident = bounded_websocket_incident || incident_bounded; + if bounded_websocket_incident.is_none() { + bounded_websocket_incident = incident_bounded; + } } } } @@ -1333,6 +1332,7 @@ impl crate::RawTransactionIngestHttpBlockPollingSource { mut stop_receiver: tokio::sync::watch::Receiver, admission_sender: tokio::sync::mpsc::Sender, processing_frontier_sender: tokio::sync::watch::Sender, + continuity_contracts: std::sync::Arc>, ) -> ksp_core_lib::Result<()> { let context_config = ksp_onchain_transport_lib::SolanaContextConfig::new(std::option::Option::Some(self.commitment), std::option::Option::None); let get_block_config = ksp_onchain_transport_lib::SolanaGetBlockConfig::new( @@ -1390,6 +1390,7 @@ impl crate::RawTransactionIngestHttpBlockPollingSource { std::option::Option::None => u64::MAX, }; let window_end = current_tip.min(candidate_end); + let window_start = next_scan_slot; let discovered = discover_http_block_window( &self.http_pool, &self.polling_role, @@ -1469,6 +1470,17 @@ impl crate::RawTransactionIngestHttpBlockPollingSource { } } if !blocked_by_null && let std::option::Option::Some(proven_end_slot) = discovery.proven_end_slot { + let coverage_result = { + let mut contracts = match continuity_contracts.lock() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + contracts.record_coverage_epoch(self.source_key, window_start, proven_end_slot) + }; + if let std::result::Result::Err(error) = coverage_result { + fault = std::option::Option::Some(error); + break; + } next_scan_slot = match proven_end_slot.checked_add(1) { std::option::Option::Some(value) => value, std::option::Option::None => { @@ -1680,8 +1692,8 @@ impl crate::RawTransactionIngestStandardBlockSource { fault = std::option::Option::Some(error); break; } - if incident_bounded { - fault = std::option::Option::Some(crate::runtime_error("source.continuity_gap_proven")); + if let std::option::Option::Some((start_slot, end_slot)) = incident_bounded { + fault = std::option::Option::Some(continuity_range_error("source.continuity_gap_proven", start_slot, end_slot)); break; } continue; @@ -1706,8 +1718,8 @@ impl crate::RawTransactionIngestStandardBlockSource { fault = std::option::Option::Some(error); break; } - if incident_bounded { - fault = std::option::Option::Some(crate::runtime_error("source.continuity_gap_proven")); + if let std::option::Option::Some((start_slot, end_slot)) = incident_bounded { + fault = std::option::Option::Some(continuity_range_error("source.continuity_gap_proven", start_slot, end_slot)); break; } } @@ -1834,6 +1846,7 @@ impl crate::RawTransactionIngestStandardLogsSource { protocol: RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_PROTOCOL, route: self.route.clone(), route_prefix: "ws", + source_key: self.source_key, source_key_domain: RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_SOURCE_KEY_DOMAIN, }; } @@ -1845,10 +1858,10 @@ impl crate::RawTransactionIngestStandardLogsSource { mut stop_receiver: tokio::sync::watch::Receiver, admission_sender: tokio::sync::mpsc::Sender, processing_frontier_sender: tokio::sync::watch::Sender, - global_hydration_registry: std::sync::Arc, - hydration_pending_limit: usize, - hydration_in_flight_limit: usize, + shared: RawTransactionIngestSourceRuntimeShared, ) -> ksp_core_lib::Result<()> { + let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } = + shared; let connected = tokio::select! { biased; _ = stop_receiver.changed() => { @@ -1886,7 +1899,7 @@ impl crate::RawTransactionIngestStandardLogsSource { return std::result::Result::Err(error); } let mut fault = std::option::Option::None; - let mut bounded_websocket_incident = false; + let mut bounded_websocket_incident = std::option::Option::None; loop { if *stop_receiver.borrow() { break; @@ -1895,11 +1908,14 @@ impl crate::RawTransactionIngestStandardLogsSource { fault = std::option::Option::Some(error); break; } - if bounded_websocket_incident && coordinator.pending_signal_count == 0 && coordinator.tasks.is_empty() { - fault = std::option::Option::Some(crate::runtime_error("source.continuity_gap_proven")); + if let std::option::Option::Some((start_slot, end_slot)) = bounded_websocket_incident + && coordinator.pending_signal_count == 0 + && coordinator.tasks.is_empty() + { + fault = std::option::Option::Some(continuity_range_error("source.continuity_gap_proven", start_slot, end_slot)); break; } - let can_receive = coordinator.can_receive() && !bounded_websocket_incident; + let can_receive = coordinator.can_receive() && bounded_websocket_incident.is_none(); if !can_receive && coordinator.tasks.is_empty() { fault = std::option::Option::Some(crate::runtime_error("source.hydration_stalled")); break; @@ -1938,6 +1954,7 @@ impl crate::RawTransactionIngestStandardLogsSource { &admission_sender, &mut stop_receiver, &mut processing_frontier, + &continuity_contracts, ) .await; match handled { @@ -1990,7 +2007,9 @@ impl crate::RawTransactionIngestStandardLogsSource { fault = std::option::Option::Some(error); break; } - bounded_websocket_incident = bounded_websocket_incident || incident_bounded; + if bounded_websocket_incident.is_none() { + bounded_websocket_incident = incident_bounded; + } } } } @@ -2233,15 +2252,15 @@ impl crate::RawTransactionIngestRuntimeResources { if self.sources.is_empty() || self.sources.len() > crate::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES { return std::result::Result::Err(crate::runtime_error("runtime_resources.source_collection_invalid")); } - let mut repair_capabilities = std::vec::Vec::with_capacity(self.sources.len()); + let mut continuity_capabilities = std::vec::Vec::with_capacity(self.sources.len()); for source in &self.sources { let capability = match source.repair_capability_descriptor() { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - repair_capabilities.push(capability); + continuity_capabilities.push(capability); } - let continuity_contracts = match crate::RawTransactionIngestContinuityContracts::new(repair_capabilities) { + let continuity_contracts = match crate::RawTransactionIngestContinuityContracts::new(continuity_capabilities) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; @@ -2249,6 +2268,7 @@ impl crate::RawTransactionIngestRuntimeResources { if let std::result::Result::Err(error) = continuity_contracts.validate_for_source_count(continuity_source_count) { return std::result::Result::Err(error); } + let continuity_contracts = std::sync::Arc::new(std::sync::Mutex::new(continuity_contracts)); let source_keys = self.sources.iter().map(RawTransactionIngestLiveSource::source_key).collect::>(); let hydration_source_count = self.sources.iter().filter(|source| return source.uses_hydration()).count(); if let std::result::Result::Err(error) = @@ -2282,24 +2302,35 @@ impl crate::RawTransactionIngestRuntimeResources { let source_admission_sender = admission_sender.clone(); let source_settings = settings.clone(); let source_stop_receiver = source_stop_receiver.clone(); - let source_global_hydration_registry = std::sync::Arc::clone(&global_hydration_registry); + let source_shared = RawTransactionIngestSourceRuntimeShared { + continuity_contracts: std::sync::Arc::clone(&continuity_contracts), + global_hydration_registry: std::sync::Arc::clone(&global_hydration_registry), + hydration_in_flight_limit, + hydration_pending_limit, + }; + let source_key = source.source_key(); let _abort_handle = children.spawn(async move { - return source - .run( - source_settings, - source_stop_receiver, - source_admission_sender, - source_global_hydration_registry, - hydration_pending_limit, - hydration_in_flight_limit, - publisher, - ) - .await; + let result = source.run(source_settings, source_stop_receiver, source_admission_sender, publisher, source_shared).await; + return (source_key, result); }); } std::mem::drop(admission_sender); - let result = supervise_live_source_tasks(stop_receiver, source_stop_sender, children).await; - if let std::result::Result::Err(error) = continuity_contracts.validate_for_source_count(continuity_source_count) { + let result = supervise_live_source_tasks( + stop_receiver, + source_stop_sender, + children, + std::sync::Arc::clone(&continuity_contracts), + std::sync::Arc::clone(&inventory), + ) + .await; + let validation = { + let contracts = match continuity_contracts.lock() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + contracts.validate_for_source_count(continuity_source_count) + }; + if let std::result::Result::Err(error) = validation { return std::result::Result::Err(error); } return result; @@ -2384,7 +2415,9 @@ fn source_projection_with_state( async fn supervise_live_source_tasks( mut stop_receiver: tokio::sync::watch::Receiver, source_stop_sender: tokio::sync::watch::Sender, - mut children: tokio::task::JoinSet>, + mut children: tokio::task::JoinSet<([u8; 32], ksp_core_lib::Result<()>)>, + continuity_contracts: std::sync::Arc>, + inventory: std::sync::Arc>, ) -> ksp_core_lib::Result<()> { loop { if *stop_receiver.borrow() { @@ -2402,21 +2435,131 @@ async fn supervise_live_source_tasks( } value = children.join_next(), if !children.is_empty() => value, }; - let first_fault = match joined { - std::option::Option::Some(std::result::Result::Ok(std::result::Result::Ok(()))) => { - std::option::Option::Some(crate::runtime_error("source.configured_source_closed")) + let (source_key, source_result) = match joined { + std::option::Option::Some(std::result::Result::Ok(value)) => value, + std::option::Option::Some(std::result::Result::Err(_)) => { + source_stop_sender.send_replace(true); + return drain_live_source_tasks(&mut children, std::option::Option::Some(crate::runtime_error("source.task_join_failed"))).await; + }, + std::option::Option::None => { + source_stop_sender.send_replace(true); + return drain_live_source_tasks(&mut children, std::option::Option::Some(crate::runtime_error("source.task_set_empty"))).await; }, - std::option::Option::Some(std::result::Result::Ok(std::result::Result::Err(error))) => std::option::Option::Some(error), - std::option::Option::Some(std::result::Result::Err(_)) => std::option::Option::Some(crate::runtime_error("source.task_join_failed")), - std::option::Option::None => std::option::Option::Some(crate::runtime_error("source.task_set_empty")), }; - source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, first_fault).await; + let source_fault = match source_result { + std::result::Result::Ok(()) => crate::runtime_error("source.configured_source_closed"), + std::result::Result::Err(error) => error, + }; + if !source_loss_is_reconcilable(&source_fault) { + source_stop_sender.send_replace(true); + return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await; + } + let (active_source_keys, processing_frontier_slot) = { + let inventory = match inventory.lock() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + match inventory.supervisor_state() { + 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; + }, + } + }; + let continuity_range = match source_loss_continuity_range(&source_fault) { + 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; + }, + }; + let continuity_range = match continuity_range { + std::option::Option::Some(value) => value, + std::option::Option::None => { + source_stop_sender.send_replace(true); + return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await; + }, + }; + let decision = { + let mut contracts = match continuity_contracts.lock() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + if let std::result::Result::Err(error) = contracts.record_source_loss_gap(source_key, continuity_range.0, continuity_range.1) { + std::result::Result::Err(error) + } else { + contracts.source_loss_decision(source_key, active_source_keys.as_slice(), processing_frontier_slot) + } + }; + let decision = match decision { + 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; + }, + }; + match decision { + crate::RawTransactionIngestSourceLossDecision::Continue => { + continue; + }, + crate::RawTransactionIngestSourceLossDecision::Fault => { + source_stop_sender.send_replace(true); + return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await; + }, + } } } +fn continuity_range_error(condition: &'static str, start_slot: u64, end_slot: u64) -> ksp_core_lib::Error { + return crate::runtime_error(condition) + .with_context("continuity_start_slot", start_slot.to_string()) + .with_context("continuity_end_slot", end_slot.to_string()); +} + +fn source_loss_continuity_range(error: &ksp_core_lib::Error) -> ksp_core_lib::Result> { + let mut start_slot = std::option::Option::None; + let mut end_slot = std::option::Option::None; + for context in error.context() { + if context.key() == "continuity_start_slot" { + start_slot = match context.value().parse::() { + std::result::Result::Ok(value) => std::option::Option::Some(value), + std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("continuity.source_loss_range_invalid")), + }; + } else if context.key() == "continuity_end_slot" { + end_slot = match context.value().parse::() { + std::result::Result::Ok(value) => std::option::Option::Some(value), + std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("continuity.source_loss_range_invalid")), + }; + } + } + return match (start_slot, end_slot) { + (std::option::Option::None, std::option::Option::None) => std::result::Result::Ok(std::option::Option::None), + (std::option::Option::Some(start), std::option::Option::Some(end)) if end >= start => std::result::Result::Ok(std::option::Option::Some((start, end))), + _ => std::result::Result::Err(crate::runtime_error("continuity.source_loss_range_invalid")), + }; +} + +fn source_loss_is_reconcilable(error: &ksp_core_lib::Error) -> bool { + if error.code() == crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED { + return true; + } + if error.code() != crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID { + return false; + } + return error.context().iter().any(|context| { + if context.key() != "condition" { + return false; + } + return matches!( + context.value(), + "source.configured_source_closed" | "source.continuity_gap_proven" | "source.replay_coverage_unproven" | "source.websocket_incident_unbounded" + ); + }); +} + async fn drain_live_source_tasks( - children: &mut tokio::task::JoinSet>, + children: &mut tokio::task::JoinSet<([u8; 32], ksp_core_lib::Result<()>)>, mut first_fault: std::option::Option, ) -> ksp_core_lib::Result<()> { while let std::option::Option::Some(joined) = children.join_next().await { @@ -2424,8 +2567,8 @@ async fn drain_live_source_tasks( continue; } first_fault = match joined { - std::result::Result::Ok(std::result::Result::Ok(())) => std::option::Option::None, - std::result::Result::Ok(std::result::Result::Err(error)) => std::option::Option::Some(error), + std::result::Result::Ok((_source_key, std::result::Result::Ok(()))) => std::option::Option::None, + std::result::Result::Ok((_source_key, std::result::Result::Err(error))) => std::option::Option::Some(error), std::result::Result::Err(_) => std::option::Option::Some(crate::runtime_error("source.task_join_failed")), }; } @@ -3930,19 +4073,20 @@ impl RawTransactionIngestProcessingFrontierReporter { return std::result::Result::Ok(()); } - fn observe_websocket_post_incident_slot(&mut self, slot: u64) -> ksp_core_lib::Result { + fn observe_websocket_post_incident_slot(&mut self, slot: u64) -> ksp_core_lib::Result> { let anchor = match self.websocket_incident_anchor.as_mut() { std::option::Option::Some(value) => value, - std::option::Option::None => return std::result::Result::Ok(false), + std::option::Option::None => return std::result::Result::Ok(std::option::Option::None), }; if anchor.end_slot().is_some() { - return std::result::Result::Ok(false); + return std::result::Result::Ok(std::option::Option::None); } if let std::result::Result::Err(error) = anchor.close_at(slot) { return std::result::Result::Err(error); } + let range = (anchor.start_slot(), slot); self.publish(); - return std::result::Result::Ok(true); + return std::result::Result::Ok(std::option::Option::Some(range)); } fn observe_source_continuity( @@ -4217,6 +4361,7 @@ impl RawTransactionIngestHydrationCoordinator { admission_sender: &tokio::sync::mpsc::Sender, stop_receiver: &mut tokio::sync::watch::Receiver, processing_frontier: &mut RawTransactionIngestProcessingFrontierReporter, + continuity_contracts: &std::sync::Arc>, ) -> ksp_core_lib::Result { let fetched = match joined { std::result::Result::Ok(std::result::Result::Ok(value)) => value, @@ -4251,6 +4396,16 @@ impl RawTransactionIngestHydrationCoordinator { RawTransactionIngestKnownReferenceMissingDisposition::AwaitCoverage | RawTransactionIngestKnownReferenceMissingDisposition::PreferBlockSlot => {}, } + let continuity_result = { + let mut contracts = match continuity_contracts.lock() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + contracts.record_known_reference_gap(hydration.source_key, signal_slot) + }; + if let std::result::Result::Err(error) = continuity_result { + return std::result::Result::Err(error); + } if let std::result::Result::Err(error) = processing_frontier.settle_pending(signal_slot) { return std::result::Result::Err(error); } 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 5efbb20..edd3695 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: 31 +// version: 32 -//! External public, security, redaction and release-boundary hardening canaries through `v0.3.14-pre.007`. +//! External public, security, redaction and release-boundary hardening canaries through `v0.3.14-pre.008`. fn network(value: &'static str) -> std::option::Option { let result = ksp_store_lib::RawNetworkId::new(value); @@ -1143,3 +1143,45 @@ fn v0_3_14_pre_007_known_reference_hydration_reuses_one_registry_and_preserves_m assert!(!root.contains("repair"), "pre.007 leaked lower-case repair responsibility through crate root"); return; } + +#[test] +fn v0_3_14_pre_008_reconciliation_and_source_loss_are_target_coverage_gated_without_respawn() { + let continuity = include_str!("../src/continuity.rs"); + let resources = include_str!("../src/runtime_resources.rs"); + let root = include_str!("../src/lib.rs"); + for required in [ + "RawTransactionIngestSourceLossDecision", + "source_loss_decision", + "is_covered_by_active_sources", + "reconcile_with_coverage_epochs", + "record_coverage_epoch", + "record_known_reference_gap", + "record_source_loss_gap", + "continuity_frontier", + "has_open_gaps", + "continuity.source_loss_active_set_invalid", + "RawTransactionIngestSourceLossDecision::Fault", + "RawTransactionIngestSourceLossDecision::Continue", + ] { + assert!(continuity.contains(required), "required pre.008 reconciliation guard missing: {required}"); + } + for required in [ + "source_loss_is_reconcilable", + "inventory.supervisor_state()", + "source_loss_continuity_range", + "contracts.record_known_reference_gap", + "contracts.record_source_loss_gap", + "contracts.source_loss_decision", + "RawTransactionIngestSourceLossDecision::Continue", + "RawTransactionIngestSourceLossDecision::Fault", + "source.websocket_incident_unbounded", + "source.replay_coverage_unproven", + ] { + assert!(resources.contains(required), "required pre.008 supervisor gate missing: {required}"); + } + assert!(!resources.contains("respawn_source")); + assert!(!resources.contains("restart_source")); + assert!(!root.contains("pub use self::continuity::RawTransactionIngestSourceLossDecision")); + assert!(!root.contains("repair"), "pre.008 leaked lower-case repair responsibility through crate root"); + 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 b9b0860..024c674 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: 26 +// version: 27 -//! Release-completeness canaries through the `v0.3.14-pre.007` known-reference hydration tranche. +//! Release-completeness canaries through the `v0.3.14-pre.008` continuity reconciliation tranche. #[test] fn pre_010_production_module_inventory_is_exact() -> std::io::Result<()> { @@ -322,3 +322,24 @@ fn v0_3_14_pre_007_known_reference_hydration_canaries_are_present_without_second assert!(!root.contains("pub use self::runtime_resources::RawTransactionIngestKnownReference")); return; } + +#[test] +fn v0_3_14_pre_008_reconciliation_canaries_are_present_without_public_surface_or_respawn_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 root = include_str!("../src/lib.rs"); + for required in [ + "pre_008_continuity_frontier_is_gap_aware_and_recovers_after_redundant_proof", + "pre_008_source_loss_requires_full_active_target_coverage", + "pre_008_open_gap_without_proven_epoch_blocks_redundant_source_continuation", + "pre_008_known_reference_missing_moves_to_continuity_ledger_and_reconciles_from_proven_full_ledger_epoch", + "v0_3_14_pre_008_proven_full_ledger_epoch_survives_filtered_peer_loss_without_worker_respawn", + "v0_3_14_pre_008_only_classified_source_loss_enters_coverage_decision", + ] { + assert!(continuity_tests.contains(required) || resource_tests.contains(required), "required pre.008 canary missing: {required}"); + } + assert!(hardening.contains("v0_3_14_pre_008_reconciliation_and_source_loss_are_target_coverage_gated_without_respawn")); + assert!(!root.contains("pub use self::continuity::RawTransactionIngestSourceLossDecision")); + 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 2c99721..8579acb 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: 5 +// version: 6 fn network() -> std::option::Option { return match ksp_store_lib::RawNetworkId::new("mainnet") { @@ -473,3 +473,96 @@ fn pre_007_known_reference_obligation_preserves_exact_reference_slot_and_commitm assert!(processed.is_err()); return; } + +#[test] +fn pre_008_continuity_frontier_is_gap_aware_and_recovers_after_redundant_proof() { + 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()); + assert_eq!(contracts.continuity_frontier(std::option::Option::Some(120)), std::option::Option::Some(99)); + assert!(contracts.record_coverage_epoch([2_u8; 32], 90, 120).is_ok()); + let decision = contracts.source_loss_decision([1_u8; 32], &[[2_u8; 32]], std::option::Option::Some(120)); + assert!(matches!(decision, std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Continue))); + assert_eq!(contracts.continuity_frontier(std::option::Option::Some(120)), std::option::Option::Some(120)); + assert!(!contracts.gap_ledger.has_open_gaps()); + return; +} + +#[test] +fn pre_008_source_loss_requires_full_active_target_coverage() { + 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("helius_transaction", 8)) { + 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 decision = contracts.source_loss_decision([1_u8; 32], &[[2_u8; 32]], std::option::Option::Some(120)); + assert!(matches!(decision, std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Fault))); + return; +} + +#[test] +fn pre_008_open_gap_without_proven_epoch_blocks_redundant_source_continuation() { + 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 gap = match gap(1, 1, 100, 110, super::RawTransactionIngestGapState::Unresolved) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("gap fixture unavailable"), + }; + contracts.gap_ledger.gaps.push(gap); + contracts.gap_ledger.next_gap_id = 2; + let decision = contracts.source_loss_decision([1_u8; 32], &[[2_u8; 32]], std::option::Option::Some(120)); + assert!(matches!(decision, std::result::Result::Ok(crate::RawTransactionIngestSourceLossDecision::Fault))); + assert_eq!(contracts.continuity_frontier(std::option::Option::Some(120)), std::option::Option::Some(99)); + return; +} + +#[test] +fn pre_008_known_reference_missing_moves_to_continuity_ledger_and_reconciles_from_proven_full_ledger_epoch() { + 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, crate::RawTransactionIngestCoverageScope::full_ledger_transactions()) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("full-ledger 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_known_reference_gap([1_u8; 32], 105).is_ok()); + assert_eq!(contracts.continuity_frontier(std::option::Option::Some(120)), std::option::Option::Some(104)); + assert!(contracts.gap_ledger.has_open_gaps()); + assert!(contracts.record_coverage_epoch([2_u8; 32], 100, 110).is_ok()); + assert!(!contracts.gap_ledger.has_open_gaps()); + assert_eq!(contracts.continuity_frontier(std::option::Option::Some(120)), std::option::Option::Some(120)); + 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 b81c6e5..6016a87 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: 28 +// version: 29 fn grpc_endpoint(cluster: &str) -> std::option::Option { return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider"); @@ -131,6 +131,62 @@ fn ws_endpoint( )); } +fn supervisor_contracts( + source_specs: &[(u8, &'static str, u8)], +) -> std::option::Option>> { + let network = match ksp_store_lib::RawNetworkId::new("devnet") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::option::Option::None, + }; + let mut capabilities = std::vec::Vec::new(); + for (source_key_byte, family, fingerprint_byte) in source_specs { + let (scope, reference_bearing, block_material) = if *family == "full_ledger" { + (crate::RawTransactionIngestCoverageScope::full_ledger_transactions(), false, true) + } else { + (crate::RawTransactionIngestCoverageScope::exact_source_scope(*family, [*fingerprint_byte; 32]), true, false) + }; + let capability = crate::RawTransactionIngestContinuityCapabilityDescriptor::new( + [*source_key_byte; 32], + network.clone(), + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + scope, + reference_bearing, + block_material, + false, + reference_bearing, + false, + false, + ); + let capability = match capability { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::option::Option::None, + }; + capabilities.push(capability); + } + return match crate::RawTransactionIngestContinuityContracts::new(capabilities) { + std::result::Result::Ok(value) => std::option::Option::Some(std::sync::Arc::new(std::sync::Mutex::new(value))), + std::result::Result::Err(_) => std::option::Option::None, + }; +} + +fn supervisor_inventory( + source_keys: std::vec::Vec<[u8; 32]>, + states: &[std::option::Option], +) -> std::option::Option>> { + if source_keys.len() != states.len() { + return std::option::Option::None; + } + let mut inventory = super::RawTransactionIngestSourceInventory::new(source_keys.clone()); + for (index, (source_key, state)) in source_keys.iter().zip(states.iter()).enumerate() { + let projection = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(100), std::option::Option::None) + .with_source_continuity(*state, 0, 0, 0); + if inventory.update(index, *source_key, projection).is_err() { + return std::option::Option::None; + } + } + return std::option::Option::Some(std::sync::Arc::new(std::sync::Mutex::new(inventory))); +} + fn standard_block_source( cluster: &str, endpoint_name: &str, @@ -1049,8 +1105,24 @@ async fn v0_3_13_pre_007_source_supervisor_joins_all_children_on_stop() { let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); let (stop_sender, stop_receiver) = tokio::sync::watch::channel(false); let (source_stop_sender, source_stop_receiver) = tokio::sync::watch::channel(false); + let source_keys = std::vec![[1_u8; 32], [2_u8; 32], [3_u8; 32]]; + let contracts = match supervisor_contracts(&[(1, "fixture-a", 1), (2, "fixture-b", 2), (3, "fixture-c", 3)]) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let inventory = match supervisor_inventory( + source_keys.clone(), + &[ + std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), + std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), + std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), + ], + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; let mut children = tokio::task::JoinSet::new(); - for _ in 0..3 { + for source_key in source_keys { let active = std::sync::Arc::clone(&active); let mut source_stop_receiver = source_stop_receiver.clone(); let _abort_handle = children.spawn(async move { @@ -1059,12 +1131,14 @@ async fn v0_3_13_pre_007_source_supervisor_joins_all_children_on_stop() { let changed = source_stop_receiver.changed().await; if changed.is_err() || *source_stop_receiver.borrow() { active.fetch_sub(1, std::sync::atomic::Ordering::AcqRel); - return std::result::Result::Ok(()); + return (source_key, std::result::Result::Ok(())); } } }); } - let supervisor = tokio::spawn(super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children)); + let supervisor = tokio::spawn(async move { + return super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory).await; + }); for _ in 0..64 { if active.load(std::sync::atomic::Ordering::Acquire) == 3 { break; @@ -1089,11 +1163,25 @@ async fn v0_3_13_pre_007_source_failure_stops_and_joins_sibling_sources() { let sibling_active = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let (_stop_sender, stop_receiver) = tokio::sync::watch::channel(false); let (source_stop_sender, source_stop_receiver) = tokio::sync::watch::channel(false); + let 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 = match supervisor_inventory( + std::vec![[1_u8; 32], [2_u8; 32]], + &[ + std::option::Option::Some(crate::RawTransactionIngestSourceState::Failed), + std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), + ], + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; let mut children = tokio::task::JoinSet::new(); let failing_barrier = std::sync::Arc::clone(&barrier); let _failing_abort_handle = children.spawn(async move { let _barrier_wait = failing_barrier.wait().await; - return std::result::Result::Err(crate::runtime_error("test.pre_007_source_failed")); + return ([1_u8; 32], std::result::Result::Err(crate::runtime_error("test.pre_007_source_failed"))); }); let sibling_barrier = std::sync::Arc::clone(&barrier); let sibling_active_for_task = std::sync::Arc::clone(&sibling_active); @@ -1105,11 +1193,11 @@ async fn v0_3_13_pre_007_source_failure_stops_and_joins_sibling_sources() { let changed = sibling_stop_receiver.changed().await; if changed.is_err() || *sibling_stop_receiver.borrow() { sibling_active_for_task.store(false, std::sync::atomic::Ordering::Release); - return std::result::Result::Ok(()); + return ([2_u8; 32], std::result::Result::Ok(())); } } }); - let result = super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children).await; + let result = super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory).await; assert!(result.is_err()); assert!(!sibling_active.load(std::sync::atomic::Ordering::Acquire)); return; @@ -3701,13 +3789,23 @@ async fn v0_3_13_pre_011_aborting_outer_source_supervisor_aborts_nested_source_t let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); let (_stop_sender, stop_receiver) = tokio::sync::watch::channel(false); let (source_stop_sender, _source_stop_receiver) = tokio::sync::watch::channel(false); + let contracts = match supervisor_contracts(&[(1, "fixture-a", 1)]) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let inventory = match supervisor_inventory(std::vec![[1_u8; 32]], &[std::option::Option::Some(crate::RawTransactionIngestSourceState::Active)]) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; let mut children = tokio::task::JoinSet::new(); let task_active = std::sync::Arc::clone(&active); let _abort_handle = children.spawn(async move { let _guard = Pre011SourceActiveGuard::new(task_active); - return std::future::pending::>().await; + 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; }); - let supervisor = tokio::spawn(super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children)); for _ in 0..64 { if active.load(std::sync::atomic::Ordering::Acquire) == 1 { break; @@ -3733,11 +3831,25 @@ async fn v0_3_13_pre_011_stop_racing_ready_source_fault_preserves_fault_and_join let fault_gate = std::sync::Arc::new(tokio::sync::Notify::new()); let (stop_sender, stop_receiver) = tokio::sync::watch::channel(false); let (source_stop_sender, source_stop_receiver) = tokio::sync::watch::channel(false); + let 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 = match supervisor_inventory( + std::vec![[1_u8; 32], [2_u8; 32]], + &[ + std::option::Option::Some(crate::RawTransactionIngestSourceState::Failed), + std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), + ], + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; let mut children = tokio::task::JoinSet::new(); let task_fault_gate = std::sync::Arc::clone(&fault_gate); let _fault_abort_handle = children.spawn(async move { task_fault_gate.notified().await; - return std::result::Result::Err(crate::runtime_error("test.pre_011_ready_source_fault")); + return ([1_u8; 32], std::result::Result::Err(crate::runtime_error("test.pre_011_ready_source_fault"))); }); let sibling_active_for_task = std::sync::Arc::clone(&sibling_active); let mut sibling_stop_receiver = source_stop_receiver.clone(); @@ -3746,11 +3858,13 @@ async fn v0_3_13_pre_011_stop_racing_ready_source_fault_preserves_fault_and_join loop { let changed = sibling_stop_receiver.changed().await; if changed.is_err() || *sibling_stop_receiver.borrow() { - return std::result::Result::Ok(()); + return ([2_u8; 32], std::result::Result::Ok(())); } } }); - let supervisor = tokio::spawn(super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children)); + let supervisor = tokio::spawn(async move { + return super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory).await; + }); for _ in 0..64 { if sibling_active.load(std::sync::atomic::Ordering::Acquire) == 1 { break; @@ -3773,6 +3887,86 @@ async fn v0_3_13_pre_011_stop_racing_ready_source_fault_preserves_fault_and_join return; } +#[tokio::test(flavor = "current_thread")] +async fn v0_3_14_pre_008_proven_full_ledger_epoch_survives_filtered_peer_loss_without_worker_respawn() { + let sibling_active = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let (stop_sender, stop_receiver) = tokio::sync::watch::channel(false); + let (source_stop_sender, source_stop_receiver) = tokio::sync::watch::channel(false); + let contracts = match supervisor_contracts(&[(1, "standard_logs", 7), (2, "full_ledger", 0)]) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + { + let mut contracts_guard = match contracts.lock() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(poisoned) => poisoned.into_inner(), + }; + assert!(contracts_guard.record_coverage_epoch([2_u8; 32], 90, 110).is_ok()); + } + let inventory = match supervisor_inventory( + std::vec![[1_u8; 32], [2_u8; 32]], + &[ + std::option::Option::Some(crate::RawTransactionIngestSourceState::Failed), + std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), + ], + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let mut children = tokio::task::JoinSet::new(); + let _lost_abort_handle = children.spawn(async move { + return ([1_u8; 32], std::result::Result::Err(super::continuity_range_error("source.continuity_gap_proven", 95, 100))); + }); + let sibling_active_for_task = std::sync::Arc::clone(&sibling_active); + let mut sibling_stop_receiver = source_stop_receiver.clone(); + let _sibling_abort_handle = children.spawn(async move { + sibling_active_for_task.store(true, std::sync::atomic::Ordering::Release); + loop { + let changed = sibling_stop_receiver.changed().await; + if changed.is_err() || *sibling_stop_receiver.borrow() { + sibling_active_for_task.store(false, std::sync::atomic::Ordering::Release); + return ([2_u8; 32], std::result::Result::Ok(())); + } + } + }); + let supervisor = tokio::spawn(async move { + return super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory).await; + }); + for _ in 0..64 { + if sibling_active.load(std::sync::atomic::Ordering::Acquire) { + 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)); + stop_sender.send_replace(true); + let result = match supervisor.await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert!(result.is_ok()); + assert!(!sibling_active.load(std::sync::atomic::Ordering::Acquire)); + return; +} + +#[test] +fn v0_3_14_pre_008_only_classified_source_loss_enters_coverage_decision() { + let transport = super::source_transport_error(ksp_onchain_transport_lib::ERROR_CODE_HTTP_CONNECTION_FAILED); + assert!(super::source_loss_is_reconcilable(&transport)); + assert!(super::source_loss_is_reconcilable(&crate::runtime_error("source.continuity_gap_proven"))); + assert!(super::source_loss_is_reconcilable(&crate::runtime_error("source.replay_coverage_unproven"))); + assert!(!super::source_loss_is_reconcilable(&crate::runtime_error("source.hydration_task_join_failed"))); + let bounded = super::continuity_range_error("source.continuity_gap_proven", 10, 20); + assert!(matches!(super::source_loss_continuity_range(&bounded), std::result::Result::Ok(std::option::Option::Some((10, 20))))); + assert!(matches!( + super::source_loss_continuity_range(&crate::runtime_error("source.continuity_gap_proven")), + std::result::Result::Ok(std::option::Option::None) + )); + return; +} + #[test] fn v0_3_14_pre_003_websocket_reconnect_is_anchored_before_terminal_gap_projection() { let (sender, receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); @@ -3790,7 +3984,7 @@ fn v0_3_14_pre_003_websocket_reconnect_is_anchored_before_terminal_gap_projectio assert!(!anchor.saw_overflow()); assert!(reporter.observe_websocket_continuity(crate::RawTransactionIngestSourceState::Active, 1, 0).is_ok()); let bounded = reporter.observe_websocket_post_incident_slot(105); - assert!(matches!(bounded, std::result::Result::Ok(true))); + assert!(matches!(bounded, std::result::Result::Ok(std::option::Option::Some((100, 105))))); let anchor = match reporter.websocket_incident_anchor { std::option::Option::Some(value) => value, std::option::Option::None => panic!("bounded reconnect incident anchor missing"), @@ -3822,7 +4016,7 @@ fn v0_3_14_pre_003_websocket_overflow_and_reconnect_storm_share_earliest_anchor( assert_eq!(projection.source_reconnect_total(), 1); assert_eq!(projection.source_continuity_gap_total(), 3); let bounded = reporter.observe_websocket_post_incident_slot(201); - assert!(matches!(bounded, std::result::Result::Ok(true))); + assert!(matches!(bounded, std::result::Result::Ok(std::option::Option::Some((200, 201))))); return; } diff --git a/deltas/0.3.14/pre.008.md b/deltas/0.3.14/pre.008.md new file mode 100644 index 0000000..72ef0d1 --- /dev/null +++ b/deltas/0.3.14/pre.008.md @@ -0,0 +1,285 @@ + + + +# Delta `0.3.14-pre.008` — réconciliation et supervisor source-loss + +## Base requise + +```text +0.3.14-pre.007-fix.001 +workspace.package.version = 0.3.14-pre.7.fix.1 +deltas/0.3.14/pre.007-fix.001.md présent +``` + +## Gate de la base + +Le gate opérateur de `0.3.14-pre.007-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 +133 unit tests : PASS +cross_layer_completeness : 8 PASS +dependency_boundary : 19 PASS +hardening : 32 PASS +public_api : 20 PASS +release_completeness : 9 PASS +``` + +## Objectif + +Implémenter strictement la tranche `pre.008` du plan `035` : + +```text +unifier les preuves replay/redondance/discovery HTTP/hydration autour du ledger de continuité +maintenir un continuity frontier distinct du processing frontier +conserver les Missing known-reference comme gaps single-slot après libération du processing frontier +réconcilier les gaps uniquement depuis des coverage epochs réellement prouvés +remplacer la terminalité « first source failure » uniquement lorsqu'une TargetCoverage présente et future est explicitement prouvée +ne jamais respawn une source Transport depuis le Worker +rester fail-closed lorsque la plage de perte est non bornée ou qu'une preuve manque +``` + +## Continuity frontier distinct du processing frontier + +Le gap ledger calcule désormais un `continuity_frontier` à partir du processing frontier courant : + +```text +aucun gap ouvert <= processing frontier + -> continuity frontier = processing frontier + +gap ouvert commençant à S <= processing frontier + -> continuity frontier = S - 1 + +gap ouvert commençant à 0 + -> aucun continuity frontier +``` + +Le processing frontier reste donc un indicateur de traitement et ne peut jamais, à lui seul, fermer une discontinuité historique. + +## Coverage epochs réellement prouvés + +Le ledger `pre.005` reçoit désormais les epochs produits pendant l'exécution. + +La source HTTP Block Polling full-ledger enregistre un epoch uniquement lorsqu'une fenêtre bornée de `pre.006` a été effectivement prouvée : + +```text +fenêtre discovery bornée ++ tous les blocs produits de la partie prouvée effectivement matérialisés ++ aucun getBlock observed = null bloquant la fenêtre += coverage epoch admissible +``` + +Un epoch adjacent ou chevauchant du même source/scope/commitment est fusionné. L'enregistrement d'un nouvel epoch déclenche immédiatement une tentative de réconciliation des gaps déjà ouverts. + +Aucune source filtrée n'obtient un epoch par sa simple présence ou sa configuration. + +## Missing known-reference raccordé au ledger + +Le résultat `Missing` introduit en `pre.007` devient un gap de continuité single-slot : + +```text +source_key exact +commitment exact +scope configuré exact de la source +range = [slot, slot] +reason = KnownReferenceMissing +``` + +Le pending de processing peut ensuite être soldé normalement. Le gap reste toutefois ouvert dans le ledger jusqu'à une preuve de coverage distincte. + +Un epoch full-ledger d'une autre source couvrant ce slot peut donc réconcilier cette obligation sans inventer une observation de transaction et sans assimiler `getTransaction = null` à une absence prouvée. + +Aucun second registre d'hydration, scheduler, semaphore ou pipeline n'est ajouté. + +## Source-loss bornée et supervisor + +Les tâches source supervisées retournent désormais leur `source_key` avec leur résultat terminal. Les projections source existantes restent l'autorité pour déterminer les sources réellement `Active` au moment de la décision. + +Les incidents WS déjà bornés en `pre.003` transportent leurs bornes inclusives jusqu'au supervisor. Une perte classifiée et bornée est enregistrée comme gap selon le scope/commitment exacts de la source perdue. + +Le supervisor peut continuer sans cette source uniquement si, après réconciliation : + +```text +aucun gap connu ne reste ouvert +continuity frontier == processing frontier +chaque requirement de TargetCoverage est couvert par au moins une source encore Active +la source perdue n'est pas utilisée comme source active de remplacement +les relations de scope restent celles de pre.005 : Exact ou Superset explicitement démontré +le commitment reste strictement identique +``` + +Sinon, la politique précédente reste fail-closed et les sources sœurs sont arrêtées puis jointes comme auparavant. + +## Pertes qui restent terminales + +Cette tranche ne fabrique aucune plage lorsqu'elle n'est pas connue. + +Restent donc terminales, entre autres : + +```text +source transport perdue sans borne de continuité sûre +replay Yellowstone dont la coverage reste non prouvée sans range sûr +fermeture configurée sans range historique démontré +join failure / task-set invariant +content conflict +persistence failure +counter/invariant exhaustion +``` + +Le simple fait qu'une autre source soit configurée ou `Active` ne constitue jamais une preuve historique de l'intervalle perdu. + +## Aucun respawn Worker + +Le Worker ne recrée aucune source terminale. Transport conserve l'ownership de ses propres reconnect/replay mechanisms. + +Après une décision `Continue`, le supervisor conserve uniquement les tâches sœurs déjà vivantes. La source perdue reste absente du run. + +Les frontières historiques restent inchangées : + +```text +aucun set_from_slot dans Worker +aucun SubscribeReplayInfo dans Worker +aucun socket WebSocket/gRPC créé par Worker +aucun retry loop réseau Worker +aucun accès Config +aucun Job Backfill +aucun backend Store physique +``` + +## Tests ajoutés + +Les unit tests couvrent notamment : + +```text +continuity frontier bloqué par un gap puis rétabli après coverage epoch redondant +source-loss refusée si les sources actives ne couvrent plus toute TargetCoverage +source-loss refusée tant qu'un gap reste ouvert malgré une source future-capable +source HTTP full-ledger prouvée permettant la survie d'une source filtrée perdue sans respawn +seules les pertes classifiées entrent dans la décision de coverage +Missing known-reference transféré au continuity ledger sans bloquer le processing frontier +Missing single-slot réconcilié seulement après epoch full-ledger réellement prouvé par une autre source +``` + +Les canaris `hardening` et `release_completeness` vérifient en plus : + +```text +présence du continuity frontier privé +présence du gate TargetCoverage dans la décision source-loss +présence de record_known_reference_gap sur le chemin runtime Missing +aucune croissance de surface publique +aucun respawn source Worker +aucun ownership replay Transport déplacé vers Worker +``` + +## Hors périmètre inchangé + +```text +aucune health policy Healthy/Degraded/Unhealthy/Faulted de pre.009 +aucune fairness spécifique pendant réconciliation de pre.010 +aucune observabilité publique détaillée des gaps de pre.011 +aucun mécanisme EARLY/shred +aucun backfill historique caller-driven +aucune nouvelle source/provider +``` + +## Fichiers ajoutés + +```text +deltas/0.3.14/pre.008.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_resources.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 +``` + +## Fichiers supprimés + +```text +aucun +``` + +## Version Cargo + +Conformément à `VER-ID-009` : + +```text +header Cargo.toml : 575 -> 576 +workspace.package.version : 0.3.14-pre.7.fix.1 -> 0.3.14-pre.8 +``` + +Versions des fichiers modifiés : + +```text +continuity.rs : 7 -> 8 +lib.rs : 31 -> 32 +runtime_resources.rs : 34 -> 35 +unit_tests/continuity.rs : 5 -> 6 +unit_tests/runtime_resources.rs : 28 -> 29 +tests/hardening.rs : 31 -> 32 +tests/release_completeness.rs : 26 -> 27 +``` + +## 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.007-fix.001 -> pre.008 : 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.009`. + +## Décisions prises + +```text +une source active n'est pas une preuve historique de coverage +les coverage epochs doivent provenir d'une observation réellement prouvée pendant le run +le processing frontier peut avancer avec un Missing, mais le continuity frontier reste bloqué par son gap +la continuation source-loss exige à la fois réconciliation historique et TargetCoverage future par les sources encore Active +les pertes sans range sûr restent fail-closed +aucun respawn Worker +``` + +## 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 +``` + +## Prochaine tranche + +`pre.009` : appliquer la health policy multi-source `Healthy / Degraded / Unhealthy / Faulted` selon coverage présente et future, avec retour à Healthy seulement après source active et gaps fermés.