v0.3.14-pre.011

crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs
-// version: 8
+// version: 9
 
 /// Runtime-neutral boxed future resolving to one newer concrete RAW transaction ingest Worker snapshot.
 pub type RawTransactionIngestSnapshotFuture<'a> =
@@ -20,14 +20,267 @@ pub enum RawTransactionIngestSourceState {
     Failed,
 }
 
-/// Private latest-value source-processing projection emitted by the productive source task.
+/// Stable source-neutral identifier of one run-local continuity gap.
+#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct RawTransactionIngestGapId(u64);
+
+impl crate::RawTransactionIngestGapId {
+    /// Creates one internal gap identifier after the continuity ledger validated monotonicity.
+    pub(crate) const fn new(value: u64) -> Self {
+        return Self(value);
+    }
+
+    /// Returns the run-local numeric identifier.
+    #[must_use]
+    pub const fn value(self) -> u64 {
+        return self.0;
+    }
+}
+
+/// Source-neutral lifecycle state of one run-local continuity gap.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub enum RawTransactionIngestGapState {
+    /// The gap is known and waiting for an admissible recovery proof.
+    Pending,
+    /// One bounded recovery mechanism is actively processing the gap.
+    Repairing,
+    /// The complete required interval has been proven covered.
+    Repaired,
+    /// The gap remains unresolved after the admissible bounded mechanisms are exhausted.
+    Unresolved,
+}
+
+/// Source-neutral reason why one run-local continuity gap was opened.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub enum RawTransactionIngestGapReason {
+    /// A slot was proven produced but its block material remained unavailable.
+    HttpProducedBlockUnavailable,
+    /// A previously observed transaction reference could not yet be materialized.
+    KnownReferenceMissing,
+    /// A logical live source became terminal before continuity was proven.
+    SourceFailure,
+    /// Transport reported bounded notification loss or overflow.
+    TransportOverflow,
+    /// A WebSocket reconnect opened a bounded continuity incident.
+    WebSocketReconnect,
+    /// Yellowstone replay retention could not cover the requested boundary.
+    YellowstoneRetention,
+}
+
+/// Source-neutral mechanism last used while attempting to close one continuity gap.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub enum RawTransactionIngestRepairMethod {
+    /// Transport-owned replay supplied material later proven sufficient.
+    Replay,
+    /// A distinct live source supplied explicit interval coverage.
+    RedundantCoverage,
+    /// A bounded HTTP slot/block discovery scan supplied coverage evidence.
+    HttpScan,
+    /// A produced slot required direct block material retrieval.
+    BlockFetch,
+    /// A known transaction reference required observed transaction hydration.
+    TransactionHydration,
+}
+
+/// Safe source-neutral snapshot of one recent or still-open run-local continuity gap.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct RawTransactionIngestGapSnapshot {
+    gap_id: crate::RawTransactionIngestGapId,
+    start_slot: u64,
+    end_slot: u64,
+    state: crate::RawTransactionIngestGapState,
+    reason: crate::RawTransactionIngestGapReason,
+    last_method: std::option::Option<crate::RawTransactionIngestRepairMethod>,
+}
+
+impl crate::RawTransactionIngestGapSnapshot {
+    /// Creates one safe gap projection from already validated continuity-ledger state.
+    pub(crate) const fn new(
+        gap_id: crate::RawTransactionIngestGapId,
+        start_slot: u64,
+        end_slot: u64,
+        state: crate::RawTransactionIngestGapState,
+        reason: crate::RawTransactionIngestGapReason,
+        last_method: std::option::Option<crate::RawTransactionIngestRepairMethod>,
+    ) -> Self {
+        return Self { gap_id, start_slot, end_slot, state, reason, last_method };
+    }
+
+    /// Returns the run-local gap identifier.
+    #[must_use]
+    pub const fn gap_id(&self) -> crate::RawTransactionIngestGapId {
+        return self.gap_id;
+    }
+
+    /// Returns the inclusive first slot of the gap.
+    #[must_use]
+    pub const fn start_slot(&self) -> u64 {
+        return self.start_slot;
+    }
+
+    /// Returns the inclusive last slot of the gap.
+    #[must_use]
+    pub const fn end_slot(&self) -> u64 {
+        return self.end_slot;
+    }
+
+    /// Returns the source-neutral gap lifecycle state.
+    #[must_use]
+    pub const fn state(&self) -> crate::RawTransactionIngestGapState {
+        return self.state;
+    }
+
+    /// Returns the source-neutral reason that opened the gap.
+    #[must_use]
+    pub const fn reason(&self) -> crate::RawTransactionIngestGapReason {
+        return self.reason;
+    }
+
+    /// Returns the last bounded recovery mechanism recorded for this gap, when any.
+    #[must_use]
+    pub const fn last_method(&self) -> std::option::Option<crate::RawTransactionIngestRepairMethod> {
+        return self.last_method;
+    }
+}
+
+/// Private bounded continuity-observability projection carried into the public latest-value Worker snapshot.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub(crate) struct RawTransactionIngestContinuitySnapshotProjection {
+    gaps: std::vec::Vec<crate::RawTransactionIngestGapSnapshot>,
+    open_gap_count: usize,
+    repairing_gap_count: usize,
+    repaired_gap_total: u64,
+    unresolved_gap_total: u64,
+    replay_repair_total: u64,
+    redundant_coverage_repair_total: u64,
+    http_scan_repair_total: u64,
+    repair_block_fetch_total: u64,
+    repair_transaction_hydration_total: u64,
+    oldest_open_gap_start_slot: std::option::Option<u64>,
+}
+
+impl crate::RawTransactionIngestContinuitySnapshotProjection {
+    /// Returns the empty continuity-observability projection before any gap exists.
+    pub(crate) fn empty() -> Self {
+        return Self {
+            gaps: std::vec::Vec::new(),
+            open_gap_count: 0,
+            repairing_gap_count: 0,
+            repaired_gap_total: 0,
+            unresolved_gap_total: 0,
+            replay_repair_total: 0,
+            redundant_coverage_repair_total: 0,
+            http_scan_repair_total: 0,
+            repair_block_fetch_total: 0,
+            repair_transaction_hydration_total: 0,
+            oldest_open_gap_start_slot: std::option::Option::None,
+        };
+    }
+
+    /// Returns a copy carrying bounded gap details and current open-gap aggregates.
+    pub(crate) fn with_gap_state(
+        mut self,
+        gaps: std::vec::Vec<crate::RawTransactionIngestGapSnapshot>,
+        open_gap_count: usize,
+        repairing_gap_count: usize,
+        oldest_open_gap_start_slot: std::option::Option<u64>,
+    ) -> Self {
+        self.gaps = gaps;
+        self.open_gap_count = open_gap_count;
+        self.repairing_gap_count = repairing_gap_count;
+        self.oldest_open_gap_start_slot = oldest_open_gap_start_slot;
+        return self;
+    }
+
+    /// Returns a copy carrying cumulative source-neutral gap recovery totals.
+    pub(crate) fn with_recovery_totals(
+        mut self,
+        repaired_gap_total: u64,
+        unresolved_gap_total: u64,
+        replay_repair_total: u64,
+        redundant_coverage_repair_total: u64,
+        http_scan_repair_total: u64,
+    ) -> Self {
+        self.repaired_gap_total = repaired_gap_total;
+        self.unresolved_gap_total = unresolved_gap_total;
+        self.replay_repair_total = replay_repair_total;
+        self.redundant_coverage_repair_total = redundant_coverage_repair_total;
+        self.http_scan_repair_total = http_scan_repair_total;
+        return self;
+    }
+
+    /// Returns a copy carrying cumulative bounded material-recovery totals.
+    pub(crate) fn with_material_totals(mut self, repair_block_fetch_total: u64, repair_transaction_hydration_total: u64) -> Self {
+        self.repair_block_fetch_total = repair_block_fetch_total;
+        self.repair_transaction_hydration_total = repair_transaction_hydration_total;
+        return self;
+    }
+
+    /// Returns the bounded recent/open gap projections.
+    pub(crate) fn gaps(&self) -> &[crate::RawTransactionIngestGapSnapshot] {
+        return self.gaps.as_slice();
+    }
+
+    /// Returns the current number of open gaps.
+    pub(crate) const fn open_gap_count(&self) -> usize {
+        return self.open_gap_count;
+    }
+
+    /// Returns the current number of actively recovering gaps.
+    pub(crate) const fn repairing_gap_count(&self) -> usize {
+        return self.repairing_gap_count;
+    }
+
+    /// Returns the cumulative repaired-gap total.
+    pub(crate) const fn repaired_gap_total(&self) -> u64 {
+        return self.repaired_gap_total;
+    }
+
+    /// Returns the cumulative unresolved-gap total.
+    pub(crate) const fn unresolved_gap_total(&self) -> u64 {
+        return self.unresolved_gap_total;
+    }
+
+    /// Returns the cumulative replay recovery total.
+    pub(crate) const fn replay_repair_total(&self) -> u64 {
+        return self.replay_repair_total;
+    }
+
+    /// Returns the cumulative distinct-source coverage recovery total.
+    pub(crate) const fn redundant_coverage_repair_total(&self) -> u64 {
+        return self.redundant_coverage_repair_total;
+    }
+
+    /// Returns the cumulative bounded HTTP scan recovery total.
+    pub(crate) const fn http_scan_repair_total(&self) -> u64 {
+        return self.http_scan_repair_total;
+    }
+
+    /// Returns the cumulative direct block material recovery total.
+    pub(crate) const fn repair_block_fetch_total(&self) -> u64 {
+        return self.repair_block_fetch_total;
+    }
+
+    /// Returns the cumulative transaction hydration recovery total.
+    pub(crate) const fn repair_transaction_hydration_total(&self) -> u64 {
+        return self.repair_transaction_hydration_total;
+    }
+
+    /// Returns the oldest first slot among currently open gaps.
+    pub(crate) const fn oldest_open_gap_start_slot(&self) -> std::option::Option<u64> {
+        return self.oldest_open_gap_start_slot;
+    }
+}
+
+/// Private latest-value source-processing projection emitted by the productive source task.
+#[derive(Clone, Debug, Eq, PartialEq)]
 pub(crate) struct RawTransactionIngestProcessingFrontierProjection {
     continuity_frontier_slot: std::option::Option<u64>,
     continuity_has_open_gaps: bool,
     continuity_policy_observed: bool,
     failed_source_losses_reconciled: bool,
     future_target_coverage: bool,
+    continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection,
     hydration_pending: usize,
     processing_frontier_slot: std::option::Option<u64>,
     oldest_pending_slot: std::option::Option<u64>,
@@ -43,13 +296,14 @@ pub(crate) struct RawTransactionIngestProcessingFrontierProjection {
 
 impl crate::RawTransactionIngestProcessingFrontierProjection {
     /// Returns the empty run-local processing projection used before the source observes work.
-    pub(crate) const fn empty() -> Self {
+    pub(crate) fn empty() -> Self {
         return Self {
             continuity_frontier_slot: std::option::Option::None,
             continuity_has_open_gaps: false,
             continuity_policy_observed: false,
             failed_source_losses_reconciled: false,
             future_target_coverage: false,
+            continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection::empty(),
             hydration_pending: 0,
             processing_frontier_slot: std::option::Option::None,
             oldest_pending_slot: std::option::Option::None,
@@ -65,17 +319,14 @@ impl crate::RawTransactionIngestProcessingFrontierProjection {
     }
 
     /// Creates one run-local processing projection from bounded source-owned state.
-    pub(crate) const fn new(
-        hydration_pending: usize,
-        processing_frontier_slot: std::option::Option<u64>,
-        oldest_pending_slot: std::option::Option<u64>,
-    ) -> Self {
+    pub(crate) fn new(hydration_pending: usize, processing_frontier_slot: std::option::Option<u64>, oldest_pending_slot: std::option::Option<u64>) -> Self {
         return Self {
             continuity_frontier_slot: std::option::Option::None,
             continuity_has_open_gaps: false,
             continuity_policy_observed: false,
             failed_source_losses_reconciled: false,
             future_target_coverage: false,
+            continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection::empty(),
             hydration_pending,
             processing_frontier_slot,
             oldest_pending_slot,
@@ -106,7 +357,7 @@ impl crate::RawTransactionIngestProcessingFrontierProjection {
     }
 
     /// Returns a copy carrying the latest source reconnect/replay projection.
-    pub(crate) const fn with_source_continuity(
+    pub(crate) fn with_source_continuity(
         mut self,
         source_state: std::option::Option<crate::RawTransactionIngestSourceState>,
         source_reconnect_total: u64,
@@ -121,7 +372,7 @@ impl crate::RawTransactionIngestProcessingFrontierProjection {
     }
 
     /// Returns a copy carrying source-neutral multi-source lifecycle counts.
-    pub(crate) const fn with_source_counts(mut self, source_total: usize, source_active: usize, source_reconnecting: usize, source_failed: usize) -> Self {
+    pub(crate) fn with_source_counts(mut self, source_total: usize, source_active: usize, source_reconnecting: usize, source_failed: usize) -> Self {
         self.source_total = source_total;
         self.source_active = source_active;
         self.source_reconnecting = source_reconnecting;
@@ -130,7 +381,7 @@ impl crate::RawTransactionIngestProcessingFrontierProjection {
     }
 
     /// Returns a copy carrying source-neutral run-local continuity evidence used only for Worker health classification.
-    pub(crate) const fn with_continuity_health(
+    pub(crate) fn with_continuity_health(
         mut self,
         continuity_frontier_slot: std::option::Option<u64>,
         continuity_has_open_gaps: bool,
@@ -143,6 +394,17 @@ impl crate::RawTransactionIngestProcessingFrontierProjection {
         return self;
     }
 
+    /// Returns a copy carrying bounded source-neutral continuity observability.
+    pub(crate) fn with_continuity_snapshot(mut self, continuity_snapshot: crate::RawTransactionIngestContinuitySnapshotProjection) -> Self {
+        self.continuity_snapshot = continuity_snapshot;
+        return self;
+    }
+
+    /// Returns the bounded source-neutral continuity observability carried by this projection.
+    pub(crate) const fn continuity_snapshot(&self) -> &crate::RawTransactionIngestContinuitySnapshotProjection {
+        return &self.continuity_snapshot;
+    }
+
     /// Returns whether a run-local continuity policy projection has been observed.
     pub(crate) const fn continuity_policy_observed(&self) -> bool {
         return self.continuity_policy_observed;
@@ -164,7 +426,7 @@ impl crate::RawTransactionIngestProcessingFrontierProjection {
     }
 
     /// Returns a copy carrying whether every terminal Failed source has a reconciled source-loss gap.
-    pub(crate) const fn with_failed_source_losses_reconciled(mut self, failed_source_losses_reconciled: bool) -> Self {
+    pub(crate) fn with_failed_source_losses_reconciled(mut self, failed_source_losses_reconciled: bool) -> Self {
         self.failed_source_losses_reconciled = failed_source_losses_reconciled;
         return self;
     }
@@ -224,6 +486,17 @@ pub struct RawTransactionIngestSnapshot {
     continuity_policy_observed: bool,
     failed_source_losses_reconciled: bool,
     future_target_coverage: bool,
+    gaps: std::vec::Vec<crate::RawTransactionIngestGapSnapshot>,
+    open_gap_count: usize,
+    repairing_gap_count: usize,
+    repaired_gap_total: u64,
+    unresolved_gap_total: u64,
+    replay_repair_total: u64,
+    redundant_coverage_repair_total: u64,
+    http_scan_repair_total: u64,
+    repair_block_fetch_total: u64,
+    repair_transaction_hydration_total: u64,
+    oldest_open_gap_start_slot: std::option::Option<u64>,
     admission_queue_capacity: usize,
     admission_queue_depth: usize,
     persistence_concurrency: usize,
@@ -374,6 +647,74 @@ impl crate::RawTransactionIngestSnapshot {
         return self.oldest_pending_slot;
     }
 
+    /// Returns the bounded source-neutral gap projections retained in this latest-value snapshot.
+    ///
+    /// Every open gap is retained while capacity remains bounded; recent repaired entries may fill unused projection capacity.
+    #[must_use]
+    pub fn gaps(&self) -> &[crate::RawTransactionIngestGapSnapshot] {
+        return self.gaps.as_slice();
+    }
+
+    /// Returns the number of currently open run-local continuity gaps.
+    #[must_use]
+    pub const fn open_gap_count(&self) -> usize {
+        return self.open_gap_count;
+    }
+
+    /// Returns the number of currently active gap recoveries.
+    #[must_use]
+    pub const fn repairing_gap_count(&self) -> usize {
+        return self.repairing_gap_count;
+    }
+
+    /// Returns the number of gaps cumulatively proven repaired during this run.
+    #[must_use]
+    pub const fn repaired_gap_total(&self) -> u64 {
+        return self.repaired_gap_total;
+    }
+
+    /// Returns the number of gaps cumulatively classified unresolved during this run.
+    #[must_use]
+    pub const fn unresolved_gap_total(&self) -> u64 {
+        return self.unresolved_gap_total;
+    }
+
+    /// Returns the number of gaps whose latest successful recovery proof used replay.
+    #[must_use]
+    pub const fn replay_repair_total(&self) -> u64 {
+        return self.replay_repair_total;
+    }
+
+    /// Returns the number of gaps whose latest successful recovery proof used distinct-source coverage.
+    #[must_use]
+    pub const fn redundant_coverage_repair_total(&self) -> u64 {
+        return self.redundant_coverage_repair_total;
+    }
+
+    /// Returns the number of gaps whose latest successful recovery proof used a bounded HTTP scan.
+    #[must_use]
+    pub const fn http_scan_repair_total(&self) -> u64 {
+        return self.http_scan_repair_total;
+    }
+
+    /// Returns the number of repaired gaps whose last recorded mechanism was direct block material retrieval.
+    #[must_use]
+    pub const fn repair_block_fetch_total(&self) -> u64 {
+        return self.repair_block_fetch_total;
+    }
+
+    /// Returns the number of repaired gaps whose last recorded mechanism was transaction hydration.
+    #[must_use]
+    pub const fn repair_transaction_hydration_total(&self) -> u64 {
+        return self.repair_transaction_hydration_total;
+    }
+
+    /// Returns the oldest first slot among currently open continuity gaps.
+    #[must_use]
+    pub const fn oldest_open_gap_start_slot(&self) -> std::option::Option<u64> {
+        return self.oldest_open_gap_start_slot;
+    }
+
     /// Returns the configured number of logical live sources for this Worker run.
     #[must_use]
     pub const fn source_total(&self) -> usize {
@@ -428,6 +769,17 @@ impl std::fmt::Debug for crate::RawTransactionIngestSnapshot {
         return formatter
             .debug_struct("RawTransactionIngestSnapshot")
             .field("worker", &self.worker)
+            .field("gaps", &self.gaps)
+            .field("open_gap_count", &self.open_gap_count)
+            .field("repairing_gap_count", &self.repairing_gap_count)
+            .field("repaired_gap_total", &self.repaired_gap_total)
+            .field("unresolved_gap_total", &self.unresolved_gap_total)
+            .field("replay_repair_total", &self.replay_repair_total)
+            .field("redundant_coverage_repair_total", &self.redundant_coverage_repair_total)
+            .field("http_scan_repair_total", &self.http_scan_repair_total)
+            .field("repair_block_fetch_total", &self.repair_block_fetch_total)
+            .field("repair_transaction_hydration_total", &self.repair_transaction_hydration_total)
+            .field("oldest_open_gap_start_slot", &self.oldest_open_gap_start_slot)
             .field("admission_queue_capacity", &self.admission_queue_capacity)
             .field("admission_queue_depth", &self.admission_queue_depth)
             .field("persistence_concurrency", &self.persistence_concurrency)
@@ -558,6 +910,17 @@ impl crate::RawTransactionIngestSnapshotPublisher {
             continuity_policy_observed: false,
             failed_source_losses_reconciled: false,
             future_target_coverage: false,
+            gaps: std::vec::Vec::new(),
+            open_gap_count: 0,
+            repairing_gap_count: 0,
+            repaired_gap_total: 0,
+            unresolved_gap_total: 0,
+            replay_repair_total: 0,
+            redundant_coverage_repair_total: 0,
+            http_scan_repair_total: 0,
+            repair_block_fetch_total: 0,
+            repair_transaction_hydration_total: 0,
+            oldest_open_gap_start_slot: std::option::Option::None,
             admission_queue_capacity: settings.admission_queue_capacity(),
             admission_queue_depth: 0,
             persistence_concurrency: settings.persistence_concurrency(),
@@ -679,6 +1042,18 @@ impl crate::RawTransactionIngestSnapshotPublisher {
         self.snapshot.continuity_policy_observed = projection.continuity_policy_observed();
         self.snapshot.failed_source_losses_reconciled = projection.failed_source_losses_reconciled();
         self.snapshot.future_target_coverage = projection.future_target_coverage();
+        let continuity_snapshot = projection.continuity_snapshot();
+        self.snapshot.gaps = continuity_snapshot.gaps().to_vec();
+        self.snapshot.open_gap_count = continuity_snapshot.open_gap_count();
+        self.snapshot.repairing_gap_count = continuity_snapshot.repairing_gap_count();
+        self.snapshot.repaired_gap_total = continuity_snapshot.repaired_gap_total();
+        self.snapshot.unresolved_gap_total = continuity_snapshot.unresolved_gap_total();
+        self.snapshot.replay_repair_total = continuity_snapshot.replay_repair_total();
+        self.snapshot.redundant_coverage_repair_total = continuity_snapshot.redundant_coverage_repair_total();
+        self.snapshot.http_scan_repair_total = continuity_snapshot.http_scan_repair_total();
+        self.snapshot.repair_block_fetch_total = continuity_snapshot.repair_block_fetch_total();
+        self.snapshot.repair_transaction_hydration_total = continuity_snapshot.repair_transaction_hydration_total();
+        self.snapshot.oldest_open_gap_start_slot = continuity_snapshot.oldest_open_gap_start_slot();
         self.snapshot.hydration_pending = projection.hydration_pending();
         self.snapshot.processing_frontier_slot = projection.processing_frontier_slot();
         self.snapshot.oldest_pending_slot = projection.oldest_pending_slot();