From 2aa64cf0a474e7cd01f1b8c551ca34f1833d4361 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Wed, 9 Sep 2026 20:42:32 +0200 Subject: [PATCH] v0.3.12-pre.008 --- Cargo.toml | 2 +- .../src/grpc_stream.rs | 45 +++- crates/ksp-onchain-transport-lib/src/lib.rs | 4 +- .../tests/public_api.rs | 6 +- .../tests/release_completeness.rs | 5 +- .../unit_tests/grpc_stream.rs | 44 +++- .../src/lib.rs | 9 +- .../src/runtime_resources.rs | 119 +++++++++- .../src/snapshot.rs | 125 ++++++++++- .../tests/dependency_boundary.rs | 51 +++-- .../tests/hardening.rs | 46 +++- .../tests/public_api.rs | 14 +- .../tests/release_completeness.rs | 8 +- .../unit_tests/runtime_resources.rs | 44 +++- .../unit_tests/snapshot.rs | 37 +++- deltas/0.3.12/pre.008.md | 205 ++++++++++++++++++ ...0_3_12_YELLOWSTONE_HYDRATION_CONTINUITY.md | 162 +++++++++++++- 17 files changed, 879 insertions(+), 47 deletions(-) create mode 100644 deltas/0.3.12/pre.008.md diff --git a/Cargo.toml b/Cargo.toml index c74f311..1c605c3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,7 +6,7 @@ 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.7.fix.2" +version = "0.3.12-pre.8" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-onchain-transport-lib/src/grpc_stream.rs b/crates/ksp-onchain-transport-lib/src/grpc_stream.rs index b74d55a..683eab3 100644 --- a/crates/ksp-onchain-transport-lib/src/grpc_stream.rs +++ b/crates/ksp-onchain-transport-lib/src/grpc_stream.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/src/grpc_stream.rs -// version: 3 +// version: 4 use tonic_prost::prost::Message; // rust-rules: trait-import @@ -39,6 +39,43 @@ pub struct YellowstoneGrpcSubscribeSnapshot { terminal_error_code: std::option::Option, } +/// Cloneable latest-value observer for one standard Yellowstone Subscribe session snapshot. +/// +/// The observer exposes only the already-safe transport snapshot and keeps the internal Tokio watch channel private. Cloning the observer does not duplicate +/// the gRPC stream, request state or reconnect actor. +#[derive(Clone)] +pub struct YellowstoneGrpcSubscribeSnapshotSource { + receiver: tokio::sync::watch::Receiver, +} + +impl crate::YellowstoneGrpcSubscribeSnapshotSource { + fn new(receiver: tokio::sync::watch::Receiver) -> Self { + return Self { receiver }; + } + + /// Returns the current safe reconnect/replay snapshot without waiting for another actor transition. + #[must_use] + pub fn current(&self) -> crate::YellowstoneGrpcSubscribeSnapshot { + return *self.receiver.borrow(); + } + + /// Waits for one newer safe reconnect/replay snapshot. + /// + /// `None` means the owning Subscribe actor dropped the latest-value publisher and no further snapshot can arrive. + pub async fn wait_for_change(&mut self) -> std::option::Option { + return match self.receiver.changed().await { + std::result::Result::Ok(()) => std::option::Option::Some(*self.receiver.borrow_and_update()), + std::result::Result::Err(_) => std::option::Option::None, + }; + } +} + +impl std::fmt::Debug for crate::YellowstoneGrpcSubscribeSnapshotSource { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + return formatter.debug_struct("YellowstoneGrpcSubscribeSnapshotSource").field("current", &self.current()).finish(); + } +} + impl YellowstoneGrpcSubscribeSnapshot { const fn new(initial_from_slot: std::option::Option) -> Self { return Self { @@ -154,6 +191,12 @@ impl SolanaYellowstoneGrpcSubscribeSession { return *self.snapshot_rx.borrow(); } + /// Returns one cloneable latest-value observer for reconnect/replay snapshot transitions. + #[must_use] + pub fn snapshot_source(&self) -> crate::YellowstoneGrpcSubscribeSnapshotSource { + return crate::YellowstoneGrpcSubscribeSnapshotSource::new(self.snapshot_rx.clone()); + } + /// Queues one complete standard Yellowstone request mutation without waiting for network dispatch. /// /// The mutation is rejected synchronously when local validation fails, the encoded request exceeds the configured outbound bound, the bounded request diff --git a/crates/ksp-onchain-transport-lib/src/lib.rs b/crates/ksp-onchain-transport-lib/src/lib.rs index 3579ee0..f1a3be0 100644 --- a/crates/ksp-onchain-transport-lib/src/lib.rs +++ b/crates/ksp-onchain-transport-lib/src/lib.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/src/lib.rs -// version: 45 +// version: 46 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -143,6 +143,8 @@ pub use self::grpc_settings::YellowstoneGrpcTransportSettings; pub use self::grpc_stream::SolanaYellowstoneGrpcSubscribeSession; /// Safe Yellowstone reconnect/replay continuity snapshot. pub use self::grpc_stream::YellowstoneGrpcSubscribeSnapshot; +/// Cloneable latest-value observer for one standard Yellowstone Subscribe session snapshot. +pub use self::grpc_stream::YellowstoneGrpcSubscribeSnapshotSource; /// Safe Yellowstone bidirectional Subscribe lifecycle state. pub use self::grpc_stream::YellowstoneGrpcSubscribeState; /// One validated standard Yellowstone account predicate. diff --git a/crates/ksp-onchain-transport-lib/tests/public_api.rs b/crates/ksp-onchain-transport-lib/tests/public_api.rs index aa721fe..d22e968 100644 --- a/crates/ksp-onchain-transport-lib/tests/public_api.rs +++ b/crates/ksp-onchain-transport-lib/tests/public_api.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/tests/public_api.rs -// version: 50 +// version: 51 //! Integration tests for the public `ksp-onchain-transport-lib` consumer contract. @@ -1013,6 +1013,10 @@ fn public_v0_2_9_pre_010_yellowstone_reconnect_snapshot_is_available_from_crate_ let _snapshot = std::any::type_name::(); let _state = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Reconnecting; let _session_snapshot = ksp_onchain_transport_lib::SolanaYellowstoneGrpcSubscribeSession::snapshot; + let _snapshot_source = ksp_onchain_transport_lib::SolanaYellowstoneGrpcSubscribeSession::snapshot_source; + let _source_type = std::any::type_name::(); + let _source_current = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshotSource::current; + let _source_wait = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshotSource::wait_for_change; let _reconnect_count = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshot::reconnect_count; let _gap_count = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshot::continuity_gap_count; let _duplicate_count = ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshot::duplicate_update_count; diff --git a/crates/ksp-onchain-transport-lib/tests/release_completeness.rs b/crates/ksp-onchain-transport-lib/tests/release_completeness.rs index 18e53c6..e675ddf 100644 --- a/crates/ksp-onchain-transport-lib/tests/release_completeness.rs +++ b/crates/ksp-onchain-transport-lib/tests/release_completeness.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/tests/release_completeness.rs -// version: 42 +// version: 43 //! Release-level completeness canaries for staged HTTP and WebSocket Transport coverage. @@ -1307,6 +1307,9 @@ fn release_v0_2_9_pre_010_adds_bounded_reconnect_replay_and_conservative_continu for required in [ "YellowstoneGrpcSubscribeState::Reconnecting", "YellowstoneGrpcSubscribeSnapshot", + "YellowstoneGrpcSubscribeSnapshotSource", + "snapshot_source", + "wait_for_change", "reconnect_subscribe_stream", "replay_first_available", "SubscribeReplayInfo", diff --git a/crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs b/crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs index 3ee937c..03ce308 100644 --- a/crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs +++ b/crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs @@ -1,5 +1,5 @@ // file: crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs -// version: 4 +// version: 5 #[derive(Clone, Copy)] enum FixtureMode { @@ -159,6 +159,9 @@ impl yellowstone_grpc_proto::geyser::geyser_server::Geyser for FixtureGeyser { }, } }); + if matches!(mode, FixtureMode::ReconnectReplay) && subscribe_call == 2 { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } return std::result::Result::Ok(tonic::Response::new(super::MpscStream::new(outbound_rx))); } @@ -662,6 +665,45 @@ async fn yellowstone_reconnect_budget_exhaustion_is_terminal_and_safe() { server.stop().await; } +#[tokio::test(flavor = "current_thread")] +async fn yellowstone_snapshot_source_observes_replay_attempt_before_successful_reconnect() { + let server = FixtureServer::start(FixtureMode::ReconnectReplay).await; + let defaults = crate::YellowstoneGrpcSessionSettings::default(); + let settings = fixture_settings_with_reconnect( + server.endpoint_url.as_str(), + 8, + 8, + defaults.max_inbound_message_size_bytes(), + defaults.max_outbound_message_size_bytes(), + crate::YellowstoneGrpcReconnectSettings::new(3, std::time::Duration::from_millis(5), std::time::Duration::from_millis(20)), + ); + let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); + let mut session = channel.open_standard_subscribe(initial_request()).await.expect("fixture Subscribe stream must open"); + let mut snapshots = session.snapshot_source(); + assert_eq!(snapshots.current().state(), crate::YellowstoneGrpcSubscribeState::Active); + let _ = session.next_update().await.expect("first slot must decode").expect("first slot must be present"); + let replaying = tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + let snapshot = match snapshots.wait_for_change().await { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + if snapshot.replay_attempt_count() > 0 && snapshot.reconnect_count() == 0 { + return std::option::Option::Some(snapshot); + } + } + }) + .await + .expect("snapshot source must observe reconnect progress") + .expect("snapshot source must stay open during reconnect"); + assert_eq!(replaying.state(), crate::YellowstoneGrpcSubscribeState::Reconnecting); + assert_eq!(replaying.replay_attempt_count(), 1); + assert_eq!(replaying.reconnect_count(), 0); + let _ = session.next_update().await.expect("replayed duplicate must decode").expect("replayed duplicate must be present"); + session.close().await.expect("reconnected stream must close cleanly"); + server.stop().await; +} + #[tokio::test(flavor = "current_thread")] async fn yellowstone_shutdown_interrupts_reconnect_backoff_and_mutation_is_rejected_during_reconnect() { let server = FixtureServer::start(FixtureMode::ReconnectReplay).await; 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 1868d71..8dd2f4e 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: 14 +// version: 15 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -14,8 +14,9 @@ //! 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. A bounded run-local processing frontier now projects hydration pending, -//! oldest pending slot and highest unblocked actually observed slot; reconnect/replay interpretation remains deferred. +//! under bounded in-flight and pending budgets. A bounded run-local processing frontier projects hydration pending, +//! oldest pending slot and highest unblocked actually observed slot. The productive source now also projects safe Transport reconnect/replay state and faults +//! conservatively when Transport proves a replay-retention continuity gap; historical repair remains outside this crate. mod admission; mod error; @@ -78,6 +79,8 @@ pub use self::snapshot::RawTransactionIngestSnapshot; pub use self::snapshot::RawTransactionIngestSnapshotFuture; /// Cloneable latest-value source exposing concrete and common Worker snapshots from one shared watch state. pub use self::snapshot::RawTransactionIngestSnapshotSource; +/// Source-neutral lifecycle state of the productive RAW transaction ingest source. +pub use self::snapshot::RawTransactionIngestSourceState; /// Receiver-side owner of the private bounded RAW transaction admission queue. pub(crate) use self::admission::RawTransactionAdmission; 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 86cdf78..6f57a02 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: 10 +// version: 11 use sha2::Digest; // rust-rules: trait-import @@ -382,6 +382,13 @@ impl crate::RawTransactionIngestYellowstoneSource { }; let mut coordinator = RawTransactionIngestHydrationCoordinator::new(&settings); let mut processing_frontier = RawTransactionIngestProcessingFrontierReporter::new(processing_frontier_sender); + let snapshot_source = session.snapshot_source(); + if let std::result::Result::Err(error) = processing_frontier.observe_session_snapshot(snapshot_source.current()) { + processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed); + let _closed = session.close().await; + return std::result::Result::Err(error); + } + let mut session_snapshot_source = std::option::Option::Some(snapshot_source); let mut fault = std::option::Option::None; loop { if *stop_receiver.borrow() { @@ -401,6 +408,19 @@ impl crate::RawTransactionIngestYellowstoneSource { _ = stop_receiver.changed() => { break; } + source_snapshot = wait_yellowstone_session_snapshot(&mut session_snapshot_source) => { + match source_snapshot { + std::option::Option::Some(snapshot) => { + if let std::result::Result::Err(error) = processing_frontier.observe_session_snapshot(snapshot) { + fault = std::option::Option::Some(error); + break; + } + }, + std::option::Option::None => { + session_snapshot_source = std::option::Option::None; + }, + } + } joined = coordinator.tasks.join_next(), if !coordinator.tasks.is_empty() => { let joined = match joined { std::option::Option::Some(value) => value, @@ -453,13 +473,21 @@ impl crate::RawTransactionIngestYellowstoneSource { } } coordinator.abort_all().await; + processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closing); let closed = session.close().await; if let std::option::Option::Some(error) = fault { + processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed); return std::result::Result::Err(error); } return match closed { - std::result::Result::Ok(()) => std::result::Result::Ok(()), - std::result::Result::Err(error) => std::result::Result::Err(source_transport_error(error.code())), + std::result::Result::Ok(()) => { + processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Closed); + std::result::Result::Ok(()) + }, + std::result::Result::Err(error) => { + processing_frontier.set_source_state(crate::RawTransactionIngestSourceState::Failed); + std::result::Result::Err(source_transport_error(error.code())) + }, }; } } @@ -774,12 +802,33 @@ fn current_raw_timestamp() -> ksp_core_lib::Result }; } +async fn wait_yellowstone_session_snapshot( + source: &mut std::option::Option, +) -> std::option::Option { + return match source { + std::option::Option::Some(value) => value.wait_for_change().await, + std::option::Option::None => { + return std::future::pending::>().await; + }, + }; +} + fn source_transport_error(code: ksp_core_lib::ErrorCode) -> ksp_core_lib::Error { return ksp_core_lib::Error::new(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED, "RAW transaction ingest Yellowstone source transport failed") .with_context("transport_domain", code.domain()) .with_context("transport_code", code.code()); } +fn map_yellowstone_source_state(state: ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState) -> crate::RawTransactionIngestSourceState { + return match state { + ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Active => crate::RawTransactionIngestSourceState::Active, + ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Reconnecting => crate::RawTransactionIngestSourceState::Reconnecting, + ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Closing => crate::RawTransactionIngestSourceState::Closing, + ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Closed => crate::RawTransactionIngestSourceState::Closed, + ksp_onchain_transport_lib::YellowstoneGrpcSubscribeState::Failed => crate::RawTransactionIngestSourceState::Failed, + }; +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] struct RawTransactionIngestProcessingSlotState { pending: usize, @@ -915,11 +964,22 @@ impl RawTransactionIngestProcessingFrontier { struct RawTransactionIngestProcessingFrontierReporter { frontier: RawTransactionIngestProcessingFrontier, sender: tokio::sync::watch::Sender, + source_state: std::option::Option, + source_reconnect_total: u64, + source_replay_attempt_total: u64, + source_continuity_gap_total: u64, } impl RawTransactionIngestProcessingFrontierReporter { fn new(sender: tokio::sync::watch::Sender) -> Self { - return Self { frontier: RawTransactionIngestProcessingFrontier::new(), sender }; + return Self { + frontier: RawTransactionIngestProcessingFrontier::new(), + sender, + source_state: std::option::Option::None, + source_reconnect_total: 0, + source_replay_attempt_total: 0, + source_continuity_gap_total: 0, + }; } fn observe_pending(&mut self, slot: u64) -> ksp_core_lib::Result<()> { @@ -946,8 +1006,57 @@ impl RawTransactionIngestProcessingFrontierReporter { return std::result::Result::Ok(()); } + fn observe_session_snapshot(&mut self, snapshot: ksp_onchain_transport_lib::YellowstoneGrpcSubscribeSnapshot) -> ksp_core_lib::Result<()> { + return self.observe_source_continuity( + map_yellowstone_source_state(snapshot.state()), + snapshot.reconnect_count(), + snapshot.replay_attempt_count(), + snapshot.continuity_gap_count(), + ); + } + + fn observe_source_continuity( + &mut self, + state: crate::RawTransactionIngestSourceState, + reconnect_total: u64, + replay_attempt_total: u64, + continuity_gap_total: u64, + ) -> ksp_core_lib::Result<()> { + if reconnect_total < self.source_reconnect_total + || replay_attempt_total < self.source_replay_attempt_total + || continuity_gap_total < self.source_continuity_gap_total + { + return std::result::Result::Err(crate::runtime_error("source.continuity_counter_regression")); + } + let gap_increased = continuity_gap_total > self.source_continuity_gap_total; + self.source_state = std::option::Option::Some(state); + self.source_reconnect_total = reconnect_total; + self.source_replay_attempt_total = replay_attempt_total; + self.source_continuity_gap_total = continuity_gap_total; + self.publish(); + if gap_increased { + return std::result::Result::Err(crate::runtime_error("source.continuity_gap_proven")); + } + return std::result::Result::Ok(()); + } + + fn set_source_state(&mut self, state: crate::RawTransactionIngestSourceState) { + self.source_state = std::option::Option::Some(state); + self.publish(); + return; + } + + fn projection(&self) -> crate::RawTransactionIngestProcessingFrontierProjection { + return self.frontier.projection().with_source_continuity( + self.source_state, + self.source_reconnect_total, + self.source_replay_attempt_total, + self.source_continuity_gap_total, + ); + } + fn publish(&self) { - self.sender.send_replace(self.frontier.projection()); + self.sender.send_replace(self.projection()); return; } } 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 14ec27d..58437dc 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs @@ -1,16 +1,35 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs -// version: 3 +// version: 4 /// 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. +/// Source-neutral lifecycle state of the productive RAW transaction ingest source. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum RawTransactionIngestSourceState { + /// The productive source is active. + Active, + /// Transport is performing one bounded reconnect/replay attempt. + Reconnecting, + /// Cooperative source shutdown has started. + Closing, + /// The productive source closed cleanly. + Closed, + /// The productive source reached a terminal failure. + Failed, +} + +/// Private latest-value source-processing 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, + source_state: std::option::Option, + source_reconnect_total: u64, + source_replay_attempt_total: u64, + source_continuity_gap_total: u64, } impl crate::RawTransactionIngestProcessingFrontierProjection { @@ -20,6 +39,10 @@ impl crate::RawTransactionIngestProcessingFrontierProjection { hydration_pending: 0, processing_frontier_slot: std::option::Option::None, oldest_pending_slot: std::option::Option::None, + source_state: std::option::Option::None, + source_reconnect_total: 0, + source_replay_attempt_total: 0, + source_continuity_gap_total: 0, }; } @@ -29,7 +52,15 @@ impl crate::RawTransactionIngestProcessingFrontierProjection { processing_frontier_slot: std::option::Option, oldest_pending_slot: std::option::Option, ) -> Self { - return Self { hydration_pending, processing_frontier_slot, oldest_pending_slot }; + return Self { + hydration_pending, + processing_frontier_slot, + oldest_pending_slot, + source_state: std::option::Option::None, + source_reconnect_total: 0, + source_replay_attempt_total: 0, + source_continuity_gap_total: 0, + }; } /// Returns the number of source signals still pending hydration/admission processing. @@ -46,6 +77,41 @@ impl crate::RawTransactionIngestProcessingFrontierProjection { pub(crate) const fn oldest_pending_slot(&self) -> std::option::Option { return self.oldest_pending_slot; } + + /// Returns a copy carrying the latest source reconnect/replay projection. + pub(crate) const fn with_source_continuity( + mut self, + source_state: std::option::Option, + source_reconnect_total: u64, + source_replay_attempt_total: u64, + source_continuity_gap_total: u64, + ) -> Self { + self.source_state = source_state; + self.source_reconnect_total = source_reconnect_total; + self.source_replay_attempt_total = source_replay_attempt_total; + self.source_continuity_gap_total = source_continuity_gap_total; + return self; + } + + /// Returns the latest source-neutral lifecycle state carried by this private projection. + pub(crate) const fn source_state(&self) -> std::option::Option { + return self.source_state; + } + + /// Returns successful reconnects carried by this private projection. + pub(crate) const fn source_reconnect_total(&self) -> u64 { + return self.source_reconnect_total; + } + + /// Returns replay-bearing reconnect attempts carried by this private projection. + pub(crate) const fn source_replay_attempt_total(&self) -> u64 { + return self.source_replay_attempt_total; + } + + /// Returns proven replay-retention gaps carried by this private projection. + pub(crate) const fn source_continuity_gap_total(&self) -> u64 { + return self.source_continuity_gap_total; + } } /// Complete safe latest-value snapshot of one continuous RAW transaction ingest Worker. @@ -71,6 +137,10 @@ pub struct RawTransactionIngestSnapshot { hydration_pending: usize, processing_frontier_slot: std::option::Option, oldest_pending_slot: std::option::Option, + source_state: std::option::Option, + source_reconnect_total: u64, + source_replay_attempt_total: u64, + source_continuity_gap_total: u64, } impl crate::RawTransactionIngestSnapshot { @@ -193,6 +263,30 @@ impl crate::RawTransactionIngestSnapshot { pub const fn oldest_pending_slot(&self) -> std::option::Option { return self.oldest_pending_slot; } + + /// Returns the latest source-neutral lifecycle state when the productive source has started. + #[must_use] + pub const fn source_state(&self) -> std::option::Option { + return self.source_state; + } + + /// Returns successful automatic Yellowstone reconnects observed by this Worker run. + #[must_use] + pub const fn source_reconnect_total(&self) -> u64 { + return self.source_reconnect_total; + } + + /// Returns Yellowstone replay-bearing reconnect attempts observed by this Worker run. + #[must_use] + pub const fn source_replay_attempt_total(&self) -> u64 { + return self.source_replay_attempt_total; + } + + /// Returns replay retention gaps conservatively proven by Transport during this Worker run. + #[must_use] + pub const fn source_continuity_gap_total(&self) -> u64 { + return self.source_continuity_gap_total; + } } impl std::fmt::Debug for crate::RawTransactionIngestSnapshot { @@ -219,6 +313,10 @@ impl std::fmt::Debug for crate::RawTransactionIngestSnapshot { .field("hydration_pending", &self.hydration_pending) .field("processing_frontier_slot", &self.processing_frontier_slot) .field("oldest_pending_slot", &self.oldest_pending_slot) + .field("source_state", &self.source_state) + .field("source_reconnect_total", &self.source_reconnect_total) + .field("source_replay_attempt_total", &self.source_replay_attempt_total) + .field("source_continuity_gap_total", &self.source_continuity_gap_total) .finish(); } } @@ -335,6 +433,10 @@ impl crate::RawTransactionIngestSnapshotPublisher { hydration_pending: 0, processing_frontier_slot: std::option::Option::None, oldest_pending_slot: std::option::Option::None, + source_state: std::option::Option::None, + source_reconnect_total: 0, + source_replay_attempt_total: 0, + source_continuity_gap_total: 0, }; let (sender, receiver) = tokio::sync::watch::channel(snapshot.clone()); return (Self { sender, snapshot }, crate::RawTransactionIngestSnapshotSource { receiver }); @@ -427,6 +529,10 @@ impl crate::RawTransactionIngestSnapshotPublisher { 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(); + self.snapshot.source_state = projection.source_state(); + self.snapshot.source_reconnect_total = projection.source_reconnect_total(); + self.snapshot.source_replay_attempt_total = projection.source_replay_attempt_total(); + self.snapshot.source_continuity_gap_total = projection.source_continuity_gap_total(); return self.publish(state, admission_queue_depth, in_flight_persistence); } @@ -541,7 +647,7 @@ impl crate::RawTransactionIngestSnapshotPublisher { sequence, state, health_for_state(state, self.snapshot.worker.health()), - activity_for_state(state, admission_queue_depth, in_flight_persistence, self.snapshot.hydration_pending), + activity_for_state(state, admission_queue_depth, in_flight_persistence, self.snapshot.hydration_pending, self.snapshot.source_state), ); self.snapshot.worker = worker; self.snapshot.admission_queue_depth = admission_queue_depth; @@ -556,8 +662,17 @@ fn activity_for_state( admission_queue_depth: usize, in_flight_persistence: usize, hydration_pending: usize, + source_state: std::option::Option, ) -> ksp_worker_api::WorkerActivity { - if admission_queue_depth > 0 || in_flight_persistence > 0 || hydration_pending > 0 { + if admission_queue_depth > 0 + || in_flight_persistence > 0 + || hydration_pending > 0 + || matches!( + source_state, + std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting) + | std::option::Option::Some(crate::RawTransactionIngestSourceState::Closing) + ) + { 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 2ffd07c..2082d8f 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: 15 +// version: 16 //! Dependency firewall canaries for the RAW transaction ingest Worker foundation. @@ -322,7 +322,7 @@ fn manifest_dependency_names(section: &str) -> std::vec::Vec<&str> { } #[test] -fn v0_3_12_pre_007_processing_frontier_is_run_local_latest_value_and_has_no_replay_or_backend_edge() { +fn v0_3_12_pre_007_processing_frontier_remains_run_local_and_backend_neutral_after_continuity_extension() { let runtime = include_str!("../src/runtime.rs"); let resources = include_str!("../src/runtime_resources.rs"); let snapshot = include_str!("../src/snapshot.rs"); @@ -343,20 +343,45 @@ fn v0_3_12_pre_007_processing_frontier_is_run_local_latest_value_and_has_no_repl "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::", - ] { + let frontier = match resources.split_once("struct RawTransactionIngestProcessingFrontier {") { + std::option::Option::Some((_, tail)) => match tail.split_once("struct RawTransactionIngestProcessingFrontierReporter") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => "", + }, + std::option::Option::None => "", + }; + for forbidden in ["SubscribeReplayInfo", "from_slot", "repair", "ksp_job_backfill_lib::", "ksp_store_postgres_lib::"] { + assert!(!frontier.contains(forbidden), "pre.007 processing frontier absorbed replay/backend responsibility: {forbidden}"); + } + for forbidden in ["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}" + "Worker crossed backend boundary: {forbidden}" ); } return; } + +#[test] +fn v0_3_12_pre_008_worker_observes_transport_reconnect_replay_and_faults_only_on_proven_retention_gap() { + let resources = include_str!("../src/runtime_resources.rs"); + let transport = include_str!("../../ksp-onchain-transport-lib/src/grpc_stream.rs"); + for required in [ + "snapshot_source()", + "wait_yellowstone_session_snapshot", + "source_reconnect_total", + "source_replay_attempt_total", + "source_continuity_gap_total", + "source.continuity_gap_proven", + "RawTransactionIngestSourceState::Reconnecting", + ] { + assert!(resources.contains(required), "required pre.008 Worker continuity token missing: {required}"); + } + for forbidden in ["subscribe_replay_info(", "set_from_slot(", "YellowstoneReplayInfo", "last_requested_from_slot()", "ksp_job_backfill_lib::"] { + assert!(!resources.contains(forbidden), "Worker took ownership of replay/repair responsibility: {forbidden}"); + } + for required in ["SubscribeReplayInfo", "set_from_slot(effective_from_slot)", "first_available > requested", "replay_attempt_count"] { + assert!(transport.contains(required), "Transport replay ownership token missing: {required}"); + } + 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 150d012..e1c6a66 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: 9 +// version: 10 //! External public, security, redaction and release-boundary hardening canaries for `pre.010`. @@ -472,6 +472,13 @@ fn v0_3_12_pre_007_processing_frontier_is_bounded_processing_only_and_redacted() ] { assert!(resources.contains(required) || snapshot.contains(required), "required pre.007 bounded frontier guard missing: {required}"); } + let frontier = match resources.split_once("struct RawTransactionIngestProcessingFrontier {") { + std::option::Option::Some((_, tail)) => match tail.split_once("struct RawTransactionIngestProcessingFrontierReporter") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => "", + }, + std::option::Option::None => "", + }; for forbidden in [ "signature: ksp_store_lib::RawTransactionSignature", "filter_id", @@ -480,17 +487,38 @@ fn v0_3_12_pre_007_processing_frontier_is_bounded_processing_only_and_redacted() "transaction:", "meta:", "ReplayInfo", + "from_slot", "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}"); + assert!(!frontier.contains(forbidden), "pre.007 frontier absorbed forbidden material: {forbidden}"); + } + return; +} + +#[test] +fn v0_3_12_pre_008_reconnect_projection_is_source_neutral_bounded_and_contains_no_replay_material() { + let resources = include_str!("../src/runtime_resources.rs"); + let snapshot = include_str!("../src/snapshot.rs"); + for required in [ + "RawTransactionIngestSourceState", + "source_reconnect_total", + "source_replay_attempt_total", + "source_continuity_gap_total", + "source.continuity_counter_regression", + "source.continuity_gap_proven", + ] { + assert!(resources.contains(required) || snapshot.contains(required), "required pre.008 bounded continuity guard missing: {required}"); + } + 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 => "", + }; + for forbidden in ["signature", "provider", "endpoint", "filter", "transaction", "meta", "from_slot", "first_available", "ReplayInfo"] { + assert!(!projection.contains(forbidden), "pre.008 source projection leaked replay/provider/RAW 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 890306e..28426d9 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: 10 +// version: 11 //! External public-surface proofs for the RAW transaction ingest Worker foundation. @@ -122,6 +122,18 @@ fn pre_008_snapshot_surface_and_common_projection_are_public_and_stable() { ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshotSource::wait_for_change; fn require_worker_source() {} require_worker_source::(); + let _state_type = std::any::type_name::(); + let _states = [ + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState::Active, + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState::Reconnecting, + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState::Closing, + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState::Closed, + ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSourceState::Failed, + ]; + let _source_state = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::source_state; + let _reconnect = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::source_reconnect_total; + let _replay = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::source_replay_attempt_total; + let _gap = ksp_worker_raw_transaction_ingest_lib::RawTransactionIngestSnapshot::source_continuity_gap_total; assert_eq!(ksp_worker_raw_transaction_ingest_lib::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED.domain(), "worker_raw_transaction_ingest"); assert_eq!(ksp_worker_raw_transaction_ingest_lib::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED.code(), "counter_exhausted"); 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 1bd7371..613af4b 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: 7 +// version: 8 //! Release-completeness canaries through the `pre.010` public/release/security hardening tranche. @@ -82,6 +82,7 @@ fn pre_010_public_root_export_inventory_is_exact() { "RawTransactionIngestSnapshot", "RawTransactionIngestSnapshotFuture", "RawTransactionIngestSnapshotSource", + "RawTransactionIngestSourceState", "RawTransactionIngestTerminalFuture", "RawTransactionIngestWorker", "RawTransactionIngestYellowstoneSource", @@ -107,6 +108,7 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { "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", + "v0_3_12_pre_008_reconnect_projection_is_source_neutral_bounded_and_contains_no_replay_material", "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", ] { @@ -120,7 +122,8 @@ 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")); + assert!(dependency_boundary.contains("v0_3_12_pre_007_processing_frontier_remains_run_local_and_backend_neutral_after_continuity_extension")); + assert!(dependency_boundary.contains("v0_3_12_pre_008_worker_observes_transport_reconnect_replay_and_faults_only_on_proven_retention_gap")); 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")); @@ -128,5 +131,6 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { 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")); + assert!(public_api.contains("pre_008_snapshot_surface_and_common_projection_are_public_and_stable")); 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 418e14d..ef6225f 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: 8 +// version: 9 fn grpc_endpoint(cluster: &str) -> std::option::Option { return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider"); @@ -1370,3 +1370,45 @@ fn pre_007_processing_frontier_compacts_settled_slots_to_pending_intervals() { assert_eq!(projection.processing_frontier_slot(), std::option::Option::Some(9)); return; } + +#[test] +fn pre_008_reconnect_replay_and_proven_gap_are_distinct_monotone_and_frontier_preserving() { + let (sender, receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); + let mut reporter = super::RawTransactionIngestProcessingFrontierReporter::new(sender); + assert!(reporter.observe_pending(42).is_ok()); + assert!(reporter.observe_source_continuity(crate::RawTransactionIngestSourceState::Reconnecting, 0, 1, 0).is_ok()); + let replaying = *receiver.borrow(); + assert_eq!(replaying.hydration_pending(), 1); + assert_eq!(replaying.oldest_pending_slot(), std::option::Option::Some(42)); + assert_eq!(replaying.source_state(), std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting)); + assert_eq!(replaying.source_reconnect_total(), 0); + assert_eq!(replaying.source_replay_attempt_total(), 1); + assert_eq!(replaying.source_continuity_gap_total(), 0); + assert!(reporter.observe_source_continuity(crate::RawTransactionIngestSourceState::Active, 1, 1, 0).is_ok()); + let gap = reporter.observe_source_continuity(crate::RawTransactionIngestSourceState::Reconnecting, 1, 2, 1); + assert!(gap.is_err()); + let gap = match gap { + std::result::Result::Ok(()) => return, + std::result::Result::Err(value) => value, + }; + assert_eq!(gap.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); + assert!(gap.context().iter().any(|context| { + return context.value() == "source.continuity_gap_proven"; + })); + let proven = *receiver.borrow(); + assert_eq!(proven.hydration_pending(), 1); + assert_eq!(proven.oldest_pending_slot(), std::option::Option::Some(42)); + assert_eq!(proven.source_reconnect_total(), 1); + assert_eq!(proven.source_replay_attempt_total(), 2); + assert_eq!(proven.source_continuity_gap_total(), 1); + let regression = reporter.observe_source_continuity(crate::RawTransactionIngestSourceState::Active, 0, 2, 1); + assert!(regression.is_err()); + let regression = match regression { + std::result::Result::Ok(()) => return, + std::result::Result::Err(value) => value, + }; + assert!(regression.context().iter().any(|context| { + return context.value() == "source.continuity_counter_regression"; + })); + 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 053a8f1..7ddc15d 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: 3 +// version: 4 fn snapshot_foundation() -> std::option::Option<(crate::RawTransactionIngestSnapshotPublisher, crate::RawTransactionIngestSnapshotSource)> { let network = match ksp_store_lib::RawNetworkId::new("mainnet") { @@ -53,6 +53,10 @@ fn pre_008_initial_snapshot_and_common_projection_are_exact() { 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); + assert_eq!(concrete.source_state(), std::option::Option::None); + assert_eq!(concrete.source_reconnect_total(), 0); + assert_eq!(concrete.source_replay_attempt_total(), 0); + assert_eq!(concrete.source_continuity_gap_total(), 0); let common = ksp_worker_api::WorkerSnapshotSource::current(&source); assert_eq!(&common, concrete.worker_snapshot()); return; @@ -164,3 +168,34 @@ fn pre_007_processing_frontier_projection_is_latest_value_and_activity_aware() { assert_eq!(settled.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Idle); return; } + +#[test] +fn pre_008_source_continuity_projection_preserves_processing_frontier_and_distinguishes_replay() { + let (mut publisher, source) = match snapshot_foundation() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let projection = crate::RawTransactionIngestProcessingFrontierProjection::new(2, std::option::Option::Some(40), std::option::Option::Some(41)) + .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting), 0, 1, 0); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, projection).is_ok()); + let reconnecting = source.current(); + assert_eq!(reconnecting.hydration_pending(), 2); + assert_eq!(reconnecting.processing_frontier_slot(), std::option::Option::Some(40)); + assert_eq!(reconnecting.oldest_pending_slot(), std::option::Option::Some(41)); + assert_eq!(reconnecting.source_state(), std::option::Option::Some(crate::RawTransactionIngestSourceState::Reconnecting)); + assert_eq!(reconnecting.source_reconnect_total(), 0); + assert_eq!(reconnecting.source_replay_attempt_total(), 1); + assert_eq!(reconnecting.source_continuity_gap_total(), 0); + assert_eq!(reconnecting.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Active); + let projection = crate::RawTransactionIngestProcessingFrontierProjection::new(0, std::option::Option::Some(45), std::option::Option::None) + .with_source_continuity(std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), 1, 1, 0); + assert!(publisher.record_processing_frontier(ksp_worker_api::WorkerState::Running, 0, 0, projection).is_ok()); + let resumed = source.current(); + assert_eq!(resumed.hydration_pending(), 0); + assert_eq!(resumed.processing_frontier_slot(), std::option::Option::Some(45)); + assert_eq!(resumed.source_state(), std::option::Option::Some(crate::RawTransactionIngestSourceState::Active)); + assert_eq!(resumed.source_reconnect_total(), 1); + assert_eq!(resumed.source_replay_attempt_total(), 1); + assert_eq!(resumed.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Idle); + return; +} diff --git a/deltas/0.3.12/pre.008.md b/deltas/0.3.12/pre.008.md new file mode 100644 index 0000000..2e359ff --- /dev/null +++ b/deltas/0.3.12/pre.008.md @@ -0,0 +1,205 @@ + + + +# Delta `0.3.12-pre.008` — reconnect / replay / gap de rétention prouvé + +## Base requise + +```text +0.3.12-pre.007-fix.002 +workspace.package.version = 0.3.12-pre.7.fix.2 +``` + +Le gate opérateur communiqué pour cette base est entièrement vert : audits Rust/Markdown, `cargo check --workspace`, Clippy strict, 61 tests unitaires Worker, 8 `dependency_boundary`, 14 `hardening`, 9 `public_api`, 3 `release_completeness` et doc-tests sans échec. + +## Objectif + +Brancher la continuité Yellowstone déjà possédée par Transport dans le Worker sans dupliquer la politique de replay : + +```text +Transport reconnect snapshot + -> Worker source-neutral latest-value projection + -> reconnect/replay/gap observability + -> proven retention gap => source fault +``` + +Le Worker ne calcule ni `from_slot`, ni `SubscribeReplayInfo.first_available`; il ne déclenche aucun repair historique. + +## Version + +```text +workspace.package.version = 0.3.12-pre.8 +``` + +## Façade Transport ajoutée + +`ksp-onchain-transport-lib` expose désormais : + +```text +YellowstoneGrpcSubscribeSnapshotSource +SolanaYellowstoneGrpcSubscribeSession::snapshot_source() +YellowstoneGrpcSubscribeSnapshotSource::current() +YellowstoneGrpcSubscribeSnapshotSource::wait_for_change() +``` + +Cette façade encapsule le `tokio::sync::watch::Receiver` déjà possédé par la session. Elle n'expose ni Tonic, ni protobuf upstream, ni actor interne, ni credentials. + +Le snapshot Transport reste la source autoritaire pour : + +```text +state +reconnect_count +replay_attempt_count +continuity_gap_count +``` + +La sélection/clamp de `from_slot` et l'appel `SubscribeReplayInfo` restent exclusivement dans Transport. + +## Projection Worker + +Le snapshot Worker ajoute le type public source-neutral : + +```text +RawTransactionIngestSourceState + Active + Reconnecting + Closing + Closed + Failed +``` + +et les getters : + +```text +source_state() +source_reconnect_total() +source_replay_attempt_total() +source_continuity_gap_total() +``` + +La même watch privée que la processing frontier transporte ces latest-values ; aucune nouvelle watch Worker n'est ajoutée. + +Les champs processing-only `hydration_pending`, `processing_frontier_slot` et `oldest_pending_slot` restent inchangés et sont conservés pendant reconnect. + +## Politique reconnect/replay + +Le source task : + +```text +ouvre la session Yellowstone via Transport +obtient snapshot_source() +publie le snapshot initial Active +observe les changements Transport en parallèle de next_update et des hydrations +continue les hydrations déjà en vol pendant Reconnecting +laisse le stop préempter reconnect/hydration/admission +``` + +`replay_attempt_count` et `reconnect_count` restent distincts. Un replay attempt peut être observé alors qu'aucun reconnect n'a encore réussi. + +## Gap de rétention prouvé + +Le Worker ne fault que si `continuity_gap_count` augmente. + +Transport incrémente ce compteur uniquement lorsqu'il prouve : + +```text +SubscribeReplayInfo.first_available > requested replay slot +``` + +La hausse est publiée dans le snapshot Worker, puis le source task retourne `source.continuity_gap_proven`; le supervisor classe ensuite le terminal en `ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED`. + +Cette preuve reste conservatrice : elle prouve une couverture de replay demandée devenue indisponible, pas l'existence certaine d'un update filtré perdu. + +Les compteurs Transport observés doivent être monotones ; une régression impossible est rejetée avec `source.continuity_counter_regression`. + +## Non-claims + +`pre.008` n'ajoute pas : + +```text +repair/backfill automatique +source secondaire ou failover multi-provider +checkpoint durable inter-process +getBlock de repair +retry Worker spécifique getTransaction null +exactly-once +preuve de continuité blockchain globale +ownership Worker de from_slot +ownership Worker de SubscribeReplayInfo +``` + +## Fichiers modifiés + +```text +Cargo.toml +crates/ksp-onchain-transport-lib/src/grpc_stream.rs +crates/ksp-onchain-transport-lib/src/lib.rs +crates/ksp-onchain-transport-lib/tests/public_api.rs +crates/ksp-onchain-transport-lib/tests/release_completeness.rs +crates/ksp-onchain-transport-lib/unit_tests/grpc_stream.rs +crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs +crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs +crates/ksp-worker-raw-transaction-ingest-lib/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 +``` + +## Fichier ajouté + +```text +deltas/0.3.12/pre.008.md +``` + +## Fichiers supprimés + +```text +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 +canaris statiques pre.008 replay/frontier/redaction/public API +contrôle des versions de fichiers modifiés +contrôle diff exact contre pre.007-fix.002 +unzip -t du delta final +reproduction du delta sur la base exacte +``` + +`cargo`, `rustc` et `rustfmt` ne sont pas installés dans l'environnement d'assemblage. Aucun `cargo check`, Clippy ou test Cargo local n'est déclaré PASS ici. + +## 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-onchain-transport-lib +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 +``` + +## Décisions durables + +```text +Transport reste propriétaire de reconnect/from_slot/ReplayInfo. +Worker observe les résultats sûrs, il ne réimplémente pas la stratégie Transport. +reconnect != replay attempt != successful reconnect != proven retention gap. +proven retention gap => fault/stop, jamais repair implicite dans 0.3.12. +processing frontier reste run-local et processing-only. +``` + +## Suite + +`pre.009` reste consacré au hardening races/retry/backpressure : stop pendant hydration/reconnect, duplicate storm, Store lent, source failure, counter exhaustion et no-orphan, sans élargissement fonctionnel. 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 8e3a67f..0d58325 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 @@ -1492,3 +1492,163 @@ Le correctif `pre.007-fix.002` ne modifie donc pas le runtime. Il rend le canari Aucune définition pending/settled, compaction, projection snapshot, source Yellowstone, hydration, admission, persistence, API publique, reconnect ou replay n'est modifiée. Les validations Cargo de ce correctif restent à rejouer par l'opérateur. + +## 65. Gate opérateur `pre.007-fix.002` + +Le gate opérateur communiqué pour `0.3.12-pre.7.fix.2` 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, 814 fichiers) +cargo check --workspace : PASS +cargo clippy --workspace --all-targets --all-features -- -D warnings : PASS +cargo test -p ksp-worker-raw-transaction-ingest-lib : 61 unit PASS / 0 FAIL +dependency_boundary : 8 PASS / 0 FAIL +hardening : 14 PASS / 0 FAIL +public_api : 9 PASS / 0 FAIL +release_completeness : 3 PASS / 0 FAIL +doc-tests : 0 FAIL +``` + +`pre.007-fix.002` devient donc la base autoritaire de `pre.008`. + +## 66. Observation reconnect/replay Transport `pre.008` + +Le moteur Yellowstone Transport possédait déjà la politique de reconnect bornée, la sélection `from_slot`, `SubscribeReplayInfo` et les compteurs conservateurs. `pre.008` n'en duplique aucune partie dans le Worker. + +Transport ajoute uniquement une façade latest-value clonable et sûre : + +```text +YellowstoneGrpcSubscribeSnapshotSource +current() -> YellowstoneGrpcSubscribeSnapshot +wait_for_change() -> Option +SolanaYellowstoneGrpcSubscribeSession::snapshot_source() +``` + +Le `tokio::sync::watch::Receiver` reste privé à Transport. Cloner la source de snapshot ne clone ni le stream gRPC, ni le request state, ni l'actor de reconnect. + +Un canari Transport vérifie qu'un replay attempt peut être observé pendant `Reconnecting` alors que `reconnect_count == 0`. Le Worker peut donc distinguer : + +```text +reconnect intent/progress +replay-bearing attempt +successful reconnect +proven retention gap +``` + +sans inférer ces états à partir des updates métier. + +## 67. Projection source-neutral Worker `pre.008` + +Le Worker étend la même projection latest-value privée déjà utilisée par la processing frontier. Aucun second channel Worker n'est ajouté. + +Le snapshot concret expose : + +```text +source_state : Option +source_reconnect_total : u64 +source_replay_attempt_total : u64 +source_continuity_gap_total : u64 +``` + +`RawTransactionIngestSourceState` est strictement source-neutral : + +```text +Active +Reconnecting +Closing +Closed +Failed +``` + +La projection ne contient ni signature, filtre, provider, endpoint, payload RAW, `from_slot`, `first_available` ou `ReplayInfo`. + +Pendant `Reconnecting`, les hydrations HTTP déjà en vol et la processing frontier restent intactes. `WorkerActivity` est `Active` pendant `Reconnecting` ou `Closing`, même si les queues aval sont momentanément vides. + +## 68. Politique conservative de gap `pre.008` + +Le Worker ne fault pas parce qu'un reconnect commence, qu'un replay est tenté ou qu'un `from_slot` est accepté. + +La seule preuve de discontinuité utilisée est une hausse de : + +```text +YellowstoneGrpcSubscribeSnapshot::continuity_gap_count() +``` + +Ce compteur est incrémenté exclusivement par Transport lorsqu'un `SubscribeReplayInfo.first_available` est strictement supérieur à la slot demandée. Cela prouve que la couverture de replay demandée n'est plus entièrement retenue ; cela ne prouve pas qu'un update correspondant au filtre a réellement existé ou a été perdu. + +Quand ce compteur augmente : + +```text +projection gap publiée avec les compteurs courants +source task -> erreur interne source.continuity_gap_proven +supervisor -> ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED +Worker -> fault/stop +``` + +Aucun repair, backfill implicite, `getBlock`, source secondaire ou checkpoint durable n'est déclenché. + +Les compteurs reconnect/replay/gap observés par le Worker doivent rester monotones ; une régression impossible de snapshot est classée `source.continuity_counter_regression`. + +## 69. Ownership replay et non-claims `pre.008` + +Le Worker ne contient aucun appel à : + +```text +subscribe_replay_info(...) +set_from_slot(...) +YellowstoneReplayInfo +last_requested_from_slot() +``` + +Ces primitives restent entièrement dans `ksp-onchain-transport-lib`. En particulier, le Worker ne calcule, ne clamp et ne persiste jamais de `from_slot`. + +`pre.008` ne prétend pas : + +```text +qu'un replay accepté est lossless +qu'un replay successful prouve la continuité complète +faire un repair historique automatique +faire un failover multi-provider +maintenir un checkpoint inter-process +dédupliquer exactement-once +transformer processing_frontier en preuve de durabilité ou de complétude blockchain +``` + +## 70. Canaris et gate requis pour `pre.008` + +Les canaris déterministes ajoutés couvrent : + +```text +watcher Transport latest-value clonable sans fuite Tokio publique +replay_attempt_count > 0 possible avant reconnect_count > 0 +mapping exact des cinq états source-neutral +reconnect/replay/gap counters projetés dans le snapshot Worker +processing frontier préservée pendant Reconnecting +gap prouvé -> erreur source.continuity_gap_proven +régression de compteur -> source.continuity_counter_regression +aucun ownership Worker de ReplayInfo/from_slot/repair +frontier pre.007 toujours processing-only +inventaire public Worker mis à jour exactement +redaction de la projection reconnect/replay +``` + +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-onchain-transport-lib +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é. +