From b2068e9ef28f033cb21248f29b68e43cf59b2eb5 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Fri, 11 Sep 2026 19:58:32 +0200 Subject: [PATCH] v0.3.14-pre.002 --- Cargo.toml | 4 +- .../src/continuity.rs | 435 ++++++++++++++++++ .../src/lib.rs | 19 +- .../src/runtime_resources.rs | 172 ++++++- .../tests/hardening.rs | 58 ++- .../tests/release_completeness.rs | 18 +- .../unit_tests/continuity.rs | 241 ++++++++++ deltas/0.3.14/pre.002.md | 257 +++++++++++ 8 files changed, 1194 insertions(+), 10 deletions(-) create mode 100644 crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs create mode 100644 crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/continuity.rs create mode 100644 deltas/0.3.14/pre.002.md diff --git a/Cargo.toml b/Cargo.toml index 8c817c7..58a62ec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 566 +# version: 567 [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.1" +version = "0.3.14-pre.2" 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 new file mode 100644 index 0000000..c091b74 --- /dev/null +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs @@ -0,0 +1,435 @@ +// file: crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs +// version: 1 + +/// Maximum number of simultaneously retained non-repaired run-local gaps. +pub(crate) const MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS: usize = 64; +/// Maximum number of run-local gaps that may actively repair at once. +pub(crate) const MAX_RAW_TRANSACTION_INGEST_REPAIR_ACTIVE_GAPS: usize = 1; +/// Maximum number of logical block fetches that a later repair scheduler may admit concurrently. +pub(crate) const MAX_RAW_TRANSACTION_INGEST_REPAIR_BLOCK_FETCH_IN_FLIGHT: usize = 4; +/// Maximum number of slots admitted by one later HTTP repair discovery window. +pub(crate) const MAX_RAW_TRANSACTION_INGEST_REPAIR_DISCOVERY_WINDOW_SLOTS: u64 = 512; +/// Maximum inclusive slot span admitted for one run-local repair gap. +pub(crate) const MAX_RAW_TRANSACTION_INGEST_REPAIR_RANGE_SLOTS: u64 = 4_096; + +/// Private source scope used to prove whether one configured live source can cover another run-local requirement. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) enum RawTransactionIngestCoverageScope { + /// Every transaction contained by every produced block at the configured commitment. + FullLedgerTransactions, + /// One opaque exact source-family scope identified independently from provider and endpoint identity. + ExactSourceScope(&'static str, [u8; 32]), + /// A bounded set of references already observed by the Worker; never sufficient as configured target coverage. + KnownReferences([u8; 32]), +} + +impl crate::RawTransactionIngestCoverageScope { + /// Creates one provider-neutral exact source scope from a stable family code and semantic fingerprint. + pub(crate) const fn exact_source_scope(family_code: &'static str, fingerprint: [u8; 32]) -> Self { + return Self::ExactSourceScope(family_code, fingerprint); + } + + /// Creates the full-ledger transaction scope used by complete block sources. + pub(crate) const fn full_ledger_transactions() -> Self { + return Self::FullLedgerTransactions; + } + + fn validate_configured_target_scope(&self) -> ksp_core_lib::Result<()> { + return match self { + Self::FullLedgerTransactions => std::result::Result::Ok(()), + Self::ExactSourceScope(family_code, fingerprint) => { + if family_code.is_empty() { + return std::result::Result::Err(crate::runtime_error("continuity.exact_scope_family_empty")); + } + let _fingerprint_first_byte = fingerprint[0]; + std::result::Result::Ok(()) + }, + Self::KnownReferences(fingerprint) => { + let _fingerprint_first_byte = fingerprint[0]; + std::result::Result::Err(crate::runtime_error("continuity.known_references_not_target_scope")) + }, + }; + } + + fn is_known_references(&self) -> bool { + return matches!(self, Self::KnownReferences(_)); + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct RawTransactionIngestCoverageRequirement { + commitment: ksp_onchain_transport_lib::SolanaCommitment, + scope: crate::RawTransactionIngestCoverageScope, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum RawTransactionIngestGapReason { + HttpProducedBlockUnavailable, + SourceFailure, + TransportOverflow, + WebSocketReconnect, + YellowstoneRetention, +} + +impl RawTransactionIngestGapReason { + const ALL: [Self; 5] = + [Self::HttpProducedBlockUnavailable, Self::SourceFailure, Self::TransportOverflow, Self::WebSocketReconnect, Self::YellowstoneRetention]; +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum RawTransactionIngestGapState { + Pending, + Repairing, + Repaired, + Unresolved, +} + +impl RawTransactionIngestGapState { + const ALL: [Self; 4] = [Self::Pending, Self::Repairing, Self::Repaired, Self::Unresolved]; + + const fn is_open(self) -> bool { + return !matches!(self, Self::Repaired); + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct RawTransactionIngestGapRange { + end_slot: u64, + start_slot: u64, +} + +impl RawTransactionIngestGapRange { + fn new(start_slot: u64, end_slot: u64) -> ksp_core_lib::Result { + if end_slot < start_slot { + return std::result::Result::Err(crate::runtime_error("continuity.gap_range_reversed")); + } + let slot_count = match end_slot.checked_sub(start_slot).and_then(|value| return value.checked_add(1)) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.gap_range_overflow")), + }; + if slot_count > crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_RANGE_SLOTS { + return std::result::Result::Err(crate::runtime_error("continuity.gap_range_too_large")); + } + return std::result::Result::Ok(Self { end_slot, start_slot }); + } + + const fn end_slot(self) -> u64 { + return self.end_slot; + } + + fn overlaps_or_is_adjacent(self, other: Self) -> bool { + if self.start_slot <= other.end_slot && other.start_slot <= self.end_slot { + return true; + } + let self_next = self.end_slot.checked_add(1); + if self_next == std::option::Option::Some(other.start_slot) { + return true; + } + let other_next = other.end_slot.checked_add(1); + return other_next == std::option::Option::Some(self.start_slot); + } + + const fn start_slot(self) -> u64 { + return self.start_slot; + } + + fn try_merge(self, other: Self) -> ksp_core_lib::Result> { + if !self.overlaps_or_is_adjacent(other) { + return std::result::Result::Ok(std::option::Option::None); + } + let start_slot = self.start_slot.min(other.start_slot); + let end_slot = self.end_slot.max(other.end_slot); + let slot_count = match end_slot.checked_sub(start_slot).and_then(|value| return value.checked_add(1)) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Ok(std::option::Option::None), + }; + if slot_count > crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_RANGE_SLOTS { + return std::result::Result::Ok(std::option::Option::None); + } + return std::result::Result::Ok(std::option::Option::Some(Self { end_slot, start_slot })); + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd)] +struct RawTransactionIngestGapId(u64); + +struct RawTransactionIngestGap { + commitment: ksp_onchain_transport_lib::SolanaCommitment, + coverage_requirement: crate::RawTransactionIngestCoverageScope, + gap_id: RawTransactionIngestGapId, + network: ksp_store_lib::RawNetworkId, + range: RawTransactionIngestGapRange, + reason: RawTransactionIngestGapReason, + source_key: [u8; 32], + state: RawTransactionIngestGapState, +} + +struct RawTransactionIngestGapLedger { + gaps: std::vec::Vec, + network: ksp_store_lib::RawNetworkId, + next_gap_id: u64, +} + +impl RawTransactionIngestGapLedger { + fn new(network: ksp_store_lib::RawNetworkId) -> Self { + return Self { gaps: std::vec::Vec::new(), network, next_gap_id: 1 }; + } + + 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")); + } + let mut ids = std::collections::BTreeSet::new(); + let mut open_gap_count = 0_usize; + let mut active_gap_count = 0_usize; + let mut highest_gap_id = 0_u64; + for gap in &self.gaps { + if gap.network != self.network { + return std::result::Result::Err(crate::runtime_error("continuity.gap_network_mismatch")); + } + if gap.gap_id.0 == 0 || !ids.insert(gap.gap_id) { + return std::result::Result::Err(crate::runtime_error("continuity.gap_id_invalid")); + } + highest_gap_id = highest_gap_id.max(gap.gap_id.0); + 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 { + active_gap_count = match active_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.active_gap_count")), + }; + } + if !RawTransactionIngestGapReason::ALL.contains(&gap.reason) || !RawTransactionIngestGapState::ALL.contains(&gap.state) { + return std::result::Result::Err(crate::runtime_error("continuity.gap_catalog_invalid")); + } + if gap.range.start_slot() > gap.range.end_slot() { + return std::result::Result::Err(crate::runtime_error("continuity.gap_range_reversed")); + } + } + if open_gap_count > crate::MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS { + return std::result::Result::Err(crate::runtime_error("continuity.open_gap_limit_exceeded")); + } + if active_gap_count > crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_ACTIVE_GAPS { + return std::result::Result::Err(crate::runtime_error("continuity.active_gap_limit_exceeded")); + } + if !self.gaps.is_empty() && self.next_gap_id <= highest_gap_id { + return std::result::Result::Err(crate::runtime_error("continuity.next_gap_id_invalid")); + } + for first_index in 0..self.gaps.len() { + let first = &self.gaps[first_index]; + for second in self.gaps.iter().skip(first_index + 1) { + if first.source_key != second.source_key + || first.commitment != second.commitment + || first.coverage_requirement != second.coverage_requirement + || first.reason != second.reason + || !first.state.is_open() + || !second.state.is_open() + { + continue; + } + let merged = first.range.try_merge(second.range); + match merged { + std::result::Result::Ok(std::option::Option::Some(_)) => { + return std::result::Result::Err(crate::runtime_error("continuity.coalescible_gap_ranges")); + }, + std::result::Result::Ok(std::option::Option::None) => {}, + std::result::Result::Err(error) => return std::result::Result::Err(error), + } + } + } + if RawTransactionIngestGapReason::ALL.len() != 5 || RawTransactionIngestGapState::ALL.len() != 4 { + return std::result::Result::Err(crate::runtime_error("continuity.gap_catalog_invalid")); + } + return std::result::Result::Ok(()); + } +} + +/// Private source capability descriptor prepared without network I/O from already validated runtime resources. +pub(crate) struct RawTransactionIngestRepairCapabilityDescriptor { + block_material: bool, + commitment: ksp_onchain_transport_lib::SolanaCommitment, + http_block_scan: bool, + known_reference_hydration: bool, + native_replay: bool, + network: ksp_store_lib::RawNetworkId, + reference_bearing: bool, + slot_enumerating: bool, + source_key: [u8; 32], + source_scope: crate::RawTransactionIngestCoverageScope, +} + +impl crate::RawTransactionIngestRepairCapabilityDescriptor { + /// Creates one private descriptor from source-local capabilities that were validated without issuing repair I/O. + #[allow(clippy::too_many_arguments)] + pub(crate) fn new( + source_key: [u8; 32], + network: ksp_store_lib::RawNetworkId, + commitment: ksp_onchain_transport_lib::SolanaCommitment, + source_scope: crate::RawTransactionIngestCoverageScope, + reference_bearing: bool, + block_material: bool, + slot_enumerating: bool, + known_reference_hydration: bool, + native_replay: bool, + http_block_scan: bool, + ) -> ksp_core_lib::Result { + let descriptor = Self { + block_material, + commitment, + http_block_scan, + known_reference_hydration, + native_replay, + network, + reference_bearing, + slot_enumerating, + source_key, + source_scope, + }; + if let std::result::Result::Err(error) = descriptor.validate() { + return std::result::Result::Err(error); + } + return std::result::Result::Ok(descriptor); + } + + fn validate(&self) -> ksp_core_lib::Result<()> { + if self.commitment == ksp_onchain_transport_lib::SolanaCommitment::Processed { + return std::result::Result::Err(crate::runtime_error("continuity.processed_commitment_unsupported")); + } + if let std::result::Result::Err(error) = self.source_scope.validate_configured_target_scope() { + return std::result::Result::Err(error); + } + if !self.reference_bearing + && !self.block_material + && !self.slot_enumerating + && !self.known_reference_hydration + && !self.native_replay + && !self.http_block_scan + { + return std::result::Result::Err(crate::runtime_error("continuity.source_capability_empty")); + } + if self.http_block_scan && (!self.block_material || !self.slot_enumerating) { + return std::result::Result::Err(crate::runtime_error("continuity.http_scan_capability_incomplete")); + } + return std::result::Result::Ok(()); + } +} + +struct RawTransactionIngestTargetCoverage { + requirements: std::vec::Vec, +} + +impl RawTransactionIngestTargetCoverage { + fn from_capabilities(capabilities: &[crate::RawTransactionIngestRepairCapabilityDescriptor]) -> ksp_core_lib::Result { + let mut value = Self { requirements: std::vec::Vec::new() }; + for capability in capabilities { + if let std::result::Result::Err(error) = value.include(capability.commitment, capability.source_scope.clone()) { + return std::result::Result::Err(error); + } + } + if value.requirements.is_empty() { + return std::result::Result::Err(crate::runtime_error("continuity.target_coverage_empty")); + } + return std::result::Result::Ok(value); + } + + fn include( + &mut self, + commitment: ksp_onchain_transport_lib::SolanaCommitment, + scope: crate::RawTransactionIngestCoverageScope, + ) -> ksp_core_lib::Result<()> { + if scope.is_known_references() { + return std::result::Result::Err(crate::runtime_error("continuity.known_references_not_target_scope")); + } + if matches!(&scope, crate::RawTransactionIngestCoverageScope::FullLedgerTransactions) { + self.requirements.retain(|requirement| return requirement.commitment != commitment); + self.requirements.push(RawTransactionIngestCoverageRequirement { commitment, scope }); + return std::result::Result::Ok(()); + } + if self.requirements.iter().any(|requirement| { + return requirement.commitment == commitment + && (requirement.scope == crate::RawTransactionIngestCoverageScope::FullLedgerTransactions || requirement.scope == scope); + }) { + return std::result::Result::Ok(()); + } + self.requirements.push(RawTransactionIngestCoverageRequirement { commitment, scope }); + return std::result::Result::Ok(()); + } +} + +/// Run-local continuity contracts prepared from the exact caller-composed live-source aggregate before productive source execution. +pub(crate) struct RawTransactionIngestContinuityContracts { + capabilities: std::vec::Vec, + gap_ledger: RawTransactionIngestGapLedger, + target_coverage: RawTransactionIngestTargetCoverage, +} + +impl crate::RawTransactionIngestContinuityContracts { + /// Builds the private capability inventory, conservative `TargetCoverage` and empty run-local gap ledger without issuing network or Store I/O. + pub(crate) fn new(capabilities: std::vec::Vec) -> ksp_core_lib::Result { + if capabilities.is_empty() || capabilities.len() > crate::MAX_RAW_TRANSACTION_INGEST_LIVE_SOURCES { + return std::result::Result::Err(crate::runtime_error("continuity.capability_inventory_size_invalid")); + } + let network = capabilities[0].network.clone(); + let mut source_keys = std::collections::BTreeSet::new(); + for capability in &capabilities { + if let std::result::Result::Err(error) = capability.validate() { + return std::result::Result::Err(error); + } + if capability.network != network { + return std::result::Result::Err(crate::runtime_error("continuity.capability_network_mismatch")); + } + if !source_keys.insert(capability.source_key) { + return std::result::Result::Err(crate::runtime_error("continuity.duplicate_capability_source")); + } + } + let target_coverage = match RawTransactionIngestTargetCoverage::from_capabilities(capabilities.as_slice()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let gap_ledger = RawTransactionIngestGapLedger::new(network); + if let std::result::Result::Err(error) = gap_ledger.validate_invariants() { + return std::result::Result::Err(error); + } + if !coverage_scope_catalog_is_complete() { + return std::result::Result::Err(crate::runtime_error("continuity.coverage_scope_catalog_invalid")); + } + if crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_BLOCK_FETCH_IN_FLIGHT == 0 + || crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_DISCOVERY_WINDOW_SLOTS == 0 + || crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_DISCOVERY_WINDOW_SLOTS > crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_RANGE_SLOTS + { + return std::result::Result::Err(crate::runtime_error("continuity.repair_bounds_invalid")); + } + return std::result::Result::Ok(Self { capabilities, gap_ledger, target_coverage }); + } + + /// 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() { + return std::result::Result::Err(crate::runtime_error("continuity.source_inventory_mismatch")); + } + if let std::result::Result::Err(error) = self.gap_ledger.validate_invariants() { + return std::result::Result::Err(error); + } + return std::result::Result::Ok(()); + } +} + +fn coverage_scope_catalog_is_complete() -> bool { + let full = crate::RawTransactionIngestCoverageScope::FullLedgerTransactions; + let exact = crate::RawTransactionIngestCoverageScope::ExactSourceScope("catalog", [0_u8; 32]); + let known = crate::RawTransactionIngestCoverageScope::KnownReferences([0_u8; 32]); + return full.validate_configured_target_scope().is_ok() + && exact.validate_configured_target_scope().is_ok() + && known.validate_configured_target_scope().is_err() + && !full.is_known_references() + && !exact.is_known_references() + && known.is_known_references(); +} + +#[cfg(test)] +#[path = "../unit_tests/continuity.rs"] +mod tests; 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 0568afa..5166c26 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: 26 +// version: 27 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -21,6 +21,7 @@ //! state and faults conservatively when Transport proves a replay-retention continuity gap; history remediation remains outside this crate. mod admission; +mod continuity; mod error; mod identity; mod persistence; @@ -110,6 +111,22 @@ pub use self::snapshot::RawTransactionIngestSourceState; pub(crate) use self::admission::RawTransactionAdmission; /// Crate-private source-neutral ingress sent through the bounded central admission queue. pub(crate) use self::admission::RawTransactionIngress; +/// Maximum number of simultaneously retained non-repaired run-local continuity gaps. +pub(crate) use self::continuity::MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS; +/// Maximum number of run-local continuity gaps that may actively repair at once. +pub(crate) use self::continuity::MAX_RAW_TRANSACTION_INGEST_REPAIR_ACTIVE_GAPS; +/// Maximum number of logical block fetches reserved for the later bounded repair scheduler. +pub(crate) use self::continuity::MAX_RAW_TRANSACTION_INGEST_REPAIR_BLOCK_FETCH_IN_FLIGHT; +/// Maximum number of slots reserved for one later HTTP repair discovery window. +pub(crate) use self::continuity::MAX_RAW_TRANSACTION_INGEST_REPAIR_DISCOVERY_WINDOW_SLOTS; +/// Maximum inclusive slot span accepted by one run-local repair gap. +pub(crate) use self::continuity::MAX_RAW_TRANSACTION_INGEST_REPAIR_RANGE_SLOTS; +/// Private run-local continuity aggregate containing capabilities, TargetCoverage and the gap ledger. +pub(crate) use self::continuity::RawTransactionIngestContinuityContracts; +/// Private provider-neutral coverage scope used by run-local continuity proof contracts. +pub(crate) use self::continuity::RawTransactionIngestCoverageScope; +/// Private source capability descriptor prepared without repair I/O. +pub(crate) use self::continuity::RawTransactionIngestRepairCapabilityDescriptor; /// Creates one terminal content-conflict error without copying conflicting material into diagnostics. pub(crate) use self::error::content_conflict_error; /// Creates one terminal counter-exhaustion error without exposing runtime material. 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 47378ac..03e02cd 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: 26 +// version: 27 use sha2::Digest; // rust-rules: trait-import @@ -31,6 +31,7 @@ const RAW_TRANSACTION_INGEST_STANDARD_BLOCK_PROTOCOL: &str = "solana_ws"; const RAW_TRANSACTION_INGEST_STANDARD_LOGS_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.standard_logs.filter.v1\0"; const RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_PROTOCOL: &str = "solana_ws_http"; const RAW_TRANSACTION_INGEST_STANDARD_LOGS_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.standard_logs_http.source_key.v1\0"; +const RAW_TRANSACTION_INGEST_YELLOWSTONE_COVERAGE_SCOPE_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone.coverage_scope.v1\0"; const RAW_TRANSACTION_INGEST_YELLOWSTONE_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone.filters.v1\0"; const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL: &str = "yellowstone_http"; const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone_http.source_key.v1\0"; @@ -75,6 +76,86 @@ impl RawTransactionIngestLiveSource { }; } + fn repair_capability_descriptor(&self) -> ksp_core_lib::Result { + let commitment = match self { + Self::HeliusTransaction(source) => source.commitment, + Self::HttpBlockPolling(source) => source.commitment, + Self::StandardBlock(source) => source.commitment, + Self::StandardLogs(source) => source.commitment, + Self::Yellowstone(source) => match source.subscribe_request.commitment() { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.yellowstone_commitment_missing")), + }, + }; + let source_scope = match self { + Self::HeliusTransaction(source) => crate::RawTransactionIngestCoverageScope::exact_source_scope("helius_transaction", source.filter_fingerprint), + Self::HttpBlockPolling(_) => crate::RawTransactionIngestCoverageScope::full_ledger_transactions(), + Self::StandardBlock(source) => match &source.filter { + ksp_onchain_transport_lib::SolanaBlockSubscribeFilter::All => crate::RawTransactionIngestCoverageScope::full_ledger_transactions(), + ksp_onchain_transport_lib::SolanaBlockSubscribeFilter::MentionsAccountOrProgram(_) => { + crate::RawTransactionIngestCoverageScope::exact_source_scope("standard_block", source.filter_fingerprint) + }, + }, + Self::StandardLogs(source) => crate::RawTransactionIngestCoverageScope::exact_source_scope("standard_logs", source.filter_fingerprint), + Self::Yellowstone(source) => { + let fingerprint = match yellowstone_coverage_scope_fingerprint(&source.subscribe_request) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + crate::RawTransactionIngestCoverageScope::exact_source_scope("yellowstone", fingerprint) + }, + }; + let (reference_bearing, live_block_material, native_replay, http_block_scan, known_reference_hydration) = match self { + Self::HeliusTransaction(source) => { + let http_block_scan = match http_role_supports_repair_scan(&source.http_pool, &source.hydration_role, source.network.as_str()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + (true, false, false, http_block_scan, true) + }, + Self::HttpBlockPolling(source) => { + let known_reference_hydration = + match http_role_supports_rpc_method(&source.http_pool, &source.polling_role, "getTransaction", source.network.as_str()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + (false, true, false, true, known_reference_hydration) + }, + Self::StandardBlock(_) => (false, true, false, false, false), + Self::StandardLogs(source) => { + let http_block_scan = match http_role_supports_repair_scan(&source.http_pool, &source.hydration_role, source.network.as_str()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + (true, false, false, http_block_scan, true) + }, + Self::Yellowstone(source) => { + let http_block_scan = match http_role_supports_repair_scan(&source.http_pool, &source.hydration_role, source.network.as_str()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let reference_bearing = + source.subscribe_request.transaction_filter_count() > 0 || source.subscribe_request.transaction_status_filter_count() > 0; + let live_block_material = source.subscribe_request.block_filter_count() > 0; + (reference_bearing, live_block_material, true, http_block_scan, true) + }, + }; + let block_material = live_block_material || http_block_scan; + let slot_enumerating = http_block_scan; + return crate::RawTransactionIngestRepairCapabilityDescriptor::new( + self.source_key(), + self.network().clone(), + commitment, + source_scope, + reference_bearing, + block_material, + slot_enumerating, + known_reference_hydration, + native_replay, + http_block_scan, + ); + } + fn source_key(&self) -> [u8; 32] { return match self { Self::HeliusTransaction(source) => source.source_key, @@ -2023,6 +2104,22 @@ 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()); + 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); + } + let continuity_contracts = match crate::RawTransactionIngestContinuityContracts::new(repair_capabilities) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let continuity_source_count = self.sources.len(); + if let std::result::Result::Err(error) = continuity_contracts.validate_for_source_count(continuity_source_count) { + return std::result::Result::Err(error); + } 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) = @@ -2072,7 +2169,11 @@ impl crate::RawTransactionIngestRuntimeResources { }); } std::mem::drop(admission_sender); - return supervise_live_source_tasks(stop_receiver, source_stop_sender, children).await; + 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) { + return std::result::Result::Err(error); + } + return result; } } @@ -2205,6 +2306,57 @@ async fn drain_live_source_tasks( }; } +fn http_role_supports_repair_scan( + pool: &ksp_onchain_transport_lib::HttpTransportPool, + role: &ksp_onchain_transport_lib::HttpRoleName, + expected_cluster: &str, +) -> ksp_core_lib::Result { + let has_get_block = match http_role_supports_rpc_method(pool, role, "getBlock", expected_cluster) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let has_get_blocks = match http_role_supports_rpc_method(pool, role, "getBlocks", expected_cluster) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let has_get_blocks_with_limit = match http_role_supports_rpc_method(pool, role, "getBlocksWithLimit", expected_cluster) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let has_get_slot = match http_role_supports_rpc_method(pool, role, "getSlot", expected_cluster) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok(has_get_block && has_get_slot && (has_get_blocks || has_get_blocks_with_limit)); +} + +fn http_role_supports_rpc_method( + pool: &ksp_onchain_transport_lib::HttpTransportPool, + role: &ksp_onchain_transport_lib::HttpRoleName, + method_name: &'static str, + expected_cluster: &str, +) -> ksp_core_lib::Result { + let method = match ksp_onchain_transport_lib::find_http_rpc_method(method_name) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("continuity.http_method_missing")), + }; + let snapshot = pool.snapshot(); + for endpoint in snapshot.endpoints() { + if !endpoint.enabled() || endpoint.cluster() != expected_cluster { + continue; + } + for endpoint_role in endpoint.roles() { + if !endpoint_role.enabled() || endpoint_role.role() != role.as_str() { + continue; + } + if http_role_supports_request_kind(endpoint_role, method.request_kind()) { + return std::result::Result::Ok(true); + } + } + } + return std::result::Result::Ok(false); +} + fn http_block_polling_live_source_key( network: &ksp_store_lib::RawNetworkId, polling_role: &ksp_onchain_transport_lib::HttpRoleName, @@ -2639,6 +2791,22 @@ fn standard_logs_live_source_key( return hasher.finalize().into(); } +fn yellowstone_coverage_scope_fingerprint(request: &ksp_onchain_transport_lib::YellowstoneSubscribeRequest) -> ksp_core_lib::Result<[u8; 32]> { + let mut normalized = request.clone(); + normalized.set_commitment(std::option::Option::None); + normalized.set_from_slot(std::option::Option::None); + normalized.set_ping(std::option::Option::None); + let identity = match normalized.identity() { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("continuity.yellowstone_scope_identity_invalid")), + }; + let mut hasher = sha2::Sha256::new(); + hasher.update(RAW_TRANSACTION_INGEST_YELLOWSTONE_COVERAGE_SCOPE_FINGERPRINT_DOMAIN); + let mut writer = RawTransactionIngestSourceKeyHashWriter { hasher: &mut hasher }; + std::hash::Hash::hash(&identity, &mut writer); + return std::result::Result::Ok(hasher.finalize().into()); +} + fn yellowstone_live_source_key( network: &ksp_store_lib::RawNetworkId, route: &RawTransactionIngestSourceRoute, 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 4d91639..f6af31f 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: 24 +// version: 25 -//! External public, security, redaction and release-boundary hardening canaries through `pre.012`. +//! External public, security, redaction and release-boundary hardening canaries through `v0.3.14-pre.002`. fn network(value: &'static str) -> std::option::Option { let result = ksp_store_lib::RawNetworkId::new(value); @@ -957,3 +957,57 @@ fn v0_3_13_pre_012_cross_layer_security_closure_is_complete_and_source_neutral() } return; } + +#[test] +fn v0_3_14_pre_002_continuity_contracts_are_private_bounded_and_io_free() { + 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 [ + "MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS", + "MAX_RAW_TRANSACTION_INGEST_REPAIR_ACTIVE_GAPS", + "MAX_RAW_TRANSACTION_INGEST_REPAIR_BLOCK_FETCH_IN_FLIGHT", + "MAX_RAW_TRANSACTION_INGEST_REPAIR_DISCOVERY_WINDOW_SLOTS", + "MAX_RAW_TRANSACTION_INGEST_REPAIR_RANGE_SLOTS", + "FullLedgerTransactions", + "ExactSourceScope", + "KnownReferences", + "RawTransactionIngestGapLedger", + "RawTransactionIngestTargetCoverage", + "RawTransactionIngestRepairCapabilityDescriptor", + "RawTransactionIngestContinuityContracts", + "continuity.known_references_not_target_scope", + "continuity.coalescible_gap_ranges", + ] { + assert!(continuity.contains(required), "required pre.002 continuity contract missing: {required}"); + } + for required in [ + "repair_capability_descriptor", + "yellowstone_coverage_scope_fingerprint", + "http_role_supports_repair_scan", + "RawTransactionIngestContinuityContracts::new", + "validate_for_source_count", + ] { + assert!(resources.contains(required), "required pre.002 runtime-resource capability wiring missing: {required}"); + } + for forbidden in [ + "get_block_observed", + "get_transaction_observed", + "get_blocks_with_limit", + "get_blocks(", + "open_standard_subscribe", + "SolanaStandardWsSession::connect", + "HeliusLaserStreamWsSession::connect", + "ksp_config_lib::", + "ksp_job_backfill_lib::", + "ksp_store_postgres_lib::", + "reqwest::", + "tonic::", + "yellowstone_grpc_proto::", + ] { + assert!(!continuity.contains(forbidden), "pre.002 continuity contract performed or imported forbidden I/O/boundary: {forbidden}"); + } + assert!(!root.contains("pub use self::continuity::")); + assert!(root.contains("pub(crate) use self::continuity::RawTransactionIngestContinuityContracts;")); + 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 fdd0ea3..58bb712 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: 20 +// version: 21 -//! Release-completeness canaries through the `pre.012` cross-layer completeness/security tranche. +//! Release-completeness canaries through the `v0.3.14-pre.002` continuity-contract tranche. #[test] fn pre_010_production_module_inventory_is_exact() -> std::io::Result<()> { @@ -34,7 +34,18 @@ fn pre_010_production_module_inventory_is_exact() -> std::io::Result<()> { names.sort_unstable(); assert_eq!( names, - std::vec!["admission.rs", "error.rs", "identity.rs", "lib.rs", "persistence.rs", "runtime.rs", "runtime_resources.rs", "settings.rs", "snapshot.rs",] + std::vec![ + "admission.rs", + "continuity.rs", + "error.rs", + "identity.rs", + "lib.rs", + "persistence.rs", + "runtime.rs", + "runtime_resources.rs", + "settings.rs", + "snapshot.rs", + ] ); return std::result::Result::Ok(()); } @@ -132,6 +143,7 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { "v0_3_13_pre_009_duplicate_storm_disagreement_and_starvation_guards_are_explicit", "v0_3_13_pre_010_multi_source_health_is_conservative_counted_and_redacted", "v0_3_13_pre_011_shutdown_races_are_bounded_joined_atomic_and_counter_safe", + "v0_3_14_pre_002_continuity_contracts_are_private_bounded_and_io_free", ] { assert!(hardening.contains(required), "required pre.010 hardening canary missing: {required}"); } 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 new file mode 100644 index 0000000..4ce6d00 --- /dev/null +++ b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/continuity.rs @@ -0,0 +1,241 @@ +// file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/continuity.rs +// version: 1 + +fn network() -> std::option::Option { + return match ksp_store_lib::RawNetworkId::new("mainnet") { + std::result::Result::Ok(value) => std::option::Option::Some(value), + std::result::Result::Err(_) => std::option::Option::None, + }; +} + +fn exact_scope(family: &'static str, fingerprint_byte: u8) -> crate::RawTransactionIngestCoverageScope { + return crate::RawTransactionIngestCoverageScope::exact_source_scope(family, [fingerprint_byte; 32]); +} + +fn capability( + source_key_byte: u8, + commitment: ksp_onchain_transport_lib::SolanaCommitment, + scope: crate::RawTransactionIngestCoverageScope, +) -> std::option::Option { + let network = match network() { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + return match crate::RawTransactionIngestRepairCapabilityDescriptor::new( + [source_key_byte; 32], + network, + commitment, + scope, + true, + false, + false, + true, + false, + false, + ) { + std::result::Result::Ok(value) => std::option::Option::Some(value), + std::result::Result::Err(_) => std::option::Option::None, + }; +} + +fn gap( + gap_id: u64, + source_key_byte: u8, + start_slot: u64, + end_slot: u64, + state: super::RawTransactionIngestGapState, +) -> std::option::Option { + let network = match network() { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + let range = match super::RawTransactionIngestGapRange::new(start_slot, end_slot) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::option::Option::None, + }; + return std::option::Option::Some(super::RawTransactionIngestGap { + commitment: ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + coverage_requirement: exact_scope("standard_logs", 9), + gap_id: super::RawTransactionIngestGapId(gap_id), + network, + range, + reason: super::RawTransactionIngestGapReason::WebSocketReconnect, + source_key: [source_key_byte; 32], + state, + }); +} + +#[test] +fn pre_002_repair_bounds_are_exact_and_run_local() { + assert_eq!(crate::MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS, 64); + assert_eq!(crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_ACTIVE_GAPS, 1); + assert_eq!(crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_BLOCK_FETCH_IN_FLIGHT, 4); + assert_eq!(crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_DISCOVERY_WINDOW_SLOTS, 512); + assert_eq!(crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_RANGE_SLOTS, 4_096); + return; +} + +#[test] +fn pre_002_gap_range_is_inclusive_bounded_and_overflow_safe() { + assert!(super::RawTransactionIngestGapRange::new(10, 9).is_err()); + assert!(super::RawTransactionIngestGapRange::new(10, 10 + crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_RANGE_SLOTS - 1).is_ok()); + assert!(super::RawTransactionIngestGapRange::new(10, 10 + crate::MAX_RAW_TRANSACTION_INGEST_REPAIR_RANGE_SLOTS).is_err()); + assert!(super::RawTransactionIngestGapRange::new(u64::MAX, u64::MAX).is_ok()); + return; +} + +#[test] +fn pre_002_gap_range_overlap_and_adjacency_are_explicit() { + let first = match super::RawTransactionIngestGapRange::new(100, 109) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("unexpected first range failure: {error}"), + }; + let overlap = match super::RawTransactionIngestGapRange::new(105, 115) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("unexpected overlap range failure: {error}"), + }; + let adjacent = match super::RawTransactionIngestGapRange::new(110, 120) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("unexpected adjacent range failure: {error}"), + }; + let separate = match super::RawTransactionIngestGapRange::new(111, 120) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("unexpected separate range failure: {error}"), + }; + assert!(first.overlaps_or_is_adjacent(overlap)); + assert!(first.overlaps_or_is_adjacent(adjacent)); + assert!(!first.overlaps_or_is_adjacent(separate)); + let merged = first.try_merge(adjacent); + assert!(matches!(merged, std::result::Result::Ok(std::option::Option::Some(value)) if value.start_slot() == 100 && value.end_slot() == 120)); + return; +} + +#[test] +fn pre_002_target_coverage_deduplicates_exact_scopes_without_cross_commitment_broadening() { + 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 finalized_full = + match capability(3, ksp_onchain_transport_lib::SolanaCommitment::Finalized, crate::RawTransactionIngestCoverageScope::full_ledger_transactions()) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("full capability fixture unavailable"), + }; + let contracts = match crate::RawTransactionIngestContinuityContracts::new(std::vec![first, second, finalized_full]) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("continuity contracts fixture failed: {error}"), + }; + assert_eq!(contracts.capabilities.len(), 3); + assert_eq!(contracts.target_coverage.requirements.len(), 2); + assert!(contracts.target_coverage.requirements.iter().any(|requirement| { + return requirement.commitment == ksp_onchain_transport_lib::SolanaCommitment::Confirmed && requirement.scope == exact_scope("standard_logs", 7); + })); + assert!(contracts.target_coverage.requirements.iter().any(|requirement| { + return requirement.commitment == ksp_onchain_transport_lib::SolanaCommitment::Finalized + && requirement.scope == crate::RawTransactionIngestCoverageScope::full_ledger_transactions(); + })); + return; +} + +#[test] +fn pre_002_full_ledger_scope_subsumes_only_same_commitment_targets() { + let exact = match capability(1, ksp_onchain_transport_lib::SolanaCommitment::Confirmed, exact_scope("helius_transaction", 4)) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("exact capability fixture unavailable"), + }; + let full = 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 capability fixture unavailable"), + }; + let contracts = match crate::RawTransactionIngestContinuityContracts::new(std::vec![exact, full]) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("continuity contracts fixture failed: {error}"), + }; + assert_eq!(contracts.target_coverage.requirements.len(), 1); + assert_eq!(contracts.target_coverage.requirements[0].scope, crate::RawTransactionIngestCoverageScope::full_ledger_transactions()); + assert_eq!(contracts.target_coverage.requirements[0].commitment, ksp_onchain_transport_lib::SolanaCommitment::Confirmed); + return; +} + +#[test] +fn pre_002_known_references_can_never_be_configured_target_coverage() { + let network = match network() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("network fixture unavailable"), + }; + let result = crate::RawTransactionIngestRepairCapabilityDescriptor::new( + [1_u8; 32], + network, + ksp_onchain_transport_lib::SolanaCommitment::Confirmed, + crate::RawTransactionIngestCoverageScope::KnownReferences([2_u8; 32]), + true, + false, + false, + true, + false, + false, + ); + assert!(result.is_err()); + return; +} + +#[test] +fn pre_002_gap_ledger_rejects_coalescible_open_ranges_and_active_overflow() { + let network = match network() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("network fixture unavailable"), + }; + let first = match gap(1, 1, 100, 109, super::RawTransactionIngestGapState::Pending) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("first gap fixture unavailable"), + }; + let adjacent = match gap(2, 1, 110, 120, super::RawTransactionIngestGapState::Pending) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("adjacent gap fixture unavailable"), + }; + let adjacent_ledger = super::RawTransactionIngestGapLedger { gaps: std::vec![first, adjacent], network: network.clone(), next_gap_id: 3 }; + assert!(adjacent_ledger.validate_invariants().is_err()); + let first_active = match gap(1, 1, 100, 109, super::RawTransactionIngestGapState::Repairing) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("first active gap fixture unavailable"), + }; + let second_active = match gap(2, 2, 200, 209, super::RawTransactionIngestGapState::Repairing) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("second active gap fixture unavailable"), + }; + let active_ledger = super::RawTransactionIngestGapLedger { gaps: std::vec![first_active, second_active], network, next_gap_id: 3 }; + assert!(active_ledger.validate_invariants().is_err()); + return; +} + +#[test] +fn pre_002_gap_ledger_enforces_open_gap_bound_and_next_id_monotonicity() { + let network = match network() { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("network fixture unavailable"), + }; + let mut gaps = std::vec::Vec::new(); + for index in 0..=crate::MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS { + let gap_id = (index as u64) + 1; + let slot = (index as u64) * 2; + let gap = match gap(gap_id, 1, slot, slot, super::RawTransactionIngestGapState::Pending) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("gap fixture unavailable"), + }; + gaps.push(gap); + } + let ledger = super::RawTransactionIngestGapLedger { gaps, network: network.clone(), next_gap_id: 66 }; + assert!(ledger.validate_invariants().is_err()); + let single = match gap(7, 1, 100, 100, super::RawTransactionIngestGapState::Unresolved) { + std::option::Option::Some(value) => value, + std::option::Option::None => panic!("single gap fixture unavailable"), + }; + let stale_next = super::RawTransactionIngestGapLedger { gaps: std::vec![single], network, next_gap_id: 7 }; + assert!(stale_next.validate_invariants().is_err()); + return; +} diff --git a/deltas/0.3.14/pre.002.md b/deltas/0.3.14/pre.002.md new file mode 100644 index 0000000..00cc6e1 --- /dev/null +++ b/deltas/0.3.14/pre.002.md @@ -0,0 +1,257 @@ + + + +# Delta `0.3.14-pre.002` — contrats run-local gap / range / coverage + +## Base requise + +```text +0.3.14-pre.001 +workspace.package.version = 0.3.14-pre.1 +deltas/0.3.14/pre.001.md présent +``` + +Gate opérateur communiqué sur `0.3.14-pre.1` : + +```text +cargo fmt --all -- --check : PASS +General Rust rule audit : clean +Rust export completeness audit : 0 candidate(s) +KSP workspace Rust rule audit : clean +Markdown table audit : clean (306 tables / 181 files sur le périmètre demandé) +cargo check --workspace : PASS +``` + +## Objectif + +Matérialiser uniquement la tranche `pre.002` du plan `035` : + +```text +contrats run-local de gap et plage inclusive +TargetCoverage conservative par commitment et scope +FullLedgerTransactions / ExactSourceScope / KnownReferences +inventaire privé des capabilities de repair déjà composées +ledger run-local borné et invariants de coalescence +bornes de future discovery / fetch / active repair +raccord pur aux RuntimeResources avant spawn des sources +``` + +Cette tranche ne réalise aucun repair, replay, scan HTTP, hydration de gap ou I/O Store supplémentaire. + +## Fichiers ajoutés + +```text +crates/ksp-worker-raw-transaction-ingest-lib/src/continuity.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/continuity.rs +deltas/0.3.14/pre.002.md +``` + +## Fichiers modifiés + +```text +Cargo.toml +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 +``` + +## Fichiers supprimés + +```text +aucun +``` + +## Contrats introduits + +Bornes run-local : + +```text +MAX_RAW_TRANSACTION_INGEST_OPEN_REPAIR_GAPS = 64 +MAX_RAW_TRANSACTION_INGEST_REPAIR_RANGE_SLOTS = 4096 +MAX_RAW_TRANSACTION_INGEST_REPAIR_DISCOVERY_WINDOW_SLOTS = 512 +MAX_RAW_TRANSACTION_INGEST_REPAIR_BLOCK_FETCH_IN_FLIGHT = 4 +MAX_RAW_TRANSACTION_INGEST_REPAIR_ACTIVE_GAPS = 1 +``` + +Le scope de coverage distingue explicitement : + +```text +FullLedgerTransactions +ExactSourceScope(family, semantic_fingerprint) +KnownReferences(fingerprint) +``` + +`KnownReferences` est admis comme notion de capability mais rejeté comme `TargetCoverage` configurée : connaître des signatures à hydrater ne prouve jamais qu'aucune transaction n'a été perdue. + +Les plages de gap sont inclusives, bornées, validées avec arithmétique checked et coalescibles seulement lorsqu'elles se chevauchent ou sont adjacentes sans dépasser la borne maximale. + +Le ledger impose notamment : + +```text +réseau unique du run +GapId non nul, unique et monotone +maximum 64 gaps ouverts +maximum 1 gap en Repairing +pas de ranges ouverts coalescibles du même source/scope/commitment/reason +aucun wrap implicite des compteurs ou bornes +``` + +## Capabilities source + +L'inventaire est dérivé des `RuntimeResources` déjà validées, sans I/O : + +```text +HTTP Block Polling All -> FullLedgerTransactions +Standard Block All -> FullLedgerTransactions +Standard Block filtered -> ExactSourceScope +Standard Logs -> ExactSourceScope +Helius transactionSubscribe -> ExactSourceScope +Yellowstone -> ExactSourceScope provider-neutral +``` + +La capability de scan HTTP n'est pas inférée de la simple présence d'un `HttpTransportPool`. Elle est déclarée seulement lorsque le rôle fourni expose réellement, sur le réseau attendu : + +```text +getBlock +getSlot +et au moins getBlocks ou getBlocksWithLimit +``` + +La disponibilité de `getTransaction` est détectée séparément pour la future hydration de références. + +Pour Yellowstone, le fingerprint de coverage retire du calcul les éléments de transport/continuité qui ne définissent pas le scope métier : + +```text +commitment +from_slot +ping +``` + +Le scope reste volontairement conservatif : aucune équivalence cross-family ou provider n'est inventée. + +## Intégration runtime + +Avant le spawn des sources live, `RawTransactionIngestRuntimeResources` : + +```text +construit exactement un descriptor par source +valide réseau / source keys / TargetCoverage / bornes +construit un ledger vide run-local +conserve ces contrats pendant la durée du supervisor +revalide les invariants à la fermeture +``` + +Aucune nouvelle tâche, socket, retry, requête HTTP/gRPC/WS, lecture Config ou opération Store n'est ajoutée par cette tranche. + +## Tests ajoutés / renforcés + +Les tests couvrent : + +```text +valeurs exactes des cinq bornes +plages inclusives / reversed / maximum / overflow +chevauchement et adjacency +merge borné +TargetCoverage exact dédupliquée +absence de broadening entre commitments +subsomption FullLedger seulement au même commitment +rejet de KnownReferences comme target +limites open/active gaps +monotonicité GapId +module production continuity privé +absence d'I/O et de nouvelles frontières dans continuity.rs +inventaire de modules de production mis à jour +``` + +La surface publique du crate reste inchangée ; les nouveaux contrats sont `pub(crate)` et réexportés uniquement par la façade crate-root pour usage interne conforme aux règles KSP. + +## Dépendances / features + +```text +aucune nouvelle dépendance +aucune nouvelle feature +aucun SDK provider +aucun accès Config depuis le Worker +aucun accès Job Backfill depuis le Worker +aucun backend Store physique depuis le Worker +``` + +## Version Cargo + +```text +header Cargo.toml : 566 -> 567 +workspace.package.version : 0.3.14-pre.1 -> 0.3.14-pre.2 +``` + +## Validations exécutées + +Dans le sandbox de préparation : + +```text +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/0.3.14 +scan statique frontières / I/O continuity +comparaison exacte avec la base pre.001 +contrôle contenu archive delta +unzip -t archive delta +``` + +Résultats : + +```text +General Rust rule audit : clean +Rust export completeness audit : 0 candidate(s) +KSP workspace Rust rule audit : clean +Markdown table audit : clean (306 tables / 182 files) +scan continuity I/O / frontières : PASS +scan surface publique continuity : PASS +comparaison pre.001 -> pre.002 : 3 ajouts / 5 modifications / 0 suppression +``` + +Le contrôle ZIP est exécuté après construction de l'archive de livraison ; toute archive qui échoue ce contrôle est rejetée et n'est pas livrée. + +## Validations non exécutées + +Le sandbox de préparation ne fournit pas le toolchain Cargo/Rust ; les gates suivants ne peuvent donc pas constituer une preuve locale : + +```text +cargo fmt --all -- --check +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 +``` + +Ils doivent être exécutés côté opérateur après application du delta. + +## Décisions prises + +```text +pre.002 reste sans repair I/O +TargetCoverage est conservative et commitment-sensitive +FullLedger ne broaden jamais un commitment différent +KnownReferences ne prouve jamais la complétude +HTTP repair capability est détectée par méthodes réellement routables +Yellowstone coverage identity exclut cursor replay et identité provider/endpoint +les contrats restent privés au Worker +aucune nouvelle dépendance ou frontière architecturale +``` + +## Questions ouvertes + +```text +aucune pour pre.002 +``` + +La tranche suivante reste `pre.003` : observabilité latest-value de continuité WebSocket, sans encore déclencher le repair. + +## Gate opérateur après application + +```bash +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/0.3.14 +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 +```