From ec94001ab83fb8c6409380af8924ed10d6b94460 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Sat, 12 Sep 2026 09:44:36 +0200 Subject: [PATCH] v0.3.14-pre.008-fix.001 --- Cargo.toml | 4 +- .../src/runtime_resources.rs | 146 ++++++++++----- deltas/0.3.14/pre.008-fix.001.md | 172 ++++++++++++++++++ 3 files changed, 277 insertions(+), 45 deletions(-) create mode 100644 deltas/0.3.14/pre.008-fix.001.md diff --git a/Cargo.toml b/Cargo.toml index c2cba82..38ad21d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 576 +# version: 577 [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.8" +version = "0.3.14-pre.8.fix.1" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs b/crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs index 2671f67..4ccb0dc 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: 35 +// version: 36 use sha2::Digest; // rust-rules: trait-import @@ -158,11 +158,15 @@ impl RawTransactionIngestLiveSource { (true, false, false, http_block_scan, true) }, Self::HttpBlockPolling(source) => { - let known_reference_hydration = - match http_role_supports_rpc_method(&source.http_pool, &source.polling_role, "getTransaction", source.network.as_str()) { - std::result::Result::Ok(value) => value, - std::result::Result::Err(error) => return std::result::Result::Err(error), - }; + let known_reference_hydration = match http_role_supports_rpc_method( + &source.http_pool, + &source.polling_role, + "getTransaction", + source.network.as_str(), + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; (false, true, false, true, known_reference_hydration) }, Self::StandardBlock(_) => (false, true, false, false, false), @@ -226,13 +230,21 @@ impl RawTransactionIngestLiveSource { tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty()); let mut source_future = std::boxed::Box::pin(async move { return match self { - Self::HeliusTransaction(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await, + Self::HeliusTransaction(source) => { + source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await + }, Self::HttpBlockPolling(source) => { - source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared.continuity_contracts).await + source + .run(settings, stop_receiver, admission_sender, source_frontier_sender, shared.continuity_contracts) + .await }, Self::StandardBlock(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender).await, - Self::StandardLogs(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await, - Self::Yellowstone(source) => source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await, + Self::StandardLogs(source) => { + source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await + }, + Self::Yellowstone(source) => { + source.run(settings, stop_receiver, admission_sender, source_frontier_sender, shared).await + }, }; }); loop { @@ -813,8 +825,12 @@ impl crate::RawTransactionIngestYellowstoneSource { processing_frontier_sender: tokio::sync::watch::Sender, shared: RawTransactionIngestSourceRuntimeShared, ) -> ksp_core_lib::Result<()> { - let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } = - shared; + let RawTransactionIngestSourceRuntimeShared { + continuity_contracts, + global_hydration_registry, + hydration_in_flight_limit, + hydration_pending_limit, + } = shared; let opened = tokio::select! { biased; _ = stop_receiver.changed() => { @@ -1055,8 +1071,12 @@ impl crate::RawTransactionIngestHeliusTransactionSource { processing_frontier_sender: tokio::sync::watch::Sender, shared: RawTransactionIngestSourceRuntimeShared, ) -> ksp_core_lib::Result<()> { - let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } = - shared; + let RawTransactionIngestSourceRuntimeShared { + continuity_contracts, + global_hydration_registry, + hydration_in_flight_limit, + hydration_pending_limit, + } = shared; let connected = tokio::select! { biased; _ = stop_receiver.changed() => { @@ -1469,7 +1489,9 @@ impl crate::RawTransactionIngestHttpBlockPollingSource { break 'source; } } - if !blocked_by_null && let std::option::Option::Some(proven_end_slot) = discovery.proven_end_slot { + 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() { std::result::Result::Ok(value) => value, @@ -1860,8 +1882,12 @@ impl crate::RawTransactionIngestStandardLogsSource { processing_frontier_sender: tokio::sync::watch::Sender, shared: RawTransactionIngestSourceRuntimeShared, ) -> ksp_core_lib::Result<()> { - let RawTransactionIngestSourceRuntimeShared { continuity_contracts, global_hydration_registry, hydration_in_flight_limit, hydration_pending_limit } = - shared; + let RawTransactionIngestSourceRuntimeShared { + continuity_contracts, + global_hydration_registry, + hydration_in_flight_limit, + hydration_pending_limit, + } = shared; let connected = tokio::select! { biased; _ = stop_receiver.changed() => { @@ -2310,7 +2336,9 @@ impl crate::RawTransactionIngestRuntimeResources { }; let source_key = source.source_key(); let _abort_handle = children.spawn(async move { - let result = source.run(source_settings, source_stop_receiver, source_admission_sender, publisher, source_shared).await; + let result = source + .run(source_settings, source_stop_receiver, source_admission_sender, publisher, source_shared) + .await; return (source_key, result); }); } @@ -2454,18 +2482,19 @@ async fn supervise_live_source_tasks( source_stop_sender.send_replace(true); return drain_live_source_tasks(&mut children, std::option::Option::Some(source_fault)).await; } - let (active_source_keys, processing_frontier_slot) = { + let supervisor_state = { let inventory = match inventory.lock() { std::result::Result::Ok(value) => value, std::result::Result::Err(poisoned) => poisoned.into_inner(), }; - match inventory.supervisor_state() { - 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; - }, - } + inventory.supervisor_state() + }; + let (active_source_keys, processing_frontier_slot) = match supervisor_state { + 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; + }, }; let continuity_range = match source_loss_continuity_range(&source_fault) { std::result::Result::Ok(value) => value, @@ -2535,7 +2564,9 @@ fn source_loss_continuity_range(error: &ksp_core_lib::Error) -> ksp_core_lib::Re } return match (start_slot, end_slot) { (std::option::Option::None, std::option::Option::None) => std::result::Result::Ok(std::option::Option::None), - (std::option::Option::Some(start), std::option::Option::Some(end)) if end >= start => std::result::Result::Ok(std::option::Option::Some((start, end))), + (std::option::Option::Some(start), std::option::Option::Some(end)) if end >= start => { + std::result::Result::Ok(std::option::Option::Some((start, end))) + }, _ => std::result::Result::Err(crate::runtime_error("continuity.source_loss_range_invalid")), }; } @@ -2553,7 +2584,10 @@ fn source_loss_is_reconcilable(error: &ksp_core_lib::Error) -> bool { } return matches!( context.value(), - "source.configured_source_closed" | "source.continuity_gap_proven" | "source.replay_coverage_unproven" | "source.websocket_incident_unbounded" + "source.configured_source_closed" + | "source.continuity_gap_proven" + | "source.replay_coverage_unproven" + | "source.websocket_incident_unbounded" ); }); } @@ -2802,7 +2836,10 @@ fn validate_http_block_discovery_result( } let proven_end_slot = discovered.last().copied().map(|slot| return slot.min(end_slot)); let produced_slots = discovered.iter().copied().take_while(|slot| return *slot <= end_slot).collect(); - std::result::Result::Ok(RawTransactionIngestHttpDiscoveryWindow { produced_slots, proven_end_slot }) + std::result::Result::Ok(RawTransactionIngestHttpDiscoveryWindow { + produced_slots, + proven_end_slot, + }) }, }; } @@ -3194,7 +3231,9 @@ fn standard_logs_live_source_key( return hasher.finalize().into(); } -fn yellowstone_coverage_scope_fingerprint(request: &ksp_onchain_transport_lib::YellowstoneSubscribeRequest) -> ksp_core_lib::Result<[u8; 32]> { +fn yellowstone_coverage_scope_fingerprint( + request: &ksp_onchain_transport_lib::YellowstoneSubscribeRequest, +) -> ksp_core_lib::Result<[u8; 32]> { let mut normalized = request.clone(); normalized.set_commitment(std::option::Option::None); normalized.set_ping(std::option::Option::None); @@ -4010,7 +4049,11 @@ impl RawTransactionIngestProcessingFrontierReporter { } fn observe_websocket_session_snapshot(&mut self, snapshot: ksp_onchain_transport_lib::WsSessionSnapshot) -> ksp_core_lib::Result<()> { - return self.observe_websocket_continuity(map_websocket_source_state(snapshot.state()), snapshot.continuity_gap_count(), snapshot.overflow_count()); + return self.observe_websocket_continuity( + map_websocket_source_state(snapshot.state()), + snapshot.continuity_gap_count(), + snapshot.overflow_count(), + ); } fn observe_websocket_continuity( @@ -4051,15 +4094,19 @@ impl RawTransactionIngestProcessingFrontierReporter { } }, std::option::Option::None => { - let anchor = - match crate::RawTransactionIngestWebSocketIncidentAnchor::new(start_slot, reconnect_total, overflow_total, saw_reconnect, saw_overflow) - { - std::result::Result::Ok(value) => value, - std::result::Result::Err(error) => { - self.publish(); - return std::result::Result::Err(error); - }, - }; + let anchor = match crate::RawTransactionIngestWebSocketIncidentAnchor::new( + start_slot, + reconnect_total, + overflow_total, + saw_reconnect, + saw_overflow, + ) { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + self.publish(); + return std::result::Result::Err(error); + }, + }; self.websocket_incident_anchor = std::option::Option::Some(anchor); }, } @@ -4378,7 +4425,13 @@ impl RawTransactionIngestHydrationCoordinator { self.pending_signal_count -= pending.signals.len(); for pending_signal in pending.signals { 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 = resolve_known_reference_hydration( + hydration, + settings, + pending_signal.signal, + pending_signal.received_at, + &fetched.observed, + ); let resolution = match resolution { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), @@ -4586,9 +4639,16 @@ fn resolve_known_reference_hydration( std::result::Result::Err(error) => return std::result::Result::Err(error), }; if let std::option::Option::Some(value) = ingress { - return std::result::Result::Ok(RawTransactionIngestKnownReferenceHydrationResolution::Available(std::boxed::Box::new(value))); + return std::result::Result::Ok(RawTransactionIngestKnownReferenceHydrationResolution::Available(std::boxed::Box::new( + value, + ))); } - let block_slot_supported = match http_role_supports_rpc_method(&hydration.http_pool, &hydration.hydration_role, "getBlock", hydration.network.as_str()) { + let block_slot_supported = match http_role_supports_rpc_method( + &hydration.http_pool, + &hydration.hydration_role, + "getBlock", + hydration.network.as_str(), + ) { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error), }; diff --git a/deltas/0.3.14/pre.008-fix.001.md b/deltas/0.3.14/pre.008-fix.001.md new file mode 100644 index 0000000..a862d67 --- /dev/null +++ b/deltas/0.3.14/pre.008-fix.001.md @@ -0,0 +1,172 @@ + + + +# Delta `0.3.14-pre.008-fix.001` — libération du guard avant `await` + +## Base requise + +```text +0.3.14-pre.008 +workspace.package.version = 0.3.14-pre.8 +deltas/0.3.14/pre.008.md présent +``` + +## Objectif + +Corriger strictement la tranche `pre.008` après le gate opérateur, sans modifier sa responsabilité fonctionnelle : + +```text +rendre le future du supervisor Send +ne jamais conserver un std::sync::MutexGuard à travers un await +préserver exactement la décision source-loss de pre.008 +ne pas avancer la health policy prévue en pre.009 +``` + +## Défaut observé + +Le gate opérateur de `0.3.14-pre.008` a produit : + +```text +cargo fmt --all : PASS +cargo fmt --all -- --check : PASS +audit Rust workspace rules : PASS +audit Markdown tables : PASS +cargo check --workspace : FAIL +cargo clippy --workspace --all-targets --all-features -- -D warnings : FAIL +cargo test -p ksp-worker-raw-transaction-ingest-lib --all-targets --all-features : compilation impossible sur le même défaut Send +``` + +Le compilateur refuse le future de `supervise_live_source_tasks` : + +```text +future cannot be sent between threads safely +std::sync::MutexGuard is not Send +await dans drain_live_source_tasks alors que le guard peut encore être vivant +``` + +## Cause + +`pre.008` appelait `inventory.supervisor_state()` dans un bloc qui contenait aussi la branche d'erreur avec `drain_live_source_tasks(...).await`. + +Même si le guard n'est pas utilisé après `supervisor_state()`, sa portée lexicale englobait la branche asynchrone. Le compilateur devait donc considérer que le `std::sync::MutexGuard` pouvait traverser l'`await`, ce qui rendait le future non-`Send` et interdisait son utilisation dans `JoinSet::spawn` et `tokio::spawn`. + +## Correction + +La lecture synchronisée de l'inventaire est séparée en deux étapes : + +```text +1. verrouiller inventory et calculer supervisor_state() dans un bloc strictement synchrone +2. sortir de ce bloc, donc détruire le MutexGuard +3. seulement ensuite traiter Result et éventuellement await drain_live_source_tasks(...) +``` + +Aucun `tokio::Mutex` n'est introduit. Aucun lock n'est conservé pendant une attente asynchrone. + +La valeur, les erreurs et les décisions issues de `supervisor_state()` restent identiques. + +## Fichiers ajoutés + +```text +deltas/0.3.14/pre.008-fix.001.md +``` + +## Fichiers modifiés + +```text +Cargo.toml +crates/ksp-worker-raw-transaction-ingest-lib/src/runtime_resources.rs +``` + +## Fichiers supprimés + +```text +aucun +``` + +## Version Cargo + +Le fix modifie du code Rust ; conformément à `VER-ID-007` et `VER-ID-010` : + +```text +header Cargo.toml : 576 -> 577 +workspace.package.version : 0.3.14-pre.8 -> 0.3.14-pre.8.fix.1 +``` + +Version de fichier : + +```text +src/runtime_resources.rs : 35 -> 36 +``` + +## Frontières préservées + +```text +aucun changement de TargetCoverage +aucun changement de continuity frontier +aucun changement de coverage epoch +aucun changement de gap ledger +aucun changement de source-loss classification +aucun respawn Worker +aucun nouveau scheduler ou task +aucun changement Transport +aucune nouvelle requête HTTP +aucune nouvelle dépendance ou feature +aucun accès Config +aucun Job Backfill +aucun backend Store physique +aucune croissance de surface publique +``` + +## Validations exécutées + +Dans le sandbox de préparation : + +```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 +inspection de la portée du MutexGuard autour de supervise_live_source_tasks +comparaison exacte pre.008 -> pre.008-fix.001 +contrôle du contenu de l'archive delta +unzip -t de l'archive delta +``` + +## Validations non exécutées + +Le sandbox de préparation ne fournit pas le toolchain Cargo/Rust. Les gates suivants restent à exécuter côté opérateur : + +```text +cargo fmt --all +cargo fmt --all -- --check +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 +``` + +## Décisions prises + +```text +fix strict de pre.008 +réduction de portée lexicale du std::sync::MutexGuard +aucun passage à tokio::Mutex +aucun allow/expect lint +aucune modification fonctionnelle de la réconciliation +pre.009 reste la première tranche autorisée à modifier la health policy multi-source +``` + +## Questions ouvertes + +```text +aucune pour ce fix +``` + +## 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 +```