From 8410b2f4efc0e5a4ef3b79dc4f95dd012dd472e2 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Wed, 9 Sep 2026 16:22:09 +0200 Subject: [PATCH] v0.3.12-pre.007-fix.001 --- Cargo.toml | 4 +- .../src/lib.rs | 4 +- .../src/runtime.rs | 22 +--- .../src/runtime_resources.rs | 97 ++++------------ .../src/snapshot.rs | 6 +- .../tests/dependency_boundary.rs | 31 +++-- .../tests/hardening.rs | 23 ++-- .../tests/release_completeness.rs | 5 +- .../unit_tests/runtime_resources.rs | 60 ++++------ deltas/0.3.12/pre.007-fix.001.md | 107 ++++++++++++++++++ ...0_3_12_YELLOWSTONE_HYDRATION_CONTINUITY.md | 24 +++- 11 files changed, 223 insertions(+), 160 deletions(-) create mode 100644 deltas/0.3.12/pre.007-fix.001.md diff --git a/Cargo.toml b/Cargo.toml index a90f235..e79af2e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 529 +# version: 530 [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.7" +version = "0.3.12-pre.7.fix.1" 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 7ae24bc..1868d71 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/lib.rs @@ -103,7 +103,7 @@ pub(crate) use self::persistence::RawTransactionIngestPersistenceOutcome; pub(crate) use self::persistence::RawTransactionIngestPersistencePort; /// Persists one already-canonical Worker acquisition through the private Store port in `Normal` mode. pub(crate) use self::persistence::persist_raw_transaction_ingest_acquisition; -/// Private latest-value publisher and checked counter owner shared by the Worker supervisor. -pub(crate) use self::snapshot::RawTransactionIngestSnapshotPublisher; /// Private latest-value processing-frontier projection emitted by the productive source task. pub(crate) use self::snapshot::RawTransactionIngestProcessingFrontierProjection; +/// Private latest-value publisher and checked counter owner shared by the Worker supervisor. +pub(crate) use self::snapshot::RawTransactionIngestSnapshotPublisher; 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 02c36a9..2f68bbb 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: 10 +// version: 11 type PersistencePort = std::sync::Arc; type PersistenceTasks = tokio::task::JoinSet>; @@ -189,7 +189,6 @@ async fn drain_admission_and_persistence( persistence: &mut PersistenceTasks, port: &std::option::Option, snapshots: &mut crate::RawTransactionIngestSnapshotPublisher, - processing_frontier_receiver: &mut std::option::Option, ) -> std::option::Option { let mut fault = std::option::Option::None; admission.close(); @@ -448,13 +447,7 @@ where Spawner: FnOnce(&mut SourceTasks, tokio::sync::watch::Receiver, tokio::sync::mpsc::Sender) + std::marker::Send + 'static, { - return start_foundation_with_port_source_spawner_and_frontier( - settings, - runtime, - port, - std::option::Option::None, - source_spawner, - ); + return start_foundation_with_port_source_spawner_and_frontier(settings, runtime, port, std::option::Option::None, source_spawner); } fn start_foundation_with_port_source_spawner_and_frontier( @@ -480,15 +473,7 @@ where let (stop_sender, stop_receiver) = tokio::sync::watch::channel(false); let (snapshots, snapshot_source) = crate::RawTransactionIngestSnapshotPublisher::new(&settings, &lifecycle); let handle = crate::RawTransactionIngestHandle { snapshots: snapshot_source, stop_sender, stop_token }; - std::mem::drop(runtime.spawn(run_supervisor( - settings, - lifecycle, - port, - stop_receiver, - snapshots, - processing_frontier_receiver, - source_spawner, - ))); + std::mem::drop(runtime.spawn(run_supervisor(settings, lifecycle, port, stop_receiver, snapshots, processing_frontier_receiver, source_spawner))); return std::result::Result::Ok(handle); } @@ -522,6 +507,7 @@ async fn supervise_until_stop( persistence: &mut PersistenceTasks, port: &std::option::Option, snapshots: &mut crate::RawTransactionIngestSnapshotPublisher, + processing_frontier_receiver: &mut std::option::Option, ) -> std::option::Option { let mut admission_open = true; loop { 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 2a27044..86cdf78 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 @@ -188,10 +188,7 @@ trait RawTransactionIngestYellowstoneBlockView { fn transaction_count(&self) -> usize; - fn transaction_identity( - &self, - position: usize, - ) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)>; + fn transaction_identity(&self, position: usize) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)>; } impl RawTransactionIngestYellowstoneBlockView for ksp_onchain_transport_lib::YellowstoneBlockUpdate { @@ -211,10 +208,7 @@ impl RawTransactionIngestYellowstoneBlockView for ksp_onchain_transport_lib::Yel return ksp_onchain_transport_lib::YellowstoneBlockUpdate::transactions(self).len(); } - fn transaction_identity( - &self, - position: usize, - ) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)> { + fn transaction_identity(&self, position: usize) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)> { let transaction = match ksp_onchain_transport_lib::YellowstoneBlockUpdate::transactions(self).get(position) { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.block_transaction_missing")), @@ -292,9 +286,7 @@ impl RawTransactionIngestYellowstoneContinuityView for ksp_onchain_transport_lib impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate)> for RawTransactionIngestSourceSignal { - fn from( - value: (&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate), - ) -> Self { + fn from(value: (&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionUpdate)) -> Self { return project_yellowstone_signal(value.0, value.1); } } @@ -302,9 +294,7 @@ impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onc impl std::convert::From<(&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate)> for RawTransactionIngestSourceSignal { - fn from( - value: (&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate), - ) -> Self { + fn from(value: (&crate::RawTransactionIngestYellowstoneSource, &ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate)) -> Self { return project_yellowstone_signal(value.0, value.1); } } @@ -591,9 +581,7 @@ fn ingestion_filter_count(request: &ksp_onchain_transport_lib::YellowstoneSubscr return request.transaction_filter_count().saturating_add(request.transaction_status_filter_count()).saturating_add(request.block_filter_count()); } -fn matched_filter_direct_id( - filters: &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName], -) -> std::option::Option { +fn matched_filter_direct_id(filters: &[ksp_onchain_transport_lib::YellowstoneSubscribeFilterName]) -> std::option::Option { let mut names = filters.iter().map(ksp_onchain_transport_lib::YellowstoneSubscribeFilterName::as_str).collect::>(); names.sort_unstable(); names.dedup(); @@ -724,24 +712,14 @@ fn route_yellowstone_update( std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - return coordinator.queue_signal( - source, - RawTransactionIngestSourceSignal::from((source, value.as_ref())), - received_at, - processing_frontier, - ); + return coordinator.queue_signal(source, RawTransactionIngestSourceSignal::from((source, value.as_ref())), received_at, processing_frontier); }, ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::TransactionStatus(value) => { let received_at = match current_raw_timestamp() { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - return coordinator.queue_signal( - source, - RawTransactionIngestSourceSignal::from((source, &value)), - received_at, - processing_frontier, - ); + return coordinator.queue_signal(source, RawTransactionIngestSourceSignal::from((source, &value)), received_at, processing_frontier); }, ksp_onchain_transport_lib::YellowstoneSubscribeUpdate::Block(value) => { let received_at = match current_raw_timestamp() { @@ -753,9 +731,7 @@ fn route_yellowstone_update( std::result::Result::Err(error) => return std::result::Result::Err(error), }; for signal in signals { - if let std::result::Result::Err(error) = - coordinator.queue_signal(source, signal, received_at, processing_frontier) - { + if let std::result::Result::Err(error) = coordinator.queue_signal(source, signal, received_at, processing_frontier) { return std::result::Result::Err(error); } } @@ -1064,15 +1040,12 @@ impl RawTransactionIngestHydrationCoordinator { fn start_hydrations(&mut self, source: &crate::RawTransactionIngestYellowstoneSource) -> ksp_core_lib::Result<()> { while self.tasks.len() < self.max_in_flight { - let key = self - .pending - .iter() - .find_map(|(key, pending)| { - if pending.in_flight { - return std::option::Option::None; - } - return std::option::Option::Some(key.clone()); - }); + let key = self.pending.iter().find_map(|(key, pending)| { + if pending.in_flight { + return std::option::Option::None; + } + return std::option::Option::Some(key.clone()); + }); let key = match key { std::option::Option::Some(value) => value, std::option::Option::None => break, @@ -1276,14 +1249,7 @@ fn finalize_yellowstone_hydration( if embedded_signature != signal.signature { return std::result::Result::Err(crate::runtime_error("hydration.signature_mismatch")); } - let provenance = build_hydration_provenance( - settings, - &signal, - observed.provider().as_str(), - observed.endpoint_name(), - commitment, - received_at, - ); + let provenance = build_hydration_provenance(settings, &signal, observed.provider().as_str(), observed.endpoint_name(), commitment, received_at); let provenance = match provenance { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), @@ -1304,12 +1270,7 @@ fn finalize_yellowstone_hydration( }), map_hydration_wire_field(transaction.transaction_index(), |value| return *value), ); - return std::result::Result::Ok(std::option::Option::Some(crate::RawTransactionIngress { - material, - network: signal.network, - provenance, - source_key, - })); + return std::result::Result::Ok(std::option::Option::Some(crate::RawTransactionIngress { material, network: signal.network, provenance, source_key })); } #[cfg(test)] @@ -1334,14 +1295,7 @@ async fn hydrate_yellowstone_signal( std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let fetched = fetch_yellowstone_hydration( - source.http_pool.clone(), - source.hydration_role.clone(), - source.network.clone(), - key, - commitment, - ) - .await; + let fetched = fetch_yellowstone_hydration(source.http_pool.clone(), source.hydration_role.clone(), source.network.clone(), key, commitment).await; let fetched = match fetched { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), @@ -1391,17 +1345,12 @@ fn build_hydration_provenance( std::result::Result::Err(error) => return std::result::Result::Err(error), }, }; - let mut provenance = ksp_store_lib::RawAcquisitionProvenance::new( - provider, - protocol, - acquisition_method, - ksp_store_lib::RawAcquisitionOrigin::Live, - received_at, - ) - .with_endpoint_id(endpoint_id) - .with_commitment(commitment) - .with_capture_session_id(capture_session) - .with_filter_id(filter_id); + let mut provenance = + ksp_store_lib::RawAcquisitionProvenance::new(provider, protocol, acquisition_method, ksp_store_lib::RawAcquisitionOrigin::Live, received_at) + .with_endpoint_id(endpoint_id) + .with_commitment(commitment) + .with_capture_session_id(capture_session) + .with_filter_id(filter_id); if let std::option::Option::Some(observed_at) = representable_observed_at(signal.created_at, received_at) { provenance = match provenance.try_with_observed_at(observed_at) { std::result::Result::Ok(value) => value, diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs index e6a658f..14ec27d 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs @@ -16,7 +16,11 @@ pub(crate) struct RawTransactionIngestProcessingFrontierProjection { impl crate::RawTransactionIngestProcessingFrontierProjection { /// Returns the empty run-local processing projection used before the source observes work. pub(crate) const fn empty() -> Self { - return Self { hydration_pending: 0, processing_frontier_slot: std::option::Option::None, oldest_pending_slot: std::option::Option::None }; + return Self { + hydration_pending: 0, + processing_frontier_slot: std::option::Option::None, + oldest_pending_slot: std::option::Option::None, + }; } /// Creates one run-local processing projection from bounded source-owned state. 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 c1ca276..6ff30c6 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 @@ -8,7 +8,19 @@ fn pre_002_manifest_dependency_surface_is_exact() { let manifest = include_str!("../Cargo.toml"); let dependencies = dependency_section(manifest); let names = manifest_dependency_names(dependencies); - assert_eq!(names, vec!["ksp-core-lib", "ksp-logging-lib", "ksp-onchain-transport-lib", "ksp-raw-transaction-lib", "ksp-store-lib", "ksp-worker-api", "sha2", "tokio",],); + assert_eq!( + names, + vec![ + "ksp-core-lib", + "ksp-logging-lib", + "ksp-onchain-transport-lib", + "ksp-raw-transaction-lib", + "ksp-store-lib", + "ksp-worker-api", + "sha2", + "tokio", + ], + ); for required in [ "ksp-core-lib = { path = \"../ksp-core-lib\" }", "ksp-logging-lib = { path = \"../ksp-logging-lib\" }", @@ -81,15 +93,14 @@ fn v0_3_12_pre_003_transaction_and_status_adapters_are_private_and_transport_fac ] { assert!(!resources.contains(forbidden) && !root.contains(forbidden), "pre.003 crossed a private/offline boundary: {forbidden}"); } - let status_impl = match resources - .split_once("impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate") - { - std::option::Option::Some((_, tail)) => match tail.split_once("impl std::convert::From<") { - std::option::Option::Some((value, _)) => value, - std::option::Option::None => tail, - }, - std::option::Option::None => "", - }; + let status_impl = + match resources.split_once("impl RawTransactionIngestYellowstoneSignalView for ksp_onchain_transport_lib::YellowstoneTransactionStatusUpdate") { + std::option::Option::Some((_, tail)) => match tail.split_once("impl std::convert::From<") { + std::option::Option::Some((value, _)) => value, + std::option::Option::None => tail, + }, + std::option::Option::None => "", + }; assert!(!status_impl.contains("::error(self)"), "TransactionStatus remote error material must not enter the private signal"); return; } 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 fb881aa..150d012 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs @@ -121,7 +121,16 @@ fn v0_3_12_pre_002_manifest_dependency_surface_opens_only_transport_and_remains_ } assert_eq!( normal, - std::collections::BTreeSet::from(["ksp-core-lib", "ksp-logging-lib", "ksp-onchain-transport-lib", "ksp-raw-transaction-lib", "ksp-store-lib", "ksp-worker-api", "sha2", "tokio",]) + std::collections::BTreeSet::from([ + "ksp-core-lib", + "ksp-logging-lib", + "ksp-onchain-transport-lib", + "ksp-raw-transaction-lib", + "ksp-store-lib", + "ksp-worker-api", + "sha2", + "tokio", + ]) ); assert!(dev.is_empty()); assert!(build.is_empty()); @@ -323,14 +332,7 @@ fn v0_3_12_pre_004_hydration_provenance_and_remote_material_are_bounded_and_reda ] { assert!(resources.contains(required), "required pre.004 bounded provenance/mismatch guard missing: {required}"); } - for forbidden in [ - "source_payload_hash", - "source_payload_size_bytes", - "HTTP-SECRET-CANARY", - "GRPC-SECRET-CANARY", - "TransactionStatus.error", - ".error()", - ] { + for forbidden in ["source_payload_hash", "source_payload_size_bytes", "HTTP-SECRET-CANARY", "GRPC-SECRET-CANARY", "TransactionStatus.error", ".error()"] { assert!(!resources.contains(forbidden), "pre.004 retained forbidden remote/source material: {forbidden}"); } return; @@ -383,7 +385,8 @@ fn v0_3_12_pre_005_block_and_continuity_shapes_are_bounded_redacted_and_raw_sepa fn v0_3_12_pre_006_runtime_resource_contract_opens_one_supervised_transport_source() { let resources = include_str!("../src/runtime_resources.rs"); let runtime = include_str!("../src/runtime.rs"); - for required in ["open_standard_subscribe", "next_update", "get_transaction_observed", "RawTransactionIngestHydrationCoordinator", "session.close().await"] { + for required in ["open_standard_subscribe", "next_update", "get_transaction_observed", "RawTransactionIngestHydrationCoordinator", "session.close().await"] + { assert!(resources.contains(required), "productive runtime-resource source behavior missing: {required}"); } assert!(runtime.contains("start_with_runtime_resources")); 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 b17ca14..1bd7371 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 @@ -32,7 +32,10 @@ fn pre_010_production_module_inventory_is_exact() -> std::io::Result<()> { } } names.sort_unstable(); - assert_eq!(names, std::vec!["admission.rs", "error.rs", "identity.rs", "lib.rs", "persistence.rs", "runtime.rs", "runtime_resources.rs", "settings.rs", "snapshot.rs",]); + assert_eq!( + names, + std::vec!["admission.rs", "error.rs", "identity.rs", "lib.rs", "persistence.rs", "runtime.rs", "runtime_resources.rs", "settings.rs", "snapshot.rs",] + ); return std::result::Result::Ok(()); } 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 2b540e8..418e14d 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: 7 +// version: 8 fn grpc_endpoint(cluster: &str) -> std::option::Option { return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider"); @@ -62,7 +62,9 @@ fn http_pool(cluster: &str, role_name: &str, request_kind: &str) -> std::option: }; } -fn transaction_request(commitment: std::option::Option) -> std::option::Option { +fn transaction_request( + commitment: std::option::Option, +) -> std::option::Option { let name = match ksp_onchain_transport_lib::YellowstoneSubscribeFilterName::new("tx-fixture") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::option::Option::None, @@ -134,10 +136,7 @@ impl super::RawTransactionIngestYellowstoneBlockView for BlockViewFixture { return self.transactions.len(); } - fn transaction_identity( - &self, - position: usize, - ) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)> { + fn transaction_identity(&self, position: usize) -> ksp_core_lib::Result<(ksp_onchain_transport_lib::YellowstoneTransactionSignature, u64)> { return match self.transactions.get(position) { std::option::Option::Some(value) => std::result::Result::Ok(*value), std::option::Option::None => std::result::Result::Err(crate::runtime_error("fixture.block_transaction_missing")), @@ -219,12 +218,7 @@ const PRE_004_OTHER_TRANSACTION_BASE64: &str = "AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEB const PRE_004_ZERO_SIGNATURE_TEXT: &str = "1111111111111111111111111111111111111111111111111111111111111111"; const PRE_004_ZERO_TRANSACTION_BASE64: &str = "AQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"; -fn http_pool_for_url( - url: &str, - cluster: &str, - endpoint_name: &str, - provider: &str, -) -> std::option::Option { +fn http_pool_for_url(url: &str, cluster: &str, endpoint_name: &str, provider: &str) -> std::option::Option { let url = match ksp_onchain_transport_lib::HttpEndpointUrl::parse(url) { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::option::Option::None, @@ -349,11 +343,8 @@ fn serve_http_once(body: std::string::String) -> std::io::Result<(std::string::S std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let response = std::format!( - "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", - body.len(), - body, - ); + let response = + std::format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), body,); if let std::result::Result::Err(error) = std::io::Write::write_all(&mut stream, response.as_bytes()) { return std::result::Result::Err(error); } @@ -441,10 +432,7 @@ async fn pre_002_source_accepts_matching_confirmed_transaction_and_get_transacti #[tokio::test(flavor = "current_thread")] async fn pre_002_source_rejects_processed_and_implicit_commitment() { - for commitment in [ - std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Processed), - std::option::Option::None, - ] { + for commitment in [std::option::Option::Some(ksp_onchain_transport_lib::SolanaCommitment::Processed), std::option::Option::None] { let endpoint = match grpc_endpoint("devnet") { std::option::Option::Some(value) => value, std::option::Option::None => return, @@ -637,8 +625,8 @@ async fn pre_003_transaction_signal_fixture_preserves_exact_source_neutral_ident assert_eq!( signal.matched_filter_fingerprint, [ - 0x20, 0x83, 0x1d, 0xc8, 0x37, 0x8b, 0x60, 0x70, 0xf9, 0x1d, 0x9d, 0xe6, 0xd3, 0xa1, 0xd2, 0xb3, 0xd7, 0x72, 0x98, - 0xe3, 0xad, 0xa7, 0x6c, 0x01, 0xfa, 0xa3, 0x08, 0x26, 0xff, 0xab, 0xcd, 0x84, + 0x20, 0x83, 0x1d, 0xc8, 0x37, 0x8b, 0x60, 0x70, 0xf9, 0x1d, 0x9d, 0xe6, 0xd3, 0xa1, 0xd2, 0xb3, 0xd7, 0x72, 0x98, 0xe3, 0xad, 0xa7, 0x6c, 0x01, + 0xfa, 0xa3, 0x08, 0x26, 0xff, 0xab, 0xcd, 0x84, ] ); assert_eq!(signal.created_at, std::option::Option::Some(super::RawTransactionIngestSourceTimestamp { nanos: 456_000_000, seconds: 1_760_000_123 })); @@ -883,10 +871,7 @@ fn pre_005_slot_status_projection_is_exact_and_source_neutral() { (ksp_onchain_transport_lib::YellowstoneSlotStatus::Processed, super::RawTransactionIngestContinuityStatus::Processed), (ksp_onchain_transport_lib::YellowstoneSlotStatus::Confirmed, super::RawTransactionIngestContinuityStatus::Confirmed), (ksp_onchain_transport_lib::YellowstoneSlotStatus::Finalized, super::RawTransactionIngestContinuityStatus::Finalized), - ( - ksp_onchain_transport_lib::YellowstoneSlotStatus::FirstShredReceived, - super::RawTransactionIngestContinuityStatus::FirstShredReceived, - ), + (ksp_onchain_transport_lib::YellowstoneSlotStatus::FirstShredReceived, super::RawTransactionIngestContinuityStatus::FirstShredReceived), (ksp_onchain_transport_lib::YellowstoneSlotStatus::Completed, super::RawTransactionIngestContinuityStatus::Completed), (ksp_onchain_transport_lib::YellowstoneSlotStatus::CreatedBank, super::RawTransactionIngestContinuityStatus::CreatedBank), (ksp_onchain_transport_lib::YellowstoneSlotStatus::Dead, super::RawTransactionIngestContinuityStatus::Dead), @@ -988,10 +973,7 @@ async fn pre_004_observed_get_transaction_closes_signal_to_common_raw_ingress_wi assert_eq!(acquisition.transaction().reference().network().as_str(), "devnet"); assert_eq!(acquisition.transaction().reference().signature().as_bytes(), &[0_u8; 64]); assert_eq!(acquisition.transaction().slot(), 42); - assert_eq!( - acquisition.transaction().block_time().map(|value| return value.unix_millis()), - std::option::Option::Some(1_760_000_120_000), - ); + assert_eq!(acquisition.transaction().block_time().map(|value| return value.unix_millis()), std::option::Option::Some(1_760_000_120_000),); assert_eq!(acquisition.observation().provenance().provider().as_str(), "ys.fixture-provider:http.fixture-http-provider"); return; } @@ -1229,14 +1211,8 @@ async fn pre_004_status_provenance_and_future_source_timestamp_are_bounded_and_r std::option::Option::Some(value) => value, std::option::Option::None => return, }; - let mut signal = match pre_004_signal( - &source, - super::RawTransactionIngestSourceFamily::TransactionStatus, - 42, - 7, - &["filter-b", "filter-a", "filter-a"], - 0, - ) { + let mut signal = match pre_004_signal(&source, super::RawTransactionIngestSourceFamily::TransactionStatus, 42, 7, &["filter-b", "filter-a", "filter-a"], 0) + { std::option::Option::Some(value) => value, std::option::Option::None => return, }; @@ -1302,10 +1278,12 @@ async fn pre_006_coalescence_key_merges_transaction_and_status_before_http_fanou std::result::Result::Err(_) => return, }; let mut coordinator = super::RawTransactionIngestHydrationCoordinator::new(&settings); - if coordinator.queue_signal(&source, transaction, received_at).is_err() { + let (frontier_sender, _frontier_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); + let mut processing_frontier = super::RawTransactionIngestProcessingFrontierReporter::new(frontier_sender); + if coordinator.queue_signal(&source, transaction, received_at, &mut processing_frontier).is_err() { return; } - if coordinator.queue_signal(&source, status, received_at).is_err() { + if coordinator.queue_signal(&source, status, received_at, &mut processing_frontier).is_err() { return; } assert_eq!(coordinator.pending.len(), 1); diff --git a/deltas/0.3.12/pre.007-fix.001.md b/deltas/0.3.12/pre.007-fix.001.md new file mode 100644 index 0000000..22a6fd9 --- /dev/null +++ b/deltas/0.3.12/pre.007-fix.001.md @@ -0,0 +1,107 @@ + + + +# Delta `0.3.12-pre.007-fix.001` — correction de propagation de la processing frontier + +## Base requise + +```text +0.3.12-pre.007 +workspace.package.version = 0.3.12-pre.7 +``` + +Le gate opérateur de `pre.007` a validé les audits Rust/Markdown puis a échoué pendant `cargo check`/Clippy sur la propagation interne du receiver frontier et sur deux callsites test-only encore sur l'ancienne arité de `queue_signal`. + +## Objectif + +Corriger uniquement ces défauts d'intégration sans modifier le contrat fonctionnel de la processing frontier `pre.007`. + +## Version + +```text +workspace.package.version = 0.3.12-pre.7.fix.1 +``` + +## Modifications + +### Runtime supervisor + +`processing_frontier_receiver` est désormais passé à `supervise_until_stop`, qui est le seul chemin qui le consomme via `wait_processing_frontier` et `record_processing_frontier`. + +Le paramètre ajouté par erreur à `drain_admission_and_persistence` est retiré ; le drain reste inchangé fonctionnellement. + +### Canari coalescence + +Le test `pre_006_coalescence_key_merges_transaction_and_status_before_http_fanout` construit désormais un `RawTransactionIngestProcessingFrontierReporter` privé sur un `watch` local et le passe aux deux appels `queue_signal`. + +Le comportement testé reste identique : les signaux Transaction et TransactionStatus identiques coalescent vers une seule hydration en vol. + +## Fichiers modifiés + +```text +Cargo.toml +crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs +docs/validation/029-V0_3_12_YELLOWSTONE_HYDRATION_CONTINUITY.md +``` + +## Fichier ajouté + +```text +deltas/0.3.12/pre.007-fix.001.md +``` + +## Fichiers supprimés + +```text +aucun +``` + +## Invariants préservés + +Le correctif ne modifie pas : + +```text +définition pending / settled +compaction des slots +processing_frontier_slot +oldest_pending_slot +hydration_pending +source Yellowstone productive +coalescence key +hydration HTTP +admission / persistence +API publique +reconnect / from_slot / ReplayInfo / repair +``` + +## 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 +contrôle statique des callsites drain_admission_and_persistence +contrôle statique des callsites supervise_until_stop +contrôle statique des callsites queue_signal du canari de coalescence +reconstruction du delta sur base pre.007 +unzip -t de l'archive finale +``` + +## Validations non exécutées localement + +L'environnement d'assemblage ne fournit pas Cargo/rustc/rustfmt. Les gates Cargo doivent être rejoués par l'opérateur : + +```text +cargo fmt --all +cargo check --workspace +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test -p ksp-worker-raw-transaction-ingest-lib +``` + +## Décisions + +Aucune nouvelle décision fonctionnelle. + +## Questions ouvertes + +Aucune nouvelle question ouverte introduite par ce fix. 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 a8f70cc..73735e8 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 @@ -1446,3 +1446,25 @@ cargo tree --duplicates ``` Dans l'environnement d'assemblage, `cargo`, `rustc` et `rustfmt` ne sont pas installés ; aucun gate Cargo local n'est revendiqué. + +## 63. Gate opérateur `pre.007` et correctif `pre.007-fix.001` + +Le gate opérateur de `pre.007` a confirmé les audits Rust/Markdown, puis a révélé un défaut d'intégration purement interne avant exécution des nouveaux tests : + +```text +processing_frontier_receiver absent de la signature supervise_until_stop +processing_frontier_receiver ajouté par erreur au drain admission/persistence +deux appels test-only queue_signal encore sur l'ancienne arité +``` + +Le correctif `pre.007-fix.001` ne modifie pas le contrat de frontier. Il : + +```text +propage le receiver frontier uniquement vers supervise_until_stop, où il est réellement consommé +retire le receiver inutilisé de drain_admission_and_persistence +construit un reporter frontier test-only pour les deux appels queue_signal du canari de coalescence +``` + +Aucune définition pending/settled, compaction, projection snapshot, API publique, source Yellowstone, hydration, admission, persistence, reconnect ou replay n'est modifiée. + +Les validations Cargo de ce correctif restent à rejouer par l'opérateur.