From ba69a5ad20c620e62f2d7935c55a4b5f12b1c496 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Wed, 9 Sep 2026 16:21:23 +0200 Subject: [PATCH] v0.3.12-pre.007 --- Cargo.toml | 4 +- .../src/lib.rs | 7 +- .../src/runtime.rs | 89 ++++- .../src/runtime_resources.rs | 319 ++++++++++++++++-- .../src/snapshot.rs | 93 ++++- .../tests/dependency_boundary.rs | 79 +++-- .../tests/hardening.rs | 66 +++- .../tests/public_api.rs | 17 +- .../tests/release_completeness.rs | 10 +- .../unit_tests/runtime_resources.rs | 101 +++++- .../unit_tests/snapshot.rs | 32 +- deltas/0.3.12/pre.007.md | 216 ++++++++++++ ...0_3_12_YELLOWSTONE_HYDRATION_CONTINUITY.md | 109 +++++- 13 files changed, 1037 insertions(+), 105 deletions(-) create mode 100644 deltas/0.3.12/pre.007.md diff --git a/Cargo.toml b/Cargo.toml index d7d68e4..a90f235 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 528 +# version: 529 [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.12-pre.6.fix.1" +version = "0.3.12-pre.7" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" 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 ef43474..7ae24bc 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: 13 +// version: 14 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -14,7 +14,8 @@ //! the validated Yellowstone/HTTP runtime-resource contract now drives one productive supervised source task. //! Transaction/TransactionStatus/Block updates feed bounded HTTP `getTransaction` hydration and the existing central //! admission path; BlockMeta/Slot remain continuity-only signals. Hydration is coalesced by network/signature/commitment -//! under bounded in-flight and pending budgets, while reconnect/frontier/replay interpretation remains deferred. +//! under bounded in-flight and pending budgets. A bounded run-local processing frontier now projects hydration pending, +//! oldest pending slot and highest unblocked actually observed slot; reconnect/replay interpretation remains deferred. mod admission; mod error; @@ -104,3 +105,5 @@ pub(crate) use self::persistence::RawTransactionIngestPersistencePort; pub(crate) use self::persistence::persist_raw_transaction_ingest_acquisition; /// Private latest-value publisher and checked counter owner shared by the Worker supervisor. pub(crate) use self::snapshot::RawTransactionIngestSnapshotPublisher; +/// Private latest-value processing-frontier projection emitted by the productive source task. +pub(crate) use self::snapshot::RawTransactionIngestProcessingFrontierProjection; diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs index b75b032..02c36a9 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs @@ -1,8 +1,9 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs -// version: 9 +// version: 10 type PersistencePort = std::sync::Arc; type PersistenceTasks = tokio::task::JoinSet>; +type ProcessingFrontierReceiver = tokio::sync::watch::Receiver; type SourceTasks = tokio::task::JoinSet>; /// Runtime-neutral boxed future resolving after one RAW transaction ingest Worker has fully reached a terminal lifecycle state. @@ -115,11 +116,20 @@ impl crate::RawTransactionIngestWorker { } let source = runtime_resources.into_yellowstone_source(); let source_settings = settings.clone(); - return start_foundation_with_source_spawner(settings, runtime, std::option::Option::Some(store), move |children, stop_receiver, admission_sender| { - let _abort_handle = children.spawn(async move { - return source.run(source_settings, stop_receiver, admission_sender).await; - }); - }); + let (processing_frontier_sender, processing_frontier_receiver) = + tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); + let port: PersistencePort = store; + return start_foundation_with_port_source_spawner_and_frontier( + settings, + runtime, + std::option::Option::Some(port), + std::option::Option::Some(processing_frontier_receiver), + move |children, stop_receiver, admission_sender| { + let _abort_handle = children.spawn(async move { + return source.run(source_settings, stop_receiver, admission_sender, processing_frontier_sender).await; + }); + }, + ); } } @@ -179,6 +189,7 @@ async fn drain_admission_and_persistence( persistence: &mut PersistenceTasks, port: &std::option::Option, snapshots: &mut crate::RawTransactionIngestSnapshotPublisher, + processing_frontier_receiver: &mut std::option::Option, ) -> std::option::Option { let mut fault = std::option::Option::None; admission.close(); @@ -344,6 +355,7 @@ async fn run_supervisor( port: std::option::Option, mut stop_receiver: tokio::sync::watch::Receiver, mut snapshots: crate::RawTransactionIngestSnapshotPublisher, + mut processing_frontier_receiver: std::option::Option, source_spawner: Spawner, ) where Spawner: @@ -380,6 +392,7 @@ async fn run_supervisor( &mut persistence, &port, &mut snapshots, + &mut processing_frontier_receiver, ) .await; source_stop_sender.send_replace(true); @@ -431,6 +444,26 @@ fn start_foundation_with_port_and_source_spawner( port: std::option::Option, source_spawner: Spawner, ) -> ksp_core_lib::Result +where + Spawner: + FnOnce(&mut SourceTasks, tokio::sync::watch::Receiver, tokio::sync::mpsc::Sender) + std::marker::Send + 'static, +{ + return start_foundation_with_port_source_spawner_and_frontier( + settings, + runtime, + port, + std::option::Option::None, + source_spawner, + ); +} + +fn start_foundation_with_port_source_spawner_and_frontier( + settings: crate::RawTransactionIngestSettings, + runtime: tokio::runtime::Handle, + port: std::option::Option, + processing_frontier_receiver: std::option::Option, + source_spawner: Spawner, +) -> ksp_core_lib::Result where Spawner: FnOnce(&mut SourceTasks, tokio::sync::watch::Receiver, tokio::sync::mpsc::Sender) + std::marker::Send + 'static, @@ -447,7 +480,15 @@ where let (stop_sender, stop_receiver) = tokio::sync::watch::channel(false); let (snapshots, snapshot_source) = crate::RawTransactionIngestSnapshotPublisher::new(&settings, &lifecycle); let handle = crate::RawTransactionIngestHandle { snapshots: snapshot_source, stop_sender, stop_token }; - std::mem::drop(runtime.spawn(run_supervisor(settings, lifecycle, port, stop_receiver, snapshots, source_spawner))); + std::mem::drop(runtime.spawn(run_supervisor( + settings, + lifecycle, + port, + stop_receiver, + snapshots, + processing_frontier_receiver, + source_spawner, + ))); return std::result::Result::Ok(handle); } @@ -504,6 +545,25 @@ async fn supervise_until_stop( break; } } + processing_frontier = wait_processing_frontier(processing_frontier_receiver) => { + match processing_frontier { + std::option::Option::Some(projection) => { + let published = snapshots.record_processing_frontier( + lifecycle.state(), + admission.queue_depth(), + persistence.len(), + projection, + ); + if let std::result::Result::Err(error) = published { + source_stop_sender.send_replace(true); + return std::option::Option::Some(error.code()); + } + }, + std::option::Option::None => { + *processing_frontier_receiver = std::option::Option::None; + }, + } + } joined = children.join_next(), if !children.is_empty() => { let source_fault = match joined { std::option::Option::Some(value) => source_completion(value, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots), @@ -571,6 +631,21 @@ async fn supervise_until_stop( return std::option::Option::None; } +async fn wait_processing_frontier( + receiver: &mut std::option::Option, +) -> std::option::Option { + let receiver = match receiver.as_mut() { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::future::pending::>().await; + }, + }; + if receiver.changed().await.is_err() { + return std::option::Option::None; + } + return std::option::Option::Some(*receiver.borrow_and_update()); +} + fn validate_store_network(settings: &crate::RawTransactionIngestSettings, store_network: &ksp_store_lib::RawNetworkId) -> ksp_core_lib::Result<()> { if settings.network() != store_network { return std::result::Result::Err(crate::runtime_error("start.store_network_mismatch")); 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 654828c..2a27044 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: 9 +// version: 10 use sha2::Digest; // rust-rules: trait-import @@ -188,7 +188,10 @@ trait RawTransactionIngestYellowstoneBlockView { fn transaction_count(&self) -> usize; - fn transaction_identity(&self, position: usize) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)>; + fn transaction_identity( + &self, + position: usize, + ) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)>; } impl RawTransactionIngestYellowstoneBlockView for ksp_onchain_transport_lib::YellowstoneBlockUpdate { @@ -208,7 +211,10 @@ impl RawTransactionIngestYellowstoneBlockView for ksp_onchain_transport_lib::Yel return ksp_onchain_transport_lib::YellowstoneBlockUpdate::transactions(self).len(); } - fn transaction_identity(&self, position: usize) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)> { + fn transaction_identity( + &self, + position: usize, + ) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)> { let transaction = match ksp_onchain_transport_lib::YellowstoneBlockUpdate::transactions(self).get(position) { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.block_transaction_missing")), @@ -286,7 +292,9 @@ impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate)> for RawTransactionIngestSourceSignal { - fn from(value: (&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate)) -> Self { + fn from( + value: (&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate), + ) -> Self { return project_yellowstone_signal(value.0, value.1); } } @@ -294,7 +302,9 @@ impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onc impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate)> for RawTransactionIngestSourceSignal { - fn from(value: (&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate)) -> Self { + fn from( + value: (&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate), + ) -> Self { return project_yellowstone_signal(value.0, value.1); } } @@ -367,6 +377,7 @@ impl crate::RawTransactionIngestYellowstoneSource { settings: crate::RawTransactionIngestSettings, mut stop_receiver: tokio::sync::watch::Receiver, admission_sender: tokio::sync::mpsc::Sender, + processing_frontier_sender: tokio::sync::watch::Sender, ) -> ksp_core_lib::Result<()> { let opened = tokio::select! { biased; @@ -380,6 +391,7 @@ impl crate::RawTransactionIngestYellowstoneSource { std::result::Result::Err(error) => return std::result::Result::Err(source_transport_error(error.code())), }; let mut coordinator = RawTransactionIngestHydrationCoordinator::new(&settings); + let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender); let mut fault = std::option::Option::None; loop { if *stop_receiver.borrow() { @@ -407,7 +419,16 @@ impl crate::RawTransactionIngestYellowstoneSource { break; }, }; - let handled = coordinator.handle_joined(joined, &self, &settings, &admission_sender, &mut stop_receiver).await; + let handled = coordinator + .handle_joined( + joined, + &self, + &settings, + &admission_sender, + &mut stop_receiver, + &mut processing_frontier, + ) + .await; match handled { std::result::Result::Ok(true) => {}, std::result::Result::Ok(false) => break, @@ -429,7 +450,12 @@ impl crate::RawTransactionIngestYellowstoneSource { break; }, }; - if let std::result::Result::Err(error) = route_yellowstone_update(&self, &mut coordinator, update) { + if let std::result::Result::Err(error) = route_yellowstone_update( + &self, + &mut coordinator, + &mut processing_frontier, + update, + ) { fault = std::option::Option::Some(error); break; } @@ -565,7 +591,9 @@ fn ingestion_filter_count(request: &ksp_onchain_transport_lib::YellowstoneSubscr return request.transaction_filter_count().saturating_add(request.transaction_status_filter_count()).saturating_add(request.block_filter_count()); } -fn matched_filter_direct_id(filters: &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName]) -> std::option::Option { +fn matched_filter_direct_id( + filters: &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName], +) -> std::option::Option { let mut names = filters.iter().map(ksp_onchain_transport_lib::YellowstoneSubscribeFilterName::as_str).collect::>(); names.sort_unstable(); names.dedup(); @@ -687,6 +715,7 @@ fn project_yellowstone_signal( fn route_yellowstone_update( source: &crate::RawTransactionIngestYellowstoneSource, coordinator: &mut RawTransactionIngestHydrationCoordinator, + processing_frontier: &mut RawTransactionIngestProcessingFrontierReporter, update: ksp_onchain_transport_lib::YellowstoneSubscribeUpdate, ) -> ksp_core_lib::Result<()> { match update { @@ -695,14 +724,24 @@ fn route_yellowstone_update( std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - return coordinator.queue_signal(source, RawTransactionIngestSourceSignal::from((source, value.as_ref())), received_at); + return coordinator.queue_signal( + source, + RawTransactionIngestSourceSignal::from((source, value.as_ref())), + received_at, + processing_frontier, + ); }, ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::TransactionStatus(value) => { let received_at = match current_raw_timestamp() { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - return coordinator.queue_signal(source, RawTransactionIngestSourceSignal::from((source, &value)), received_at); + return coordinator.queue_signal( + source, + RawTransactionIngestSourceSignal::from((source, &value)), + received_at, + processing_frontier, + ); }, ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Block(value) => { let received_at = match current_raw_timestamp() { @@ -714,18 +753,26 @@ fn route_yellowstone_update( std::result::Result::Err(error) => return std::result::Result::Err(error), }; for signal in signals { - if let std::result::Result::Err(error) = coordinator.queue_signal(source, signal, received_at) { + if let std::result::Result::Err(error) = + coordinator.queue_signal(source, signal, received_at, processing_frontier) + { return std::result::Result::Err(error); } } return std::result::Result::Ok(()); }, ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::BlockMeta(value) => { - let _continuity_signal = project_yellowstone_continuity_signal(source, &value); + let continuity_signal = project_yellowstone_continuity_signal(source, &value); + if let std::result::Result::Err(error) = processing_frontier.observe_settled(continuity_signal.slot) { + return std::result::Result::Err(error); + } return std::result::Result::Ok(()); }, ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Slot(value) => { - let _continuity_signal = project_yellowstone_continuity_signal(source, &value); + let continuity_signal = project_yellowstone_continuity_signal(source, &value); + if let std::result::Result::Err(error) = processing_frontier.observe_settled(continuity_signal.slot) { + return std::result::Result::Err(error); + } return std::result::Result::Ok(()); }, ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Account(_) @@ -757,6 +804,178 @@ fn source_transport_error(code: ksp_core_lib::ErrorCode) -> ksp_core_lib::Error .with_context("transport_code", code.code()); } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct RawTransactionIngestProcessingSlotState { + pending: usize, + settled: u64, +} + +struct RawTransactionIngestProcessingFrontier { + pending_total: usize, + slots: std::collections::BTreeMap, +} + +impl RawTransactionIngestProcessingFrontier { + fn new() -> Self { + return Self { pending_total: 0, slots: std::collections::BTreeMap::new() }; + } + + fn observe_pending(&mut self, slot: u64) -> ksp_core_lib::Result<()> { + let pending_total = match self.pending_total.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.frontier_pending_counter_exhausted")), + }; + let state = self.slots.entry(slot).or_insert(RawTransactionIngestProcessingSlotState { pending: 0, settled: 0 }); + state.pending = match state.pending.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.frontier_slot_pending_counter_exhausted")), + }; + self.pending_total = pending_total; + self.compact(); + return std::result::Result::Ok(()); + } + + fn observe_settled(&mut self, slot: u64) -> ksp_core_lib::Result<()> { + let state = self.slots.entry(slot).or_insert(RawTransactionIngestProcessingSlotState { pending: 0, settled: 0 }); + state.settled = match state.settled.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.frontier_slot_settled_counter_exhausted")), + }; + self.compact(); + return std::result::Result::Ok(()); + } + + fn settle_pending(&mut self, slot: u64) -> ksp_core_lib::Result<()> { + let state = match self.slots.get_mut(&slot) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.frontier_pending_slot_missing")), + }; + if state.pending == 0 || self.pending_total == 0 { + return std::result::Result::Err(crate::runtime_error("source.frontier_pending_counter_invalid")); + } + state.pending -= 1; + self.pending_total -= 1; + state.settled = match state.settled.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.frontier_slot_settled_counter_exhausted")), + }; + self.compact(); + return std::result::Result::Ok(()); + } + + fn projection(&self) -> crate::RawTransactionIngestProcessingFrontierProjection { + let oldest_pending_slot = self.slots.iter().find_map(|(slot, state)| { + if state.pending == 0 { + return std::option::Option::None; + } + return std::option::Option::Some(*slot); + }); + let processing_frontier_slot = match oldest_pending_slot { + std::option::Option::Some(pending_slot) => self.slots.range(..pending_slot).rev().find_map(|(slot, state)| { + if state.pending == 0 && state.settled > 0 { + return std::option::Option::Some(*slot); + } + return std::option::Option::None; + }), + std::option::Option::None => self.slots.iter().rev().find_map(|(slot, state)| { + if state.pending == 0 && state.settled > 0 { + return std::option::Option::Some(*slot); + } + return std::option::Option::None; + }), + }; + return crate::RawTransactionIngestProcessingFrontierProjection::new(self.pending_total, processing_frontier_slot, oldest_pending_slot); + } + + fn compact(&mut self) { + let pending_slots = self + .slots + .iter() + .filter_map(|(slot, state)| { + if state.pending > 0 { + return std::option::Option::Some(*slot); + } + return std::option::Option::None; + }) + .collect::>(); + let mut settled_interval_highs = std::collections::BTreeMap::, u64>::new(); + for (slot, state) in &self.slots { + if state.pending > 0 || state.settled == 0 { + continue; + } + let upper_pending = pending_slots.iter().find_map(|pending_slot| { + if pending_slot > slot { + return std::option::Option::Some(*pending_slot); + } + return std::option::Option::None; + }); + match settled_interval_highs.entry(upper_pending) { + std::collections::btree_map::Entry::Occupied(mut entry) => { + if *slot > *entry.get() { + entry.insert(*slot); + } + }, + std::collections::btree_map::Entry::Vacant(entry) => { + entry.insert(*slot); + }, + } + } + let mut retained = pending_slots.into_iter().collect::>(); + for slot in settled_interval_highs.into_values() { + retained.insert(slot); + } + self.slots.retain(|slot, _state| { + return retained.contains(slot); + }); + return; + } + + #[cfg(test)] + fn slot_state_count(&self) -> usize { + return self.slots.len(); + } +} + +struct RawTransactionIngestProcessingFrontierReporter { + frontier: RawTransactionIngestProcessingFrontier, + sender: tokio::sync::watch::Sender, +} + +impl RawTransactionIngestProcessingFrontierReporter { + fn new(sender: tokio::sync::watch::Sender) -> Self { + return Self { frontier: RawTransactionIngestProcessingFrontier::new(), sender }; + } + + fn observe_pending(&mut self, slot: u64) -> ksp_core_lib::Result<()> { + if let std::result::Result::Err(error) = self.frontier.observe_pending(slot) { + return std::result::Result::Err(error); + } + self.publish(); + return std::result::Result::Ok(()); + } + + fn observe_settled(&mut self, slot: u64) -> ksp_core_lib::Result<()> { + if let std::result::Result::Err(error) = self.frontier.observe_settled(slot) { + return std::result::Result::Err(error); + } + self.publish(); + return std::result::Result::Ok(()); + } + + fn settle_pending(&mut self, slot: u64) -> ksp_core_lib::Result<()> { + if let std::result::Result::Err(error) = self.frontier.settle_pending(slot) { + return std::result::Result::Err(error); + } + self.publish(); + return std::result::Result::Ok(()); + } + + fn publish(&self) { + self.sender.send_replace(self.frontier.projection()); + return; + } +} + #[derive(Clone, Eq, Ord, PartialEq, PartialOrd)] struct RawTransactionIngestHydrationKey { commitment: &'static str, @@ -812,6 +1031,7 @@ impl RawTransactionIngestHydrationCoordinator { source: &crate::RawTransactionIngestYellowstoneSource, signal: RawTransactionIngestSourceSignal, received_at: ksp_store_lib::RawTimestamp, + processing_frontier: &mut RawTransactionIngestProcessingFrontierReporter, ) -> ksp_core_lib::Result<()> { if self.pending_signal_count >= self.max_pending_signals { return std::result::Result::Err(crate::runtime_error("source.hydration_pending_saturated")); @@ -820,6 +1040,7 @@ impl RawTransactionIngestHydrationCoordinator { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; + let signal_slot = signal.slot; match self.pending.entry(key) { std::collections::btree_map::Entry::Occupied(mut entry) => { entry.get_mut().signals.push(RawTransactionIngestPendingSignal { received_at, signal }); @@ -835,17 +1056,23 @@ impl RawTransactionIngestHydrationCoordinator { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.hydration_pending_counter_exhausted")), }; + if let std::result::Result::Err(error) = processing_frontier.observe_pending(signal_slot) { + return std::result::Result::Err(error); + } return self.start_hydrations(source); } fn start_hydrations(&mut self, source: &crate::RawTransactionIngestYellowstoneSource) -> ksp_core_lib::Result<()> { while self.tasks.len() < self.max_in_flight { - let key = self.pending.iter().find_map(|(key, pending)| { - if pending.in_flight { - return std::option::Option::None; - } - return std::option::Option::Some(key.clone()); - }); + let key = self + .pending + .iter() + .find_map(|(key, pending)| { + if pending.in_flight { + return std::option::Option::None; + } + return std::option::Option::Some(key.clone()); + }); let key = match key { std::option::Option::Some(value) => value, std::option::Option::None => break, @@ -877,6 +1104,7 @@ impl RawTransactionIngestHydrationCoordinator { settings: &crate::RawTransactionIngestSettings, admission_sender: &tokio::sync::mpsc::Sender, stop_receiver: &mut tokio::sync::watch::Receiver, + processing_frontier: &mut RawTransactionIngestProcessingFrontierReporter, ) -> ksp_core_lib::Result { let fetched = match joined { std::result::Result::Ok(std::result::Result::Ok(value)) => value, @@ -892,6 +1120,7 @@ impl RawTransactionIngestHydrationCoordinator { } self.pending_signal_count -= pending.signals.len(); for pending_signal in pending.signals { + let signal_slot = pending_signal.signal.slot; let ingress = finalize_yellowstone_hydration(source, settings, pending_signal.signal, pending_signal.received_at, &fetched.observed); let ingress = match ingress { std::result::Result::Ok(value) => value, @@ -899,7 +1128,12 @@ impl RawTransactionIngestHydrationCoordinator { }; let ingress = match ingress { std::option::Option::Some(value) => value, - std::option::Option::None => continue, + std::option::Option::None => { + if let std::result::Result::Err(error) = processing_frontier.settle_pending(signal_slot) { + return std::result::Result::Err(error); + } + continue; + }, }; let sent = tokio::select! { biased; @@ -914,6 +1148,9 @@ impl RawTransactionIngestHydrationCoordinator { } return std::result::Result::Err(crate::runtime_error("source.admission_closed")); } + if let std::result::Result::Err(error) = processing_frontier.settle_pending(signal_slot) { + return std::result::Result::Err(error); + } } if let std::result::Result::Err(error) = self.start_hydrations(source) { return std::result::Result::Err(error); @@ -1039,7 +1276,14 @@ fn finalize_yellowstone_hydration( if embedded_signature != signal.signature { return std::result::Result::Err(crate::runtime_error("hydration.signature_mismatch")); } - let provenance = build_hydration_provenance(settings, &signal, observed.provider().as_str(), observed.endpoint_name(), commitment, received_at); + let provenance = build_hydration_provenance( + settings, + &signal, + observed.provider().as_str(), + observed.endpoint_name(), + commitment, + received_at, + ); let provenance = match provenance { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), @@ -1060,7 +1304,12 @@ fn finalize_yellowstone_hydration( }), map_hydration_wire_field(transaction.transaction_index(), |value| return *value), ); - return std::result::Result::Ok(std::option::Option::Some(crate::RawTransactionIngress { material, network: signal.network, provenance, source_key })); + return std::result::Result::Ok(std::option::Option::Some(crate::RawTransactionIngress { + material, + network: signal.network, + provenance, + source_key, + })); } #[cfg(test)] @@ -1085,7 +1334,14 @@ async fn hydrate_yellowstone_signal( std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let fetched = fetch_yellowstone_hydration(source.http_pool.clone(), source.hydration_role.clone(), source.network.clone(), key, commitment).await; + let fetched = fetch_yellowstone_hydration( + source.http_pool.clone(), + source.hydration_role.clone(), + source.network.clone(), + key, + commitment, + ) + .await; let fetched = match fetched { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), @@ -1135,12 +1391,17 @@ fn build_hydration_provenance( std::result::Result::Err(error) => return std::result::Result::Err(error), }, }; - let mut provenance = - ksp_store_lib::RawAcquisitionProvenance::new(provider, protocol, acquisition_method, ksp_store_lib::RawAcquisitionOrigin::Live, received_at) - .with_endpoint_id(endpoint_id) - .with_commitment(commitment) - .with_capture_session_id(capture_session) - .with_filter_id(filter_id); + let mut provenance = ksp_store_lib::RawAcquisitionProvenance::new( + provider, + protocol, + acquisition_method, + ksp_store_lib::RawAcquisitionOrigin::Live, + received_at, + ) + .with_endpoint_id(endpoint_id) + .with_commitment(commitment) + .with_capture_session_id(capture_session) + .with_filter_id(filter_id); if let std::option::Option::Some(observed_at) = representable_observed_at(signal.created_at, received_at) { provenance = match provenance.try_with_observed_at(observed_at) { std::result::Result::Ok(value) => value, diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs index 941ae18..e6a658f 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs @@ -1,10 +1,49 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs -// version: 2 +// version: 3 /// Runtime-neutral boxed future resolving to one newer concrete RAW transaction ingest Worker snapshot. pub type RawTransactionIngestSnapshotFuture<'a> = std::pin::Pin + std::marker::Send + 'a>>; +/// Private latest-value processing-frontier projection emitted by the productive source task. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct RawTransactionIngestProcessingFrontierProjection { + hydration_pending: usize, + processing_frontier_slot: std::option::Option, + oldest_pending_slot: std::option::Option, +} + +impl crate::RawTransactionIngestProcessingFrontierProjection { + /// Returns the empty run-local processing projection used before the source observes work. + pub(crate) const fn empty() -> Self { + return Self { hydration_pending: 0, processing_frontier_slot: std::option::Option::None, oldest_pending_slot: std::option::Option::None }; + } + + /// Creates one run-local processing projection from bounded source-owned state. + pub(crate) const fn new( + hydration_pending: usize, + processing_frontier_slot: std::option::Option, + oldest_pending_slot: std::option::Option, + ) -> Self { + return Self { hydration_pending, processing_frontier_slot, oldest_pending_slot }; + } + + /// Returns the number of source signals still pending hydration/admission processing. + pub(crate) const fn hydration_pending(&self) -> usize { + return self.hydration_pending; + } + + /// Returns the highest actually observed slot not blocked by older pending source work. + pub(crate) const fn processing_frontier_slot(&self) -> std::option::Option { + return self.processing_frontier_slot; + } + + /// Returns the oldest actually observed slot that still owns pending source work. + pub(crate) const fn oldest_pending_slot(&self) -> std::option::Option { + return self.oldest_pending_slot; + } +} + /// Complete safe latest-value snapshot of one continuous RAW transaction ingest Worker. #[derive(Clone, Eq, PartialEq)] pub struct RawTransactionIngestSnapshot { @@ -25,6 +64,9 @@ pub struct RawTransactionIngestSnapshot { store_failure_total: u64, source_failure_total: u64, backpressure_wait_total: u64, + hydration_pending: usize, + processing_frontier_slot: std::option::Option, + oldest_pending_slot: std::option::Option, } impl crate::RawTransactionIngestSnapshot { @@ -129,6 +171,24 @@ impl crate::RawTransactionIngestSnapshot { pub const fn backpressure_wait_total(&self) -> u64 { return self.backpressure_wait_total; } + + /// Returns the latest number of source signals pending hydration/admission processing. + #[must_use] + pub const fn hydration_pending(&self) -> usize { + return self.hydration_pending; + } + + /// Returns the run-local processing frontier over actually observed source work. + #[must_use] + pub const fn processing_frontier_slot(&self) -> std::option::Option { + return self.processing_frontier_slot; + } + + /// Returns the oldest actually observed slot that still owns pending source work. + #[must_use] + pub const fn oldest_pending_slot(&self) -> std::option::Option { + return self.oldest_pending_slot; + } } impl std::fmt::Debug for crate::RawTransactionIngestSnapshot { @@ -152,6 +212,9 @@ impl std::fmt::Debug for crate::RawTransactionIngestSnapshot { .field("store_failure_total", &self.store_failure_total) .field("source_failure_total", &self.source_failure_total) .field("backpressure_wait_total", &self.backpressure_wait_total) + .field("hydration_pending", &self.hydration_pending) + .field("processing_frontier_slot", &self.processing_frontier_slot) + .field("oldest_pending_slot", &self.oldest_pending_slot) .finish(); } } @@ -265,6 +328,9 @@ impl crate::RawTransactionIngestSnapshotPublisher { store_failure_total: 0, source_failure_total: 0, backpressure_wait_total: 0, + hydration_pending: 0, + processing_frontier_slot: std::option::Option::None, + oldest_pending_slot: std::option::Option::None, }; let (sender, receiver) = tokio::sync::watch::channel(snapshot.clone()); return (Self { sender, snapshot }, crate::RawTransactionIngestSnapshotSource { receiver }); @@ -346,6 +412,20 @@ impl crate::RawTransactionIngestSnapshotPublisher { return self.publish(state, admission_queue_depth, in_flight_persistence); } + /// Publishes one latest run-local processing-frontier projection emitted by the source task. + pub(crate) fn record_processing_frontier( + &mut self, + state: ksp_worker_api::WorkerState, + admission_queue_depth: usize, + in_flight_persistence: usize, + projection: crate::RawTransactionIngestProcessingFrontierProjection, + ) -> ksp_core_lib::Result<()> { + 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(); + return self.publish(state, admission_queue_depth, in_flight_persistence); + } + /// Records one failed private source task without retaining provider-specific error material. pub(crate) fn record_source_failure( &mut self, @@ -457,7 +537,7 @@ impl crate::RawTransactionIngestSnapshotPublisher { sequence, state, health_for_state(state, self.snapshot.worker.health()), - activity_for_state(state, admission_queue_depth, in_flight_persistence), + activity_for_state(state, admission_queue_depth, in_flight_persistence, self.snapshot.hydration_pending), ); self.snapshot.worker = worker; self.snapshot.admission_queue_depth = admission_queue_depth; @@ -467,8 +547,13 @@ impl crate::RawTransactionIngestSnapshotPublisher { } } -fn activity_for_state(state: ksp_worker_api::WorkerState, admission_queue_depth: usize, in_flight_persistence: usize) -> ksp_worker_api::WorkerActivity { - if admission_queue_depth > 0 || in_flight_persistence > 0 { +fn activity_for_state( + state: ksp_worker_api::WorkerState, + admission_queue_depth: usize, + in_flight_persistence: usize, + hydration_pending: usize, +) -> ksp_worker_api::WorkerActivity { + if admission_queue_depth > 0 || in_flight_persistence > 0 || hydration_pending > 0 { return ksp_worker_api::WorkerActivity::Active; } return match state { diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs index 9ad20c2..c1ca276 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs -// version: 13 +// version: 14 //! Dependency firewall canaries for the RAW transaction ingest Worker foundation. @@ -8,19 +8,7 @@ fn pre_002_manifest_dependency_surface_is_exact() { let manifest = include_str!("../Cargo.toml"); let dependencies = dependency_section(manifest); let names = manifest_dependency_names(dependencies); - assert_eq!( - names, - vec![ - "ksp-core-lib", - "ksp-logging-lib", - "ksp-onchain-transport-lib", - "ksp-raw-transaction-lib", - "ksp-store-lib", - "ksp-worker-api", - "sha2", - "tokio", - ], - ); + assert_eq!(names, vec!["ksp-core-lib", "ksp-logging-lib", "ksp-onchain-transport-lib", "ksp-raw-transaction-lib", "ksp-store-lib", "ksp-worker-api", "sha2", "tokio",],); for required in [ "ksp-core-lib = { path = \"../ksp-core-lib\" }", "ksp-logging-lib = { path = \"../ksp-logging-lib\" }", @@ -93,14 +81,15 @@ fn v0_3_12_pre_003_transaction_and_status_adapters_are_private_and_transport_fac ] { assert!(!resources.contains(forbidden) && !root.contains(forbidden), "pre.003 crossed a private/offline boundary: {forbidden}"); } - let status_impl = - match resources.split_once("impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate") { - std::option::Option::Some((_, tail)) => match tail.split_once("impl std::convert::From<") { - std::option::Option::Some((value, _)) => value, - std::option::Option::None => tail, - }, - std::option::Option::None => "", - }; + let status_impl = match resources + .split_once("impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate") + { + std::option::Option::Some((_, tail)) => match tail.split_once("impl std::convert::From<") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => tail, + }, + std::option::Option::None => "", + }; assert!(!status_impl.contains("::error(self)"), "TransactionStatus remote error material must not enter the private signal"); return; } @@ -175,7 +164,11 @@ fn v0_3_12_pre_005_block_and_continuity_adapters_remain_private_and_transport_fa fn v0_3_12_pre_006_productive_source_uses_transport_session_bounded_coalescence_and_existing_admission() { let runtime = include_str!("../src/runtime.rs"); let resources = include_str!("../src/runtime_resources.rs"); - for required in ["runtime_resources.into_yellowstone_source()", "children.spawn", "source.run(source_settings, stop_receiver, admission_sender)"] { + for required in [ + "runtime_resources.into_yellowstone_source()", + "children.spawn", + "source.run(source_settings, stop_receiver, admission_sender, processing_frontier_sender)", + ] { assert!(runtime.contains(required), "required pre.006 supervisor wiring missing: {required}"); } for required in [ @@ -306,3 +299,43 @@ fn manifest_dependency_names(section: &str) -> std::vec::Vec<&str> { names.sort_unstable(); return names; } + +#[test] +fn v0_3_12_pre_007_processing_frontier_is_run_local_latest_value_and_has_no_replay_or_backend_edge() { + let runtime = include_str!("../src/runtime.rs"); + let resources = include_str!("../src/runtime_resources.rs"); + let snapshot = include_str!("../src/snapshot.rs"); + for required in [ + "RawTransactionIngestProcessingFrontierProjection", + "processing_frontier_sender", + "wait_processing_frontier", + "record_processing_frontier", + "hydration_pending", + "processing_frontier_slot", + "oldest_pending_slot", + "RawTransactionIngestProcessingFrontier", + "settle_pending", + "observe_settled", + ] { + assert!( + runtime.contains(required) || resources.contains(required) || snapshot.contains(required), + "required pre.007 frontier contract missing: {required}" + ); + } + for forbidden in [ + "SubscribeReplayInfo", + "last_requested_from_slot", + "continuity_gap_count", + "ksp_job_backfill_lib::", + "ksp_store_postgres_lib::", + "reqwest::", + "tonic::", + "yellowstone_grpc_proto::", + ] { + assert!( + !runtime.contains(forbidden) && !resources.contains(forbidden) && !snapshot.contains(forbidden), + "pre.007 crossed replay/backend boundary: {forbidden}" + ); + } + return; +} 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 84d6158..fb881aa 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs -// version: 8 +// version: 9 //! External public, security, redaction and release-boundary hardening canaries for `pre.010`. @@ -121,16 +121,7 @@ fn v0_3_12_pre_002_manifest_dependency_surface_opens_only_transport_and_remains_ } assert_eq!( normal, - std::collections::BTreeSet::from([ - "ksp-core-lib", - "ksp-logging-lib", - "ksp-onchain-transport-lib", - "ksp-raw-transaction-lib", - "ksp-store-lib", - "ksp-worker-api", - "sha2", - "tokio", - ]) + std::collections::BTreeSet::from(["ksp-core-lib", "ksp-logging-lib", "ksp-onchain-transport-lib", "ksp-raw-transaction-lib", "ksp-store-lib", "ksp-worker-api", "sha2", "tokio",]) ); assert!(dev.is_empty()); assert!(build.is_empty()); @@ -332,7 +323,14 @@ fn v0_3_12_pre_004_hydration_provenance_and_remote_material_are_bounded_and_reda ] { assert!(resources.contains(required), "required pre.004 bounded provenance/mismatch guard missing: {required}"); } - for forbidden in ["source_payload_hash", "source_payload_size_bytes", "HTTP-SECRET-CANARY", "GRPC-SECRET-CANARY", "TransactionStatus.error", ".error()"] { + for forbidden in [ + "source_payload_hash", + "source_payload_size_bytes", + "HTTP-SECRET-CANARY", + "GRPC-SECRET-CANARY", + "TransactionStatus.error", + ".error()", + ] { assert!(!resources.contains(forbidden), "pre.004 retained forbidden remote/source material: {forbidden}"); } return; @@ -385,12 +383,11 @@ fn v0_3_12_pre_005_block_and_continuity_shapes_are_bounded_redacted_and_raw_sepa fn v0_3_12_pre_006_runtime_resource_contract_opens_one_supervised_transport_source() { let resources = include_str!("../src/runtime_resources.rs"); let runtime = include_str!("../src/runtime.rs"); - for required in ["open_standard_subscribe", "next_update", "get_transaction_observed", "RawTransactionIngestHydrationCoordinator", "session.close().await"] - { + for required in ["open_standard_subscribe", "next_update", "get_transaction_observed", "RawTransactionIngestHydrationCoordinator", "session.close().await"] { assert!(resources.contains(required), "productive runtime-resource source behavior missing: {required}"); } assert!(runtime.contains("start_with_runtime_resources")); - assert!(runtime.contains("source.run(source_settings, stop_receiver, admission_sender)")); + assert!(runtime.contains("source.run(source_settings, stop_receiver, admission_sender, processing_frontier_sender)")); for forbidden in ["get_block_observed", "ksp_config_lib::", "ksp_store_postgres_lib::", "reqwest::", "tonic::", "yellowstone_grpc_proto::"] { assert!(!resources.contains(forbidden) && !runtime.contains(forbidden), "pre.006 runtime source crossed a forbidden boundary: {forbidden}"); } @@ -455,3 +452,42 @@ fn v0_3_12_pre_002_public_root_exposes_contract_types_without_transport_implemen } return; } + +#[test] +fn v0_3_12_pre_007_processing_frontier_is_bounded_processing_only_and_redacted() { + let resources = include_str!("../src/runtime_resources.rs"); + let snapshot = include_str!("../src/snapshot.rs"); + for required in [ + "pending_total", + "std::collections::BTreeMap", + "settled_interval_highs", + "oldest_pending_slot", + "processing_frontier_slot", + "hydration_pending", + "processing_frontier.observe_pending", + "processing_frontier.settle_pending", + ] { + assert!(resources.contains(required) || snapshot.contains(required), "required pre.007 bounded frontier guard missing: {required}"); + } + for forbidden in [ + "signature: ksp_store_lib::RawTransactionSignature", + "filter_id", + "provider", + "endpoint_id", + "transaction:", + "meta:", + "ReplayInfo", + "continuity_gap", + "repair", + ] { + let projection = match snapshot.split_once("struct RawTransactionIngestProcessingFrontierProjection {") { + std::option::Option::Some((_, tail)) => match tail.split_once("impl crate::RawTransactionIngestProcessingFrontierProjection") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => "", + }, + std::option::Option::None => "", + }; + assert!(!projection.contains(forbidden), "frontier projection leaked forbidden material: {forbidden}"); + } + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs index 423891d..890306e 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs -// version: 9 +// version: 10 //! External public-surface proofs for the RAW transaction ingest Worker foundation. @@ -169,3 +169,18 @@ fn v0_3_12_pre_002_runtime_resource_types_are_consumable_without_client_escape_h } return; } + +#[test] +fn v0_3_12_pre_007_processing_frontier_snapshot_getters_are_public_and_processing_only() { + let _hydration_pending: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> usize = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::hydration_pending; + let _processing_frontier_slot: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> std::option::Option = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::processing_frontier_slot; + let _oldest_pending_slot: fn(&ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot) -> std::option::Option = + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::oldest_pending_slot; + let root = include_str!("../src/lib.rs"); + for forbidden in ["ReplayInfo", "from_slot", "repair", "backfill"] { + assert!(!root.contains(forbidden), "pre.007 public root overclaims continuity/replay: {forbidden}"); + } + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs index acbf950..b17ca14 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,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs -// version: 6 +// version: 7 //! Release-completeness canaries through the `pre.010` public/release/security hardening tranche. @@ -32,10 +32,7 @@ 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",] - ); + assert_eq!(names, std::vec!["admission.rs", "error.rs", "identity.rs", "lib.rs", "persistence.rs", "runtime.rs", "runtime_resources.rs", "settings.rs", "snapshot.rs",]); return std::result::Result::Ok(()); } @@ -106,6 +103,7 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { "v0_3_12_pre_002_production_sources_keep_transport_confined_to_runtime_resources", "v0_3_12_pre_006_runtime_resource_contract_opens_one_supervised_transport_source", "v0_3_12_pre_006_source_coalescence_is_bounded_stop_preemptible_and_redacted", + "v0_3_12_pre_007_processing_frontier_is_bounded_processing_only_and_redacted", "pre_010_lower_layers_have_no_dependency_return_to_concrete_worker", "v0_3_12_pre_002_public_root_exposes_contract_types_without_transport_implementation_paths", ] { @@ -119,11 +117,13 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { assert!(dependency_boundary.contains("v0_3_12_pre_003_transaction_and_status_adapters_are_private_and_transport_facade_only")); assert!(dependency_boundary.contains("v0_3_12_pre_004_hydration_contract_uses_only_transport_facade_common_raw_and_existing_ingress")); assert!(dependency_boundary.contains("v0_3_12_pre_005_block_and_continuity_adapters_remain_private_and_transport_facade_only")); + assert!(dependency_boundary.contains("v0_3_12_pre_007_processing_frontier_is_run_local_latest_value_and_has_no_replay_or_backend_edge")); let public_api = include_str!("public_api.rs"); assert!(public_api.contains("pre_003_kind_code_and_settings_are_consumable_from_crate_root")); assert!(public_api.contains("pre_004_start_handle_and_terminal_future_are_consumable_without_public_join_handle")); assert!(public_api.contains("pre_008_snapshot_surface_and_common_projection_are_public_and_stable")); assert!(public_api.contains("pre_009_source_and_drain_timeout_error_codes_are_public_and_stable")); assert!(public_api.contains("v0_3_12_pre_002_runtime_resource_types_are_consumable_without_client_escape_hatch")); + assert!(public_api.contains("v0_3_12_pre_007_processing_frontier_snapshot_getters_are_public_and_processing_only")); return; } diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs index 89a3951..2b540e8 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs -// version: 6 +// version: 7 fn grpc_endpoint(cluster: &str) -> std::option::Option { return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider"); @@ -62,9 +62,7 @@ fn http_pool(cluster: &str, role_name: &str, request_kind: &str) -> std::option: }; } -fn transaction_request( - commitment: std::option::Option, -) -> std::option::Option { +fn transaction_request(commitment: std::option::Option) -> std::option::Option { let name = match ksp_onchain_transport_lib::YellowstoneSubscribeFilterName::new("tx-fixture") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::option::Option::None, @@ -136,7 +134,10 @@ impl super::RawTransactionIngestYellowstoneBlockView for BlockViewFixture { return self.transactions.len(); } - fn transaction_identity(&self, position: usize) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)> { + fn transaction_identity( + &self, + position: usize, + ) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)> { return match self.transactions.get(position) { std::option::Option::Some(value) => std::result::Result::Ok(*value), std::option::Option::None => std::result::Result::Err(crate::runtime_error("fixture.block_transaction_missing")), @@ -218,7 +219,12 @@ const PRE_004_OTHER_TRANSACTION_BASE64: &str = "AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEB const PRE_004_ZERO_SIGNATURE_TEXT: &str = "1111111111111111111111111111111111111111111111111111111111111111"; const PRE_004_ZERO_TRANSACTION_BASE64: &str = "AQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"; -fn http_pool_for_url(url: &str, cluster: &str, endpoint_name: &str, provider: &str) -> std::option::Option { +fn http_pool_for_url( + url: &str, + cluster: &str, + endpoint_name: &str, + provider: &str, +) -> std::option::Option { let url = match ksp_onchain_transport_lib::HttpEndpointUrl::parse(url) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::option::Option::None, @@ -343,8 +349,11 @@ fn serve_http_once(body: std::string::String) -> std::io::Result<(std::string::S std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let response = - std::format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), body,); + let response = std::format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body, + ); if let std::result::Result::Err(error) = std::io::Write::write_all(&mut stream, response.as_bytes()) { return std::result::Result::Err(error); } @@ -432,7 +441,10 @@ async fn pre_002_source_accepts_matching_confirmed_transaction_and_get_transacti #[tokio::test(flavor = "current_thread")] async fn pre_002_source_rejects_processed_and_implicit_commitment() { - for commitment in [std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Processed), std::option::Option::None] { + for commitment in [ + std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Processed), + std::option::Option::None, + ] { let endpoint = match grpc_endpoint("devnet") { std::option::Option::Some(value) => value, std::option::Option::None => return, @@ -625,8 +637,8 @@ async fn pre_003_transaction_signal_fixture_preserves_exact_source_neutral_ident assert_eq!( signal.matched_filter_fingerprint, [ - 0x20, 0x83, 0x1d, 0xc8, 0x37, 0x8b, 0x60, 0x70, 0xf9, 0x1d, 0x9d, 0xe6, 0xd3, 0xa1, 0xd2, 0xb3, 0xd7, 0x72, 0x98, 0xe3, 0xad, 0xa7, 0x6c, 0x01, - 0xfa, 0xa3, 0x08, 0x26, 0xff, 0xab, 0xcd, 0x84, + 0x20, 0x83, 0x1d, 0xc8, 0x37, 0x8b, 0x60, 0x70, 0xf9, 0x1d, 0x9d, 0xe6, 0xd3, 0xa1, 0xd2, 0xb3, 0xd7, 0x72, 0x98, + 0xe3, 0xad, 0xa7, 0x6c, 0x01, 0xfa, 0xa3, 0x08, 0x26, 0xff, 0xab, 0xcd, 0x84, ] ); assert_eq!(signal.created_at, std::option::Option::Some(super::RawTransactionIngestSourceTimestamp { nanos: 456_000_000, seconds: 1_760_000_123 })); @@ -871,7 +883,10 @@ fn pre_005_slot_status_projection_is_exact_and_source_neutral() { (ksp_onchain_transport_lib::YellowstoneSlotStatus::Processed, super::RawTransactionIngestContinuityStatus::Processed), (ksp_onchain_transport_lib::YellowstoneSlotStatus::Confirmed, super::RawTransactionIngestContinuityStatus::Confirmed), (ksp_onchain_transport_lib::YellowstoneSlotStatus::Finalized, super::RawTransactionIngestContinuityStatus::Finalized), - (ksp_onchain_transport_lib::YellowstoneSlotStatus::FirstShredReceived, super::RawTransactionIngestContinuityStatus::FirstShredReceived), + ( + ksp_onchain_transport_lib::YellowstoneSlotStatus::FirstShredReceived, + super::RawTransactionIngestContinuityStatus::FirstShredReceived, + ), (ksp_onchain_transport_lib::YellowstoneSlotStatus::Completed, super::RawTransactionIngestContinuityStatus::Completed), (ksp_onchain_transport_lib::YellowstoneSlotStatus::CreatedBank, super::RawTransactionIngestContinuityStatus::CreatedBank), (ksp_onchain_transport_lib::YellowstoneSlotStatus::Dead, super::RawTransactionIngestContinuityStatus::Dead), @@ -973,7 +988,10 @@ async fn pre_004_observed_get_transaction_closes_signal_to_common_raw_ingress_wi assert_eq!(acquisition.transaction().reference().network().as_str(), "devnet"); assert_eq!(acquisition.transaction().reference().signature().as_bytes(), &[0_u8; 64]); assert_eq!(acquisition.transaction().slot(), 42); - assert_eq!(acquisition.transaction().block_time().map(|value| return value.unix_millis()), std::option::Option::Some(1_760_000_120_000),); + assert_eq!( + acquisition.transaction().block_time().map(|value| return value.unix_millis()), + std::option::Option::Some(1_760_000_120_000), + ); assert_eq!(acquisition.observation().provenance().provider().as_str(), "ys.fixture-provider:http.fixture-http-provider"); return; } @@ -1211,8 +1229,14 @@ async fn pre_004_status_provenance_and_future_source_timestamp_are_bounded_and_r std::option::Option::Some(value) => value, std::option::Option::None => return, }; - let mut signal = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::TransactionStatus, 42, 7, &["filter-b", "filter-a", "filter-a"], 0) - { + let mut signal = match pre_004_signal( + &source, + super::RawTransactionIngestSourceFamily::TransactionStatus, + 42, + 7, + &["filter-b", "filter-a", "filter-a"], + 0, + ) { std::option::Option::Some(value) => value, std::option::Option::None => return, }; @@ -1321,3 +1345,50 @@ fn pre_006_coalescence_key_separates_network_signature_and_commitment() { assert_eq!(first_key.network.as_str(), "devnet"); return; } + +#[test] +fn pre_007_processing_frontier_never_advances_through_pending_work() { + let mut frontier = super::RawTransactionIngestProcessingFrontier::new(); + assert!(frontier.observe_settled(100).is_ok()); + let projection = frontier.projection(); + assert_eq!(projection.hydration_pending(), 0); + assert_eq!(projection.processing_frontier_slot(), std::option::Option::Some(100)); + assert_eq!(projection.oldest_pending_slot(), std::option::Option::None); + assert!(frontier.observe_pending(102).is_ok()); + assert!(frontier.observe_settled(105).is_ok()); + assert!(frontier.observe_pending(104).is_ok()); + let projection = frontier.projection(); + assert_eq!(projection.hydration_pending(), 2); + assert_eq!(projection.processing_frontier_slot(), std::option::Option::Some(100)); + assert_eq!(projection.oldest_pending_slot(), std::option::Option::Some(102)); + assert!(frontier.settle_pending(102).is_ok()); + let projection = frontier.projection(); + assert_eq!(projection.hydration_pending(), 1); + assert_eq!(projection.processing_frontier_slot(), std::option::Option::Some(102)); + assert_eq!(projection.oldest_pending_slot(), std::option::Option::Some(104)); + assert!(frontier.settle_pending(104).is_ok()); + let projection = frontier.projection(); + assert_eq!(projection.hydration_pending(), 0); + assert_eq!(projection.processing_frontier_slot(), std::option::Option::Some(105)); + assert_eq!(projection.oldest_pending_slot(), std::option::Option::None); + return; +} + +#[test] +fn pre_007_processing_frontier_compacts_settled_slots_to_pending_intervals() { + let mut frontier = super::RawTransactionIngestProcessingFrontier::new(); + assert!(frontier.observe_pending(10).is_ok()); + assert!(frontier.observe_pending(20).is_ok()); + for slot in 0_u64..=1000_u64 { + if slot == 10 || slot == 20 { + continue; + } + assert!(frontier.observe_settled(slot).is_ok()); + } + assert!(frontier.slot_state_count() <= 5); + let projection = frontier.projection(); + assert_eq!(projection.hydration_pending(), 2); + assert_eq!(projection.oldest_pending_slot(), std::option::Option::Some(10)); + assert_eq!(projection.processing_frontier_slot(), std::option::Option::Some(9)); + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs index 0bca54e..053a8f1 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs -// version: 2 +// version: 3 fn snapshot_foundation() -> std::option::Option<(crate::RawTransactionIngestSnapshotPublisher, crate::RawTransactionIngestSnapshotSource)> { let network = match ksp_store_lib::RawNetworkId::new("mainnet") { @@ -50,6 +50,9 @@ fn pre_008_initial_snapshot_and_common_projection_are_exact() { assert_eq!(concrete.store_failure_total(), 0); assert_eq!(concrete.source_failure_total(), 0); assert_eq!(concrete.backpressure_wait_total(), 0); + assert_eq!(concrete.hydration_pending(), 0); + assert_eq!(concrete.processing_frontier_slot(), std::option::Option::None); + assert_eq!(concrete.oldest_pending_slot(), std::option::Option::None); let common = ksp_worker_api::WorkerSnapshotSource::current(&source); assert_eq!(&common, concrete.worker_snapshot()); return; @@ -134,3 +137,30 @@ fn pre_009_source_failure_and_backpressure_counters_advance_without_changing_pro assert_eq!(&common, concrete.worker_snapshot()); return; } + +#[test] +fn pre_007_processing_frontier_projection_is_latest_value_and_activity_aware() { + let (mut publisher, source) = match snapshot_foundation() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let initial = source.current(); + assert_eq!(initial.hydration_pending(), 0); + assert_eq!(initial.processing_frontier_slot(), std::option::Option::None); + assert_eq!(initial.oldest_pending_slot(), std::option::Option::None); + let projection = crate::RawTransactionIngestProcessingFrontierProjection::new(2, std::option::Option::Some(40), std::option::Option::Some(41)); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, projection).is_ok()); + let pending = source.current(); + assert_eq!(pending.hydration_pending(), 2); + assert_eq!(pending.processing_frontier_slot(), std::option::Option::Some(40)); + assert_eq!(pending.oldest_pending_slot(), std::option::Option::Some(41)); + assert_eq!(pending.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Active); + let projection = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(45), std::option::Option::None); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, projection).is_ok()); + let settled = source.current(); + assert_eq!(settled.hydration_pending(), 0); + assert_eq!(settled.processing_frontier_slot(), std::option::Option::Some(45)); + assert_eq!(settled.oldest_pending_slot(), std::option::Option::None); + assert_eq!(settled.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Idle); + return; +} diff --git a/deltas/0.3.12/pre.007.md b/deltas/0.3.12/pre.007.md new file mode 100644 index 0000000..7a5b7f3 --- /dev/null +++ b/deltas/0.3.12/pre.007.md @@ -0,0 +1,216 @@ + + + +# Delta `0.3.12-pre.007` — processing frontier run-local bornée + +## Base requise + +```text +0.3.12-pre.006-fix.001 +workspace.package.version = 0.3.12-pre.6.fix.1 +``` + +Le gate opérateur communiqué pour cette base est entièrement vert : audits Rust/Markdown, `cargo check --workspace`, Clippy strict, 58 tests Worker et toutes les suites d'intégration associées. + +## Objectif + +Ajouter la processing frontier run-local prévue au plan sans anticiper reconnect/replay/repair : + +```text +source work observed +pending / settled par slot +oldest pending slot +highest unblocked actually-observed settled slot +latest-value snapshot projection +``` + +La frontier reste strictement processing-only. + +## Version + +```text +workspace.package.version = 0.3.12-pre.7 +``` + +## Définition pending / settled + +Un signal transactionnel devient pending seulement après validation de sa clé d'hydration et insertion dans le coordinator borné. + +Il devient settled lorsque : + +```text +hydration HTTP -> Missing +ou +hydration HTTP -> Available puis ingress accepté par la queue centrale +``` + +`BlockMeta` et `Slot` sont continuity-only et deviennent settled dès leur projection locale, sans créer de RAW. + +Un signal non admis à cause d'un stop ou d'un fault reste pending dans la dernière projection de run. `settled` ne signifie jamais Store persisted/durable. + +## Frontier + +Le tracker privé maintient : + +```text +pending_total +pending par slot +settled par slot retenu +oldest_pending_slot +processing_frontier_slot +``` + +La frontier est la plus haute slot réellement observée/settled située avant la plus ancienne slot encore pending. Sans pending, elle devient la plus haute slot settled effectivement observée. + +Ainsi, une slot pending empêche explicitement toute progression à travers elle. + +## Borne mémoire + +Les slots pending sont conservées exactement. Les slots settled-only sont compactées par intervalles autour des slots pending : un seul maximum settled est conservé par intervalle. + +La représentation est donc bornée linéairement par le nombre de slots pending distinctes, lui-même borné par le coordinator d'hydration. Un flux continuity-only sans pending se compacte à une seule candidate. + +Aucun historique de slots non borné n'est conservé. + +## Projection snapshot + +Le source task émet une projection privée latest-value par `tokio::sync::watch` : + +```text +hydration_pending +processing_frontier_slot +oldest_pending_slot +``` + +Le supervisor reste l'unique owner/mutateur du snapshot concret et applique ces valeurs via `RawTransactionIngestSnapshotPublisher`. + +Les trois getters deviennent publics sur `RawTransactionIngestSnapshot`. Aucun nouveau type public ni module public n'est ajouté. + +`WorkerActivity` considère désormais `hydration_pending > 0` comme activité réelle même si la queue centrale et la persistence sont momentanément vides. + +## Sécurité / redaction + +La projection ne transporte jamais : + +```text +signature +filter id/value +provider +endpoint +transaction/meta +URL/header/credential +remote error text +``` + +Les erreurs de compteur/frontier utilisent uniquement des codes/contextes internes sûrs. + +## Canaris ajoutés/ajustés + +```text +frontier ne traverse jamais une slot pending +oldest pending avance après settlement +frontier rejoint la plus haute slot settled lorsque tout pending est résolu +compaction bornée des settled-only +snapshot initial frontier vide +snapshot latest-value des trois getters +WorkerActivity active sur hydration pending +public getters exacts +source.run inclut le publisher frontier privé +aucun ReplayInfo/from_slot/continuity_gap/repair +aucun backend/direct protocol edge +release completeness inclut les nouveaux canaris +``` + +## Décisions prises + +```text +frontier = source-processing, pas Store durability +Missing = settled source-processing +Available = settled après send admission réussi +continuity-only = settled immédiat local +projection supervisor via watch latest-value +pas de checkpoint persistant +pas de claim monotone de blockchain completeness +``` + +## Questions ouvertes + +Aucune question ne bloque `pre.007`. + +Restent réservés : + +```text +reconnect / from_slot / ReplayInfo / retention gap : pre.008 +races/retry/backpressure final : pre.009 +cross-layer completeness/security : pre.010 +live opt-in : pre.011 +``` + +## Fichiers ajoutés + +```text +deltas/0.3.12/pre.007.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.rs +crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs +crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs +docs/validation/029-V0_3_12_YELLOWSTONE_HYDRATION_CONTINUITY.md +``` + +## Fichiers supprimés + +Aucun. + +## Validations exécutées dans l'environnement d'assemblage + +```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 +``` + +Des scanners statiques ciblés vérifient également la borne du tracker, le blocage par oldest pending, les getters publics, le confinement Transport et l'absence de responsabilités `pre.008`. + +## Validations non exécutées dans l'environnement d'assemblage + +`cargo`, `rustc` et `rustfmt` ne sont pas disponibles dans l'environnement d'assemblage. Aucun gate Cargo local n'est revendiqué. + +## Gate opérateur requis + +```bash +cargo fmt --all +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas +cargo check --workspace +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test -p ksp-worker-raw-transaction-ingest-lib +cargo tree -p ksp-worker-raw-transaction-ingest-lib --edges normal +cargo tree -p ksp-worker-raw-transaction-ingest-lib -e features +cargo tree --duplicates +``` + +## Non-claims + +```text +pas de reconnect policy Worker +pas de from_slot Worker +pas d'interprétation ReplayInfo +pas de preuve replay +pas de continuity gap fault +pas de repair/backfill implicite +pas de checkpoint inter-process +pas de getBlock +pas de nouveau backend/Config edge +pas de smoke live revendiqué +``` diff --git a/docs/validation/029-V0_3_12_YELLOWSTONE_HYDRATION_CONTINUITY.md b/docs/validation/029-V0_3_12_YELLOWSTONE_HYDRATION_CONTINUITY.md index 52f684a..a8f70cc 100644 --- a/docs/validation/029-V0_3_12_YELLOWSTONE_HYDRATION_CONTINUITY.md +++ b/docs/validation/029-V0_3_12_YELLOWSTONE_HYDRATION_CONTINUITY.md @@ -1,5 +1,5 @@ - + # Validation v0.3.12 — Yellowstone + hydration HTTP + continuité de run du Worker RawTransaction @@ -1339,3 +1339,110 @@ cargo clippy --workspace --all-targets --all-features -- -D warnings cargo test -p ksp-worker-raw-transaction-ingest-lib ``` +## 57. Gate opérateur `pre.006-fix.001` + +Le gate opérateur communiqué pour `0.3.12-pre.6.fix.1` est entièrement vert : + +```text +General Rust rule audit : clean +Rust export completeness audit : 0 candidate(s) +KSP workspace Rust rule audit : clean +Markdown table audit : clean (340 tables, 811 fichiers) +cargo check --workspace : PASS +cargo clippy --workspace --all-targets --all-features -- -D warnings : PASS +cargo test -p ksp-worker-raw-transaction-ingest-lib : 58 PASS / 0 FAIL +``` + +Les suites `dependency_boundary`, `hardening`, `public_api`, `release_completeness` et les doc-tests passent également. `pre.006-fix.001` devient donc la base autoritaire de `pre.007`. + +## 58. Contrat processing frontier `pre.007` + +`pre.007` introduit une frontier strictement run-local liée uniquement au travail réellement observé par le source task : + +```text +source signal Transaction/Status/Block validé -> pending +getTransaction -> Missing -> settled +getTransaction -> Available + ingress accepté par admission -> settled +BlockMeta/Slot continuity-only observé -> settled immédiat +``` + +`settled` ne signifie ni persisté, ni durable, ni complet sur la blockchain. L'admission queue depth et `in_flight_persistence` restent les métriques séparées qui décrivent les étages aval. + +La projection publique ajoute uniquement : + +```text +hydration_pending +processing_frontier_slot +oldest_pending_slot +``` + +`processing_frontier_slot` est la plus haute slot réellement observée et settled qui n'est pas située au-delà de la plus ancienne slot pending. Une slot pending bloque donc toujours l'avancement de la frontier. + +## 59. Borne mémoire et compaction + +Le tracker privé conserve les slots pending exactes et compacte les slots settled-only par intervalles autour des slots pending. Il ne conserve qu'un maximum settled par intervalle, plus les slots qui possèdent encore du pending. + +Avec `N` slots pending distinctes, la représentation reste donc bornée linéairement par le même travail déjà borné par le coordinator d'hydration ; un flux continuity-only sans pending se compacte à une seule slot settled candidate. + +Aucune map de slots historique/non bornée n'est créée. + +## 60. Projection supervisor et activité + +Le source task publie sa projection par un `tokio::sync::watch` latest-value privé. Le supervisor est l'unique owner du `RawTransactionIngestSnapshotPublisher` et applique la projection dans le snapshot concret existant. + +La projection peut coalescer des valeurs intermédiaires ; elle ne perd pas de compteur durable car elle ne transporte que des gauges/latest-values calculées à partir de l'état source courant. + +`WorkerActivity` devient `Active` lorsque `hydration_pending > 0`, même si l'admission queue et les persistences sont momentanément à zéro. + +Aucune signature, filter, provider, endpoint, transaction/meta ou erreur distante n'entre dans la projection. + +## 61. Canaris `pre.007` + +Les canaris déterministes ajoutés vérifient : + +```text +frontier initiale vide +slot settled 100 -> frontier 100 +pending 102 + settled 105 -> frontier reste 100 +pending 104 puis settle 102 -> frontier 102, oldest pending 104 +settle 104 -> frontier 105 +compaction de 0..1000 avec pending 10/20 -> <= 5 états de slot +snapshot latest-value hydration_pending/frontier/oldest_pending +WorkerActivity active quand hydration_pending > 0 +getters publics exacts +aucun ReplayInfo/from_slot/continuity_gap/repair +aucun backend/Config/Backfill/direct reqwest/tonic/proto +``` + +## 62. Non-claims `pre.007` + +`pre.007` ne prétend pas : + +```text +interpréter reconnect Transport +modifier ou décider from_slot +interpréter SubscribeReplayInfo +prouver un replay réussi +prouver une continuité blockchain +maintenir un checkpoint inter-process +fault sur gap de rétention +lancer un repair/backfill +``` + +Ces responsabilités restent `pre.008` et les releases de repair ultérieures. + +## 63. Gate opérateur requis pour `pre.007` + +```bash +cargo fmt --all +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas +cargo check --workspace +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test -p ksp-worker-raw-transaction-ingest-lib +cargo tree -p ksp-worker-raw-transaction-ingest-lib --edges normal +cargo tree -p ksp-worker-raw-transaction-ingest-lib -e features +cargo tree --duplicates +``` + +Dans l'environnement d'assemblage, `cargo`, `rustc` et `rustfmt` ne sont pas installés ; aucun gate Cargo local n'est revendiqué.