From 901a0515b776868823716df9af1876798e5c5ee5 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Wed, 9 Sep 2026 13:30:15 +0200 Subject: [PATCH] v0.3.12-pre.006 --- Cargo.toml | 4 +- .../src/lib.rs | 10 +- .../src/runtime.rs | 12 +- .../src/runtime_resources.rs | 517 +++++++++++++++--- .../tests/dependency_boundary.rs | 56 +- .../tests/hardening.rs | 38 +- .../tests/release_completeness.rs | 12 +- .../unit_tests/runtime_resources.rs | 78 ++- deltas/0.3.12/pre.006.md | 219 ++++++++ ...0_3_12_YELLOWSTONE_HYDRATION_CONTINUITY.md | 180 +++++- 10 files changed, 1011 insertions(+), 115 deletions(-) create mode 100644 deltas/0.3.12/pre.006.md diff --git a/Cargo.toml b/Cargo.toml index c441b60..e113919 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 526 +# version: 527 [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.5.fix.1" +version = "0.3.12-pre.6" 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 a37739c..ef43474 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: 12 +// version: 13 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -11,10 +11,10 @@ //! and the caller-runtime-owned lifecycle with private child-task supervision. This tranche also //! owns bounded source-neutral admission, common RAW canonicalization/assembly and backend-neutral //! Store persistence in normal mode plus concrete latest-value snapshots projected onto Worker API; -//! the first validated Yellowstone/HTTP runtime-resource contract also closes the deterministic hydration -//! qualification path in test configuration. Transaction/TransactionStatus/Block signal projection, BlockMeta/Slot -//! continuity projection and HTTP hydration remain inactive in production until the productive source task is wired; -//! no live stream is opened yet. +//! 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. mod admission; mod error; 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 0bebc64..b75b032 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs -// version: 8 +// version: 9 type PersistencePort = std::sync::Arc; type PersistenceTasks = tokio::task::JoinSet>; @@ -96,7 +96,7 @@ impl crate::RawTransactionIngestWorker { return start_foundation(settings, runtime, std::option::Option::Some(store)); } - /// Starts one Worker with caller-composed runtime resources after synchronous network validation, without opening the live source before its dedicated tranche. + /// Starts one Worker with caller-composed runtime resources and one supervised productive Yellowstone source task. pub fn start_with_runtime_resources( settings: crate::RawTransactionIngestSettings, store: std::sync::Arc, @@ -113,7 +113,13 @@ impl crate::RawTransactionIngestWorker { if let std::result::Result::Err(error) = runtime_resources.validate_network(settings.network()) { return std::result::Result::Err(error); } - return start_foundation(settings, runtime, std::option::Option::Some(store)); + 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; + }); + }); } } 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 5c3084f..d345c51 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,17 +1,12 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs -// version: 7 +// version: 8 -#[cfg(test)] use sha2::Digest; // rust-rules: trait-import -#[cfg(test)] const RAW_TRANSACTION_INGEST_YELLOWSTONE_FILTER_FINGERPRINT_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone.filters.v1\0"; -#[cfg(test)] const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_PROTOCOL: &str = "yellowstone_http"; -#[cfg(test)] const RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN: &[u8] = b"ksp.raw_transaction_ingest.yellowstone_http.source_key.v1\0"; -#[cfg(test)] #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum RawTransactionIngestSourceFamily { Block, @@ -19,14 +14,12 @@ enum RawTransactionIngestSourceFamily { TransactionStatus, } -#[cfg(test)] #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum RawTransactionIngestContinuityFamily { BlockMeta, Slot, } -#[cfg(test)] #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum RawTransactionIngestContinuityStatus { Completed, @@ -38,7 +31,6 @@ enum RawTransactionIngestContinuityStatus { Processed, } -#[cfg(test)] #[derive(Clone, Eq, PartialEq)] struct RawTransactionIngestContinuitySignal { created_at: std::option::Option, @@ -53,7 +45,6 @@ struct RawTransactionIngestContinuitySignal { status: std::option::Option, } -#[cfg(test)] impl std::fmt::Debug for RawTransactionIngestContinuitySignal { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter @@ -78,21 +69,18 @@ struct RawTransactionIngestSourceRoute { provider: ksp_store_lib::RawProvenanceCode, } -#[cfg(test)] #[derive(Clone, Copy, Eq, PartialEq)] struct RawTransactionIngestSourceTimestamp { nanos: u32, seconds: i64, } -#[cfg(test)] impl std::fmt::Debug for RawTransactionIngestSourceTimestamp { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter.debug_struct("RawTransactionIngestSourceTimestamp").field("nanos", &self.nanos).field("seconds", &self.seconds).finish(); } } -#[cfg(test)] #[derive(Clone, Eq, PartialEq)] struct RawTransactionIngestSourceSignal { created_at: std::option::Option, @@ -107,7 +95,6 @@ struct RawTransactionIngestSourceSignal { transaction_index: std::option::Option, } -#[cfg(test)] impl std::fmt::Debug for RawTransactionIngestSourceSignal { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { return formatter @@ -126,7 +113,6 @@ impl std::fmt::Debug for RawTransactionIngestSourceSignal { } } -#[cfg(test)] trait RawTransactionIngestYellowstoneSignalView { fn created_at(&self) -> std::option::Option; @@ -141,7 +127,6 @@ trait RawTransactionIngestYellowstoneSignalView { fn slot(&self) -> u64; } -#[cfg(test)] impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::YellowstoneTransactionUpdate { fn created_at(&self) -> std::option::Option { return ksp_onchain_transport_lib::YellowstoneTransactionUpdate::created_at(self); @@ -168,7 +153,6 @@ impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::Ye } } -#[cfg(test)] impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate { fn created_at(&self) -> std::option::Option { return ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate::created_at(self); @@ -195,7 +179,6 @@ impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::Ye } } -#[cfg(test)] trait RawTransactionIngestYellowstoneBlockView { fn created_at(&self) -> std::option::Option; @@ -208,7 +191,6 @@ trait RawTransactionIngestYellowstoneBlockView { fn transaction_identity(&self, position: usize) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)>; } -#[cfg(test)] impl RawTransactionIngestYellowstoneBlockView for ksp_onchain_transport_lib::YellowstoneBlockUpdate { fn created_at(&self) -> std::option::Option { return ksp_onchain_transport_lib::YellowstoneBlockUpdate::created_at(self); @@ -235,7 +217,6 @@ impl RawTransactionIngestYellowstoneBlockView for ksp_onchain_transport_lib::Yel } } -#[cfg(test)] trait RawTransactionIngestYellowstoneContinuityView { fn created_at(&self) -> std::option::Option; @@ -250,7 +231,6 @@ trait RawTransactionIngestYellowstoneContinuityView { fn status(&self) -> std::option::Option; } -#[cfg(test)] impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib::YellowstoneBlockMetaUpdate { fn created_at(&self) -> std::option::Option { return ksp_onchain_transport_lib::YellowstoneBlockMetaUpdate::created_at(self); @@ -277,7 +257,6 @@ impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib } } -#[cfg(test)] impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib::YellowstoneSlotUpdate { fn created_at(&self) -> std::option::Option { return ksp_onchain_transport_lib::YellowstoneSlotUpdate::created_at(self); @@ -304,7 +283,6 @@ impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib } } -#[cfg(test)] impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate)> for RawTransactionIngestSourceSignal { @@ -313,7 +291,6 @@ impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onc } } -#[cfg(test)] impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate)> for RawTransactionIngestSourceSignal { @@ -324,8 +301,8 @@ impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onc /// Validated Yellowstone plus HTTP runtime source owned by the continuous RAW transaction ingest Worker. /// -/// The Transport-owned channel, subscribe request, HTTP pool and hydration role remain private. Construction validates only deterministic source-composition -/// invariants and performs no network I/O. +/// The Transport-owned channel, subscribe request, HTTP pool and hydration role remain private. Construction validates deterministic source-composition +/// invariants without network I/O; runtime execution opens exactly one Transport-owned Yellowstone session. pub struct RawTransactionIngestYellowstoneSource { yellowstone_channel: ksp_onchain_transport_lib::YellowstoneGrpcChannel, subscribe_request: ksp_onchain_transport_lib::YellowstoneSubscribeRequest, @@ -383,6 +360,92 @@ impl crate::RawTransactionIngestYellowstoneSource { } return std::result::Result::Ok(Self { yellowstone_channel, subscribe_request, http_pool, hydration_role, network, route }); } + + /// Runs the productive Yellowstone source task until cooperative stop or one safe terminal source failure. + pub(crate) async fn run( + self, + settings: crate::RawTransactionIngestSettings, + mut stop_receiver: tokio::sync::watch::Receiver, + admission_sender: tokio::sync::mpsc::Sender, + ) -> ksp_core_lib::Result<()> { + let opened = tokio::select! { + biased; + _ = stop_receiver.changed() => { + return std::result::Result::Ok(()); + } + result = self.yellowstone_channel.open_standard_subscribe(self.subscribe_request.clone()) => result, + }; + let mut session = match opened { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(source_transport_error(error.code())), + }; + let mut coordinator = RawTransactionIngestHydrationCoordinator::new(&settings); + let mut fault = std::option::Option::None; + loop { + if *stop_receiver.borrow() { + break; + } + if let std::result::Result::Err(error) = coordinator.start_hydrations(&self) { + fault = std::option::Option::Some(error); + break; + } + let can_receive = coordinator.can_receive(); + if !can_receive && coordinator.tasks.is_empty() { + fault = std::option::Option::Some(crate::runtime_error("source.hydration_stalled")); + break; + } + tokio::select! { + biased; + _ = stop_receiver.changed() => { + break; + } + joined = coordinator.tasks.join_next(), if !coordinator.tasks.is_empty() => { + let joined = match joined { + std::option::Option::Some(value) => value, + std::option::Option::None => { + fault = std::option::Option::Some(crate::runtime_error("source.hydration_join_missing")); + break; + }, + }; + let handled = coordinator.handle_joined(joined, &self, &settings, &admission_sender, &mut stop_receiver).await; + match handled { + std::result::Result::Ok(true) => {}, + std::result::Result::Ok(false) => break, + std::result::Result::Err(error) => { + fault = std::option::Option::Some(error); + break; + }, + } + } + update = session.next_update(), if can_receive => { + let update = match update { + std::result::Result::Ok(std::option::Option::Some(value)) => value, + std::result::Result::Ok(std::option::Option::None) => { + fault = std::option::Option::Some(crate::runtime_error("source.session_closed")); + break; + }, + std::result::Result::Err(error) => { + fault = std::option::Option::Some(source_transport_error(error.code())); + break; + }, + }; + if let std::result::Result::Err(error) = route_yellowstone_update(&self, &mut coordinator, update) { + fault = std::option::Option::Some(error); + break; + } + } + } + } + coordinator.abort_all().await; + let closed = session.close().await; + if let std::option::Option::Some(error) = fault { + 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())), + }; + } } impl std::fmt::Debug for crate::RawTransactionIngestYellowstoneSource { @@ -416,7 +479,7 @@ pub struct RawTransactionIngestRuntimeResources { } impl crate::RawTransactionIngestRuntimeResources { - /// Owns the first validated productive-source contract for a future runtime start. + /// Owns the validated productive Yellowstone source contract consumed by `start_with_runtime_resources`. #[must_use] pub fn new(yellowstone_source: crate::RawTransactionIngestYellowstoneSource) -> Self { return Self { yellowstone_source }; @@ -429,6 +492,11 @@ impl crate::RawTransactionIngestRuntimeResources { } return std::result::Result::Ok(()); } + + /// Consumes the runtime-resource aggregate into its single validated Yellowstone source for private supervisor wiring. + pub(crate) fn into_yellowstone_source(self) -> crate::RawTransactionIngestYellowstoneSource { + return self.yellowstone_source; + } } impl std::fmt::Debug for crate::RawTransactionIngestRuntimeResources { @@ -497,7 +565,6 @@ 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()); } -#[cfg(test)] 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(); @@ -511,7 +578,6 @@ fn matched_filter_direct_id(filters: &[ksp_onchain_transport_lib::YellowstoneSub }; } -#[cfg(test)] fn matched_filter_fingerprint(filters: &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName]) -> [u8; 32] { let mut names = filters.iter().map(ksp_onchain_transport_lib::YellowstoneSubscribeFilterName::as_str).collect::>(); names.sort_unstable(); @@ -526,7 +592,6 @@ fn matched_filter_fingerprint(filters: &[ksp_onchain_transport_lib::YellowstoneS return hasher.finalize().into(); } -#[cfg(test)] fn project_yellowstone_block_signals( source: &crate::RawTransactionIngestYellowstoneSource, update: &T, @@ -558,7 +623,6 @@ fn project_yellowstone_block_signals( source: &crate::RawTransactionIngestYellowstoneSource, update: &T, @@ -578,7 +642,6 @@ fn project_yellowstone_continuity_signal RawTransactionIngestContinuityStatus { return match status { ksp_onchain_transport_lib::YellowstoneSlotStatus::Processed => RawTransactionIngestContinuityStatus::Processed, @@ -591,7 +654,6 @@ fn map_yellowstone_slot_status(status: ksp_onchain_transport_lib::YellowstoneSlo }; } -#[cfg(test)] fn map_yellowstone_source_timestamp( created_at: std::option::Option, ) -> std::option::Option { @@ -600,7 +662,6 @@ fn map_yellowstone_source_timestamp( }); } -#[cfg(test)] fn project_yellowstone_signal( source: &crate::RawTransactionIngestYellowstoneSource, update: &T, @@ -623,52 +684,326 @@ fn project_yellowstone_signal( }; } -#[cfg(test)] -enum RawTransactionIngestHydrationOutcome { - Available(std::boxed::Box), - Missing(ksp_store_lib::RawTransactionReference), +fn route_yellowstone_update( + source: &crate::RawTransactionIngestYellowstoneSource, + coordinator: &mut RawTransactionIngestHydrationCoordinator, + update: ksp_onchain_transport_lib::YellowstoneSubscribeUpdate, +) -> ksp_core_lib::Result<()> { + match update { + ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Transaction(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.as_ref())), received_at); + }, + 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); + }, + ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Block(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), + }; + let signals = match project_yellowstone_block_signals(source, &value) { + std::result::Result::Ok(value) => value, + 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) { + 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); + return std::result::Result::Ok(()); + }, + ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Slot(value) => { + let _continuity_signal = project_yellowstone_continuity_signal(source, &value); + return std::result::Result::Ok(()); + }, + ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Account(_) + | ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Ping(_) + | ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Pong(_) + | ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Entry(_) => std::result::Result::Ok(()), + } } -#[cfg(test)] -async fn hydrate_yellowstone_signal( +fn current_raw_timestamp() -> ksp_core_lib::Result { + let duration = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH); + let duration = match duration { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.clock_before_epoch")), + }; + let millis = match u64::try_from(duration.as_millis()) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.clock_unrepresentable")), + }; + return match ksp_store_lib::RawTimestamp::from_unix_millis(millis) { + std::result::Result::Ok(value) => std::result::Result::Ok(value), + std::result::Result::Err(_) => std::result::Result::Err(crate::runtime_error("source.clock_out_of_bounds")), + }; +} + +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()); +} + +#[derive(Clone, Eq, Ord, PartialEq, PartialOrd)] +struct RawTransactionIngestHydrationKey { + commitment: &'static str, + network: ksp_store_lib::RawNetworkId, + signature: ksp_store_lib::RawTransactionSignature, +} + +struct RawTransactionIngestPendingSignal { + received_at: ksp_store_lib::RawTimestamp, + signal: RawTransactionIngestSourceSignal, +} + +struct RawTransactionIngestPendingHydration { + in_flight: bool, + signals: std::vec::Vec, +} + +type RawTransactionIngestObservedTransaction = + ksp_onchain_transport_lib::HttpObservedValue>; + +struct RawTransactionIngestHydrationFetch { + key: RawTransactionIngestHydrationKey, + observed: RawTransactionIngestObservedTransaction, +} + +type RawTransactionIngestHydrationTasks = tokio::task::JoinSet>; + +struct RawTransactionIngestHydrationCoordinator { + max_in_flight: usize, + max_pending_signals: usize, + pending_signal_count: usize, + pending: std::collections::BTreeMap, + tasks: RawTransactionIngestHydrationTasks, +} + +impl RawTransactionIngestHydrationCoordinator { + fn new(settings: &crate::RawTransactionIngestSettings) -> Self { + return Self { + max_in_flight: settings.persistence_concurrency(), + max_pending_signals: crate::MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY, + pending_signal_count: 0, + pending: std::collections::BTreeMap::new(), + tasks: RawTransactionIngestHydrationTasks::new(), + }; + } + + fn can_receive(&self) -> bool { + return self.pending_signal_count < self.max_pending_signals; + } + + fn queue_signal( + &mut self, + source: &crate::RawTransactionIngestYellowstoneSource, + signal: RawTransactionIngestSourceSignal, + received_at: ksp_store_lib::RawTimestamp, + ) -> 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")); + } + let key = match hydration_key(source, &signal) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + match self.pending.entry(key) { + std::collections::btree_map::Entry::Occupied(mut entry) => { + entry.get_mut().signals.push(RawTransactionIngestPendingSignal { received_at, signal }); + }, + std::collections::btree_map::Entry::Vacant(entry) => { + entry.insert(RawTransactionIngestPendingHydration { + in_flight: false, + signals: std::vec![RawTransactionIngestPendingSignal { received_at, signal }], + }); + }, + } + self.pending_signal_count = match self.pending_signal_count.checked_add(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.hydration_pending_counter_exhausted")), + }; + 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 { std::option::Option::None } else { std::option::Option::Some(key.clone()) }); + let key = match key { + std::option::Option::Some(value) => value, + std::option::Option::None => break, + }; + let commitment = match hydration_commitment(source) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let pending = match self.pending.get_mut(&key) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.hydration_pending_missing")), + }; + pending.in_flight = true; + let pool = source.http_pool.clone(); + let role = source.hydration_role.clone(); + let expected_network = source.network.clone(); + let task_key = key.clone(); + let _abort_handle = self.tasks.spawn(async move { + return fetch_yellowstone_hydration(pool, role, expected_network, task_key, commitment).await; + }); + } + return std::result::Result::Ok(()); + } + + async fn handle_joined( + &mut self, + joined: std::result::Result, tokio::task::JoinError>, + source: &crate::RawTransactionIngestYellowstoneSource, + settings: &crate::RawTransactionIngestSettings, + admission_sender: &tokio::sync::mpsc::Sender, + stop_receiver: &mut tokio::sync::watch::Receiver, + ) -> ksp_core_lib::Result { + let fetched = match joined { + std::result::Result::Ok(std::result::Result::Ok(value)) => value, + std::result::Result::Ok(std::result::Result::Err(error)) => return std::result::Result::Err(error), + std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.hydration_task_join_failed")), + }; + let pending = match self.pending.remove(&fetched.key) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.hydration_result_without_pending")), + }; + if pending.signals.len() > self.pending_signal_count { + return std::result::Result::Err(crate::runtime_error("source.hydration_pending_counter_invalid")); + } + self.pending_signal_count -= pending.signals.len(); + for pending_signal in pending.signals { + 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, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let ingress = match ingress { + std::option::Option::Some(value) => value, + std::option::Option::None => continue, + }; + let sent = tokio::select! { + biased; + _ = stop_receiver.changed() => { + return std::result::Result::Ok(false); + } + result = admission_sender.send(ingress) => result, + }; + if sent.is_err() { + if *stop_receiver.borrow() { + return std::result::Result::Ok(false); + } + return std::result::Result::Err(crate::runtime_error("source.admission_closed")); + } + } + if let std::result::Result::Err(error) = self.start_hydrations(source) { + return std::result::Result::Err(error); + } + return std::result::Result::Ok(true); + } + + async fn abort_all(&mut self) { + self.tasks.abort_all(); + while self.tasks.join_next().await.is_some() {} + self.pending.clear(); + self.pending_signal_count = 0; + return; + } +} + +fn hydration_commitment(source: &crate::RawTransactionIngestYellowstoneSource) -> ksp_core_lib::Result { + return match source.subscribe_request.commitment() { + std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed) => { + std::result::Result::Ok(ksp_onchain_transport_lib::SolanaCommitment::Confirmed) + }, + std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Finalized) => { + std::result::Result::Ok(ksp_onchain_transport_lib::SolanaCommitment::Finalized) + }, + std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Processed) | std::option::Option::None => { + std::result::Result::Err(crate::runtime_error("hydration.commitment_invalid")) + }, + }; +} + +fn hydration_key( + source: &crate::RawTransactionIngestYellowstoneSource, + signal: &RawTransactionIngestSourceSignal, +) -> ksp_core_lib::Result { + let commitment = match hydration_commitment(source) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok(RawTransactionIngestHydrationKey { + commitment: commitment.as_str(), + network: signal.network.clone(), + signature: signal.signature, + }); +} + +async fn fetch_yellowstone_hydration( + http_pool: ksp_onchain_transport_lib::HttpTransportPool, + hydration_role: ksp_onchain_transport_lib::HttpRoleName, + expected_network: ksp_store_lib::RawNetworkId, + key: RawTransactionIngestHydrationKey, + commitment: ksp_onchain_transport_lib::SolanaCommitment, +) -> ksp_core_lib::Result { + if key.network != expected_network { + return std::result::Result::Err(crate::runtime_error("hydration.network_key_mismatch")); + } + if key.commitment != commitment.as_str() { + return std::result::Result::Err(crate::runtime_error("hydration.commitment_key_mismatch")); + } + let config = ksp_onchain_transport_lib::SolanaGetTransactionConfig::new( + std::option::Option::Some(commitment), + std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionEncoding::Base64), + std::option::Option::Some(0), + ); + let signature_text = ksp_raw_transaction_lib::format_raw_transaction_signature(&key.signature); + let observed = http_pool.get_transaction_observed(&hydration_role, signature_text.as_str(), std::option::Option::Some(&config)).await; + let observed = match observed { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(hydration_transport_error(error.code())), + }; + return std::result::Result::Ok(RawTransactionIngestHydrationFetch { key, observed }); +} + +fn finalize_yellowstone_hydration( source: &crate::RawTransactionIngestYellowstoneSource, settings: &crate::RawTransactionIngestSettings, signal: RawTransactionIngestSourceSignal, received_at: ksp_store_lib::RawTimestamp, -) -> ksp_core_lib::Result { + observed: &RawTransactionIngestObservedTransaction, +) -> ksp_core_lib::Result> { if &signal.network != settings.network() || signal.network != source.network { return std::result::Result::Err(crate::runtime_error("hydration.network_mismatch")); } if signal.route != source.route { return std::result::Result::Err(crate::runtime_error("hydration.source_route_mismatch")); } - let commitment = match source.subscribe_request.commitment() { - std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Confirmed) => ksp_onchain_transport_lib::SolanaCommitment::Confirmed, - std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Finalized) => ksp_onchain_transport_lib::SolanaCommitment::Finalized, - std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Processed) | std::option::Option::None => { - return std::result::Result::Err(crate::runtime_error("hydration.commitment_invalid")); - }, - }; - let config = ksp_onchain_transport_lib::SolanaGetTransactionConfig::new( - std::option::Option::Some(commitment), - std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionEncoding::Base64), - std::option::Option::Some(0), - ); - let signature_text = ksp_raw_transaction_lib::format_raw_transaction_signature(&signal.signature); - let observed = source.http_pool.get_transaction_observed(&source.hydration_role, signature_text.as_str(), std::option::Option::Some(&config)).await; - let observed = match observed { + let commitment = match hydration_commitment(source) { std::result::Result::Ok(value) => value, - std::result::Result::Err(error) => return std::result::Result::Err(hydration_transport_error(error.code())), + std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let http_provider = observed.provider().as_str().to_owned(); - let http_endpoint = observed.endpoint_name().to_owned(); - let transaction = observed.into_value(); - let transaction = match transaction { + let transaction = match observed.value().as_ref() { std::option::Option::Some(value) => value, - std::option::Option::None => { - let reference = ksp_store_lib::RawTransactionReference::new(signal.network, signal.signature); - return std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Missing(reference)); - }, + std::option::Option::None => return std::result::Result::Ok(std::option::Option::None), }; if transaction.slot() != signal.slot { return std::result::Result::Err(crate::runtime_error("hydration.slot_mismatch")); @@ -699,7 +1034,7 @@ async fn hydrate_yellowstone_signal( if embedded_signature != signal.signature { return std::result::Result::Err(crate::runtime_error("hydration.signature_mismatch")); } - let provenance = build_hydration_provenance(settings, &signal, http_provider.as_str(), http_endpoint.as_str(), 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), @@ -720,15 +1055,46 @@ async fn hydrate_yellowstone_signal( }), map_hydration_wire_field(transaction.transaction_index(), |value| return *value), ); - return std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Available(std::boxed::Box::new(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)] +enum RawTransactionIngestHydrationOutcome { + Available(std::boxed::Box), + Missing(ksp_store_lib::RawTransactionReference), +} + +#[cfg(test)] +async fn hydrate_yellowstone_signal( + source: &crate::RawTransactionIngestYellowstoneSource, + settings: &crate::RawTransactionIngestSettings, + signal: RawTransactionIngestSourceSignal, + received_at: ksp_store_lib::RawTimestamp, +) -> ksp_core_lib::Result { + let reference = ksp_store_lib::RawTransactionReference::new(signal.network.clone(), signal.signature); + let key = match hydration_key(source, &signal) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let commitment = match hydration_commitment(source) { + 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 = match fetched { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let ingress = finalize_yellowstone_hydration(source, settings, signal, received_at, &fetched.observed); + return match ingress { + std::result::Result::Ok(std::option::Option::Some(value)) => { + std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Available(std::boxed::Box::new(value))) + }, + std::result::Result::Ok(std::option::Option::None) => std::result::Result::Ok(RawTransactionIngestHydrationOutcome::Missing(reference)), + std::result::Result::Err(error) => std::result::Result::Err(error), + }; +} + fn build_hydration_provenance( settings: &crate::RawTransactionIngestSettings, signal: &RawTransactionIngestSourceSignal, @@ -779,7 +1145,6 @@ fn build_hydration_provenance( return std::result::Result::Ok(provenance); } -#[cfg(test)] fn hydration_method_code(family: RawTransactionIngestSourceFamily) -> &'static str { return match family { RawTransactionIngestSourceFamily::Block => "block_get_transaction", @@ -788,7 +1153,6 @@ fn hydration_method_code(family: RawTransactionIngestSourceFamily) -> &'static s }; } -#[cfg(test)] fn fingerprint_filter_code(fingerprint: &[u8; 32]) -> ksp_core_lib::Result { const HEX: &[u8; 16] = b"0123456789abcdef"; let mut value = std::string::String::with_capacity(71); @@ -803,7 +1167,6 @@ fn fingerprint_filter_code(fingerprint: &[u8; 32]) -> ksp_core_lib::Result, received_at: ksp_store_lib::RawTimestamp, @@ -830,7 +1193,6 @@ fn representable_observed_at( return std::option::Option::Some(observed_at); } -#[cfg(test)] fn hydration_source_key(provenance: &ksp_store_lib::RawAcquisitionProvenance) -> [u8; 32] { let mut hasher = sha2::Sha256::new(); hasher.update(RAW_TRANSACTION_INGEST_YELLOWSTONE_HTTP_SOURCE_KEY_DOMAIN); @@ -852,21 +1214,18 @@ fn hydration_source_key(provenance: &ksp_store_lib::RawAcquisitionProvenance) -> return hasher.finalize().into(); } -#[cfg(test)] fn hash_hydration_source_key(hasher: &mut sha2::Sha256, value: &[u8]) { hasher.update((value.len() as u64).to_be_bytes()); hasher.update(value); return; } -#[cfg(test)] fn hydration_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 hydration transport failed") .with_context("transport_domain", code.domain()) .with_context("transport_code", code.code()); } -#[cfg(test)] fn map_hydration_wire_field( field: &ksp_onchain_transport_lib::SolanaWireField, mut map_value: F, 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 a024989..9ad20c2 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: 12 +// version: 13 //! Dependency firewall canaries for the RAW transaction ingest Worker foundation. @@ -64,7 +64,7 @@ fn v0_3_12_pre_002_manifest_opens_only_the_onchain_transport_live_source_edge() } #[test] -fn v0_3_12_pre_003_transaction_and_status_adapters_are_private_offline_and_transport_facade_only() { +fn v0_3_12_pre_003_transaction_and_status_adapters_are_private_and_transport_facade_only() { let root = include_str!("../src/lib.rs"); let resources = include_str!("../src/runtime_resources.rs"); for required in [ @@ -85,8 +85,6 @@ fn v0_3_12_pre_003_transaction_and_status_adapters_are_private_offline_and_trans "pub struct RawTransactionIngestSourceSignal", "pub(crate) struct RawTransactionIngestSourceSignal", "pub use self::runtime_resources::RawTransactionIngestSourceSignal", - "open_standard_subscribe", - "next_update", "get_block_observed", "ksp_store_postgres_lib::", "reqwest::", @@ -127,10 +125,9 @@ fn v0_3_12_pre_004_hydration_contract_uses_only_transport_facade_common_raw_and_ ] { assert!(resources.contains(required), "required pre.004 hydration contract missing: {required}"); } - assert!(resources.contains("#[cfg(test)]\nasync fn hydrate_yellowstone_signal")); + assert!(resources.contains("fn finalize_yellowstone_hydration")); + assert!(resources.contains("fn fetch_yellowstone_hydration")); for forbidden in [ - "open_standard_subscribe", - "next_update", "get_block_observed", "ksp_config_lib::", "ksp_job_backfill_lib::", @@ -145,7 +142,7 @@ fn v0_3_12_pre_004_hydration_contract_uses_only_transport_facade_common_raw_and_ } #[test] -fn v0_3_12_pre_005_block_and_continuity_adapters_remain_private_offline_and_transport_facade_only() { +fn v0_3_12_pre_005_block_and_continuity_adapters_remain_private_and_transport_facade_only() { let root = include_str!("../src/lib.rs"); let resources = include_str!("../src/runtime_resources.rs"); for required in [ @@ -161,8 +158,6 @@ fn v0_3_12_pre_005_block_and_continuity_adapters_remain_private_offline_and_tran assert!(resources.contains(required), "required pre.005 private adapter contract missing: {required}"); } for forbidden in [ - "open_standard_subscribe", - "next_update", "get_block_observed", "ksp_config_lib::", "ksp_job_backfill_lib::", @@ -177,7 +172,46 @@ fn v0_3_12_pre_005_block_and_continuity_adapters_remain_private_offline_and_tran } #[test] -fn v0_3_12_pre_002_source_surface_hardens_shutdown_and_faults_without_backend_or_premature_live_io() { +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)"] { + assert!(runtime.contains(required), "required pre.006 supervisor wiring missing: {required}"); + } + for required in [ + "open_standard_subscribe", + "next_update", + "RawTransactionIngestHydrationCoordinator", + "std::collections::BTreeMap", + "tokio::task::JoinSet", + "MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY", + "get_transaction_observed", + "admission_sender.send(ingress)", + "tasks.abort_all()", + "session.close().await", + "commitment: commitment.as_str()", + "network: signal.network.clone()", + "signature: signal.signature", + ] { + assert!(resources.contains(required), "required pre.006 productive-source contract missing: {required}"); + } + for forbidden in [ + "get_block_observed", + "ksp_config_lib::", + "ksp_job_backfill_lib::", + "ksp_store_postgres_lib::", + "reqwest::", + "tonic::", + "yellowstone_grpc_proto::", + "unbounded_channel", + ] { + assert!(!runtime.contains(forbidden) && !resources.contains(forbidden), "pre.006 crossed a forbidden productive-source boundary: {forbidden}"); + } + return; +} + +#[test] +fn v0_3_12_pre_006_source_surface_hardens_shutdown_and_faults_without_backend_edges() { let root = include_str!("../src/lib.rs"); let runtime = include_str!("../src/runtime.rs"); let admission = include_str!("../src/admission.rs"); 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 9307910..84d6158 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: 7 +// version: 8 //! External public, security, redaction and release-boundary hardening canaries for `pre.010`. @@ -382,17 +382,39 @@ fn v0_3_12_pre_005_block_and_continuity_shapes_are_bounded_redacted_and_raw_sepa } #[test] -fn v0_3_12_pre_002_runtime_resource_contract_performs_no_live_io_or_source_spawn() { +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 forbidden in ["open_standard_subscribe", "next_update", "get_block_observed", "tokio::spawn", "JoinSet"] { - assert!(!resources.contains(forbidden), "runtime-resource contract opened premature live source behavior: {forbidden}"); + 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!(resources.contains("#[cfg(test)]\nasync fn hydrate_yellowstone_signal")); - assert!(resources.contains("get_transaction_observed")); assert!(runtime.contains("start_with_runtime_resources")); - for forbidden in ["open_standard_subscribe", "next_update", "get_transaction_observed", "get_block_observed"] { - assert!(!runtime.contains(forbidden), "pre.002 runtime start opened premature live behavior: {forbidden}"); + assert!(runtime.contains("source.run(source_settings, stop_receiver, admission_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}"); + } + return; +} + +#[test] +fn v0_3_12_pre_006_source_coalescence_is_bounded_stop_preemptible_and_redacted() { + let resources = include_str!("../src/runtime_resources.rs"); + for required in [ + "max_in_flight: settings.persistence_concurrency()", + "max_pending_signals: crate::MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY", + "pending_signal_count", + "stop_receiver.changed()", + "tasks.abort_all()", + "source.hydration_pending_saturated", + "source.hydration_task_join_failed", + "hydration_transport_error(error.code())", + "source_transport_error(error.code())", + ] { + assert!(resources.contains(required), "required pre.006 bounded/stop-safe source guard missing: {required}"); + } + for forbidden in ["remote_message", "response_body", "dead_error", "TransactionStatus.error", ".error()", "Authorization", "Bearer "] { + assert!(!resources.contains(forbidden), "remote/secret material leaked into productive source: {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 e859bc1..acbf950 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: 5 +// version: 6 //! Release-completeness canaries through the `pre.010` public/release/security hardening tranche. @@ -104,7 +104,8 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { "pre_010_source_visibility_contract_uses_crate_root_for_shared_items", "pre_010_production_surface_has_no_historical_backfill_or_retriever_contract", "v0_3_12_pre_002_production_sources_keep_transport_confined_to_runtime_resources", - "v0_3_12_pre_002_runtime_resource_contract_performs_no_live_io_or_source_spawn", + "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", "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", ] { @@ -112,11 +113,12 @@ fn pre_010_external_hardening_suite_is_present_and_scoped() { } let dependency_boundary = include_str!("dependency_boundary.rs"); assert!(dependency_boundary.contains("pre_002_manifest_dependency_surface_is_exact")); - assert!(dependency_boundary.contains("v0_3_12_pre_002_source_surface_hardens_shutdown_and_faults_without_backend_or_premature_live_io")); + assert!(dependency_boundary.contains("v0_3_12_pre_006_source_surface_hardens_shutdown_and_faults_without_backend_edges")); + assert!(dependency_boundary.contains("v0_3_12_pre_006_productive_source_uses_transport_session_bounded_coalescence_and_existing_admission")); assert!(dependency_boundary.contains("v0_3_12_pre_002_manifest_opens_only_the_onchain_transport_live_source_edge")); - assert!(dependency_boundary.contains("v0_3_12_pre_003_transaction_and_status_adapters_are_private_offline_and_transport_facade_only")); + 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_offline_and_transport_facade_only")); + assert!(dependency_boundary.contains("v0_3_12_pre_005_block_and_continuity_adapters_remain_private_and_transport_facade_only")); 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")); 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 02cbc48..89a3951 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: 5 +// version: 6 fn grpc_endpoint(cluster: &str) -> std::option::Option { return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider"); @@ -1245,3 +1245,79 @@ async fn pre_004_status_provenance_and_future_source_timestamp_are_bounded_and_r assert!(!debug.contains(PRE_004_ZERO_SIGNATURE_TEXT)); return; } + +#[tokio::test(flavor = "current_thread")] +async fn pre_006_coalescence_key_merges_transaction_and_status_before_http_fanout() { + let source = match signal_source() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let settings = match pre_004_settings() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let transaction = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::Transaction, 91, 3, &["transaction-filter"], 0x44) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let status = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::TransactionStatus, 91, 3, &["status-filter"], 0x44) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let transaction_key = match super::hydration_key(&source, &transaction) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let status_key = match super::hydration_key(&source, &status) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert!(transaction_key == status_key); + let received_at = match ksp_store_lib::RawTimestamp::from_unix_millis(1_760_000_200_000) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let mut coordinator = super::RawTransactionIngestHydrationCoordinator::new(&settings); + if coordinator.queue_signal(&source, transaction, received_at).is_err() { + return; + } + if coordinator.queue_signal(&source, status, received_at).is_err() { + return; + } + assert_eq!(coordinator.pending.len(), 1); + assert_eq!(coordinator.pending_signal_count, 2); + assert_eq!(coordinator.tasks.len(), 1); + coordinator.abort_all().await; + assert_eq!(coordinator.pending_signal_count, 0); + assert!(coordinator.pending.is_empty()); + assert!(coordinator.tasks.is_empty()); + return; +} + +#[test] +fn pre_006_coalescence_key_separates_network_signature_and_commitment() { + let source = match signal_source() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let first = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::Transaction, 12, 1, &["same"], 0x10) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let second = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::Block, 12, 1, &["same"], 0x11) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let first_key = match super::hydration_key(&source, &first) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let second_key = match super::hydration_key(&source, &second) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert!(first_key != second_key); + assert_eq!(first_key.commitment, "confirmed"); + assert_eq!(first_key.network.as_str(), "devnet"); + return; +} diff --git a/deltas/0.3.12/pre.006.md b/deltas/0.3.12/pre.006.md new file mode 100644 index 0000000..1dc8c01 --- /dev/null +++ b/deltas/0.3.12/pre.006.md @@ -0,0 +1,219 @@ + + + +# Delta `0.3.12-pre.006` — source Yellowstone productive + coalescence bornée + +## Base requise + +```text +0.3.12-pre.005-fix.001 +workspace.package.version = 0.3.12-pre.5.fix.1 +``` + +Le gate opérateur communiqué pour cette base est entièrement vert : audits Rust/Markdown, `cargo check --workspace`, Clippy strict et 56 tests Worker avec toutes les suites d'intégration associées. + +## Objectif + +Ouvrir la première source Yellowstone réellement productive en réutilisant exclusivement les moteurs déjà qualifiés : + +```text +Transport Yellowstone session +adapters Transaction/Status/Block +hydration HTTP getTransaction observed +coalescence bornée +RawTransactionIngress existant +SourceTasks JoinSet existant +supervisor/admission/persistence existants +``` + +La tranche ne crée ni processing frontier, ni politique Worker de reconnect/from_slot/ReplayInfo, ni repair historique. + +## Version + +```text +workspace.package.version = 0.3.12-pre.6 +``` + +## Source task productive + +`RawTransactionIngestWorker::start_with_runtime_resources` consomme maintenant les ressources validées et ajoute exactement un child source au `JoinSet` privé existant. + +Le child appelle uniquement la façade Transport : + +```text +YellowstoneGrpcChannel::open_standard_subscribe +SolanaYellowstoneGrpcSubscribeSession::next_update +SolanaYellowstoneGrpcSubscribeSession::close +``` + +Aucun client `tonic`, `reqwest`, proto Yellowstone ou SDK provider n'est instancié par le Worker. + +L'ouverture est stop-preemptible. Après stop/fault, les hydrations privées sont abort/join et la session est fermée via Transport. Toute erreur distante est réduite à un `ErrorCode` KSP sûr avant de remonter au supervisor, qui conserve son mapping stable `source_failed`. + +## Activation des adapters + +Les adapters `Transaction`, `TransactionStatus`, `Block`, `BlockMeta` et `Slot` sortent de `#[cfg(test)]` car ils possèdent désormais un consumer productif réel, conformément à `RUST-API-008`. + +Le routage est : + +```text +Transaction/TransactionStatus/Block -> candidat hydration +BlockMeta/Slot -> continuity-only, aucune admission RAW +Account/Entry/Ping/Pong -> ignorés par le vertical RAW +``` + +Le wrapper fixture `hydrate_yellowstone_signal` reste test-only ; le chemin productif sépare fetch HTTP et finalisation afin de permettre le fan-out coalescé. + +## Coalescence + +La clé est exactement `(network, signature, commitment)`. + +Un `RawTransactionIngestHydrationCoordinator` privé possède : + +```text +BTreeMap pending +JoinSet hydration +max_in_flight = settings.persistence_concurrency() +max_pending_signals = MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY +``` + +Des signaux Transaction/Status/Block identiques partagent donc un seul `getTransaction` in-flight. À la complétion, la réponse HTTP observée est finalisée séparément pour chaque signal afin de conserver family/filter/observed_at et donc des provenances distinctes. + +Aucun cache terminé n'est conservé. La saturation applique de la backpressure au stream et ne provoque aucun drop silencieux. + +## Hydration/admission + +Le fetch reste strictement : + +```text +get_transaction_observed +base64 +confirmed | finalized +maxSupportedTransactionVersion = 0 +``` + +La finalisation conserve tous les guards `pre.004` : network, source route, slot, transaction index, signature wire et provenance composite depuis le winner HTTP réel. `null` ne produit aucun ingress. + +Les ingresses disponibles passent uniquement par le sender central créé par `RawTransactionAdmission::new`; la persistence Normal reste inchangée. + +## Continuité + +`BlockMeta` et `Slot` sont réellement reçus/projetés dans le source task, mais restent continuity-only. Aucune frontier n'est maintenue et aucun gap n'est interprété dans cette tranche. + +Le snapshot reconnect/replay de Transport n'est pas encore consommé par le Worker ; cette responsabilité reste `pre.008`. + +## API / dépendances + +```text +aucun nouvel export public +aucun nouveau module +aucune nouvelle dépendance Cargo +Worker -> Transport reste l'unique edge live +Worker -> Config/Backfill/Store backend reste interdit +``` + +## Canaris ajoutés/ajustés + +```text +source runtime réellement supervisée par SourceTasks +open_standard_subscribe/next_update/close présents +coalescence key network/signature/commitment +Transaction + Status identiques -> une pending key / une hydration task +bounded pending/in-flight +stop preemptible +abort/join hydration +aucun getBlock/get_block_observed +aucun remote error/dead_error/secret +public API et module inventory inchangés +anciens canaris pre.003-pre.005 ajustés de offline -> private/Transport-only +``` + +## Décisions prises + +```text +coalescence productive dans le Worker, retry réseau reste Transport +concurrence hydration bornée par le budget de concurrence runtime déjà validé +pending total borné par la borne maximale d'admission publiée +fan-out après fetch partagé, avant admission +source close via Transport uniquement +frontier/reconnect/replay non anticipés +``` + +## Questions ouvertes + +Aucune question ouverte ne bloque `pre.006`. + +Restent réservés : + +```text +processing frontier run-local : pre.007 +reconnect/from_slot/ReplayInfo : 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.006.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/tests/dependency_boundary.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.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 le confinement Transport, l'absence de `get_block_observed`, l'absence de dépendances directes interdites, la clé de coalescence et l'inventaire public/module inchangé. + +## 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 processing frontier +pas d'interprétation reconnect/replay Worker +pas de from_slot Worker +pas de ReplayInfo Worker +pas de détection de gap de rétention +pas de repair/backfill implicite +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 3a06722..ecd7e27 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 @@ -1128,3 +1128,181 @@ cargo clippy --workspace --all-targets --all-features -- -D warnings cargo test -p ksp-worker-raw-transaction-ingest-lib ``` +## 47. Gate opérateur `pre.005-fix.001` + +Le gate opérateur communiqué pour `0.3.12-pre.5.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, 809 fichiers) +cargo check --workspace: PASS +cargo clippy --workspace --all-targets --all-features -- -D warnings: PASS +cargo test -p ksp-worker-raw-transaction-ingest-lib: PASS +56 unit tests Worker: PASS +6 dependency-boundary tests: PASS +12 hardening tests: PASS +8 public-api tests: PASS +3 release-completeness tests: PASS +Doc-tests: PASS +``` + +Cette base ferme donc `pre.005` et son fix de conversion `YellowstoneTransactionSignature::as_bytes()` avant ouverture de la source productive. + +## 48. `pre.006` — source Yellowstone productive + +`start_with_runtime_resources` ne délègue plus au foundation sans source. Après les validations synchrones Store/settings/runtime-resources déjà qualifiées, il : + +```text +consomme RawTransactionIngestRuntimeResources +extrait exactement une RawTransactionIngestYellowstoneSource +clone les settings nécessaires au child privé +spawn exactement un child dans le SourceTasks JoinSet existant +laisse le supervisor 0.3.11 posséder stop/drain/fault +``` + +Le child ouvre la session uniquement via : + +```text +YellowstoneGrpcChannel::open_standard_subscribe +SolanaYellowstoneGrpcSubscribeSession::next_update +SolanaYellowstoneGrpcSubscribeSession::close +``` + +Le Worker ne crée aucun `tonic::Channel`, client Geyser, client Reqwest ou actor gRPC parallèle. Transport reste propriétaire des metadata, credentials, ping/pong, message-size bounds et reconnect interne. + +Le stop privé est prioritaire pendant l'ouverture, pendant la lecture de stream et pendant l'envoi vers l'admission. Un stop avant ouverture abandonne le futur d'ouverture sans session orpheline. À la sortie, les hydrations privées sont abort/join puis la session Transport est fermée via son close borné. + +## 49. Routage productif des updates + +Le dispatch productif est désormais : + +```text +Transaction -> 1 RawTransactionIngestSourceSignal -> hydration +TransactionStatus -> 1 RawTransactionIngestSourceSignal -> hydration +Block -> N RawTransactionIngestSourceSignal dans l'ordre source -> hydration +BlockMeta -> RawTransactionIngestContinuitySignal continuity-only +Slot -> RawTransactionIngestContinuitySignal continuity-only +Account/Entry/Ping/Pong -> aucun RAW Worker +``` + +`BlockMeta` et `Slot` restent délibérément sans admission RAW. `dead_error` et `TransactionStatus.error` ne sont pas lus. Aucun `getBlock`/`get_block_observed` n'est ajouté. + +Les adapters `pre.003-pre.005`, précédemment gardés sous `#[cfg(test)]` faute de consumer productif, sont maintenant compilés dans le build normal. Le wrapper fixture `hydrate_yellowstone_signal` reste test-only ; la production utilise les deux étapes séparées `fetch_yellowstone_hydration` puis `finalize_yellowstone_hydration`. + +## 50. Coalescence bornée d'hydration + +La clé privée est exactement : + +```text +network +signature +commitment +``` + +Elle ne contient ni family ni filter. Ainsi `Transaction`, `TransactionStatus` et `Block` portant la même signature/network/commitment partagent au plus un `getTransaction` in-flight. + +Le coordinator privé conserve : + +```text +BTreeMap +JoinSet +max_in_flight = persistence_concurrency validée +max_pending_signals = MAX_RAW_TRANSACTION_INGEST_ADMISSION_QUEUE_CAPACITY +``` + +Le nombre de tâches HTTP simultanées est donc borné par le réglage existant de concurrence, sans nouveau knob public. Le nombre de signaux coalescés/pending reste borné par la borne maximale déjà publiée du Worker. Quand cette borne est atteinte, la lecture Yellowstone est backpressurée ; un dépassement à l'intérieur d'une update composite produit un fault source sûr au lieu d'un drop silencieux. + +À la complétion d'un unique fetch HTTP, chaque signal coalescé est finalisé séparément. Le fan-out reconstruit donc sa propre provenance : + +```text +transaction_get_transaction +status_get_transaction +block_get_transaction +filter_id direct ou fingerprint propre au signal +observed_at propre au signal quand représentable +``` + +Store idempotence reste ensuite la seconde barrière. Aucun cache de déduplication terminé/non borné n'est conservé. + +## 51. Hydration productive et admission + +Le fetch utilise toujours le contrat `pre.004` : + +```text +get_transaction_observed +encoding = base64 +commitment = confirmed | finalized de la source +maxSupportedTransactionVersion = 0 +``` + +Le résultat observé est partagé uniquement entre les signaux ayant la même clé. Chaque finalisation réapplique avant admission : + +```text +network cohérent +route Yellowstone cohérente +slot HTTP == slot signal +transaction_index cohérent lorsqu'il est présent des deux côtés +signature embarquée == signature signal +provenance composite sûre depuis le winner HTTP réel +``` + +`null` reste `Missing` : aucun ingress n'est créé. Une erreur Transport est réduite à son `ErrorCode` KSP sûr. Aucun texte distant n'entre dans l'erreur Worker. + +Les ingresses disponibles sont envoyés uniquement via le `tokio::sync::mpsc::Sender` déjà créé par `RawTransactionAdmission::new`. Aucun second canal d'admission ou chemin Store n'est introduit. + +## 52. Canaris `pre.006` + +Les nouveaux canaris vérifient au minimum : + +```text +start_with_runtime_resources -> source.run -> children.spawn +open_standard_subscribe/next_update/close présents uniquement via la façade Transport +get_transaction_observed reste l'unique hydration HTTP +aucun get_block_observed +aucun Config/Backfill/backend/reqwest/tonic/proto direct +coalescence key = network/signature/commitment +Transaction + Status de même signature -> une seule entrée pending et une seule tâche HTTP in-flight +signature différente -> clé différente +pending et in-flight bornés +stop preemptible pendant admission +abort_all + join des hydrations privées +aucun remote error/dead_error/secret dans la source productive +public root inchangé +module inventory inchangé +``` + +## 53. Non-claims `pre.006` + +`pre.006` ne prétend pas encore : + +```text +maintenir une processing frontier +interpréter le snapshot reconnect Transport +modifier from_slot au niveau Worker +interpréter SubscribeReplayInfo +prouver une continuité blockchain +fault sur gap de rétention prouvé +faire un repair/backfill implicite +qualifier les races duplicate storm/Store lent finales de pre.009 +réussir un smoke provider live sans credentials opérateur +``` + +Ces responsabilités restent respectivement `pre.007`, `pre.008`, `pre.009` et `pre.011` conformément au plan. + +## 54. Gate opérateur requis pour `pre.006` + +```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 donc revendiqué.