diff --git a/Cargo.toml b/Cargo.toml index 9ad0691..cf99985 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 586 +# version: 587 [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.14-pre.11.fix.2" +version = "0.3.14-pre.12" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" 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 20831fb..b4a5b71 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs -// version: 42 +// version: 43 use sha2::Digest; // rust-rules: trait-import @@ -1726,6 +1726,9 @@ impl crate::RawTransactionIngestHttpBlockPollingSource { break 'source; } } + if *stop_receiver.borrow() { + break; + } if !blocked_by_null && let std::option::Option::Some(proven_end_slot) = discovery.proven_end_slot { let coverage_result = { let mut contracts = match continuity_contracts.lock() { @@ -2584,6 +2587,7 @@ impl crate::RawTransactionIngestRuntimeResources { std::sync::Arc::clone(&continuity_contracts), std::sync::Arc::clone(&inventory), processing_frontier_sender, + settings.shutdown_drain_timeout(), ) .await; let validation = { @@ -2682,18 +2686,19 @@ async fn supervise_live_source_tasks( continuity_contracts: std::sync::Arc>, inventory: std::sync::Arc>, processing_frontier_sender: tokio::sync::watch::Sender, + shutdown_drain_timeout: std::time::Duration, ) -> ksp_core_lib::Result<()> { loop { if *stop_receiver.borrow() { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::None).await; + return drain_live_source_tasks(&mut children, std::option::Option::None, shutdown_drain_timeout).await; } let joined = tokio::select! { biased; changed = stop_receiver.changed() => { if changed.is_err() || *stop_receiver.borrow() { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::None).await; + return drain_live_source_tasks(&mut children, std::option::Option::None, shutdown_drain_timeout).await; } continue; } @@ -2703,11 +2708,21 @@ async fn supervise_live_source_tasks( std::option::Option::Some(std::result::Result::Ok(value)) => value, std::option::Option::Some(std::result::Result::Err(_)) => { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::Some(crate::runtime_error("source.task_join_failed"))).await; + return drain_live_source_tasks( + &mut children, + std::option::Option::Some(crate::runtime_error("source.task_join_failed")), + shutdown_drain_timeout, + ) + .await; }, std::option::Option::None => { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::Some(crate::runtime_error("source.task_set_empty"))).await; + return drain_live_source_tasks( + &mut children, + std::option::Option::Some(crate::runtime_error("source.task_set_empty")), + shutdown_drain_timeout, + ) + .await; }, }; let source_fault = match source_result { @@ -2716,7 +2731,7 @@ async fn supervise_live_source_tasks( }; if !source_loss_is_reconcilable(&source_fault) { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await; + return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault), shutdown_drain_timeout).await; } let supervisor_state = { let inventory = match inventory.lock() { @@ -2729,21 +2744,21 @@ async fn supervise_live_source_tasks( std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await; + return drain_live_source_tasks(&mut children, std::option::Option::Some(error), shutdown_drain_timeout).await; }, }; let continuity_range = match source_loss_continuity_range(&source_fault) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await; + return drain_live_source_tasks(&mut children, std::option::Option::Some(error), shutdown_drain_timeout).await; }, }; let continuity_range = match continuity_range { std::option::Option::Some(value) => value, std::option::Option::None => { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await; + return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault), shutdown_drain_timeout).await; }, }; let decision = { @@ -2761,7 +2776,7 @@ async fn supervise_live_source_tasks( std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await; + return drain_live_source_tasks(&mut children, std::option::Option::Some(error), shutdown_drain_timeout).await; }, }; match decision { @@ -2770,7 +2785,7 @@ async fn supervise_live_source_tasks( std::result::Result::Ok(value) => value, std::result::Result::Err(error) => { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::Some(error)).await; + return drain_live_source_tasks(&mut children, std::option::Option::Some(error), shutdown_drain_timeout).await; }, }; processing_frontier_sender.send_replace(aggregate); @@ -2778,7 +2793,7 @@ async fn supervise_live_source_tasks( }, crate::RawTransactionIngestSourceLossDecision::Fault => { source_stop_sender.send_replace(true); - return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await; + return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault), shutdown_drain_timeout).await; }, } } @@ -2834,16 +2849,29 @@ fn source_loss_is_reconcilable(error: &ksp_core_lib::Error) -> bool { async fn drain_live_source_tasks( children: &mut tokio::task::JoinSet<([u8; 32], ksp_core_lib::Result<()>)>, mut first_fault: std::option::Option, + shutdown_drain_timeout: std::time::Duration, ) -> ksp_core_lib::Result<()> { - while let std::option::Option::Some(joined) = children.join_next().await { - if first_fault.is_some() { - continue; + let drain = async { + while let std::option::Option::Some(joined) = children.join_next().await { + if first_fault.is_some() { + continue; + } + first_fault = match joined { + std::result::Result::Ok((_source_key, std::result::Result::Ok(()))) => std::option::Option::None, + std::result::Result::Ok((_source_key, std::result::Result::Err(error))) => std::option::Option::Some(error), + std::result::Result::Err(_) => std::option::Option::Some(crate::runtime_error("source.task_join_failed")), + }; + } + }; + if tokio::time::timeout(shutdown_drain_timeout, drain).await.is_err() { + children.abort_all(); + while children.join_next().await.is_some() {} + if first_fault.is_none() { + first_fault = std::option::Option::Some(ksp_core_lib::Error::new( + crate::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT, + "RAW transaction ingest source drain timed out", + )); } - first_fault = match joined { - std::result::Result::Ok((_source_key, std::result::Result::Ok(()))) => std::option::Option::None, - std::result::Result::Ok((_source_key, std::result::Result::Err(error))) => std::option::Option::Some(error), - std::result::Result::Err(_) => std::option::Option::Some(crate::runtime_error("source.task_join_failed")), - }; } return match first_fault { std::option::Option::Some(error) => std::result::Result::Err(error), @@ -4656,6 +4684,9 @@ impl RawTransactionIngestHydrationCoordinator { std::result::Result::Ok(std::result::Result::Err(error)) => return std::result::Result::Err(error), std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("source.hydration_task_join_failed")), }; + if *stop_receiver.borrow() { + return std::result::Result::Ok(false); + } let pending = match self.pending.remove(&fetched.key) { std::option::Option::Some(value) => value, std::option::Option::None => return std::result::Result::Err(crate::runtime_error("source.hydration_result_without_pending")), @@ -4665,6 +4696,9 @@ impl RawTransactionIngestHydrationCoordinator { } self.pending_signal_count -= pending.signals.len(); for pending_signal in pending.signals { + if *stop_receiver.borrow() { + return std::result::Result::Ok(false); + } let signal_slot = pending_signal.signal.slot; let resolution = resolve_known_reference_hydration(hydration, settings, pending_signal.signal, pending_signal.received_at, &fetched.observed); let resolution = match resolution { 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 814b7c0..b9140b8 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: 29 +// version: 30 //! Dependency firewall canaries for the RAW transaction ingest Worker foundation. @@ -732,3 +732,29 @@ fn v0_3_13_pre_012_cross_layer_closure_does_not_expand_worker_dependency_graph() } return; } + +#[test] +fn v0_3_14_pre_012_shutdown_hardening_stays_worker_local_and_facade_only() { + let runtime = include_str!("../src/runtime.rs"); + let resources = include_str!("../src/runtime_resources.rs"); + for required in [ + "tokio::time::timeout(settings.shutdown_drain_timeout(), drain).await", + "tokio::time::timeout(shutdown_drain_timeout, drain).await", + "children.abort_all()", + "persistence.abort_all()", + ] { + assert!(runtime.contains(required) || resources.contains(required), "required pre.012 bounded shutdown guard missing: {required}"); + } + for forbidden in [ + "ksp_store_postgres_lib::", + "ksp_config_lib::", + "ksp_job_backfill_lib::", + "reqwest::", + "tokio_tungstenite::", + "tonic::", + "yellowstone_grpc_proto::", + ] { + assert!(!runtime.contains(forbidden) && !resources.contains(forbidden), "pre.012 added forbidden Worker dependency edge: {forbidden}"); + } + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs index 98e8df0..fc5c600 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs -// version: 35 +// version: 36 //! External public, security, redaction and release-boundary hardening canaries through `v0.3.14-pre.011`. @@ -1313,3 +1313,42 @@ fn v0_3_14_pre_011_gap_observability_is_bounded_checked_and_redacted() { assert!(!root.contains("source_key")); return; } + +#[test] +fn v0_3_14_pre_012_shutdown_races_are_bounded_across_source_and_worker_drains() { + let runtime = include_str!("../src/runtime.rs"); + let resources = include_str!("../src/runtime_resources.rs"); + for required in [ + "tokio::time::timeout(settings.shutdown_drain_timeout(), drain).await", + "persistence.abort_all()", + "while persistence.join_next().await.is_some() {}", + "while children.join_next().await.is_some() {}", + ] { + assert!(runtime.contains(required), "required pre.012 outer shutdown guard missing: {required}"); + } + for required in [ + "shutdown_drain_timeout: std::time::Duration", + "tokio::time::timeout(shutdown_drain_timeout, drain).await", + "children.abort_all()", + "ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT", + "coordinator.abort_all(&mut processing_frontier).await", + "discover_http_block_window(", + "admission_sender.send(ingress)", + "if *stop_receiver.borrow()", + ] { + assert!(resources.contains(required), "required pre.012 source shutdown guard missing: {required}"); + } + for forbidden in [ + "unbounded_channel", + "ksp_store_postgres_lib::", + "ksp_config_lib::", + "ksp_job_backfill_lib::", + "reqwest::", + "tokio_tungstenite::", + "tonic::", + "yellowstone_grpc_proto::", + ] { + assert!(!runtime.contains(forbidden) && !resources.contains(forbidden), "pre.012 crossed a Worker/facade boundary: {forbidden}"); + } + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs b/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs index e3d862e..39f0d4c 100644 --- a/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs +++ b/crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs @@ -1,5 +1,5 @@ // file: crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs -// version: 30 +// version: 31 //! Release-completeness canaries through the `v0.3.14-pre.011` gap observability tranche. @@ -422,3 +422,22 @@ fn v0_3_14_pre_011_gap_observability_canaries_are_present_without_source_identit assert!(!root.contains("source_key")); return; } + +#[test] +fn v0_3_14_pre_012_shutdown_race_canaries_cover_bounded_source_and_worker_drains() { + let dependency_boundary = include_str!("dependency_boundary.rs"); + let hardening = include_str!("hardening.rs"); + let resource_tests = include_str!("../unit_tests/runtime_resources.rs"); + let root = include_str!("../src/lib.rs"); + for required in [ + "v0_3_14_pre_012_source_fault_preserves_first_fault_and_aborts_non_cooperative_sibling", + "v0_3_14_pre_012_stop_aborts_non_cooperative_source_after_bounded_drain", + ] { + assert!(resource_tests.contains(required), "required pre.012 source shutdown canary missing: {required}"); + } + assert!(hardening.contains("v0_3_14_pre_012_shutdown_races_are_bounded_across_source_and_worker_drains")); + assert!(dependency_boundary.contains("v0_3_14_pre_012_shutdown_hardening_stays_worker_local_and_facade_only")); + assert!(!root.contains("RawTransactionIngestShutdown")); + assert!(!root.contains("SourceDrain")); + return; +} diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs b/crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs index 6e4b8ca..a6b064d 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: 33 +// version: 34 fn grpc_endpoint(cluster: &str) -> std::option::Option { return grpc_endpoint_with_identity(cluster, "yellowstone-fixture", "fixture-provider"); @@ -1155,6 +1155,7 @@ async fn v0_3_13_pre_007_source_supervisor_joins_all_children_on_stop() { contracts, inventory, tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0, + std::time::Duration::from_secs(5), ) .await; }); @@ -1223,6 +1224,7 @@ async fn v0_3_13_pre_007_source_failure_stops_and_joins_sibling_sources() { contracts, inventory, tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0, + std::time::Duration::from_secs(5), ) .await; assert!(result.is_err()); @@ -3838,6 +3840,7 @@ async fn v0_3_13_pre_011_aborting_outer_source_supervisor_aborts_nested_source_t contracts, inventory, tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0, + std::time::Duration::from_secs(5), ) .await; }); @@ -3905,6 +3908,7 @@ async fn v0_3_13_pre_011_stop_racing_ready_source_fault_preserves_fault_and_join contracts, inventory, tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0, + std::time::Duration::from_secs(5), ) .await; }); @@ -3930,6 +3934,110 @@ async fn v0_3_13_pre_011_stop_racing_ready_source_fault_preserves_fault_and_join return; } +#[tokio::test(flavor = "current_thread")] +async fn v0_3_14_pre_012_source_fault_preserves_first_fault_and_aborts_non_cooperative_sibling() { + let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(2)); + let (_stop_sender, stop_receiver) = tokio::sync::watch::channel(false); + let (source_stop_sender, _source_stop_receiver) = tokio::sync::watch::channel(false); + let contracts = match supervisor_contracts(&[(1, "fixture-a", 1), (2, "fixture-b", 2)]) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let inventory = match supervisor_inventory( + std::vec![[1_u8; 32], [2_u8; 32]], + &[ + std::option::Option::Some(crate::RawTransactionIngestSourceState::Failed), + std::option::Option::Some(crate::RawTransactionIngestSourceState::Active), + ], + ) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let mut children = tokio::task::JoinSet::new(); + let failing_barrier = std::sync::Arc::clone(&barrier); + let _failing_abort_handle = children.spawn(async move { + let _barrier_wait = failing_barrier.wait().await; + return ([1_u8; 32], std::result::Result::Err(crate::runtime_error("test.pre_012_source_fault"))); + }); + let sibling_barrier = std::sync::Arc::clone(&barrier); + let sibling_active = std::sync::Arc::clone(&active); + let _sibling_abort_handle = children.spawn(async move { + let _guard = Pre011SourceActiveGuard::new(sibling_active); + let _barrier_wait = sibling_barrier.wait().await; + return std::future::pending::<([u8; 32], ksp_core_lib::Result<()>)>().await; + }); + let result = super::supervise_live_source_tasks( + stop_receiver, + source_stop_sender, + children, + contracts, + inventory, + tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0, + std::time::Duration::from_millis(5), + ) + .await; + let error = match result { + std::result::Result::Ok(()) => return, + std::result::Result::Err(value) => value, + }; + assert_eq!(error.context()[0].value(), "test.pre_012_source_fault"); + assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 0); + return; +} + +#[tokio::test(flavor = "current_thread")] +async fn v0_3_14_pre_012_stop_aborts_non_cooperative_source_after_bounded_drain() { + let active = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let (stop_sender, stop_receiver) = tokio::sync::watch::channel(false); + let (source_stop_sender, _source_stop_receiver) = tokio::sync::watch::channel(false); + let contracts = match supervisor_contracts(&[(1, "fixture-a", 1)]) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let inventory = match supervisor_inventory(std::vec![[1_u8; 32]], &[std::option::Option::Some(crate::RawTransactionIngestSourceState::Active)]) { + std::option::Option::Some(value) => value, + std::option::Option::None => return, + }; + let mut children = tokio::task::JoinSet::new(); + let source_active = std::sync::Arc::clone(&active); + let _abort_handle = children.spawn(async move { + let _guard = Pre011SourceActiveGuard::new(source_active); + return std::future::pending::<([u8; 32], ksp_core_lib::Result<()>)>().await; + }); + let supervisor = tokio::spawn(async move { + return super::supervise_live_source_tasks( + stop_receiver, + source_stop_sender, + children, + contracts, + inventory, + tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()).0, + std::time::Duration::from_millis(5), + ) + .await; + }); + for _ in 0..64 { + if active.load(std::sync::atomic::Ordering::Acquire) == 1 { + break; + } + tokio::task::yield_now().await; + } + assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 1); + stop_sender.send_replace(true); + let result = match supervisor.await { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return, + }; + let error = match result { + std::result::Result::Ok(()) => return, + std::result::Result::Err(value) => value, + }; + assert_eq!(error.code(), crate::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT); + assert_eq!(active.load(std::sync::atomic::Ordering::Acquire), 0); + return; +} + #[tokio::test(flavor = "current_thread")] async fn v0_3_14_pre_008_proven_full_ledger_epoch_survives_filtered_peer_loss_without_worker_respawn() { let sibling_active = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); @@ -3974,7 +4082,16 @@ async fn v0_3_14_pre_008_proven_full_ledger_epoch_survives_filtered_peer_loss_wi }); let (health_sender, health_receiver) = tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); let supervisor = tokio::spawn(async move { - return super::supervise_live_source_tasks(stop_receiver, source_stop_sender, children, contracts, inventory, health_sender).await; + return super::supervise_live_source_tasks( + stop_receiver, + source_stop_sender, + children, + contracts, + inventory, + health_sender, + std::time::Duration::from_secs(5), + ) + .await; }); for _ in 0..64 { if sibling_active.load(std::sync::atomic::Ordering::Acquire) && health_receiver.borrow().failed_source_losses_reconciled() { diff --git a/deltas/0.3.14/pre.012.md b/deltas/0.3.14/pre.012.md new file mode 100644 index 0000000..ee2ff9c --- /dev/null +++ b/deltas/0.3.14/pre.012.md @@ -0,0 +1,237 @@ + + + +# Delta `0.3.14-pre.012` — hardening races / shutdown + +## Base requise + +```text +0.3.14-pre.011-fix.002 +workspace.package.version = 0.3.14-pre.11.fix.2 +deltas/0.3.14/pre.011-fix.002.md présent +``` + +## Gate de la base + +Le gate opérateur de `0.3.14-pre.011-fix.002` est validé avant ouverture de cette tranche : + +```text +cargo fmt --all : PASS +cargo fmt --all -- --check : PASS +audit Rust workspace rules : PASS +audit Markdown tables : PASS +cargo check --workspace : PASS +cargo clippy --workspace --all-targets --all-features -- -D warnings : PASS +cargo test -p ksp-worker-raw-transaction-ingest-lib --all-targets --all-features : PASS +``` + +Le gate Worker comprend notamment : + +```text +150 unit tests : PASS +cross_layer_completeness : 8 PASS +dependency_boundary : 19 PASS +hardening : 36 PASS +public_api : 21 PASS +release_completeness : 13 PASS +``` + +## Objectif + +Implémenter strictement la tranche `pre.012` du plan `035` : + +```text +borner stop/fault pendant les sous-tâches source et les phases de continuité +préserver la première faute quand un sibling ne coopère pas au shutdown +abandonner et rejoindre les tâches source restantes après expiration de la deadline +empêcher une mutation de coverage après observation d'un stop HTTP +empêcher une hydration déjà jointe de poursuivre admission/continuity après observation d'un stop +conserver le drain/abort/join global déjà validé pour admission/persistence +ne créer aucun task pool, pipeline, timeout ou ownership parallèle +``` + +## Drain interne des sources désormais borné + +Avant cette tranche, le drain global Worker était borné par `shutdown_drain_timeout`, mais le superviseur interne des sources pouvait encore attendre indéfiniment un sibling non coopératif après une faute de source. + +`supervise_live_source_tasks` reçoit maintenant la même deadline déjà configurée dans `RawTransactionIngestSettings` : + +```text +settings.shutdown_drain_timeout() +``` + +`drain_live_source_tasks` applique : + +```text +tokio::time::timeout(shutdown_drain_timeout, drain) +``` + +En cas d'expiration : + +```text +children.abort_all() +join de tous les children annulés +``` + +Aucune tâche source ne reste donc détenue après retour du superviseur interne. + +## Priorité de faute + +Si le shutdown interne commence après une faute source déjà observée et que le sibling ne coopère pas avant la deadline : + +```text +la première faute source reste la faute retournée +``` + +Le timeout ne remplace pas une cause terminale déjà connue. + +Si aucun défaut antérieur n'existe et qu'un stop coopératif ne peut pas joindre une source avant la deadline : + +```text +ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT +``` + +est retourné. + +Cette règle aligne le superviseur source avec le drain Worker global déjà validé. + +## Stop avant mutation de coverage HTTP + +Le polling HTTP vérifie maintenant explicitement le stop après le traitement des blocs découverts et avant : + +```text +record_coverage_epoch(...) +processing_frontier.publish() +``` + +Un stop déjà observé ne peut donc pas être suivi d'une nouvelle preuve de coverage de cette fenêtre. + +Les appels réseau restent stop-preemptible via les `tokio::select!` existants ; aucun retry Worker n'est ajouté. + +## Stop pendant hydration/admission + +`RawTransactionIngestHydrationCoordinator::handle_joined` vérifie maintenant le stop : + +```text +avant de retirer/appliquer le résultat d'hydration joint +avant chaque signal coalescé de ce résultat +``` + +Les `tokio::select!` existants restent en place autour de l'admission async. + +Un stop déjà observé ne poursuit donc pas la mutation continuity ou l'admission d'un résultat d'hydration devenu prêt au même moment. + +## Invariants préservés + +```text +Transport reste propriétaire des sockets, reconnects et replay natif +Worker ne modifie jamais from_slot +aucun respawn de source +aucun second scheduler ou sémaphore +fairness pre.010 inchangée +observabilité pre.011 inchangée +Common RAW/admission/persistence restent uniques +Store reste accessible via ksp-store-lib seulement +aucun Worker -> Config/Job/backend Store +aucune nouvelle dépendance +``` + +## Tests ajoutés + +Deux canaris runtime reproduisent les races que cette tranche ferme : + +```text +v0_3_14_pre_012_source_fault_preserves_first_fault_and_aborts_non_cooperative_sibling +v0_3_14_pre_012_stop_aborts_non_cooperative_source_after_bounded_drain +``` + +Le premier prouve : + +```text +sibling réellement démarré +faute source observée +sibling volontairement non coopératif +deadline interne atteinte +abort + join du sibling +première faute préservée +aucune tâche active après retour +``` + +Le second prouve : + +```text +stop coopératif demandé +source volontairement non coopérative +deadline interne atteinte +ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT +abort + join +aucune tâche active après retour +``` + +Les canaris `hardening`, `dependency_boundary` et `release_completeness` vérifient en plus : + +```text +deadline source et deadline Worker présentes +abort_all + join présents aux deux niveaux +checkpoints stop avant les mutations repair/admission concernées +aucune nouvelle dépendance ni frontière backend +aucune croissance de surface publique shutdown +``` + +## Hors périmètre inchangé + +```text +aucune nouvelle stratégie de repair +aucune nouvelle méthode HTTP +aucun changement de health policy +aucun changement de snapshot public +aucun changement de TargetCoverage +aucun changement de fairness +aucun changement de persistence semantics +aucun EARLY/shred +aucun Job Backfill depuis Worker +``` + +## Fichiers ajoutés + +```text +deltas/0.3.14/pre.012.md +``` + +## Fichiers modifiés + +```text +Cargo.toml +crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs +crates/ksp-worker-raw-transaction-ingest-lib/unit_tests/runtime_resources.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/dependency_boundary.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/hardening.rs +crates/ksp-worker-raw-transaction-ingest-lib/tests/release_completeness.rs +``` + +## Fichiers supprimés + +```text +aucun +``` + +## Version Cargo + +Conformément au workflow prerelease non-fix : + +```text +header Cargo.toml : 586 -> 587 +workspace.package.version : 0.3.14-pre.11.fix.2 -> 0.3.14-pre.12 +``` + +## Gate opérateur après application + +```bash +cargo fmt --all +cargo fmt --all -- --check +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 --all-targets --all-features +```