From 822e8fcd863e97f329a8edfbe936325682cd6e23 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Tue, 8 Sep 2026 16:22:40 +0200 Subject: [PATCH] v0.3.11-pre.009 --- Cargo.toml | 4 +- .../src/admission.rs | 15 +- .../src/error.rs | 8 +- .../src/lib.rs | 6 +- .../src/runtime.rs | 176 +++++++---- .../src/snapshot.rs | 40 ++- .../tests/dependency_boundary.rs | 18 +- .../tests/public_api.rs | 11 +- .../unit_tests/admission.rs | 23 +- .../unit_tests/runtime.rs | 274 ++++++++++++++++-- .../unit_tests/snapshot.rs | 25 +- deltas/0.3.11/pre.009.md | 135 +++++++++ ...ANSACTION_INGEST_WORKER_FOUNDATION_PLAN.md | 6 +- ...AW_TRANSACTION_INGEST_WORKER_FOUNDATION.md | 105 ++++++- 14 files changed, 747 insertions(+), 99 deletions(-) create mode 100644 deltas/0.3.11/pre.009.md diff --git a/Cargo.toml b/Cargo.toml index db30dfa..90a4d37 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 510 +# version: 511 [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.11-pre.8.fix.1" +version = "0.3.11-pre.9" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/admission.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/admission.rs index 3318b59..ef9bd7b 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/admission.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/admission.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/admission.rs -// version: 2 +// version: 3 use sha2::Digest; // rust-rules: trait-import @@ -19,6 +19,8 @@ pub(crate) struct RawTransactionIngress { /// Receiver side of the bounded central RAW transaction admission queue owned by the Worker supervisor. pub(crate) struct RawTransactionAdmission { + backpressure_wait_observed: bool, + capacity: usize, receiver: tokio::sync::mpsc::Receiver, } @@ -27,7 +29,7 @@ impl crate::RawTransactionAdmission { #[must_use] pub(crate) fn new(capacity: usize) -> (crate::RawTransactionAdmission, tokio::sync::mpsc::Sender) { let (sender, receiver) = tokio::sync::mpsc::channel(capacity); - return (Self { receiver }, sender); + return (Self { backpressure_wait_observed: false, capacity, receiver }, sender); } /// Returns the latest receiver-side queue depth without consuming an ingress entry. @@ -47,6 +49,7 @@ impl crate::RawTransactionAdmission { &mut self, expected_network: &ksp_store_lib::RawNetworkId, ) -> ksp_core_lib::Result> { + self.backpressure_wait_observed = self.receiver.len() == self.capacity; let ingress = match self.receiver.recv().await { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Ok(std::option::Option::None), @@ -57,6 +60,14 @@ impl crate::RawTransactionAdmission { std::result::Result::Err(error) => std::result::Result::Err(error), }; } + + /// Takes and clears the source-neutral signal that the previous dequeue observed the bounded queue at full capacity. + #[must_use] + pub(crate) fn take_backpressure_wait_observed(&mut self) -> bool { + let observed = self.backpressure_wait_observed; + self.backpressure_wait_observed = false; + return observed; + } } fn canonicalize_ingress( diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/error.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/error.rs index cf2fe86..cadd1c2 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/error.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/error.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/error.rs -// version: 5 +// version: 6 /// Error code used when durable Store content conflicts with one admitted canonical RAW transaction. pub const ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT: ksp_core_lib::ErrorCode = @@ -7,12 +7,18 @@ pub const ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT: ksp_core_lib::Erro /// Error code used when one monotone RAW transaction ingest Worker counter or snapshot sequence cannot advance without wrapping. pub const ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("worker_raw_transaction_ingest", "counter_exhausted"); +/// Error code used when private Worker tasks cannot drain before the configured shutdown deadline. +pub const ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT: ksp_core_lib::ErrorCode = + ksp_core_lib::ErrorCode::new("worker_raw_transaction_ingest", "drain_timeout"); /// Error code used when the RAW transaction ingest Worker reaches an invalid runtime or lifecycle condition. pub const ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("worker_raw_transaction_ingest", "runtime_invalid"); /// Error code used when RAW transaction ingest Worker settings violate one bounded runtime invariant. pub const ERROR_CODE_RAW_TRANSACTION_INGEST_SETTINGS_INVALID: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("worker_raw_transaction_ingest", "settings_invalid"); +/// Error code used when one private source task terminates with a classified failure. +pub const ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED: ksp_core_lib::ErrorCode = + ksp_core_lib::ErrorCode::new("worker_raw_transaction_ingest", "source_failed"); /// Error code used when Store persistence fails for one admitted canonical RAW transaction. pub const ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED: ksp_core_lib::ErrorCode = ksp_core_lib::ErrorCode::new("worker_raw_transaction_ingest", "store_failed"); 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 95b02db..37cb340 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: 7 +// version: 8 #![warn(missing_docs)] #![deny(unreachable_pub)] @@ -25,10 +25,14 @@ mod snapshot; pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_CONTENT_CONFLICT; /// Error code used when one monotone Worker counter or snapshot sequence cannot advance without wrapping. pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED; +/// Error code used when private Worker tasks cannot drain before the configured shutdown deadline. +pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT; /// Error code used when the RAW transaction ingest Worker reaches an invalid runtime or lifecycle condition. pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID; /// Error code used when RAW transaction ingest Worker settings violate one bounded runtime invariant. pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_SETTINGS_INVALID; +/// Error code used when one private source task terminates with a classified failure. +pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED; /// Error code used when Store persistence fails for one admitted canonical RAW transaction. pub use self::error::ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED; /// Stable Worker kind code used by the continuous RAW transaction ingest vertical. 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 de0e846..35a7500 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs @@ -1,8 +1,9 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs -// version: 6 +// version: 7 type PersistencePort = std::sync::Arc; type PersistenceTasks = tokio::task::JoinSet>; +type SourceTasks = tokio::task::JoinSet>; /// Runtime-neutral boxed future resolving after one RAW transaction ingest Worker has fully reached a terminal lifecycle state. pub type RawTransactionIngestTerminalFuture<'a> = @@ -118,12 +119,15 @@ fn current_runtime_handle() -> ksp_core_lib::Result { }; } -async fn drain_children(children: &mut tokio::task::JoinSet<()>) -> std::option::Option { +async fn drain_children( + state: ksp_worker_api::WorkerState, + children: &mut SourceTasks, + snapshots: &mut crate::RawTransactionIngestSnapshotPublisher, +) -> std::option::Option { let mut fault = std::option::Option::None; while let std::option::Option::Some(joined) = children.join_next().await { - if joined.is_err() && fault.is_none() { - fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); - } + let current = source_completion(joined, state, 0, 0, snapshots); + fault = merge_fault(fault, current); } return fault; } @@ -137,9 +141,7 @@ async fn drain_persistence( let mut fault = std::option::Option::None; while let std::option::Option::Some(joined) = persistence.join_next().await { let current = persistence_completion(joined, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots); - if fault.is_none() { - fault = current; - } + fault = merge_fault(fault, current); } return fault; } @@ -161,15 +163,14 @@ async fn drain_admission_and_persistence( std::option::Option::Some(value) => persistence_completion(value, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots), std::option::Option::None => std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID), }; - if fault.is_none() { - fault = current; - } + fault = merge_fault(fault, current); } let received = admission.receive(settings.network()).await; + let backpressure_wait_observed = admission.take_backpressure_wait_observed(); match received { std::result::Result::Ok(std::option::Option::Some(acquisition)) => { let spawned = spawn_persistence(persistence, port, acquisition); - let published = snapshots.record_admission_success(lifecycle.state(), admission.queue_depth(), persistence.len()); + let published = snapshots.record_admission_success(lifecycle.state(), admission.queue_depth(), persistence.len(), backpressure_wait_observed); if !spawned && fault.is_none() { fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); } @@ -181,24 +182,56 @@ async fn drain_admission_and_persistence( }, std::result::Result::Ok(std::option::Option::None) => break, std::result::Result::Err(_) => { - let published = snapshots.record_admission_failure(lifecycle.state(), admission.queue_depth(), persistence.len()); - if let std::result::Result::Err(error) = published { - if fault.is_none() { - fault = std::option::Option::Some(error.code()); - } - } else if fault.is_none() { - fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); + let published = snapshots.record_admission_failure(lifecycle.state(), admission.queue_depth(), persistence.len(), backpressure_wait_observed); + match published { + std::result::Result::Ok(()) => { + if fault.is_none() { + fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); + } + }, + std::result::Result::Err(error) => { + if fault.is_none() { + fault = std::option::Option::Some(error.code()); + } + }, } }, } } let persistence_fault = drain_persistence(lifecycle, admission, persistence, snapshots).await; - if fault.is_none() { - fault = persistence_fault; - } + fault = merge_fault(fault, persistence_fault); return fault; } +async fn drain_owned_work( + settings: &crate::RawTransactionIngestSettings, + lifecycle: &ksp_worker_api::WorkerLifecycle, + children: &mut SourceTasks, + admission: &mut crate::RawTransactionAdmission, + persistence: &mut PersistenceTasks, + port: &std::option::Option, + snapshots: &mut crate::RawTransactionIngestSnapshotPublisher, +) -> std::option::Option { + let drain = async { + let mut fault = drain_admission_and_persistence(settings, lifecycle, admission, persistence, port, snapshots).await; + let child_fault = drain_children(lifecycle.state(), children, snapshots).await; + fault = merge_fault(fault, child_fault); + return fault; + }; + let timed = tokio::time::timeout(settings.shutdown_drain_timeout(), drain).await; + return match timed { + std::result::Result::Ok(fault) => fault, + std::result::Result::Err(_) => { + admission.close(); + persistence.abort_all(); + children.abort_all(); + while persistence.join_next().await.is_some() {} + while children.join_next().await.is_some() {} + std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT) + }, + }; +} + fn finish_faulted( lifecycle: &mut ksp_worker_api::WorkerLifecycle, snapshots: &mut crate::RawTransactionIngestSnapshotPublisher, @@ -261,6 +294,24 @@ fn persistence_completion( }; } +fn source_completion( + joined: std::result::Result, tokio::task::JoinError>, + state: ksp_worker_api::WorkerState, + admission_queue_depth: usize, + in_flight_persistence: usize, + snapshots: &mut crate::RawTransactionIngestSnapshotPublisher, +) -> std::option::Option { + match joined { + std::result::Result::Ok(std::result::Result::Ok(())) => return std::option::Option::None, + std::result::Result::Ok(std::result::Result::Err(_)) | std::result::Result::Err(_) => {}, + } + let published = snapshots.record_source_failure(state, admission_queue_depth, in_flight_persistence); + return match published { + std::result::Result::Ok(()) => std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED), + std::result::Result::Err(error) => std::option::Option::Some(error.code()), + }; +} + async fn run_supervisor( settings: crate::RawTransactionIngestSettings, mut lifecycle: ksp_worker_api::WorkerLifecycle, @@ -269,9 +320,8 @@ async fn run_supervisor( mut snapshots: crate::RawTransactionIngestSnapshotPublisher, source_spawner: Spawner, ) where - Spawner: FnOnce(&mut tokio::task::JoinSet<()>, tokio::sync::watch::Receiver, tokio::sync::mpsc::Sender) - + std::marker::Send - + 'static, + Spawner: + FnOnce(&mut SourceTasks, tokio::sync::watch::Receiver, tokio::sync::mpsc::Sender) + std::marker::Send + 'static, { if *stop_receiver.borrow() { let stopping_fault = begin_stopping(&mut lifecycle, &mut snapshots, 0, 0); @@ -289,7 +339,7 @@ async fn run_supervisor( finish_faulted(&mut lifecycle, &mut snapshots, error.code()); return; } - let mut children = tokio::task::JoinSet::new(); + let mut children = SourceTasks::new(); let mut persistence = PersistenceTasks::new(); let (source_stop_sender, source_stop_receiver) = tokio::sync::watch::channel(false); let (mut admission, admission_sender) = crate::RawTransactionAdmission::new(settings.admission_queue_capacity()); @@ -306,15 +356,19 @@ async fn run_supervisor( &mut snapshots, ) .await; - if fault.is_some() { - source_stop_sender.send_replace(true); - } + source_stop_sender.send_replace(true); let stopping_fault = begin_stopping(&mut lifecycle, &mut snapshots, admission.queue_depth(), persistence.len()); fault = merge_fault(fault, stopping_fault); - let drain_fault = drain_admission_and_persistence(&settings, &lifecycle, &mut admission, &mut persistence, &port, &mut snapshots).await; - fault = merge_fault(fault, drain_fault); - let child_fault = drain_children(&mut children).await; - fault = merge_fault(fault, child_fault); + let drain_fault = drain_owned_work(&settings, &lifecycle, &mut children, &mut admission, &mut persistence, &port, &mut snapshots).await; + match drain_fault { + std::option::Option::Some(code) if code == crate::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT => { + fault = std::option::Option::Some(code); + }, + std::option::Option::Some(code) => { + fault = merge_fault(fault, std::option::Option::Some(code)); + }, + std::option::Option::None => {}, + } match fault { std::option::Option::Some(code) => finish_faulted(&mut lifecycle, &mut snapshots, code), std::option::Option::None => finish_stopped(&mut lifecycle, &mut snapshots), @@ -352,9 +406,8 @@ fn start_foundation_with_port_and_source_spawner( source_spawner: Spawner, ) -> ksp_core_lib::Result where - Spawner: FnOnce(&mut tokio::task::JoinSet<()>, tokio::sync::watch::Receiver, tokio::sync::mpsc::Sender) - + std::marker::Send - + 'static, + Spawner: + FnOnce(&mut SourceTasks, tokio::sync::watch::Receiver, tokio::sync::mpsc::Sender) + std::marker::Send + 'static, { let kind = match ksp_worker_api::WorkerKindCode::new(crate::RAW_TRANSACTION_INGEST_WORKER_KIND_CODE) { std::result::Result::Ok(value) => value, @@ -379,9 +432,8 @@ fn start_foundation_with_source_spawner( source_spawner: Spawner, ) -> ksp_core_lib::Result where - Spawner: FnOnce(&mut tokio::task::JoinSet<()>, tokio::sync::watch::Receiver, tokio::sync::mpsc::Sender) - + std::marker::Send - + 'static, + Spawner: + FnOnce(&mut SourceTasks, tokio::sync::watch::Receiver, tokio::sync::mpsc::Sender) + std::marker::Send + 'static, { let port = match store_guard { std::option::Option::Some(store) => { @@ -398,7 +450,7 @@ async fn supervise_until_stop( lifecycle: &ksp_worker_api::WorkerLifecycle, source_stop_sender: &tokio::sync::watch::Sender, stop_receiver: &mut tokio::sync::watch::Receiver, - children: &mut tokio::task::JoinSet<()>, + children: &mut SourceTasks, admission: &mut crate::RawTransactionAdmission, persistence: &mut PersistenceTasks, port: &std::option::Option, @@ -427,25 +479,38 @@ async fn supervise_until_stop( } } joined = children.join_next(), if !children.is_empty() => { - if let std::option::Option::Some(std::result::Result::Err(_)) = joined { + let source_fault = match joined { + std::option::Option::Some(value) => source_completion(value, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots), + std::option::Option::None => std::option::Option::None, + }; + if source_fault.is_some() { source_stop_sender.send_replace(true); - return std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); + return source_fault; } } joined = persistence.join_next(), if !persistence.is_empty() => { - if let std::option::Option::Some(value) = joined { - let fault = persistence_completion(value, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots); - if fault.is_some() { - source_stop_sender.send_replace(true); - return fault; - } + match joined { + std::option::Option::Some(value) => { + let fault = persistence_completion(value, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots); + if fault.is_some() { + source_stop_sender.send_replace(true); + return fault; + } + }, + std::option::Option::None => {}, } } received = admission.receive(settings.network()), if admission_open && persistence.len() < settings.persistence_concurrency() => { match received { std::result::Result::Ok(std::option::Option::Some(acquisition)) => { + let backpressure_wait_observed = admission.take_backpressure_wait_observed(); let spawned = spawn_persistence(persistence, port, acquisition); - let published = snapshots.record_admission_success(lifecycle.state(), admission.queue_depth(), persistence.len()); + let published = snapshots.record_admission_success( + lifecycle.state(), + admission.queue_depth(), + persistence.len(), + backpressure_wait_observed, + ); if !spawned { source_stop_sender.send_replace(true); return std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID); @@ -455,9 +520,18 @@ async fn supervise_until_stop( return std::option::Option::Some(error.code()); } }, - std::result::Result::Ok(std::option::Option::None) => admission_open = false, + std::result::Result::Ok(std::option::Option::None) => { + let _backpressure_wait_observed = admission.take_backpressure_wait_observed(); + admission_open = false; + }, std::result::Result::Err(_) => { - let published = snapshots.record_admission_failure(lifecycle.state(), admission.queue_depth(), persistence.len()); + let backpressure_wait_observed = admission.take_backpressure_wait_observed(); + let published = snapshots.record_admission_failure( + lifecycle.state(), + admission.queue_depth(), + persistence.len(), + backpressure_wait_observed, + ); source_stop_sender.send_replace(true); return match published { std::result::Result::Ok(()) => std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID), 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 739df6f..941ae18 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/snapshot.rs -// version: 1 +// version: 2 /// Runtime-neutral boxed future resolving to one newer concrete RAW transaction ingest Worker snapshot. pub type RawTransactionIngestSnapshotFuture<'a> = @@ -124,7 +124,7 @@ impl crate::RawTransactionIngestSnapshot { return self.source_failure_total; } - /// Returns the number of source-side admission waits explicitly classified as backpressure waits. + /// Returns the number of supervisor dequeues that observed the bounded admission queue at full capacity. #[must_use] pub const fn backpressure_wait_total(&self) -> u64 { return self.backpressure_wait_total; @@ -303,12 +303,19 @@ impl crate::RawTransactionIngestSnapshotPublisher { state: ksp_worker_api::WorkerState, admission_queue_depth: usize, in_flight_persistence: usize, + backpressure_wait_observed: bool, ) -> ksp_core_lib::Result<()> { let admitted_total = match checked_counter(self.snapshot.admitted_total, "admitted_total") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; + let backpressure_wait_total = + match checked_optional_counter(self.snapshot.backpressure_wait_total, backpressure_wait_observed, "backpressure_wait_total") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; self.snapshot.admitted_total = admitted_total; + self.snapshot.backpressure_wait_total = backpressure_wait_total; return self.publish(state, admission_queue_depth, in_flight_persistence); } @@ -318,20 +325,42 @@ impl crate::RawTransactionIngestSnapshotPublisher { state: ksp_worker_api::WorkerState, admission_queue_depth: usize, in_flight_persistence: usize, + backpressure_wait_observed: bool, ) -> ksp_core_lib::Result<()> { let admitted_total = match checked_counter(self.snapshot.admitted_total, "admitted_total") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; + let backpressure_wait_total = + match checked_optional_counter(self.snapshot.backpressure_wait_total, backpressure_wait_observed, "backpressure_wait_total") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; let canonicalized_total = match checked_counter(self.snapshot.canonicalized_total, "canonicalized_total") { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; self.snapshot.admitted_total = admitted_total; + self.snapshot.backpressure_wait_total = backpressure_wait_total; self.snapshot.canonicalized_total = canonicalized_total; return self.publish(state, admission_queue_depth, in_flight_persistence); } + /// Records one failed private source task without retaining provider-specific error material. + pub(crate) fn record_source_failure( + &mut self, + state: ksp_worker_api::WorkerState, + admission_queue_depth: usize, + in_flight_persistence: usize, + ) -> ksp_core_lib::Result<()> { + let source_failure_total = match checked_counter(self.snapshot.source_failure_total, "source_failure_total") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + self.snapshot.source_failure_total = source_failure_total; + return self.publish(state, admission_queue_depth, in_flight_persistence); + } + /// Records one successful Store persistence outcome and publishes the resulting latest value. pub(crate) fn record_persistence_success( &mut self, @@ -458,6 +487,13 @@ fn checked_counter(current: u64, field: &'static str) -> ksp_core_lib::Result ksp_core_lib::Result { + if !increment { + return std::result::Result::Ok(current); + } + return checked_counter(current, field); +} + fn health_for_state(state: ksp_worker_api::WorkerState, previous: ksp_worker_api::WorkerHealth) -> ksp_worker_api::WorkerHealth { return match state { ksp_worker_api::WorkerState::Running => ksp_worker_api::WorkerHealth::Healthy, 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 15bb238..c219ebb 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: 7 +// version: 8 //! Dependency firewall canaries for the RAW transaction ingest Worker foundation. @@ -52,7 +52,7 @@ fn pre_002_manifest_keeps_forbidden_layers_and_live_sources_out() { } #[test] -fn pre_008_source_surface_adds_one_latest_value_snapshot_watch_without_backend_or_live_source() { +fn pre_009_source_surface_hardens_shutdown_and_faults_without_backend_or_live_source() { let root = include_str!("../src/lib.rs"); let runtime = include_str!("../src/runtime.rs"); let admission = include_str!("../src/admission.rs"); @@ -69,6 +69,12 @@ fn pre_008_source_surface_adds_one_latest_value_snapshot_watch_without_backend_o "WorkerSnapshotSource", "tokio::sync::watch::channel", "ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED", + "ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT", + "ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED", + "record_source_failure", + "tokio::time::timeout", + "abort_all", + "backpressure_wait_total", ] { assert!( root.contains(required) @@ -76,7 +82,7 @@ fn pre_008_source_surface_adds_one_latest_value_snapshot_watch_without_backend_o || admission.contains(required) || persistence.contains(required) || snapshot.contains(required), - "required pre.008 snapshot contract missing: {required}" + "required pre.009 hardening contract missing: {required}" ); } for forbidden in [ @@ -87,8 +93,8 @@ fn pre_008_source_surface_adds_one_latest_value_snapshot_watch_without_backend_o "ksp_store_postgres_lib::", "ksp_onchain_transport_lib::", "ForceRehydrate", - "source_failed", - "drain_timeout", + "ksp_config_lib::", + "reqwest::", ] { assert!( !root.contains(forbidden) @@ -96,7 +102,7 @@ fn pre_008_source_surface_adds_one_latest_value_snapshot_watch_without_backend_o && !admission.contains(forbidden) && !persistence.contains(forbidden) && !snapshot.contains(forbidden), - "pre.008 crossed a forbidden runtime boundary: {forbidden}" + "pre.009 crossed a forbidden runtime boundary: {forbidden}" ); } assert_eq!(runtime.matches("tokio::sync::watch::channel").count(), 2, "runtime retains only the external-control and private-source stop watches"); diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs index e4771f0..c41f321 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs -// version: 7 +// version: 8 //! External public-surface proofs for the RAW transaction ingest Worker foundation. @@ -126,3 +126,12 @@ fn pre_008_snapshot_surface_and_common_projection_are_public_and_stable() { assert_eq!(ksp_worker_raw_transaction_ingest_lib::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED.code(), "counter_exhausted"); return; } + +#[test] +fn pre_009_source_and_drain_timeout_error_codes_are_public_and_stable() { + assert_eq!(ksp_worker_raw_transaction_ingest_lib::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT.domain(), "worker_raw_transaction_ingest"); + assert_eq!(ksp_worker_raw_transaction_ingest_lib::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT.code(), "drain_timeout"); + assert_eq!(ksp_worker_raw_transaction_ingest_lib::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED.domain(), "worker_raw_transaction_ingest"); + assert_eq!(ksp_worker_raw_transaction_ingest_lib::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED.code(), "source_failed"); + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/admission.rs b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/admission.rs index 1476379..d2304c2 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/admission.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/admission.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/admission.rs -// version: 2 +// version: 3 fn material( network: &str, @@ -239,3 +239,24 @@ fn pre_006_network_guards_reject_ingress_or_material_mismatch_without_echoing_va } return; } + +#[tokio::test(flavor = "current_thread")] +async fn pre_009_full_queue_dequeue_marks_source_neutral_backpressure_observation() { + let (network, material) = match material("mainnet", 13, "AQID") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let provenance = match provenance() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let (mut admission, sender) = crate::RawTransactionAdmission::new(1); + let ingress = crate::RawTransactionIngress { material, network, provenance, source_key: [13; 32] }; + assert!(sender.send(ingress).await.is_ok()); + assert!(admission.receive().await.is_some()); + assert!(admission.take_backpressure_wait_observed()); + admission.close(); + assert!(admission.receive().await.is_none()); + assert!(!admission.take_backpressure_wait_observed()); + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs index f2ece97..6df90e4 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs -// version: 5 +// version: 6 struct ActiveTaskGuard { active: std::sync::Arc, @@ -169,7 +169,7 @@ async fn pre_005_supervisor_joins_all_cooperative_source_tasks_before_terminal() loop { let changed = child_stop.changed().await; if changed.is_err() || *child_stop.borrow() { - return; + return std::result::Result::Ok(()); } } }); @@ -209,7 +209,7 @@ async fn pre_005_supervisor_reaps_completed_source_task_and_still_joins_live_chi let completed = std::sync::Arc::clone(&source_completed); let _completed_abort_handle = children.spawn(async move { completed.fetch_add(1, std::sync::atomic::Ordering::AcqRel); - return; + return std::result::Result::Ok(()); }); let active = std::sync::Arc::clone(&source_active); let mut child_stop = stop_receiver.clone(); @@ -218,7 +218,7 @@ async fn pre_005_supervisor_reaps_completed_source_task_and_still_joins_live_chi loop { let changed = child_stop.changed().await; if changed.is_err() || *child_stop.borrow() { - return; + return std::result::Result::Ok(()); } } }); @@ -250,9 +250,21 @@ async fn pre_005_supervisor_reaps_completed_source_task_and_still_joins_live_chi enum RuntimePortResponse { Conflict, StoreFailure, + StoreFailureBlocked, SuccessBlocked, } +struct RuntimePortActiveGuard<'a> { + active: &'a std::sync::atomic::AtomicUsize, +} + +impl std::ops::Drop for RuntimePortActiveGuard<'_> { + fn drop(&mut self) { + self.active.fetch_sub(1, std::sync::atomic::Ordering::AcqRel); + return; + } +} + struct RuntimePersistencePort { active: std::sync::atomic::AtomicUsize, completed: std::sync::atomic::AtomicUsize, @@ -310,18 +322,20 @@ impl crate::RawTransactionIngestPersistencePort for RuntimePersistencePort { self.normal_mode_seen.store(mode == ksp_store_lib::RawTransactionAcquisitionMode::Normal, std::sync::atomic::Ordering::Release); let response = self.response; return std::boxed::Box::pin(async move { - if matches!(response, RuntimePortResponse::SuccessBlocked) { + if matches!(response, RuntimePortResponse::StoreFailureBlocked | RuntimePortResponse::SuccessBlocked) { let current = self.active.fetch_add(1, std::sync::atomic::Ordering::AcqRel) + 1; + let _active_guard = RuntimePortActiveGuard { active: &self.active }; self.update_max_active(current); while !self.released.load(std::sync::atomic::Ordering::Acquire) { self.notify.notified().await; } - self.active.fetch_sub(1, std::sync::atomic::Ordering::AcqRel); - self.completed.fetch_add(1, std::sync::atomic::Ordering::AcqRel); - return std::result::Result::Ok(ksp_store_lib::RawAcquisitionWriteOutcome::new( - ksp_store_lib::RawEntityWriteOutcome::Inserted, - ksp_store_lib::RawObservationWriteOutcome::Inserted, - )); + if matches!(response, RuntimePortResponse::SuccessBlocked) { + self.completed.fetch_add(1, std::sync::atomic::Ordering::AcqRel); + return std::result::Result::Ok(ksp_store_lib::RawAcquisitionWriteOutcome::new( + ksp_store_lib::RawEntityWriteOutcome::Inserted, + ksp_store_lib::RawObservationWriteOutcome::Inserted, + )); + } } if matches!(response, RuntimePortResponse::Conflict) { self.completed.fetch_add(1, std::sync::atomic::Ordering::AcqRel); @@ -369,6 +383,14 @@ fn runtime_ingress(network: &ksp_store_lib::RawNetworkId, signature_byte: u8) -> } fn settings_with_persistence_concurrency(concurrency: usize) -> std::option::Option { + return settings_with_runtime_limits(8, concurrency, std::time::Duration::from_secs(5)); +} + +fn settings_with_runtime_limits( + admission_queue_capacity: usize, + persistence_concurrency: usize, + shutdown_drain_timeout: std::time::Duration, +) -> std::option::Option { let network = match ksp_store_lib::RawNetworkId::new("mainnet") { std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::option::Option::None, @@ -377,7 +399,7 @@ fn settings_with_persistence_concurrency(concurrency: usize) -> std::option::Opt std::result::Result::Ok(value) => value, std::result::Result::Err(_) => return std::option::Option::None, }; - return match crate::RawTransactionIngestSettings::new(network, worker_id, 8, concurrency, std::time::Duration::from_secs(5)) { + return match crate::RawTransactionIngestSettings::new(network, worker_id, admission_queue_capacity, persistence_concurrency, shutdown_drain_timeout) { std::result::Result::Ok(value) => std::option::Option::Some(value), std::result::Result::Err(_) => std::option::Option::None, }; @@ -421,13 +443,13 @@ async fn pre_007_runtime_bounds_in_flight_store_persistence_to_configured_concur for signature_byte in 1..=4 { let ingress = match runtime_ingress(&network, signature_byte) { std::option::Option::Some(value) => value, - std::option::Option::None => return, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")), }; if admission_sender.send(ingress).await.is_err() { - return; + return std::result::Result::Err(crate::runtime_error("test.source_failed")); } } - return; + return std::result::Result::Ok(()); }); }, ) { @@ -469,10 +491,12 @@ async fn pre_007_content_conflict_becomes_terminal_after_private_drain() { let _abort_handle = children.spawn(async move { let ingress = match runtime_ingress(&network, 11) { std::option::Option::Some(value) => value, - std::option::Option::None => return, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")), }; - let _sent = admission_sender.send(ingress).await; - return; + if admission_sender.send(ingress).await.is_err() { + return std::result::Result::Err(crate::runtime_error("test.source_failed")); + } + return std::result::Result::Ok(()); }); }, ) { @@ -511,10 +535,12 @@ async fn pre_007_store_failure_becomes_terminal_without_exposing_remote_error_te let _abort_handle = children.spawn(async move { let ingress = match runtime_ingress(&network, 12) { std::option::Option::Some(value) => value, - std::option::Option::None => return, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")), }; - let _sent = admission_sender.send(ingress).await; - return; + if admission_sender.send(ingress).await.is_err() { + return std::result::Result::Err(crate::runtime_error("test.source_failed")); + } + return std::result::Result::Ok(()); }); }, ) { @@ -550,13 +576,13 @@ async fn pre_008_runtime_snapshots_count_successful_pipeline_and_retain_terminal for signature_byte in 21..=23 { let ingress = match runtime_ingress(&network, signature_byte) { std::option::Option::Some(value) => value, - std::option::Option::None => return, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")), }; if admission_sender.send(ingress).await.is_err() { - return; + return std::result::Result::Err(crate::runtime_error("test.source_failed")); } } - return; + return std::result::Result::Ok(()); }); }, ) { @@ -615,10 +641,12 @@ async fn pre_008_fault_snapshots_count_classified_store_failures_and_project_unh let _abort_handle = children.spawn(async move { let ingress = match runtime_ingress(&network, 24) { std::option::Option::Some(value) => value, - std::option::Option::None => return, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.source_failed")), }; - let _sent = admission_sender.send(ingress).await; - return; + if admission_sender.send(ingress).await.is_err() { + return std::result::Result::Err(crate::runtime_error("test.source_failed")); + } + return std::result::Result::Ok(()); }); }, ) { @@ -643,3 +671,195 @@ async fn pre_008_fault_snapshots_count_classified_store_failures_and_project_unh assert_eq!(snapshot.content_conflict_total(), 0); return; } + +#[tokio::test(flavor = "current_thread")] +async fn pre_009_source_failure_is_counted_and_late_stop_cannot_replace_terminal_fault() { + let settings = match settings("mainnet") { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let handle = + match super::start_foundation_with_source_spawner(settings, tokio::runtime::Handle::current(), move |children, _stop_receiver, _admission_sender| { + let _abort_handle = children.spawn(async move { std::result::Result::Err(crate::runtime_error("test.source_failed")) }); + }) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let source = handle.snapshot_source(); + let terminal = match handle.wait_terminal().await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert_eq!(terminal, ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED)); + let snapshot = source.current(); + assert_eq!(snapshot.source_failure_total(), 1); + assert_eq!(snapshot.backpressure_wait_total(), 0); + assert_eq!(snapshot.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy); + assert!(!handle.request_stop()); + let retained = match handle.wait_terminal().await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert_eq!(retained, terminal); + return; +} + +#[tokio::test(flavor = "current_thread")] +async fn pre_009_stop_does_not_hide_store_failure_observed_during_bounded_drain() { + let settings = match settings_with_persistence_concurrency(1) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let network = settings.network().clone(); + let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::StoreFailureBlocked, false)); + let runtime_port: super::PersistencePort = port.clone(); + let handle = match super::start_foundation_with_port_and_source_spawner( + settings, + tokio::runtime::Handle::current(), + std::option::Option::Some(runtime_port), + move |children, _stop_receiver, admission_sender| { + let _abort_handle = children.spawn(async move { + let ingress = match runtime_ingress(&network, 41) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.ingress_invalid")), + }; + if admission_sender.send(ingress).await.is_err() { + return std::result::Result::Err(crate::runtime_error("test.source_failed")); + } + return std::result::Result::Ok(()); + }); + }, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert!(wait_for_max_active(&port, 1).await); + assert!(handle.request_stop()); + port.release(); + let terminal = match handle.wait_terminal().await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert_eq!(terminal, ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_STORE_FAILED)); + let snapshot = handle.snapshot_source().current(); + assert_eq!(snapshot.store_failure_total(), 1); + assert_eq!(snapshot.in_flight_persistence(), 0); + assert_eq!(port.active.load(std::sync::atomic::Ordering::Acquire), 0); + return; +} + +#[tokio::test(flavor = "current_thread")] +async fn pre_009_saturation_is_observable_without_drop_or_unbounded_admission() { + let settings = match settings_with_runtime_limits(1, 1, std::time::Duration::from_secs(5)) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let network = settings.network().clone(); + let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::SuccessBlocked, false)); + let runtime_port: super::PersistencePort = port.clone(); + let source_stage = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let source_stage_for_task = source_stage.clone(); + let handle = match super::start_foundation_with_port_and_source_spawner( + settings, + tokio::runtime::Handle::current(), + std::option::Option::Some(runtime_port), + move |children, _stop_receiver, admission_sender| { + let _abort_handle = children.spawn(async move { + for signature_byte in 51..=53 { + let ingress = match runtime_ingress(&network, signature_byte) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.ingress_invalid")), + }; + if admission_sender.send(ingress).await.is_err() { + return std::result::Result::Err(crate::runtime_error("test.source_failed")); + } + source_stage_for_task.fetch_add(1, std::sync::atomic::Ordering::AcqRel); + } + return std::result::Result::Ok(()); + }); + }, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert!(wait_for_max_active(&port, 1).await); + for _ in 0..128 { + if source_stage.load(std::sync::atomic::Ordering::Acquire) == 2 { + break; + } + tokio::task::yield_now().await; + } + assert_eq!(source_stage.load(std::sync::atomic::Ordering::Acquire), 2); + port.release(); + assert!(wait_for_completed(&port, 3).await); + for _ in 0..128 { + if source_stage.load(std::sync::atomic::Ordering::Acquire) == 3 { + break; + } + tokio::task::yield_now().await; + } + assert_eq!(source_stage.load(std::sync::atomic::Ordering::Acquire), 3); + assert!(handle.request_stop()); + let terminal = match handle.wait_terminal().await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert_eq!(terminal, ksp_worker_api::WorkerState::Stopped); + let snapshot = handle.snapshot_source().current(); + assert_eq!(snapshot.persisted_total(), 3); + assert_eq!(snapshot.source_failure_total(), 0); + assert!(snapshot.backpressure_wait_total() >= 1); + assert_eq!(snapshot.admission_queue_depth(), 0); + assert_eq!(snapshot.in_flight_persistence(), 0); + return; +} + +#[tokio::test(flavor = "current_thread")] +async fn pre_009_drain_timeout_aborts_and_joins_all_owned_source_and_persistence_tasks() { + let settings = match settings_with_runtime_limits(1, 1, crate::MIN_RAW_TRANSACTION_INGEST_SHUTDOWN_DRAIN_TIMEOUT) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let network = settings.network().clone(); + let port = std::sync::Arc::new(RuntimePersistencePort::new(network.clone(), RuntimePortResponse::SuccessBlocked, false)); + let runtime_port: super::PersistencePort = port.clone(); + let source_active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let source_active_for_task = source_active.clone(); + let handle = match super::start_foundation_with_port_and_source_spawner( + settings, + tokio::runtime::Handle::current(), + std::option::Option::Some(runtime_port), + move |children, _stop_receiver, admission_sender| { + let _abort_handle = children.spawn(async move { + let ingress = match runtime_ingress(&network, 61) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::result::Result::Err(crate::runtime_error("test.ingress_invalid")), + }; + if admission_sender.send(ingress).await.is_err() { + return std::result::Result::Err(crate::runtime_error("test.source_failed")); + } + let _guard = ActiveTaskGuard::new(source_active_for_task); + return std::future::pending::>().await; + }); + }, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert!(wait_for_max_active(&port, 1).await); + assert!(wait_for_active_count(&source_active, 1).await); + assert!(handle.request_stop()); + let terminal = match handle.wait_terminal().await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + assert_eq!(terminal, ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT)); + let snapshot = handle.snapshot_source().current(); + assert_eq!(snapshot.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Unhealthy); + assert_eq!(snapshot.admission_queue_depth(), 0); + assert_eq!(snapshot.in_flight_persistence(), 0); + assert_eq!(source_active.load(std::sync::atomic::Ordering::Acquire), 0); + assert_eq!(port.active.load(std::sync::atomic::Ordering::Acquire), 0); + assert_eq!(port.completed.load(std::sync::atomic::Ordering::Acquire), 0); + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs index 1a0bfd2..0bca54e 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs -// version: 1 +// version: 2 fn snapshot_foundation() -> std::option::Option<(crate::RawTransactionIngestSnapshotPublisher, crate::RawTransactionIngestSnapshotSource)> { let network = match ksp_store_lib::RawNetworkId::new("mainnet") { @@ -78,7 +78,7 @@ async fn pre_008_slow_concrete_and_common_listeners_coalesce_to_same_latest_sequ }; let observed = source.current().worker_snapshot().sequence(); assert!(publisher.publish_state(ksp_worker_api::WorkerState::Running, 0, 0).is_ok()); - assert!(publisher.record_admission_success(ksp_worker_api::WorkerState::Running, 1, 1).is_ok()); + assert!(publisher.record_admission_success(ksp_worker_api::WorkerState::Running, 1, 1, false).is_ok()); let concrete = source.wait_for_change(observed).await; let common = ksp_worker_api::WorkerSnapshotSource::wait_for_change(&source, observed).await; assert_eq!(concrete.worker_snapshot().sequence(), common.sequence()); @@ -113,3 +113,24 @@ async fn pre_008_terminal_latest_value_is_retained_after_publisher_drop() { assert_eq!(common, retained.worker_snapshot().clone()); return; } + +#[test] +fn pre_009_source_failure_and_backpressure_counters_advance_without_changing_projection_contract() { + let (mut publisher, source) = match snapshot_foundation() { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + assert!(publisher.publish_state(ksp_worker_api::WorkerState::Running, 0, 0).is_ok()); + assert!(publisher.record_admission_success(ksp_worker_api::WorkerState::Running, 0, 1, true).is_ok()); + assert!(publisher.record_source_failure(ksp_worker_api::WorkerState::Running, 0, 1).is_ok()); + let concrete = source.current(); + assert_eq!(concrete.admitted_total(), 1); + assert_eq!(concrete.canonicalized_total(), 1); + assert_eq!(concrete.backpressure_wait_total(), 1); + assert_eq!(concrete.source_failure_total(), 1); + assert_eq!(concrete.worker_snapshot().health(), ksp_worker_api::WorkerHealth::Healthy); + assert_eq!(concrete.worker_snapshot().activity(), ksp_worker_api::WorkerActivity::Active); + let common = ksp_worker_api::WorkerSnapshotSource::current(&source); + assert_eq!(&common, concrete.worker_snapshot()); + return; +} diff --git a/deltas/0.3.11/pre.009.md b/deltas/0.3.11/pre.009.md new file mode 100644 index 0000000..8d18eec --- /dev/null +++ b/deltas/0.3.11/pre.009.md @@ -0,0 +1,135 @@ + + + +# Delta `0.3.11-pre.009` — hardening shutdown, backpressure et races fault/stop + +## Base requise + +```text +0.3.11-pre.008-fix.001 +workspace.package.version = 0.3.11-pre.8.fix.1 +``` + +Le gate opérateur du 8 septembre 2026 est vert sur `fmt`, audits Rust/Markdown, `cargo check --workspace`, Clippy strict, 38 tests de la crate Worker et les doc-tests. Aucune dépendance/feature n'ayant changé, aucun `cargo tree` intermédiaire n'était requis. + +## Objectif + +Fermer uniquement le hardening `pre.009` du plan `032` : faults source sûrs, saturation observable, ordering stop/fault déterministe, deadline de drain, abort+join au timeout et absence de tâches orphelines. Aucune source live n'est introduite. + +## Version + +```text +identifiant de livraison : 0.3.11-pre.009 +workspace.package.version : 0.3.11-pre.9 +``` + +## Faults source + +Les tâches source privées retournent `ksp_core_lib::Result<()>`. Une erreur source ou un `JoinError` devient : + +```text +worker_raw_transaction_ingest.source_failed +``` + +Le détail distant/provider n'est pas propagé et `source_failure_total` est incrémenté via checked-add. + +## Shutdown borné + +Le drain complet est enveloppé par `shutdown_drain_timeout`. En cas de timeout : + +```text +admission.close() +persistence.abort_all() +sources.abort_all() +join complet des persistences +join complet des sources +Faulted(worker_raw_transaction_ingest.drain_timeout) +``` + +Le terminal n'est donc jamais publié avant la récolte de toutes les tâches possédées. + +## Ordering stop/fault + +Le stop reste prioritaire dans le `select!`, mais un Store fault découvert pendant le drain n'est pas masqué. Un fault terminal déjà décidé n'est pas remplacé par un stop tardif. Le timeout de drain prévaut lorsqu'il empêche la clôture coopérative. + +## Backpressure source-neutral + +L'admission conserve une unique queue `mpsc` bornée. Le dequeue constate si cette queue était pleine à sa capacité configurée et incrémente `backpressure_wait_total`. Aucun drop, aucune queue secondaire et aucun `unbounded_channel` ne sont ajoutés. + +## Tests ajoutés ou étendus + +```text +source failure terminal + compteur +late stop ne remplace pas source_failed +stop puis Store fault pendant drain +saturation capacity=1 sans drop +backpressure_wait_total observable +drain timeout + abort/join des sources et persistences +aucune tâche active après terminal timeout +projection snapshot source/backpressure +error codes source_failed/drain_timeout publics +dependency firewall inchangé +``` + +## Fichiers ajoutés + +```text +deltas/0.3.11/pre.009.md +``` + +## Fichiers modifiés + +```text +Cargo.toml +crates/ksp-worker-raw-transaction-ingest-lib/src/admission.rs +crates/ksp-worker-raw-transaction-ingest-lib/src/error.rs +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/snapshot.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/public_api.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/admission.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/snapshot.rs +docs/plans/032-V0_3_11_RAW_TRANSACTION_INGEST_WORKER_FOUNDATION_PLAN.md +docs/validation/028-V0_3_11_RAW_TRANSACTION_INGEST_WORKER_FOUNDATION.md +``` + +## Fichiers supprimés + +Aucun. + +## Dépendances + +Aucune dépendance ni feature n'est ajoutée, supprimée ou modifiée. Le `Cargo.toml` de la crate Worker reste identique à la base. + +## 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 +scan statique des frontières runtime/dependencies +``` + +L'environnement d'assemblage ne fournit ni `cargo`, ni `rustc`, ni `rustfmt`. Aucun gate Cargo de `pre.009` n'est déclaré PASS localement. + +## Gate opérateur demandé + +Aucune dépendance ni feature n'ayant changé, aucun `cargo tree` intermédiaire n'est 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 +``` + +## Décision + +`pre.010` reste bloquée jusqu'à validation opérateur verte du hardening shutdown/backpressure/fault races. + +## Questions ouvertes + +Aucune nouvelle question architecturale. Les sources live restent explicitement hors scope de `0.3.11`. diff --git a/docs/plans/032-V0_3_11_RAW_TRANSACTION_INGEST_WORKER_FOUNDATION_PLAN.md b/docs/plans/032-V0_3_11_RAW_TRANSACTION_INGEST_WORKER_FOUNDATION_PLAN.md index b793c9b..1ccb8bc 100644 --- a/docs/plans/032-V0_3_11_RAW_TRANSACTION_INGEST_WORKER_FOUNDATION_PLAN.md +++ b/docs/plans/032-V0_3_11_RAW_TRANSACTION_INGEST_WORKER_FOUNDATION_PLAN.md @@ -1,5 +1,5 @@ - + # Plan v0.3.11 — fondation runtime du Worker RawTransaction ingest @@ -738,12 +738,14 @@ Budget cible : **15–20 min**. Port privé Store, mode Normal, concurrency born Budget cible : **15–20 min**. `watch` latest-value, compteurs checked, common projection, slow/no listener et terminal retained. -État après matérialisation : **implémenté, gate opérateur requis**. La crate expose `RawTransactionIngestSnapshot` et `RawTransactionIngestSnapshotSource`, ce dernier implémentant directement `ksp-worker-api::WorkerSnapshotSource`. Un unique `watch` porte la valeur latest-value concrète ; `wait_terminal()` lit ce même flux, supprimant le watch terminal spécialisé de `pre.004`. La séquence `WorkerSnapshotSequence` est donc identique pour la vue concrète et la projection commune. Les compteurs `u64` sont incrémentés par checked-add au point exact où le supervisor dequeue/canonicalise ou récolte une persistence ; l'épuisement devient `worker_raw_transaction_ingest.counter_exhausted`. Les profondeurs de queue et de persistence sont dérivées des structures bornées existantes, sans tâche de monitoring ni seconde event queue. `source_failure_total` et `backpressure_wait_total` existent dans le contrat sûr mais restent à zéro dans cette tranche source-neutral ; `pre.009` est propriétaire du fault source et de l'instrumentation de saturation. Les tests couvrent projection common/concrete, listener lent coalescé, terminal retenu, counters de pipeline réussi et Store fault projeté `Unhealthy`. +État après matérialisation : **implémenté et gate opérateur validé après `pre.008-fix.001`**. La crate expose `RawTransactionIngestSnapshot` et `RawTransactionIngestSnapshotSource`, ce dernier implémentant directement `ksp-worker-api::WorkerSnapshotSource`. Un unique `watch` porte la valeur latest-value concrète ; `wait_terminal()` lit ce même flux. La séquence `WorkerSnapshotSequence` est identique pour la vue concrète et la projection commune. Les compteurs `u64` utilisent checked-add ; `source_failure_total` et `backpressure_wait_total` existent mais restent à zéro jusqu'à `pre.009`. Le correctif `pre.008-fix.001` a uniquement résolu `clippy::collapsible_if` dans `runtime.rs`, sans modifier comportement, dépendance ni feature. Le gate communiqué le 8 septembre 2026 est vert sur `fmt`, audits, `check`, Clippy strict, 38 tests de crate et doc-tests. Aucun arbre Cargo n'a été requis car aucune dépendance/feature n'avait changé. ### `pre.009` — hardening shutdown/backpressure/fault races Budget cible : **15–20 min**. Drain deadline, stop/fault ordering, saturation, source failure, Store slow/failure, abort+join au timeout, no orphan tasks. Scinder immédiatement si le gate réel dépasse le budget. +État après matérialisation : **implémenté, gate opérateur requis**. Les tâches source privées retournent désormais `ksp_core_lib::Result<()>` dans le `JoinSet` du supervisor ; une erreur source ou un `JoinError` devient `worker_raw_transaction_ingest.source_failed` et incrémente `source_failure_total` sans exposer de texte provider. Le shutdown possède une deadline stricte `shutdown_drain_timeout` ; au timeout, admission, persistences et sources sont fermées/abortées puis toutes rejointes avant `Faulted(worker_raw_transaction_ingest.drain_timeout)`. Le stop reste prioritaire dans le `select!`, mais un Store fault découvert pendant le drain n'est pas masqué par un stop antérieur ; un fault terminal déjà décidé n'est pas remplacé par un stop tardif. La saturation est instrumentée de façon source-neutral au dequeue d'une queue `mpsc` pleine via `backpressure_wait_total`, sans drop ni queue non bornée. Les tests déterministes couvrent source failure, stop/Store fault, saturation et timeout avec absence de tâches orphelines. Aucune source live, aucun Transport, aucun backend direct et aucune nouvelle dépendance/feature ne sont introduits. + ### `pre.010` — hardening public/release/security Budget cible : **10–15 min**. Tests externes exacts : API root, dependencies, historical-surface absence, redaction, error codes, module inventory et scans Config/secret/backend. diff --git a/docs/validation/028-V0_3_11_RAW_TRANSACTION_INGEST_WORKER_FOUNDATION.md b/docs/validation/028-V0_3_11_RAW_TRANSACTION_INGEST_WORKER_FOUNDATION.md index 8bc1287..78c6f21 100644 --- a/docs/validation/028-V0_3_11_RAW_TRANSACTION_INGEST_WORKER_FOUNDATION.md +++ b/docs/validation/028-V0_3_11_RAW_TRANSACTION_INGEST_WORKER_FOUNDATION.md @@ -1,5 +1,5 @@ - + # Validation v0.3.11 — fondation runtime du Worker RawTransaction ingest @@ -1301,3 +1301,106 @@ cargo test -p ksp-worker-raw-transaction-ingest-lib Critère de passage : snapshots common/concrete latest-value, compteurs checked, terminal retenu et absence de nouveau boundary crossing sont verts. +## 20. `pre.009` — hardening shutdown/backpressure/fault races + +### 20.1 Clôture opérateur de `pre.008-fix.001` + +Le gate communiqué le 8 septembre 2026 est 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 table(s), 785 file(s)) +cargo check --workspace: PASS +cargo clippy --workspace --all-targets --all-features -- -D warnings: PASS +29 unit tests: PASS +3 dependency-boundary tests: PASS +6 public API tests: PASS +doc-tests: PASS +``` + +Aucune dépendance ni feature n'avait changé ; aucun `cargo tree` intermédiaire n'était requis. + +### 20.2 Classification des faults source + +Les tâches source privées sont possédées par : + +```text +tokio::task::JoinSet> +``` + +Une `Err` source ou un `JoinError` est réduite à : + +```text +worker_raw_transaction_ingest.source_failed +``` + +Le texte de l'erreur source n'est jamais recopié. `source_failure_total` est incrémenté avec la même discipline checked que les autres compteurs. + +### 20.3 Deadline de drain et absence d'orphelins + +Le drain complet est borné par `shutdown_drain_timeout` via `tokio::time::timeout`. Si la deadline expire : + +```text +fermeture admission +abort_all des persistences +abort_all des tâches source +join complet des deux JoinSet +Faulted(worker_raw_transaction_ingest.drain_timeout) +``` + +Aucun terminal n'est publié avant la récolte des tâches possédées. + +### 20.4 Priorité stop/fault + +Le stop reste la branche prioritaire du `select!`. Cette priorité n'efface cependant pas un Store fault découvert pendant le drain déjà engagé. Réciproquement, un fault terminal décidé avant un stop tardif demeure le terminal retenu. Le timeout de drain prévaut lorsqu'il empêche la fermeture coopérative complète. + +### 20.5 Saturation et backpressure + +La queue d'admission demeure le `mpsc` borné de `pre.006`. Le supervisor observe, avant dequeue, si la queue est à sa capacité configurée ; cette observation source-neutral incrémente `backpressure_wait_total`. Aucun élément n'est droppé, aucune seconde queue et aucun `unbounded_channel` ne sont introduits. + +### 20.6 Preuves ajoutées + +Les tests `pre.009` couvrent : + +```text +source failure -> source_failed + source_failure_total +stop tardif ne remplace pas source_failed +stop avant Store fault -> Store fault terminal après drain +queue capacity=1 + persistence concurrency=1 -> saturation observable sans drop +drain timeout -> abort + join source et persistence +aucune tâche source/persistence active après terminal timeout +projection snapshot conserve les compteurs source/backpressure +error codes source_failed/drain_timeout publics et stables +aucun nouvel edge de dépendance +``` + +### 20.7 Hors scope conservé + +`pre.009` n'introduit toujours aucun : + +```text +source live +Transport +Config +backend Store direct +retry / reconnect +gap repair / replay live +provider capability public +``` + +### 20.8 Gate opérateur demandé pour `pre.009` + +Aucune dépendance ni feature n'est modifiée ; aucun `cargo tree` intermédiaire n'est 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 +``` + +Critère de passage : races stop/fault déterministes, timeout borné sans tâche orpheline, saturation observable sans drop et frontières de dépendances inchangées.