v0.3.14-pre.002

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<Self> {
+        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<std::option::Option<Self>> {
+        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<RawTransactionIngestGap>,
+    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<Self> {
+        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<RawTransactionIngestCoverageRequirement>,
+}
+
+impl RawTransactionIngestTargetCoverage {
+    fn from_capabilities(capabilities: &[crate::RawTransactionIngestRepairCapabilityDescriptor]) -> ksp_core_lib::Result<Self> {
+        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<crate::RawTransactionIngestRepairCapabilityDescriptor>,
+    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<crate::RawTransactionIngestRepairCapabilityDescriptor>) -> ksp_core_lib::Result<Self> {
+        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;