From cf43cb840c80ba96a844abfbb9f0a8f6b058567c Mon Sep 17 00:00:00 2001 From: roman Date: Sun, 5 Jul 2026 23:25:39 -0600 Subject: [PATCH 01/13] fix(zakura): keep Zakura reactors alive when the stall watchdog falls back to legacy sync Killing the reactors turned every fallback into a fleet-wide Zakura outage: the fallback node is often the only peer with working block ingest, and with its reactors dead it stops advertising, serving, and forwarding, pinning every Zakura peer's frontier. Legacy ChainSync now resumes as the body-sync driver while Zakura quiesces into a serving/advertising bridge. Also makes inbound block-gossip ingest failures and queue drops visible at info with counters, so tip-ingest freezes are attributable from logs. --- CHANGELOG.md | 13 +++ zebrad/src/commands/start.rs | 17 ++-- zebrad/src/components/inbound/downloads.rs | 31 +++++-- zebrad/src/components/sync.rs | 66 +++++++------- zebrad/src/components/sync/tests/fallback.rs | 91 +++++++------------- 5 files changed, 106 insertions(+), 112 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b38ae2c84c1..7e99337d58b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,19 @@ and this project adheres to [Semantic Versioning](https://semver.org). - Fixed a near-tip sync restart loop when a timed-out `AwaitUtxo` lookup in the transaction verifier was converted to `InternalDowncastError` instead of a missing transparent input. +- The dual-stack Zakura stall watchdog no longer shuts down the Zakura header- + and block-sync reactors when it falls back to legacy `ChainSync`. Killing + them turned every fallback into a fleet-wide Zakura outage: the fallback + node — often the only peer with working block ingest — stopped advertising + statuses, serving headers and bodies, and forwarding `NewBlock`s, which + pinned every Zakura peer's frontier and starved zakura-only nodes until a + manual restart. Legacy `ChainSync` now resumes as the body-sync driver while + the Zakura reactors stay alive, follow local commits through the chain-tip + mirror, and keep serving the Zakura network as a bridge. Fallback engagement + is counted in `sync.zakura.legacy_fallback.engaged`. Inbound block-gossip + ingest failures and queue drops — previously silent at default log levels, + which made tip-ingest freezes unattributable — are now logged at info and + counted (`gossip.ingest.failed.count`, `gossip.ingest.duplicate.count`). - Fixed Zakura nodes gossiping and following side-chain blocks. An inbound `NewBlock` that committed as a side chain (for example a testnet min-difficulty branch) still advanced the node's Zakura header and verified diff --git a/zebrad/src/commands/start.rs b/zebrad/src/commands/start.rs index 6e20001c166..8f378bcb8a3 100644 --- a/zebrad/src/commands/start.rs +++ b/zebrad/src/commands/start.rs @@ -73,7 +73,7 @@ //! //! Some of the diagnostic features are optional, and need to be enabled at compile-time. -mod zakura; +pub(crate) mod zakura; use std::{net::SocketAddr, path::Path, sync::Arc}; @@ -980,20 +980,13 @@ impl StartCmd { let syncer_task_handle = if use_zakura_block_sync(&config.network) { info!("Zakura block sync is replacing the legacy ChainSync body downloader"); // Only dual-stack nodes (Zakura + legacy peers) fall back to legacy ChainSync on a - // Zakura stall; a Zakura-only node has no legacy peers to drive body sync. When the - // fallback fires it first cancels the Zakura endpoint shutdown token (stopping the - // header- and block-sync drivers) so the two commit pipelines never run at once. + // Zakura stall; a Zakura-only node has no legacy peers to drive body sync. The + // fallback resumes legacy ChainSync as the body-sync driver while the Zakura + // reactors stay alive as a serving/advertising bridge for zakura-only peers. let legacy_fallback = config.network.v2_p2p && config.network.legacy_p2p; tokio::spawn( syncer - .bootstrap_genesis_then_pause( - read_only_state_service.clone(), - legacy_fallback, - zakura_endpoint.clone(), - zakura_endpoint - .as_ref() - .and_then(|endpoint| endpoint.header_sync_shutdown()), - ) + .bootstrap_genesis_then_pause(read_only_state_service.clone(), legacy_fallback) .in_current_span(), ) } else { diff --git a/zebrad/src/components/inbound/downloads.rs b/zebrad/src/components/inbound/downloads.rs index 57bc3000739..95faffe9208 100644 --- a/zebrad/src/components/inbound/downloads.rs +++ b/zebrad/src/components/inbound/downloads.rs @@ -214,7 +214,26 @@ where let (result, hash) = match join_result.expect("block download and verify tasks must not panic") { Ok(hash) => (Ok(hash), hash), - Err((e, hash, advertiser_addr)) => (Err((e, advertiser_addr)), hash), + Err((e, hash, advertiser_addr)) => { + // A failed gossip ingest is the only way a node whose + // syncer is parked (Zakura mode) falls behind the + // network, so every non-duplicate failure must be + // visible at default log levels: silent failures here + // previously made fleet-wide tip-ingest freezes + // unattributable after the fact. + if crate::commands::start::zakura::block_verify_error_is_duplicate(&e) { + debug!(?hash, "gossiped block was already known"); + metrics::counter!("gossip.ingest.duplicate.count").increment(1); + } else { + info!( + ?hash, + error = ?e, + "gossiped block download or verify failed" + ); + metrics::counter!("gossip.ingest.failed.count").increment(1); + } + (Err((e, advertiser_addr)), hash) + } }; if let Some((_, Some(source))) = this.cancel_handles.remove(&hash) { let source_count = this @@ -310,7 +329,7 @@ where } if self.queue_len() >= self.full_verify_concurrency_limit { - debug!( + info!( ?hash, queue_len = self.queue_len(), concurrency_limit = self.full_verify_concurrency_limit, @@ -328,7 +347,7 @@ where let source_count = self.source_counts.get(source).copied().unwrap_or_default(); let source_limit = source.max_in_flight(self.full_verify_concurrency_limit); if source_count >= source_limit { - debug!( + info!( ?hash, ?source, source_count, @@ -499,7 +518,7 @@ where let block_height = block .coinbase_height() .ok_or_else(|| { - debug!( + info!( ?hash, "gossiped block with no height: dropped downloaded block" ); @@ -510,7 +529,7 @@ where .map_err(|e| (e, None))?; if block_height > max_lookahead_height { - debug!( + info!( ?hash, ?block_height, ?tip_height, @@ -522,7 +541,7 @@ where Err("gossiped block height too far ahead").map_err(|e| (e.into(), None))?; } else if block_height < min_accepted_height { - debug!( + info!( ?hash, ?block_height, ?tip_height, diff --git a/zebrad/src/components/sync.rs b/zebrad/src/components/sync.rs index 37af7aa76a4..d24e8582cd1 100644 --- a/zebrad/src/components/sync.rs +++ b/zebrad/src/components/sync.rs @@ -516,22 +516,23 @@ where } } -/// Cancels and drains the Zakura sync drivers before the legacy [`ChainSync::sync`] loop resumes. +/// Records that legacy [`ChainSync::sync`] is resuming as the body-sync driver +/// while the Zakura reactors stay alive. /// -/// The cancellation token stops new Zakura sync work. Awaiting the endpoint-owned tasks makes the -/// hand-off a commit barrier: already-started Zakura block applies finish before legacy can submit -/// commits through the same verifier and state pipeline. No-op when Zakura networking is absent. -async fn stop_zakura_sync( - zakura_endpoint: Option<&zn::zakura::ZakuraEndpoint>, - zakura_shutdown: &Option, -) { - if let Some(token) = zakura_shutdown { - token.cancel(); - } - - if let Some(endpoint) = zakura_endpoint { - endpoint.shutdown_sync_tasks().await; - } +/// The Zakura header- and block-sync reactors are deliberately NOT cancelled: +/// killing them turned every fallback into a fleet-wide Zakura outage — the +/// fallback node is often the only peer with working ingest, and with its +/// reactors dead it stops advertising statuses, serving headers/bodies, and +/// forwarding `NewBlock`s, starving every zakura-only peer and pinning the +/// other nodes' frontiers (they then fall back too). Both engines coexist +/// safely: the reactors are frontier-driven, so once legacy commits ahead of +/// the Zakura frontier they stop requesting and quiesce into a serving role, +/// following local commits through the chain-tip mirror — the same way +/// gossip-driven commits already coexist with the legacy syncer. Duplicate +/// commits from the overlap window are rejected by the verifier as usual. +fn engage_legacy_fallback_alongside_zakura() { + metrics::counter!("sync.zakura.legacy_fallback.engaged").increment(1); + metrics::gauge!("sync.zakura.legacy_fallback.active").set(1.0); } /// Sync configuration section. @@ -953,26 +954,25 @@ where /// Zakura body-sync peers, and parking forever there leaves it stuck at genesis. /// /// `legacy_fallback` (set when the node runs both stacks, `v2_p2p && legacy_p2p`) - /// controls the recovery path. When `true`, a Zakura stall first cancels - /// `zakura_shutdown` — the endpoint shutdown token shared by the Zakura header- and - /// block-sync drivers — so they stop before the legacy [`ChainSync::sync`] loop - /// resumes, ensuring only one body-sync committer is ever active (two at once break - /// the state-commit pipeline's accounting and can deadlock the node). When `false` - /// (a Zakura-only node, where falling back to absent legacy peers is pointless), - /// the watchdog never switches: it parks and warns (once per stall window) so a - /// stalled, eclipsed, or peerless node is visible in the logs. + /// controls the recovery path. When `true`, a Zakura stall resumes the legacy + /// [`ChainSync::sync`] loop as the body-sync driver while the Zakura header- and + /// block-sync reactors stay alive: they follow local commits through the chain-tip + /// mirror and quiesce into a serving/advertising role, so the fallback node keeps + /// acting as a Zakura bridge for zakura-only peers (see + /// [`engage_legacy_fallback_alongside_zakura`]). When `false` (a Zakura-only node, + /// where falling back to absent legacy peers is pointless), the watchdog never + /// switches: it parks and warns (once per stall window) so a stalled, eclipsed, or + /// peerless node is visible in the logs. /// /// `read_state` answers [`ReadRequest::BestHeaderTip`](zs::ReadRequest::BestHeaderTip) /// for the legacy-informed cross-check, which only probes legacy peers when /// the verified tip is frozen and the node looks caught up to its own header /// frontier. - #[instrument(skip(self, read_state, zakura_endpoint, zakura_shutdown))] + #[instrument(skip(self, read_state))] pub async fn bootstrap_genesis_then_pause( mut self, mut read_state: RS, legacy_fallback: bool, - zakura_endpoint: Option, - zakura_shutdown: Option, ) -> Result<(), Report> where RS: Service @@ -1029,11 +1029,11 @@ where verified_tip = ?verified_height, header_tip = ?header_tip_height, stall = ?ZAKURA_BODY_SYNC_STALL_TIMEOUT, - "Zakura body sync is not closing the gap to the network tip; stopping \ - Zakura sync drivers and falling back to legacy ChainSync so legacy peers \ - can drive body sync" + "Zakura body sync is not closing the gap to the network tip; resuming \ + legacy ChainSync as the body-sync driver while Zakura keeps serving \ + peers and following local commits" ); - stop_zakura_sync(zakura_endpoint.as_ref(), &zakura_shutdown).await; + engage_legacy_fallback_alongside_zakura(); return self.sync().await; } ZakuraWatchdogAction::ProbeLegacyPeers => { @@ -1049,10 +1049,10 @@ where header_tip = ?header_tip_height, ?blocks_ahead, "Zakura body sync is frozen while legacy peers advertise a much \ - higher tip; stopping Zakura sync drivers and falling back to legacy \ - ChainSync so it can drive body sync" + higher tip; resuming legacy ChainSync as the body-sync driver \ + while Zakura keeps serving peers and following local commits" ); - stop_zakura_sync(zakura_endpoint.as_ref(), &zakura_shutdown).await; + engage_legacy_fallback_alongside_zakura(); return self.sync().await; } } diff --git a/zebrad/src/components/sync/tests/fallback.rs b/zebrad/src/components/sync/tests/fallback.rs index ae67059bf04..b885e14655a 100644 --- a/zebrad/src/components/sync/tests/fallback.rs +++ b/zebrad/src/components/sync/tests/fallback.rs @@ -1,18 +1,15 @@ //! Tests for the Zakura body-sync stall watchdog //! ([`ChainSync::bootstrap_genesis_then_pause`]). //! -//! These exercise the pure decision function [`zakura_block_sync_stalled`] and the -//! [`stop_zakura_sync`] hand-off helper directly, so they are deterministic and need -//! no clock, services, or live `ChainTip`. - -use tokio_util::sync::CancellationToken; +//! These exercise the pure decision function [`zakura_block_sync_stalled`] directly, +//! so they are deterministic and need no clock, services, or live `ChainTip`. use zebra_chain::{block::Height, chain_sync_status::ChainSyncStatus}; use super::super::{ - legacy_probe_supports_fallback, stop_zakura_sync, zakura_block_sync_stalled, - zakura_sync_status_length, zakura_watchdog_action, SyncStatus, ZakuraLegacyProbe, - ZakuraStallTracker, ZakuraWatchdogAction, ZAKURA_LEGACY_BEHIND_THRESHOLD, + legacy_probe_supports_fallback, zakura_block_sync_stalled, zakura_sync_status_length, + zakura_watchdog_action, SyncStatus, ZakuraLegacyProbe, ZakuraStallTracker, + ZakuraWatchdogAction, ZAKURA_LEGACY_BEHIND_THRESHOLD, }; /// The original height-only rule, reproduced here only to demonstrate the F-88602 @@ -214,11 +211,14 @@ fn frozen_but_materially_behind_leaves_probe_to_gap_rule() { } } -#[tokio::test] -async fn stalled_zakura_with_legacy_fallback_cancels_the_shutdown_token() { +/// The fallback decision fires on a frozen verified tip, and the hand-off keeps +/// the Zakura reactors alive: legacy ChainSync resumes as the body-sync driver +/// while Zakura quiesces into a serving/advertising bridge. Killing the +/// reactors here previously turned every fallback into a fleet-wide Zakura +/// outage (the fallback node is often the only peer with working ingest). +#[test] +fn stalled_zakura_with_legacy_fallback_keeps_zakura_reactors_alive() { let max_idle_polls = 3; - let token = CancellationToken::new(); - let driver_view = token.child_token(); let mut tracker = ZakuraStallTracker::new(Some(Height(0))); let mut legacy_probe = ZakuraLegacyProbe::new(Some(Height(0))); @@ -241,18 +241,14 @@ async fn stalled_zakura_with_legacy_fallback_cancels_the_shutdown_token() { ZakuraWatchdogAction::FallbackToLegacy, "a frozen verified tip must trigger legacy fallback when it is enabled" ); - stop_zakura_sync(None, &Some(token)).await; - assert!( - driver_view.is_cancelled(), - "falling back to legacy must cancel the Zakura sync drivers' shutdown token" - ); + // The hand-off is now driver-only: nothing in the fallback path may cancel + // Zakura work, so there is no shutdown token to assert on. + super::super::engage_legacy_fallback_alongside_zakura(); } #[test] fn stalled_zakura_without_legacy_fallback_keeps_waiting() { let max_idle_polls = 3; - let token = CancellationToken::new(); - let driver_view = token.child_token(); let mut tracker = ZakuraStallTracker::new(Some(Height(0))); let mut legacy_probe = ZakuraLegacyProbe::new(Some(Height(0))); @@ -281,18 +277,15 @@ fn stalled_zakura_without_legacy_fallback_keeps_waiting() { saw_warn_only, "Zakura-only stalls should still produce the warn-only watchdog action" ); - assert!( - !driver_view.is_cancelled(), - "warn-only Zakura stalls must not cancel the Zakura shutdown token" - ); } -#[tokio::test] -async fn frozen_zero_gap_with_legacy_peers_ahead_cancels_the_shutdown_token() { +/// A frozen tip that looks caught up cross-checks legacy peers, and a probe at +/// or above the behind threshold engages the legacy fallback — which keeps the +/// Zakura reactors alive rather than cancelling anything. +#[test] +fn frozen_zero_gap_with_legacy_peers_ahead_engages_fallback() { let max_idle_polls = 5; let frozen = Some(Height(1_000)); - let token = CancellationToken::new(); - let driver_view = token.child_token(); let mut tracker = ZakuraStallTracker::new(frozen); let mut legacy_probe = ZakuraLegacyProbe::new(frozen); @@ -318,11 +311,7 @@ async fn frozen_zero_gap_with_legacy_peers_ahead_cancels_the_shutdown_token() { "legacy peers at or above the behind threshold must trigger fallback" ); - stop_zakura_sync(None, &Some(token)).await; - assert!( - driver_view.is_cancelled(), - "legacy-informed fallback must cancel the Zakura sync drivers' shutdown token" - ); + super::super::engage_legacy_fallback_alongside_zakura(); } #[test] @@ -405,33 +394,13 @@ fn zakura_sync_status_lengths_drive_existing_mempool_gate() { ); } -/// The point of this test is to lock in the fallback behavior: when Zebra decides to stop using -/// Zakura sync and fall back to legacy sync, it must signal the running Zakura driver tasks to shut down. -/// This asserts that the shutdown token is cancelled when the fallback occurs. -#[tokio::test] -async fn fallback_cancels_the_zakura_shutdown_token() { - let token = CancellationToken::new(); - assert!( - !token.is_cancelled(), - "precondition: a fresh token is not cancelled" - ); - - // A child token stands in for the drivers' observed shutdown: cancelling the shared token the - // watchdog holds must propagate to what the drivers actually await. - let driver_view = token.child_token(); - - stop_zakura_sync(None, &Some(token)).await; - - assert!( - driver_view.is_cancelled(), - "falling back to legacy must cancel the Zakura sync drivers' shutdown token" - ); -} - -/// On a Zakura-only node there is no endpoint shutdown token, so the hand-off helper must be a -/// no-op rather than panic. -#[tokio::test] -async fn stop_zakura_sync_is_a_noop_without_a_token() { - // Must not panic. - stop_zakura_sync(None, &None).await; +/// Locks in the fallback behavior: engaging legacy fallback must not signal any +/// Zakura shutdown — the reactors stay alive as a serving bridge. This test +/// documents the intentional inversion of the old "fallback cancels the Zakura +/// shutdown token" contract. +#[test] +fn fallback_does_not_cancel_any_zakura_shutdown_token() { + // The hand-off helper takes no token and cancels nothing; it only records + // the mode switch. Must not panic. + super::super::engage_legacy_fallback_alongside_zakura(); } From a1d409226d886221012fb0d384a677c9f84dfe46 Mon Sep 17 00:00:00 2001 From: roman Date: Sun, 5 Jul 2026 23:52:54 -0600 Subject: [PATCH 02/13] fix(state): don't panic the write task when a fork marker is above a truncated chain tip Invalidating a chain suffix (Zakura fork recovery) can leave last_fork_height above the new tip; recent_fork_length deliberately returns None for that, but the progress-bar metrics caller expected Some and panicked the block write task, crash-looping the node. Metrics display code must never panic. --- zebra-state/src/service/non_finalized_state.rs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/zebra-state/src/service/non_finalized_state.rs b/zebra-state/src/service/non_finalized_state.rs index 337cf01ff1c..2baff78c2be 100644 --- a/zebra-state/src/service/non_finalized_state.rs +++ b/zebra-state/src/service/non_finalized_state.rs @@ -1002,11 +1002,14 @@ impl NonFinalizedState { // For now, we don't show any work here, see the deleted code in PR #7087. let mut desc = String::new(); - if let Some(recent_fork_height) = chain.recent_fork_height() { - let recent_fork_length = chain - .recent_fork_length() - .expect("just checked recent fork height"); - + // Both are `None`-checked: invalidating a chain suffix can leave + // `last_fork_height` above the truncated tip, making the length + // `None` while the height is `Some`. This is metrics-only display + // code and must never panic the block write task (it crash-looped + // a node when the Zakura fork-recovery invalidation first hit it). + if let (Some(recent_fork_height), Some(recent_fork_length)) = + (chain.recent_fork_height(), chain.recent_fork_length()) + { let mut plural = "s"; if recent_fork_length == 1 { plural = ""; From 86895c61fccceddf48dad87c2ca1d5380f3d86fd Mon Sep 17 00:00:00 2001 From: roman Date: Mon, 6 Jul 2026 00:10:26 -0600 Subject: [PATCH 03/13] fix(zakura): reconcile a body suffix stranded across a crashed fork recovery If the node exits between a durable header reorg and its body invalidation, later header commits find the rows already replaced and report no conflict, so the event-based invalidation never fires again and the body chain stays stranded on the abandoned branch forever. Committed ranges overlapping the body tip are now compared against the best body chain, and the first stranded body block is invalidated (one extra state read per commit only when the range overlaps committed bodies). --- zebra-state/src/request.rs | 11 ++ zebra-state/src/service.rs | 4 + .../service/finalized_state/zebra_db/block.rs | 6 +- zebrad/src/commands/start.rs | 109 ++++++++++++++- .../start/zakura/header_sync_driver.rs | 124 ++++++++++++++++++ zebrad/src/commands/start/zakura/mod.rs | 4 +- 6 files changed, 250 insertions(+), 8 deletions(-) diff --git a/zebra-state/src/request.rs b/zebra-state/src/request.rs index b834aa935cd..8718db6075b 100644 --- a/zebra-state/src/request.rs +++ b/zebra-state/src/request.rs @@ -1568,6 +1568,16 @@ pub enum ReadRequest { /// * [`ReadResponse::BlockHash(None)`](ReadResponse::BlockHash) otherwise. BestChainBlockHash(block::Height), + /// Looks up the hash in the stored Zakura header chain at the given height. + /// + /// Unlike [`ReadRequest::BestChainBlockHash`], this never falls back to the + /// best body chain: it answers what the Zakura header store holds, so the + /// two chains can be compared to detect a body suffix stranded off the + /// header chain. + /// + /// Returns [`ReadResponse::BlockHash(Option)`](crate::ReadResponse::BlockHash). + ZakuraHeaderHash(block::Height), + /// Get state information from the best block chain. /// /// Returns [`ReadResponse::ChainInfo(info)`](ReadResponse::ChainInfo) where `info` is a @@ -1653,6 +1663,7 @@ impl ReadRequest { } ReadRequest::BestChainNextMedianTimePast => "best_chain_next_median_time_past", ReadRequest::BestChainBlockHash(_) => "best_chain_block_hash", + ReadRequest::ZakuraHeaderHash(_) => "zakura_header_hash", #[cfg(feature = "indexer")] ReadRequest::SpendingTransactionId(_) => "spending_transaction_id", ReadRequest::ChainInfo => "chain_info", diff --git a/zebra-state/src/service.rs b/zebra-state/src/service.rs index 9b4b4279f84..e9f26409754 100644 --- a/zebra-state/src/service.rs +++ b/zebra-state/src/service.rs @@ -2037,6 +2037,10 @@ impl Service for ReadStateService { read::hash_by_height(state.latest_best_chain(), &state.db, height), )), + ReadRequest::ZakuraHeaderHash(height) => { + Ok(ReadResponse::BlockHash(state.db.zakura_header_hash(height))) + } + // Used by get_block_template and getblockchaininfo RPCs. ReadRequest::ChainInfo => { // # Correctness diff --git a/zebra-state/src/service/finalized_state/zebra_db/block.rs b/zebra-state/src/service/finalized_state/zebra_db/block.rs index 53fb8ae2c99..82b0a8770a5 100644 --- a/zebra-state/src/service/finalized_state/zebra_db/block.rs +++ b/zebra-state/src/service/finalized_state/zebra_db/block.rs @@ -609,7 +609,11 @@ impl ZebraDb { } #[allow(clippy::unwrap_in_result)] - fn zakura_header_hash(&self, height: block::Height) -> Option { + /// Returns the hash stored in the Zakura header chain at `height`, if any. + /// + /// This reads the header store directly, without preferring the best body + /// chain like the serving reads do, so callers can compare the two chains. + pub fn zakura_header_hash(&self, height: block::Height) -> Option { let hash_by_height = self.db.cf_handle(ZAKURA_HEADER_HASH_BY_HEIGHT).unwrap(); self.db.zs_get(&hash_by_height, &height) } diff --git a/zebrad/src/commands/start.rs b/zebrad/src/commands/start.rs index 8f378bcb8a3..a3a8241bbad 100644 --- a/zebrad/src/commands/start.rs +++ b/zebrad/src/commands/start.rs @@ -2123,11 +2123,11 @@ mod zakura_header_sync_driver_tests { coalesce_ready_needed_block_queries, coalesce_stale_needed_block_queries, commit_block_sync_body, drive_block_sync_actions, drive_zakura_header_sync_actions, header_range_commit_failure_kind, notify_block_sync_header_tip, query_block_sync_frontiers, - query_block_sync_needed_blocks, root_covered_query_best_header_tip, - tree_aux_roots_for_served_header_range, verified_block_tip_from_state, BlockApplyClass, - BlocksyncThroughputProbe, ZakuraHeaderSyncDriverHandles, - ZAKURA_BLOCK_SYNC_CHECKPOINT_FRONTIER_REFRESH_INTERVAL, ZAKURA_BLOCK_SYNC_DRIVER_TIMEOUT, - ZAKURA_BLOCK_SYNC_MISSING_BODY_WINDOW, + query_block_sync_needed_blocks, reconcile_stranded_body_suffix, + root_covered_query_best_header_tip, tree_aux_roots_for_served_header_range, + verified_block_tip_from_state, BlockApplyClass, BlocksyncThroughputProbe, + ZakuraHeaderSyncDriverHandles, ZAKURA_BLOCK_SYNC_CHECKPOINT_FRONTIER_REFRESH_INTERVAL, + ZAKURA_BLOCK_SYNC_DRIVER_TIMEOUT, ZAKURA_BLOCK_SYNC_MISSING_BODY_WINDOW, }; fn mainnet_block(bytes: &[u8]) -> Arc { @@ -2903,6 +2903,105 @@ mod zakura_header_sync_driver_tests { endpoint.shutdown().await; } + /// `reconcile_stranded_body_suffix` is a no-op when the body chain matches + /// the Zakura header store, and invalidates the first stranded body block + /// when the committed suffix sits on a branch the header chain reorged away + /// from (e.g. after a crash between a durable header reorg and its body + /// invalidation, when later commits report no header conflict). + #[tokio::test] + async fn stranded_body_suffix_is_reconciled_by_invalidation() { + use std::sync::atomic::{AtomicBool, Ordering}; + + let fork_height = block::Height(99); + let tip_height = block::Height(100); + let shared_hash = block::Hash([0x99; 32]); + let honest_hash_100 = block::Hash([0xAA; 32]); + let stranded_hash_100 = block::Hash([0xBB; 32]); + + // Case 1: the header store matches the body tip; no write happens. + let agree_read = service_fn(move |request: zebra_state::ReadRequest| async move { + match request { + zebra_state::ReadRequest::Tip => Ok::<_, zebra_state::BoxError>( + zebra_state::ReadResponse::Tip(Some((tip_height, honest_hash_100))), + ), + zebra_state::ReadRequest::ZakuraHeaderHash(height) if height == tip_height => { + Ok(zebra_state::ReadResponse::BlockHash(Some(honest_hash_100))) + } + request => panic!("unexpected read request: {request:?}"), + } + }); + let no_write = service_fn(|request: zebra_state::Request| async move { + panic!("agreeing chains must not trigger state writes: {request:?}"); + #[allow(unreachable_code)] + Ok::<_, zebra_state::BoxError>(zebra_state::Response::Committed(block::Hash([0; 32]))) + }); + reconcile_stranded_body_suffix( + no_write, + agree_read, + &zebra_network::zakura::ZakuraTrace::noop(), + ) + .await; + + // Case 2: the body tip diverges from the header store at 100 and the + // chains agree at 99, so the stranded body block at 100 is invalidated. + let stranded_read = service_fn(move |request: zebra_state::ReadRequest| async move { + match request { + zebra_state::ReadRequest::Tip => Ok::<_, zebra_state::BoxError>( + zebra_state::ReadResponse::Tip(Some((tip_height, stranded_hash_100))), + ), + zebra_state::ReadRequest::ZakuraHeaderHash(height) => { + let hash = if height == tip_height { + honest_hash_100 + } else if height == fork_height { + shared_hash + } else { + panic!("unexpected header height: {height:?}"); + }; + Ok(zebra_state::ReadResponse::BlockHash(Some(hash))) + } + zebra_state::ReadRequest::BestChainBlockHash(height) => { + let hash = if height == fork_height { + shared_hash + } else if height == tip_height { + stranded_hash_100 + } else { + panic!("unexpected body height: {height:?}"); + }; + Ok(zebra_state::ReadResponse::BlockHash(Some(hash))) + } + request => panic!("unexpected read request: {request:?}"), + } + }); + let invalidated = Arc::new(AtomicBool::new(false)); + let saw_invalidate = invalidated.clone(); + let write = service_fn(move |request: zebra_state::Request| { + let saw_invalidate = saw_invalidate.clone(); + async move { + match request { + zebra_state::Request::InvalidateBlock(hash) => { + assert_eq!( + hash, stranded_hash_100, + "the stranded body block at the first divergent height is invalidated" + ); + saw_invalidate.store(true, Ordering::SeqCst); + Ok::<_, zebra_state::BoxError>(zebra_state::Response::Invalidated(hash)) + } + request => panic!("unexpected state request: {request:?}"), + } + } + }); + reconcile_stranded_body_suffix( + write, + stranded_read, + &zebra_network::zakura::ZakuraTrace::noop(), + ) + .await; + assert!( + invalidated.load(Ordering::SeqCst), + "a stranded body suffix must be invalidated" + ); + } + #[tokio::test] async fn block_sync_driver_coalesces_stale_needed_queries() { let (action_tx, mut action_rx) = mpsc::channel(8); diff --git a/zebrad/src/commands/start/zakura/header_sync_driver.rs b/zebrad/src/commands/start/zakura/header_sync_driver.rs index 3f69c5f641a..10549fb247b 100644 --- a/zebrad/src/commands/start/zakura/header_sync_driver.rs +++ b/zebrad/src/commands/start/zakura/header_sync_driver.rs @@ -949,6 +949,12 @@ pub(crate) async fn drive_zakura_header_sync_actions { publish_header_frontier( @@ -1210,6 +1216,124 @@ where } } +/// Detects and repairs a body chain stranded off the stored Zakura header chain. +/// +/// Compares the best body chain against the Zakura header store at the body +/// tip. When they disagree, the committed body suffix is on a branch the +/// header chain reorged away from — a state that can outlive the event-based +/// invalidation in the commit path (for example when the node exits between a +/// durable header reorg and its body invalidation: later commits find the +/// header rows already replaced, report no conflict, and nothing else repairs +/// the bodies). Walks down to the fork point (bounded by the state's reorg +/// window) and invalidates the first stranded body block, resetting the chain +/// tip so block sync re-downloads the header chain's branch. +/// +/// Cheap when the chains agree (the normal case): two state reads. +pub(crate) async fn reconcile_stranded_body_suffix( + state: State, + read_state: ReadState, + trace: &ZakuraTrace, +) where + State: Service< + zebra_state::Request, + Response = zebra_state::Response, + Error = zebra_state::BoxError, + > + Clone + + Send + + 'static, + State::Future: Send + 'static, + ReadState: Service< + zebra_state::ReadRequest, + Response = zebra_state::ReadResponse, + Error = zebra_state::BoxError, + > + Clone + + Send + + 'static, + ReadState::Future: Send + 'static, +{ + let header_hash_at = |height: block::Height| { + let read_state = read_state.clone(); + async move { + match read_state + .oneshot(zebra_state::ReadRequest::ZakuraHeaderHash(height)) + .await + { + Ok(zebra_state::ReadResponse::BlockHash(hash)) => hash, + _ => None, + } + } + }; + let body_hash_at = |height: block::Height| { + let read_state = read_state.clone(); + async move { + match read_state + .oneshot(zebra_state::ReadRequest::BestChainBlockHash(height)) + .await + { + Ok(zebra_state::ReadResponse::BlockHash(hash)) => hash, + _ => None, + } + } + }; + + let body_tip = match read_state + .clone() + .oneshot(zebra_state::ReadRequest::Tip) + .await + { + Ok(zebra_state::ReadResponse::Tip(tip)) => tip, + _ => None, + }; + let Some((body_tip_height, body_tip_hash)) = body_tip else { + return; + }; + + // Fast path: no header row (header sync has not covered this height) or a + // matching one means the body suffix is on the header chain. + let Some(header_tip_hash) = header_hash_at(body_tip_height).await else { + return; + }; + if header_tip_hash == body_tip_hash { + return; + } + + // The body suffix is stranded. Walk down to the fork point; the state + // cannot reorg deeper than its reorg window, so bound the walk. + let mut stranded_at = body_tip_height; + let mut stranded_header_hash = header_tip_hash; + for _ in 0..zebra_state::MAX_BLOCK_REORG_HEIGHT { + let Ok(parent) = stranded_at.previous() else { + break; + }; + let (Some(body_hash), Some(header_hash)) = + (body_hash_at(parent).await, header_hash_at(parent).await) + else { + break; + }; + if body_hash == header_hash { + break; + } + stranded_at = parent; + stranded_header_hash = header_hash; + } + + metrics::counter!("sync.header.fork_recovery.body_suffix_reconciled").increment(1); + warn!( + ?body_tip_height, + ?body_tip_hash, + ?stranded_at, + "committed body suffix is stranded off the header chain; reconciling" + ); + invalidate_reorged_body_suffix( + state, + read_state, + stranded_at, + Some(stranded_header_hash), + trace, + ) + .await; +} + /// Drops the committed body suffix stranded by a header-chain reorg. /// /// `reorged_at` is the first height where a committed header range replaced a diff --git a/zebrad/src/commands/start/zakura/mod.rs b/zebrad/src/commands/start/zakura/mod.rs index bff8f791fd8..9c8397d7ffc 100644 --- a/zebrad/src/commands/start/zakura/mod.rs +++ b/zebrad/src/commands/start/zakura/mod.rs @@ -26,8 +26,8 @@ pub(crate) use frontier::{query_block_sync_frontiers, verified_block_tip_from_st pub(crate) use header_sync_driver::{ block_roots_cover_range, block_sync_chain_tip_event, body_sizes_for_served_header_range, chain_tip_mirror_frontier_change, header_range_commit_failure_kind, - notify_block_sync_header_tip, root_covered_query_best_header_tip, - tree_aux_roots_for_served_header_range, + notify_block_sync_header_tip, reconcile_stranded_body_suffix, + root_covered_query_best_header_tip, tree_aux_roots_for_served_header_range, }; pub(crate) use header_sync_driver::{ drive_zakura_header_sync_actions, mirror_zakura_full_block_commits, From 4d771657f68b9a9e44c0e6860ad8414a2915f617 Mon Sep 17 00:00:00 2001 From: roman Date: Mon, 6 Jul 2026 00:57:00 -0600 Subject: [PATCH 04/13] fix(zakura): drain the Zakura apply pipeline before legacy fallback drives commits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two sync engines submitting bulk commits concurrently race in the applying queue, so the fallback is now a commit barrier again — without killing the reactors: an apply gate stops new Zakura block applies once fallback engages and the watchdog waits for in-flight applies to finish (bounded; each apply has its own driver timeout) before legacy ChainSync takes the pipeline. Serving, statuses, header commits, and NewBlock accepts continue unaffected. --- zebrad/src/commands/start.rs | 25 ++++- .../start/zakura/block_sync_driver.rs | 54 ++++++++--- zebrad/src/commands/start/zakura/mod.rs | 95 +++++++++++++++++++ zebrad/src/components/sync.rs | 20 +++- zebrad/src/components/sync/tests/fallback.rs | 59 +++++++++--- 5 files changed, 223 insertions(+), 30 deletions(-) diff --git a/zebrad/src/commands/start.rs b/zebrad/src/commands/start.rs index a3a8241bbad..29925e33dd9 100644 --- a/zebrad/src/commands/start.rs +++ b/zebrad/src/commands/start.rs @@ -663,6 +663,10 @@ impl StartCmd { ) .await; + // Gates the Zakura bulk-apply pipeline so the legacy fallback can drain + // in-flight applies before driving commits through the same pipeline. + let zakura_apply_gate = zakura::ZakuraApplyGate::new(); + if let Some(endpoint) = zakura_endpoint.clone() { let trace = endpoint.trace(); if let (Some(header_sync), Some(shutdown), Some(actions)) = ( @@ -706,6 +710,7 @@ impl StartCmd { config.sync.zakura_block_apply_concurrency_limit, trace.clone(), blocksync_throughput_probe.clone(), + zakura_apply_gate.clone(), shutdown.clone().cancelled_owned(), ) .in_current_span(), @@ -986,7 +991,11 @@ impl StartCmd { let legacy_fallback = config.network.v2_p2p && config.network.legacy_p2p; tokio::spawn( syncer - .bootstrap_genesis_then_pause(read_only_state_service.clone(), legacy_fallback) + .bootstrap_genesis_then_pause( + read_only_state_service.clone(), + legacy_fallback, + zakura_apply_gate.clone(), + ) .in_current_span(), ) } else { @@ -3178,6 +3187,7 @@ mod zakura_header_sync_driver_tests { sync::DEFAULT_ZAKURA_BLOCK_APPLY_CONCURRENCY_LIMIT, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -3320,6 +3330,7 @@ mod zakura_header_sync_driver_tests { sync::DEFAULT_ZAKURA_BLOCK_APPLY_CONCURRENCY_LIMIT, trace.clone(), Some(probe), + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -3469,6 +3480,7 @@ mod zakura_header_sync_driver_tests { sync::DEFAULT_ZAKURA_BLOCK_APPLY_CONCURRENCY_LIMIT, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -3582,6 +3594,7 @@ mod zakura_header_sync_driver_tests { sync::DEFAULT_ZAKURA_BLOCK_APPLY_CONCURRENCY_LIMIT, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -3683,6 +3696,7 @@ mod zakura_header_sync_driver_tests { 1, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -3785,6 +3799,7 @@ mod zakura_header_sync_driver_tests { 1, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -3896,6 +3911,7 @@ mod zakura_header_sync_driver_tests { 1, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -4297,6 +4313,7 @@ mod zakura_header_sync_driver_tests { sync::DEFAULT_ZAKURA_BLOCK_APPLY_CONCURRENCY_LIMIT, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -4390,6 +4407,7 @@ mod zakura_header_sync_driver_tests { zebra_consensus::MAX_CHECKPOINT_HEIGHT_GAP, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -4510,6 +4528,7 @@ mod zakura_header_sync_driver_tests { sync::DEFAULT_ZAKURA_BLOCK_APPLY_CONCURRENCY_LIMIT, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -4633,6 +4652,7 @@ mod zakura_header_sync_driver_tests { sync::DEFAULT_ZAKURA_BLOCK_APPLY_CONCURRENCY_LIMIT, trace, None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -4769,6 +4789,7 @@ mod zakura_header_sync_driver_tests { sync::DEFAULT_ZAKURA_BLOCK_APPLY_CONCURRENCY_LIMIT, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -4898,6 +4919,7 @@ mod zakura_header_sync_driver_tests { sync::DEFAULT_ZAKURA_BLOCK_APPLY_CONCURRENCY_LIMIT, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -5052,6 +5074,7 @@ mod zakura_header_sync_driver_tests { sync::DEFAULT_ZAKURA_BLOCK_APPLY_CONCURRENCY_LIMIT, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, diff --git a/zebrad/src/commands/start/zakura/block_sync_driver.rs b/zebrad/src/commands/start/zakura/block_sync_driver.rs index c20b2c5f7e4..6928ce3b70f 100644 --- a/zebrad/src/commands/start/zakura/block_sync_driver.rs +++ b/zebrad/src/commands/start/zakura/block_sync_driver.rs @@ -111,6 +111,7 @@ pub(crate) async fn drive_block_sync_actions( combined_apply_limit: usize, trace: ZakuraTrace, throughput_probe: Option, + apply_gate: std::sync::Arc, shutdown: impl Future + Send + 'static, ) where ReadState: Service< @@ -159,6 +160,7 @@ pub(crate) async fn drive_block_sync_actions( if shutting_down { if let Some(completed) = in_flight_applies.next().await { handle_completed_block_apply( + &apply_gate, completed, &mut pending_applies, &mut in_flight_applies, @@ -185,6 +187,7 @@ pub(crate) async fn drive_block_sync_actions( if !in_flight_applies.is_empty() { if let Some(Some(completed)) = in_flight_applies.next().now_or_never() { handle_completed_block_apply( + &apply_gate, completed, &mut pending_applies, &mut in_flight_applies, @@ -226,6 +229,7 @@ pub(crate) async fn drive_block_sync_actions( continue; }; handle_completed_block_apply( + &apply_gate, completed, &mut pending_applies, &mut in_flight_applies, @@ -535,6 +539,7 @@ pub(crate) async fn drive_block_sync_actions( block, }); drain_pending_block_applies( + &apply_gate, &mut pending_applies, &mut in_flight_applies, &mut checkpoint_in_flight, @@ -651,6 +656,7 @@ pub(crate) fn coalesce_stale_needed_block_queries( #[allow(clippy::too_many_arguments)] fn handle_completed_block_apply( + apply_gate: &std::sync::Arc, completed: BlockApplyCompletion, pending_applies: &mut VecDeque, in_flight_applies: &mut FuturesUnordered>, @@ -685,6 +691,7 @@ fn handle_completed_block_apply( observe_block_apply_completion(completed, checkpoint_frontier_refresh); drain_pending_block_applies( + apply_gate, pending_applies, in_flight_applies, checkpoint_in_flight, @@ -704,6 +711,7 @@ fn handle_completed_block_apply( #[allow(clippy::too_many_arguments)] fn drain_pending_block_applies( + apply_gate: &std::sync::Arc, pending_applies: &mut VecDeque, in_flight_applies: &mut FuturesUnordered>, checkpoint_in_flight: &mut usize, @@ -732,6 +740,14 @@ fn drain_pending_block_applies( BlockVerifier::Error: std::fmt::Debug + Send + Sync + 'static, BlockVerifier::Future: Send + 'static, { + // Once the pipeline has yielded to legacy ChainSync, start no new applies: + // two engines driving bulk commits concurrently race in the applying + // queue. Queued blocks stay pending; legacy commits cover their heights + // and the sequencer releases them as its frontier advances. + if apply_gate.is_yielded() { + return; + } + // The checkpoint verifier can hold a complete range until its checkpoint is // reached. Keep room for the current range and the next complete range. let checkpoint_pipeline_apply_limit = checkpoint_apply_limit.saturating_mul(2); @@ -763,19 +779,33 @@ fn drain_pending_block_applies( } let class = pending.class; + let Some(permit) = apply_gate.begin_apply() else { + // Yielded while draining: put the block back and stop starting + // applies. Its in-flight accounting was already incremented above, + // so hand it back too. + decrement_in_flight_apply_count(class, checkpoint_in_flight, full_in_flight); + pending_applies.push_front(pending); + return; + }; + let apply = apply_block_sync_body( + block_verifier.clone(), + latest_chain_tip.clone(), + endpoint.clone(), + read_state.clone(), + block_sync.clone(), + pending.token, + pending.block, + class, + trace.clone(), + throughput_probe.clone(), + ); in_flight_applies.push( - apply_block_sync_body( - block_verifier.clone(), - latest_chain_tip.clone(), - endpoint.clone(), - read_state.clone(), - block_sync.clone(), - pending.token, - pending.block, - class, - trace.clone(), - throughput_probe.clone(), - ) + async move { + // Holds the apply-gate slot for the whole apply, so the legacy + // fallback's drain barrier sees it until completion. + let _permit = permit; + apply.await + } .boxed(), ); } diff --git a/zebrad/src/commands/start/zakura/mod.rs b/zebrad/src/commands/start/zakura/mod.rs index 9c8397d7ffc..11acfefd6a0 100644 --- a/zebrad/src/commands/start/zakura/mod.rs +++ b/zebrad/src/commands/start/zakura/mod.rs @@ -7,6 +7,101 @@ use zebra_network::zakura::{ ZakuraTrace, COMMIT_STATE_TABLE, }; +/// Gates the Zakura block-sync bulk-apply pipeline so the legacy `ChainSync` +/// fallback can drain it before driving commits through the same verifier and +/// state pipeline. +/// +/// Two sync engines submitting bulk commits concurrently race in the applying +/// queue (ordering races can poison the parent-error map and trip the +/// same-hash commit lockout), so the fallback must be a commit barrier: once +/// yielded, the block-sync driver starts no new applies, and the watchdog +/// waits for in-flight applies to finish before resuming legacy sync. The +/// Zakura reactors stay alive throughout — serving, statuses, and header +/// commits are unaffected; only bulk body applies are gated. +#[derive(Debug)] +pub(crate) struct ZakuraApplyGate { + yielded: std::sync::atomic::AtomicBool, + in_flight: std::sync::atomic::AtomicUsize, + drained: tokio::sync::Notify, +} + +/// Tracks one in-flight Zakura block apply; dropping it releases the slot and +/// wakes a pending [`ZakuraApplyGate::yield_and_drain`]. +#[derive(Debug)] +pub(crate) struct ZakuraApplyPermit(std::sync::Arc); + +impl ZakuraApplyGate { + pub(crate) fn new() -> std::sync::Arc { + std::sync::Arc::new(Self { + yielded: std::sync::atomic::AtomicBool::new(false), + in_flight: std::sync::atomic::AtomicUsize::new(0), + drained: tokio::sync::Notify::new(), + }) + } + + /// Whether the pipeline has been yielded to legacy sync. + pub(crate) fn is_yielded(&self) -> bool { + self.yielded.load(std::sync::atomic::Ordering::SeqCst) + } + + /// Returns a permit for one block apply, or `None` once the pipeline has + /// been yielded to legacy sync. + pub(crate) fn begin_apply(self: &std::sync::Arc) -> Option { + if self.is_yielded() { + return None; + } + self.in_flight + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + // Re-check after reserving so a yield that lands between the check and + // the increment still sees an accurate in-flight count. + if self.is_yielded() { + self.release(); + return None; + } + Some(ZakuraApplyPermit(self.clone())) + } + + fn release(&self) { + if self + .in_flight + .fetch_sub(1, std::sync::atomic::Ordering::SeqCst) + == 1 + { + self.drained.notify_waiters(); + } + } + + /// Yields the apply pipeline to legacy sync and waits until in-flight + /// applies drain, bounded by `timeout` (each apply already has its own + /// driver timeout, so the bound is a backstop, not the primary limit). + pub(crate) async fn yield_and_drain(&self, timeout: Duration) { + self.yielded + .store(true, std::sync::atomic::Ordering::SeqCst); + let deadline = tokio::time::Instant::now() + timeout; + loop { + let drained = self.drained.notified(); + let in_flight = self.in_flight.load(std::sync::atomic::Ordering::SeqCst); + if in_flight == 0 { + return; + } + if tokio::time::timeout_at(deadline, drained).await.is_err() { + tracing::warn!( + in_flight, + "timed out draining Zakura block applies before legacy fallback; \ + remaining applies resolve through their own driver timeouts" + ); + return; + } + } + } +} + +impl Drop for ZakuraApplyPermit { + fn drop(&mut self) { + self.0.release(); + } +} + pub(crate) mod block_sync_driver; pub(crate) mod frontier; pub(crate) mod header_sync_driver; diff --git a/zebrad/src/components/sync.rs b/zebrad/src/components/sync.rs index d24e8582cd1..d6326c1fad5 100644 --- a/zebrad/src/components/sync.rs +++ b/zebrad/src/components/sync.rs @@ -530,9 +530,18 @@ where /// following local commits through the chain-tip mirror — the same way /// gossip-driven commits already coexist with the legacy syncer. Duplicate /// commits from the overlap window are rejected by the verifier as usual. -fn engage_legacy_fallback_alongside_zakura() { +async fn engage_legacy_fallback_alongside_zakura( + apply_gate: &crate::commands::start::zakura::ZakuraApplyGate, +) { metrics::counter!("sync.zakura.legacy_fallback.engaged").increment(1); metrics::gauge!("sync.zakura.legacy_fallback.active").set(1.0); + // Commit barrier: two engines driving bulk commits concurrently race in + // the applying queue, so stop new Zakura applies and drain the in-flight + // ones before legacy ChainSync takes the pipeline. Each apply has its own + // driver timeout, so the drain bound is a backstop. + apply_gate + .yield_and_drain(std::time::Duration::from_secs(60)) + .await; } /// Sync configuration section. @@ -968,11 +977,12 @@ where /// for the legacy-informed cross-check, which only probes legacy peers when /// the verified tip is frozen and the node looks caught up to its own header /// frontier. - #[instrument(skip(self, read_state))] - pub async fn bootstrap_genesis_then_pause( + #[instrument(skip(self, read_state, apply_gate))] + pub(crate) async fn bootstrap_genesis_then_pause( mut self, mut read_state: RS, legacy_fallback: bool, + apply_gate: std::sync::Arc, ) -> Result<(), Report> where RS: Service @@ -1033,7 +1043,7 @@ where legacy ChainSync as the body-sync driver while Zakura keeps serving \ peers and following local commits" ); - engage_legacy_fallback_alongside_zakura(); + engage_legacy_fallback_alongside_zakura(&apply_gate).await; return self.sync().await; } ZakuraWatchdogAction::ProbeLegacyPeers => { @@ -1052,7 +1062,7 @@ where higher tip; resuming legacy ChainSync as the body-sync driver \ while Zakura keeps serving peers and following local commits" ); - engage_legacy_fallback_alongside_zakura(); + engage_legacy_fallback_alongside_zakura(&apply_gate).await; return self.sync().await; } } diff --git a/zebrad/src/components/sync/tests/fallback.rs b/zebrad/src/components/sync/tests/fallback.rs index b885e14655a..9132a97d608 100644 --- a/zebrad/src/components/sync/tests/fallback.rs +++ b/zebrad/src/components/sync/tests/fallback.rs @@ -241,9 +241,18 @@ fn stalled_zakura_with_legacy_fallback_keeps_zakura_reactors_alive() { ZakuraWatchdogAction::FallbackToLegacy, "a frozen verified tip must trigger legacy fallback when it is enabled" ); - // The hand-off is now driver-only: nothing in the fallback path may cancel - // Zakura work, so there is no shutdown token to assert on. - super::super::engage_legacy_fallback_alongside_zakura(); + // The hand-off drains the apply gate but cancels no Zakura work: with no + // in-flight applies this returns immediately, and the gate ends yielded. + let gate = crate::commands::start::zakura::ZakuraApplyGate::new(); + futures::executor::block_on(super::super::engage_legacy_fallback_alongside_zakura(&gate)); + assert!( + gate.is_yielded(), + "fallback must yield the Zakura apply gate" + ); + assert!( + gate.begin_apply().is_none(), + "no new Zakura applies may start after the fallback engages" + ); } #[test] @@ -311,7 +320,12 @@ fn frozen_zero_gap_with_legacy_peers_ahead_engages_fallback() { "legacy peers at or above the behind threshold must trigger fallback" ); - super::super::engage_legacy_fallback_alongside_zakura(); + let gate = crate::commands::start::zakura::ZakuraApplyGate::new(); + futures::executor::block_on(super::super::engage_legacy_fallback_alongside_zakura(&gate)); + assert!( + gate.is_yielded(), + "fallback must yield the Zakura apply gate" + ); } #[test] @@ -395,12 +409,33 @@ fn zakura_sync_status_lengths_drive_existing_mempool_gate() { } /// Locks in the fallback behavior: engaging legacy fallback must not signal any -/// Zakura shutdown — the reactors stay alive as a serving bridge. This test -/// documents the intentional inversion of the old "fallback cancels the Zakura -/// shutdown token" contract. -#[test] -fn fallback_does_not_cancel_any_zakura_shutdown_token() { - // The hand-off helper takes no token and cancels nothing; it only records - // the mode switch. Must not panic. - super::super::engage_legacy_fallback_alongside_zakura(); +/// Zakura shutdown — the reactors stay alive as a serving bridge — but it must +/// be a commit barrier: in-flight Zakura applies drain before it returns, and +/// no new applies can start afterwards. This documents the intentional +/// inversion of the old "fallback cancels the Zakura shutdown token" contract. +#[tokio::test(start_paused = true)] +async fn fallback_drains_the_apply_gate_without_cancelling_zakura() { + let gate = crate::commands::start::zakura::ZakuraApplyGate::new(); + let permit = gate.begin_apply().expect("applies run before the fallback"); + + let drain_gate = gate.clone(); + let drain = tokio::spawn(async move { + super::super::engage_legacy_fallback_alongside_zakura(&drain_gate).await + }); + + // The barrier must wait for the in-flight apply... + tokio::task::yield_now().await; + assert!( + !drain.is_finished(), + "the drain waits for in-flight applies" + ); + assert!( + gate.begin_apply().is_none(), + "no new Zakura applies may start once the fallback begins" + ); + + // ...and complete as soon as it finishes. + drop(permit); + drain.await.expect("drain task completes"); + assert!(gate.is_yielded()); } From 1fa3595e212d078a37ac8a6be089e7e3665be95e Mon Sep 17 00:00:00 2001 From: roman Date: Mon, 6 Jul 2026 01:51:50 -0600 Subject: [PATCH 05/13] feat(zakura): re-promote Zakura to body-sync driver after legacy catch-up Fallback was permanent per process lifetime: bootstrap_genesis_then_pause exited into the legacy sync loop and nothing ever handed the driver role back, so every min-difficulty burst demoted one more node's Zakura sync to a serve-only bridge until restart. The watchdog is now a cycle: on fallback it yields+drains the apply gate, legacy drives until caught up and stable (three consecutive exhausted sync rounds advancing the tip by at most two blocks), then the gate is returned to Zakura and the watchdog resumes. Hand-back happens between legacy rounds, so no legacy bulk work is in flight when Zakura resumes driving. --- CHANGELOG.md | 7 + zebrad/src/commands/start/zakura/mod.rs | 10 + zebrad/src/components/sync.rs | 213 ++++++++++++++----- zebrad/src/components/sync/tests/fallback.rs | 47 ++++ 4 files changed, 221 insertions(+), 56 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7e99337d58b..5ada0ce846e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,13 @@ and this project adheres to [Semantic Versioning](https://semver.org). - Fixed a near-tip sync restart loop when a timed-out `AwaitUtxo` lookup in the transaction verifier was converted to `InternalDowncastError` instead of a missing transparent input. +- After a legacy fallback, the dual-stack watchdog now re-promotes Zakura to + the body-sync driver once legacy `ChainSync` is caught up and stable (three + consecutive exhausted sync rounds with the tip advancing by at most two + blocks). Fallback was previously permanent per process lifetime, so every + min-difficulty burst demoted one more node's Zakura sync to a serve-only + bridge until restart. The hand-back reuses the apply-gate commit barrier in + reverse and is counted in `sync.zakura.repromoted`. - The dual-stack Zakura stall watchdog no longer shuts down the Zakura header- and block-sync reactors when it falls back to legacy `ChainSync`. Killing them turned every fallback into a fleet-wide Zakura outage: the fallback diff --git a/zebrad/src/commands/start/zakura/mod.rs b/zebrad/src/commands/start/zakura/mod.rs index 11acfefd6a0..3aba77053f6 100644 --- a/zebrad/src/commands/start/zakura/mod.rs +++ b/zebrad/src/commands/start/zakura/mod.rs @@ -71,6 +71,16 @@ impl ZakuraApplyGate { } } + /// Returns the apply pipeline to Zakura after a legacy catch-up completes. + /// + /// Safe without a drain in this direction: the caller only re-promotes + /// after the legacy sync loop has returned between rounds, so no legacy + /// bulk work is in flight when Zakura resumes driving. + pub(crate) fn unyield(&self) { + self.yielded + .store(false, std::sync::atomic::Ordering::SeqCst); + } + /// Yields the apply pipeline to legacy sync and waits until in-flight /// applies drain, bounded by `timeout` (each apply already has its own /// driver timeout, so the bound is a backstop, not the primary limit). diff --git a/zebrad/src/components/sync.rs b/zebrad/src/components/sync.rs index d6326c1fad5..fa9a1a79304 100644 --- a/zebrad/src/components/sync.rs +++ b/zebrad/src/components/sync.rs @@ -530,6 +530,53 @@ where /// following local commits through the chain-tip mirror — the same way /// gossip-driven commits already coexist with the legacy syncer. Duplicate /// commits from the overlap window are rejected by the verifier as usual. +/// Consecutive stable legacy sync rounds required before Zakura is +/// re-promoted to the driver role after a fallback. +const ZAKURA_REPROMOTE_STABLE_ROUNDS: usize = 3; + +/// Maximum state-tip advance (in blocks) a legacy sync round may make and +/// still count as "stable" for re-promotion: at-tip rounds only pick up the +/// occasional gossip-adjacent block, while catch-up rounds move hundreds. +const ZAKURA_REPROMOTE_MAX_ROUND_PROGRESS: u32 = 2; + +/// Counts consecutive stable legacy sync rounds for the re-promotion decision. +#[derive(Debug, Default)] +struct RepromotionTracker { + stable_rounds: usize, +} + +impl RepromotionTracker { + /// Records one legacy sync round; returns `true` when enough consecutive + /// stable rounds have accumulated to re-promote Zakura. + fn record_round( + &mut self, + exhausted: bool, + before: Option, + after: Option, + ) -> bool { + let progressed = match (before, after) { + (Some(before), Some(after)) => after.0.saturating_sub(before.0), + // Unknown tips cannot prove stability. + _ => u32::MAX, + }; + if exhausted && progressed <= ZAKURA_REPROMOTE_MAX_ROUND_PROGRESS { + self.stable_rounds += 1; + } else { + self.stable_rounds = 0; + } + self.stable_rounds >= ZAKURA_REPROMOTE_STABLE_ROUNDS + } +} + +/// Returns the bulk-apply pipeline to Zakura after a completed legacy +/// catch-up, and records the mode switch. +fn repromote_zakura(apply_gate: &crate::commands::start::zakura::ZakuraApplyGate) { + apply_gate.unyield(); + metrics::counter!("sync.zakura.repromoted").increment(1); + metrics::gauge!("sync.zakura.legacy_fallback.active").set(0.0); + warn!("re-promoted Zakura sync to the body-sync driver after legacy catch-up"); +} + async fn engage_legacy_fallback_alongside_zakura( apply_gate: &crate::commands::start::zakura::ZakuraApplyGate, ) { @@ -1002,74 +1049,128 @@ where / ZAKURA_BODY_SYNC_STALL_POLL.as_secs()) .max(1); - let initial_tip = self.latest_chain_tip.best_tip_height(); - let mut tracker = ZakuraStallTracker::new(initial_tip); - let mut legacy_probe = ZakuraLegacyProbe::new(initial_tip); - loop { - sleep(ZAKURA_BODY_SYNC_STALL_POLL).await; - - let verified_height = self.latest_chain_tip.best_tip_height(); - let header_tip_height = best_header_tip_height(&mut read_state).await; - if let Some(sync_length) = zakura_sync_status_length(verified_height, header_tip_height) - { - self.recent_syncs.push_extend_tips_length(sync_length); - } - - match zakura_watchdog_action( - &mut tracker, - &mut legacy_probe, - verified_height, - header_tip_height, - max_idle_polls, - legacy_fallback, - ) { - ZakuraWatchdogAction::ContinueWaiting => continue, - ZakuraWatchdogAction::WarnOnly => { - warn!( - verified_tip = ?verified_height, - header_tip = ?header_tip_height, - stall = ?ZAKURA_BODY_SYNC_STALL_TIMEOUT, - "Zakura body sync is not closing the gap to the network tip; legacy \ - fallback disabled (legacy_p2p is off), continuing to wait for Zakura" - ); - continue; - } - ZakuraWatchdogAction::FallbackToLegacy => { - warn!( - verified_tip = ?verified_height, - header_tip = ?header_tip_height, - stall = ?ZAKURA_BODY_SYNC_STALL_TIMEOUT, - "Zakura body sync is not closing the gap to the network tip; resuming \ - legacy ChainSync as the body-sync driver while Zakura keeps serving \ - peers and following local commits" - ); - engage_legacy_fallback_alongside_zakura(&apply_gate).await; - return self.sync().await; + // Fallback is a cycle, not a ratchet: when Zakura stalls, legacy + // ChainSync drives until the node is caught up and stable, then the + // apply gate is returned to Zakura and the watchdog resumes. Without + // re-promotion, every burst permanently demoted one more node's Zakura + // sync to a serve-only bridge. + 'drive_cycle: loop { + let initial_tip = self.latest_chain_tip.best_tip_height(); + let mut tracker = ZakuraStallTracker::new(initial_tip); + let mut legacy_probe = ZakuraLegacyProbe::new(initial_tip); + loop { + sleep(ZAKURA_BODY_SYNC_STALL_POLL).await; + + let verified_height = self.latest_chain_tip.best_tip_height(); + let header_tip_height = best_header_tip_height(&mut read_state).await; + if let Some(sync_length) = + zakura_sync_status_length(verified_height, header_tip_height) + { + self.recent_syncs.push_extend_tips_length(sync_length); } - ZakuraWatchdogAction::ProbeLegacyPeers => { - let blocks_ahead = self.legacy_peers_blocks_ahead().await; - info!( - verified_tip = ?verified_height, - ?blocks_ahead, - "Zakura watchdog probed legacy peers for connected blocks ahead" - ); - if legacy_probe_supports_fallback(blocks_ahead) { + + match zakura_watchdog_action( + &mut tracker, + &mut legacy_probe, + verified_height, + header_tip_height, + max_idle_polls, + legacy_fallback, + ) { + ZakuraWatchdogAction::ContinueWaiting => continue, + ZakuraWatchdogAction::WarnOnly => { warn!( verified_tip = ?verified_height, header_tip = ?header_tip_height, - ?blocks_ahead, - "Zakura body sync is frozen while legacy peers advertise a much \ - higher tip; resuming legacy ChainSync as the body-sync driver \ - while Zakura keeps serving peers and following local commits" + stall = ?ZAKURA_BODY_SYNC_STALL_TIMEOUT, + "Zakura body sync is not closing the gap to the network tip; legacy \ + fallback disabled (legacy_p2p is off), continuing to wait for Zakura" + ); + continue; + } + ZakuraWatchdogAction::FallbackToLegacy => { + warn!( + verified_tip = ?verified_height, + header_tip = ?header_tip_height, + stall = ?ZAKURA_BODY_SYNC_STALL_TIMEOUT, + "Zakura body sync is not closing the gap to the network tip; resuming \ + legacy ChainSync as the body-sync driver while Zakura keeps serving \ + peers and following local commits" ); engage_legacy_fallback_alongside_zakura(&apply_gate).await; - return self.sync().await; + self.sync_until_caught_up_and_stable().await?; + repromote_zakura(&apply_gate); + continue 'drive_cycle; + } + ZakuraWatchdogAction::ProbeLegacyPeers => { + let blocks_ahead = self.legacy_peers_blocks_ahead().await; + info!( + verified_tip = ?verified_height, + ?blocks_ahead, + "Zakura watchdog probed legacy peers for connected blocks ahead" + ); + if legacy_probe_supports_fallback(blocks_ahead) { + warn!( + verified_tip = ?verified_height, + header_tip = ?header_tip_height, + ?blocks_ahead, + "Zakura body sync is frozen while legacy peers advertise a much \ + higher tip; resuming legacy ChainSync as the body-sync driver \ + while Zakura keeps serving peers and following local commits" + ); + engage_legacy_fallback_alongside_zakura(&apply_gate).await; + self.sync_until_caught_up_and_stable().await?; + repromote_zakura(&apply_gate); + continue 'drive_cycle; + } } } } } } + /// Drives legacy sync rounds until the node is caught up and the tip is + /// stable, then returns so Zakura can be re-promoted to the driver role. + /// + /// "Caught up and stable" means [`ZAKURA_REPROMOTE_STABLE_ROUNDS`] + /// consecutive sync rounds each exhausted their prospective tip set while + /// advancing the state tip by at most + /// [`ZAKURA_REPROMOTE_MAX_ROUND_PROGRESS`] blocks — legacy is no longer + /// adding anything gossip would not. Returns between rounds, so no legacy + /// bulk downloads are in flight when the caller re-promotes Zakura. + async fn sync_until_caught_up_and_stable(&mut self) -> Result<(), Report> { + let mut tracker = RepromotionTracker::default(); + loop { + let before = self.latest_chain_tip.best_tip_height(); + let round = self.try_to_sync().await; + if round.is_err() { + self.downloads.cancel_all(); + } + let after = self.latest_chain_tip.best_tip_height(); + + if tracker.record_round(round.is_ok(), before, after) { + info!( + state_tip = ?after, + "legacy catch-up is complete and stable; re-promoting Zakura sync" + ); + return Ok(()); + } + + self.update_metrics(); + let restart_delay = if self.is_regtest { + REGTEST_SYNC_RESTART_DELAY + } else { + SYNC_RESTART_SLEEP + }; + info!( + timeout = ?restart_delay, + state_tip = ?after, + "waiting to restart sync" + ); + sleep(restart_delay).await; + } + } + /// Probes the legacy peer set for how far ahead the network is on **our** /// chain, returning the longest run of peer-offered headers that extends a /// block we already have (`None` if no peer answered). diff --git a/zebrad/src/components/sync/tests/fallback.rs b/zebrad/src/components/sync/tests/fallback.rs index 9132a97d608..15ee7f8457f 100644 --- a/zebrad/src/components/sync/tests/fallback.rs +++ b/zebrad/src/components/sync/tests/fallback.rs @@ -439,3 +439,50 @@ async fn fallback_drains_the_apply_gate_without_cancelling_zakura() { drain.await.expect("drain task completes"); assert!(gate.is_yielded()); } + +/// Re-promotion requires consecutive stable rounds: catch-up rounds (large +/// tip advances) and failed rounds reset the count; unknown tips never count. +#[test] +fn repromotion_requires_consecutive_stable_rounds() { + use super::super::RepromotionTracker; + + let mut tracker = RepromotionTracker::default(); + + // Catch-up rounds: big progress, never re-promotes. + assert!(!tracker.record_round(true, Some(Height(0)), Some(Height(500)))); + assert!(!tracker.record_round(true, Some(Height(500)), Some(Height(900)))); + + // Two stable rounds are not enough... + assert!(!tracker.record_round(true, Some(Height(900)), Some(Height(901)))); + assert!(!tracker.record_round(true, Some(Height(901)), Some(Height(901)))); + // ...a failed round resets... + assert!(!tracker.record_round(false, Some(Height(901)), Some(Height(901)))); + assert!(!tracker.record_round(true, Some(Height(901)), Some(Height(902)))); + assert!(!tracker.record_round(true, Some(Height(902)), Some(Height(902)))); + // ...and the third consecutive stable round re-promotes. + assert!(tracker.record_round(true, Some(Height(902)), Some(Height(902)))); + + // Unknown tips cannot prove stability. + let mut tracker = RepromotionTracker::default(); + assert!(!tracker.record_round(true, None, Some(Height(1)))); + assert!(!tracker.record_round(true, Some(Height(1)), None)); +} + +/// The apply gate round-trips: yielded blocks new applies, unyield restores them. +#[tokio::test] +async fn apply_gate_unyield_restores_zakura_applies() { + let gate = crate::commands::start::zakura::ZakuraApplyGate::new(); + assert!(gate.begin_apply().is_some()); + + gate.yield_and_drain(std::time::Duration::from_secs(1)) + .await; + assert!(gate.is_yielded()); + assert!(gate.begin_apply().is_none()); + + gate.unyield(); + assert!(!gate.is_yielded()); + assert!( + gate.begin_apply().is_some(), + "re-promotion must let Zakura applies start again" + ); +} From a433abe623912261425865bf83d93df4a8f11be3 Mon Sep 17 00:00:00 2001 From: roman Date: Mon, 6 Jul 2026 02:28:16 -0600 Subject: [PATCH 06/13] fix(zakura): stop floor-GC from leaving stale block-sync liveness deadlines MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the download floor passed a request (other peers delivered its heights), the GC removed it but only disarmed the liveness deadline for peers whose last delivery was newer than their last request. A healthy peer with an older delivery kept a deadline it could no longer answer, was disconnected for 'no accepted block progress' with zero outstanding, and parked in the 180s no-progress cooldown — at tip this serially exiled peers until the block-sync peer set collapsed (reproduced live 5→0 after a fleet-wide restart, wedging body sync at the next gap). Heights satisfied below the floor now always clear an idle peer's deadline; unresponsive peers are still caught by request timeouts, which deliberately keep the deadline armed. --- CHANGELOG.md | 14 ++ .../src/zakura/block_sync/peer_routine.rs | 132 +++++++++++++++++- 2 files changed, 143 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ada0ce846e..fcd12ba349a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,20 @@ and this project adheres to [Semantic Versioning](https://semver.org). - Fixed a near-tip sync restart loop when a timed-out `AwaitUtxo` lookup in the transaction verifier was converted to `InternalDowncastError` instead of a missing transparent input. +- Fixed Zakura block-sync peers being serially disconnected and parked at the + chain tip by a stale liveness deadline. When the download floor passed a + request because other peers' deliveries satisfied its heights, the floor GC + removed the request but only cleared the liveness deadline for peers whose + last delivery was newer than their last request — a healthy peer with an + older delivery kept a deadline it could no longer answer, was disconnected + for "no accepted block progress" with zero outstanding requests, and parked + in the 180-second no-progress cooldown. At tip this exiled peers one by one + until the block-sync peer set collapsed (reproduced live: 5 → 0 peers within + ~25 minutes of a fleet-wide restart, wedging body sync when the next gap + formed). Heights satisfied below the floor now always clear an idle peer's + deadline; genuinely unresponsive peers are still caught by request timeouts, + which deliberately keep the deadline armed. The liveness disconnect log was + also promoted to info. - After a legacy fallback, the dual-stack watchdog now re-promotes Zakura to the body-sync driver once legacy `ChainSync` is caught up and stable (three consecutive exhausted sync rounds with the tip advancing by at most two diff --git a/zebra-network/src/zakura/block_sync/peer_routine.rs b/zebra-network/src/zakura/block_sync/peer_routine.rs index 3625d439246..0b2d44bb229 100644 --- a/zebra-network/src/zakura/block_sync/peer_routine.rs +++ b/zebra-network/src/zakura/block_sync/peer_routine.rs @@ -1141,7 +1141,7 @@ impl PeerRoutine { now + self.config.effective_no_progress_peer_cooldown(), ); self.trace_protocol_reject_liveness(error); - tracing::debug!( + tracing::info!( peer = ?self.peer, outstanding = self.window.outstanding.len(), "disconnecting Zakura block-sync peer after no accepted block progress" @@ -1185,7 +1185,18 @@ impl PeerRoutine { } if removed { self.publish_outstanding(); - self.window.disarm_liveness_after_progress_if_idle(); + // Heights the floor passed were satisfied by the network; this peer + // no longer owes them, so a liveness deadline armed by those + // requests must not survive the drain. The after-progress disarm is + // not enough: a peer that delivered earlier and then had a NEWER + // request GC'd here has `request_at > block_at`, kept its stale + // deadline, and was disconnected at tip with zero outstanding — + // serially exiling healthy peers into the no-progress cooldown + // until the block-sync peer set collapsed. Dead-session detection + // is unaffected: an unanswered request expires through + // `expire_due_timeouts`, which deliberately leaves the deadline + // armed. + self.window.clear_liveness_if_idle(); } } @@ -2324,8 +2335,11 @@ mod tests { use super::super::peer_registry::PeerRegistry; use super::super::request::BlockSizeEstimate; + use super::super::request::{BlockRangeRequest, ExpectedBlock}; use super::super::sequencer_task::{initial_view, SequencerControlInput}; - use super::super::state::{ByteBudget, ThroughputMeter}; + use super::super::state::{ + ByteBudget, LivenessOutcome, OutstandingBlockRange, ReceivedBlockTracker, ThroughputMeter, + }; use super::super::work_queue::WorkQueue; use super::super::{BlockSyncFrontiers, BlockSyncPeerSession, ZakuraBlockSyncConfig}; use super::PeerRoutine; @@ -2591,4 +2605,116 @@ mod tests { request_timeout )); } + + /// Floor-GC of a request satisfied by other peers must clear the liveness + /// deadline: the peer no longer owes those heights. Before the fix, a peer + /// that had delivered earlier and then had a NEWER request GC'd below the + /// floor kept a stale deadline (`request_at > block_at` fails the + /// after-progress disarm) and was disconnected at tip with zero + /// outstanding, then parked in the 180 s no-progress cooldown — serially + /// exiling healthy peers until the block-sync peer set collapsed + /// (reproduced live: 5 → 0 peers after a fleet-wide restart). + #[tokio::test] + async fn floor_gc_clears_stale_liveness_deadline() { + let config = ZakuraBlockSyncConfig::default(); + let budget = ByteBudget::new(1_000_000); + let work = Arc::new(WorkQueue::new(block::Height(0))); + + let cancel = CancellationToken::new(); + let (out_send, _out_recv) = framed_channel(16); + let (_in_send, in_recv) = framed_channel(16); + let peer = ZakuraPeerId::new(vec![8u8; 32]).expect("test peer id is within bounds"); + let session = BlockSyncPeerSession::for_test(peer.clone(), out_send, cancel.clone()); + + let (sequencer_input_tx, _sequencer_input_rx) = mpsc::channel(16); + let (control_tx, _control_rx) = mpsc::unbounded_channel(); + let (actions_tx, _actions_rx) = mpsc::channel(16); + let (routine_to_reactor_tx, _routine_to_reactor_rx) = mpsc::channel(16); + // The download floor sits above the request below, as if other peers + // delivered those heights and the sequencer committed them. + let (_view_tx, view_rx) = watch::channel(initial_view(BlockSyncFrontiers { + finalized_height: block::Height(0), + verified_block_tip: block::Height(100), + verified_block_hash: block::Hash([0; 32]), + })); + + let mut routine = PeerRoutine::new( + peer, + session, + in_recv, + config, + 0, + budget, + work, + Arc::new(PeerRegistry::new()), + Arc::new(Mutex::new(ThroughputMeter::new(Instant::now()))), + sequencer_input_tx, + Arc::new(AtomicU64::new(0)), + control_tx, + actions_tx, + routine_to_reactor_tx, + view_rx, + cancel, + ZakuraTrace::noop(), + ); + + let now = Instant::now(); + let liveness = Duration::from_secs(30); + + // The peer delivered a block for an earlier request... + routine + .window + .arm_liveness(now - Duration::from_secs(120), liveness); + routine + .window + .note_block_progress(now - Duration::from_secs(90), liveness); + + // ...then received a newer request (deadline re-armed) whose heights + // the floor later passed. + routine + .window + .arm_liveness(now - Duration::from_secs(60), liveness); + routine.window.outstanding.push(OutstandingBlockRange { + request: BlockRangeRequest { + start_height: block::Height(99), + count: 2, + anchor_hash: block::Hash([9; 32]), + estimated_bytes: 0, + expected_blocks: vec![ + ExpectedBlock { + height: block::Height(99), + hash: block::Hash([99; 32]), + estimated_bytes: 0, + }, + ExpectedBlock { + height: block::Height(100), + hash: block::Hash([100; 32]), + estimated_bytes: 0, + }, + ], + }, + queued_at: now - Duration::from_secs(60), + deadline: now + Duration::from_secs(60), + delivery_snapshot: routine + .window + .delivery_snapshot(now - Duration::from_secs(60)), + delivered_bytes: 0, + received: ReceivedBlockTracker::default(), + }); + + routine.gc_committed_outstanding(); + + assert!( + routine.window.outstanding.is_empty(), + "the floor passed the whole request, so GC must remove it" + ); + assert_eq!( + routine.window.block_liveness_deadline, None, + "a request satisfied below the floor must not leave a stale liveness deadline" + ); + assert!( + matches!(routine.window.check_liveness(now), LivenessOutcome::Ok), + "the peer owes nothing and must not be disconnected" + ); + } } From 600a67e2fa3199c53af6309c598e8c118745e80f Mon Sep 17 00:00:00 2001 From: roman Date: Mon, 6 Jul 2026 02:45:04 -0600 Subject: [PATCH 07/13] fix(zakura): publish chain-tip resets verbatim from the chain-tip mirror MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mirror maxed a TipAction::Reset tip against the latest_chain_tip watch, which may not have observed the reset yet; the stale higher tip turned the downstream VerifiedReset into a no-op, pinning the block-sync sequencer one block above the real tip after a fork-recovery invalidation — the missing parent was never requested and every body commit timed out until restart. On Reset the action's tip is authoritative; anti-regression maxing now applies only to Grow. --- CHANGELOG.md | 9 +++++ zebrad/src/commands/start.rs | 25 +++++++++++++ .../start/zakura/header_sync_driver.rs | 36 +++++++++++++++---- zebrad/src/commands/start/zakura/mod.rs | 4 +-- 4 files changed, 66 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fcd12ba349a..7c706cd9b54 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,15 @@ and this project adheres to [Semantic Versioning](https://semver.org). - Fixed a near-tip sync restart loop when a timed-out `AwaitUtxo` lookup in the transaction verifier was converted to `InternalDowncastError` instead of a missing transparent input. +- Fixed a chain-tip reset (fork-recovery invalidation, `invalidateblock`) + being silently discarded by the Zakura chain-tip mirror. The mirror maxed + the reset tip against a `latest_chain_tip` watch that may not have observed + the reset yet, republished the stale higher tip, and turned the downstream + `VerifiedReset` into a no-op — the block-sync sequencer then pinned one + block above the real tip, never requested the missing parent, and every + body commit timed out until restart. On `TipAction::Reset` the action's tip + is now published verbatim; the anti-regression maxing only applies to + `Grow`. - Fixed Zakura block-sync peers being serially disconnected and parked at the chain tip by a stale liveness deadline. When the download floor passed a request because other peers' deliveries satisfied its heights, the floor GC diff --git a/zebrad/src/commands/start.rs b/zebrad/src/commands/start.rs index 29925e33dd9..9e6784ec79c 100644 --- a/zebrad/src/commands/start.rs +++ b/zebrad/src/commands/start.rs @@ -2917,6 +2917,31 @@ mod zakura_header_sync_driver_tests { /// when the committed suffix sits on a branch the header chain reorged away /// from (e.g. after a crash between a durable header reorg and its body /// invalidation, when later commits report no header conflict). + /// On `TipAction::Reset` the mirror must publish the action's tip verbatim: + /// maxing it against a `latest_chain_tip` watch that has not yet observed + /// the reset republishes the stale higher tip, turns the `VerifiedReset` + /// into a no-op, and pins the block-sync sequencer one block above the + /// real tip (observed live after a fork-recovery invalidation: every body + /// commit then timed out waiting for a parent that was never requested). + #[test] + fn chain_tip_mirror_reset_tip_is_authoritative() { + use super::zakura::chain_tip_mirror_verified_tip; + + let reset_tip = (block::Height(100), block::Hash([1; 32])); + let stale_watch_tip = Some((block::Height(101), block::Hash([2; 32]))); + let finalized = Some((block::Height(90), block::Hash([3; 32]))); + let reset = zebra_state::TipAction::Reset { + height: reset_tip.0, + hash: reset_tip.1, + }; + + assert_eq!( + chain_tip_mirror_verified_tip(&reset, finalized, reset_tip, stale_watch_tip), + reset_tip, + "a reset's tip is authoritative and must not be maxed against stale sources" + ); + } + #[tokio::test] async fn stranded_body_suffix_is_reconciled_by_invalidation() { use std::sync::atomic::{AtomicBool, Ordering}; diff --git a/zebrad/src/commands/start/zakura/header_sync_driver.rs b/zebrad/src/commands/start/zakura/header_sync_driver.rs index 10549fb247b..b60aef2c34f 100644 --- a/zebrad/src/commands/start/zakura/header_sync_driver.rs +++ b/zebrad/src/commands/start/zakura/header_sync_driver.rs @@ -1553,13 +1553,11 @@ pub(crate) async fn mirror_zakura_full_block_commits( insert_cs_height(row, cs_trace::FINALIZED_HEIGHT, finalized_height); }, ); - let action_tip = Some((height, hash)); - let verified_block_tip = - verified_block_tip_from_state(finalized_tip, action_tip, (height, hash)); - let verified_block_tip = verified_block_tip_from_state( - Some(verified_block_tip), + let verified_block_tip = chain_tip_mirror_verified_tip( + &action, + finalized_tip, + (height, hash), latest_chain_tip.best_tip_height_and_hash(), - verified_block_tip, ); emit_commit_state( @@ -1704,6 +1702,32 @@ pub(crate) fn block_sync_chain_tip_event( } } +/// The verified-body tip the chain-tip mirror publishes for a tip action. +/// +/// On `Grow`, the action tip is maxed against the finalized tip and the +/// `latest_chain_tip` watch so a lagging source cannot regress the frontier. +/// On `Reset` the action's tip IS the authoritative post-reset chain tip: +/// maxing it against a watch that has not yet observed the reset republishes +/// the stale higher tip, turns the downstream `VerifiedReset` into a no-op, +/// and pins the block-sync sequencer one block above the real tip — it then +/// never requests the missing parent and every body commit times out +/// (observed live after a fork-recovery invalidation). +pub(crate) fn chain_tip_mirror_verified_tip( + action: &zebra_state::TipAction, + finalized_tip: Option<(block::Height, block::Hash)>, + action_tip: (block::Height, block::Hash), + latest_tip: Option<(block::Height, block::Hash)>, +) -> (block::Height, block::Hash) { + match action { + zebra_state::TipAction::Reset { .. } => action_tip, + zebra_state::TipAction::Grow { .. } => { + let verified_block_tip = + verified_block_tip_from_state(finalized_tip, Some(action_tip), action_tip); + verified_block_tip_from_state(Some(verified_block_tip), latest_tip, verified_block_tip) + } + } +} + pub(crate) fn chain_tip_mirror_frontier_change( action: &zebra_state::TipAction, previous_verified_body: block::Height, diff --git a/zebrad/src/commands/start/zakura/mod.rs b/zebrad/src/commands/start/zakura/mod.rs index 3aba77053f6..750ba4242a0 100644 --- a/zebrad/src/commands/start/zakura/mod.rs +++ b/zebrad/src/commands/start/zakura/mod.rs @@ -130,8 +130,8 @@ pub(crate) use frontier::{query_block_sync_frontiers, verified_block_tip_from_st #[cfg(test)] pub(crate) use header_sync_driver::{ block_roots_cover_range, block_sync_chain_tip_event, body_sizes_for_served_header_range, - chain_tip_mirror_frontier_change, header_range_commit_failure_kind, - notify_block_sync_header_tip, reconcile_stranded_body_suffix, + chain_tip_mirror_frontier_change, chain_tip_mirror_verified_tip, + header_range_commit_failure_kind, notify_block_sync_header_tip, reconcile_stranded_body_suffix, root_covered_query_best_header_tip, tree_aux_roots_for_served_header_range, }; pub(crate) use header_sync_driver::{ From 9c967e992d71ec6b0a64405e777b854ddd74a110 Mon Sep 17 00:00:00 2001 From: roman Date: Mon, 6 Jul 2026 03:48:37 -0600 Subject: [PATCH 08/13] debug(consensus): attribute slow block commits to their pipeline stage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A ~45-minute uniform commit-stall (every Zakura block-sync commit timing out while the state write task, runtime workers, and parent-wait queue were all idle) could not be attributed from logs: the driver's timeout says only that the verifier did not answer. Log the two stages that can silently hold a block — the transaction-verification drain (UTXO awaits on out-of-order parents serialize here, bounded by the 6-minute UTXO_LOOKUP_TIMEOUT) and the state commit await (parent-waiting queue / write-task backlog) — at warn when they exceed 20s, and log timed-out and slow (>20s) transparent UTXO lookups with their outpoint. --- zebra-consensus/src/block.rs | 31 +++++++++++++++++++++++++++--- zebra-consensus/src/transaction.rs | 21 +++++++++++++++++++- 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/zebra-consensus/src/block.rs b/zebra-consensus/src/block.rs index ad8c16556f8..dd5920a581f 100644 --- a/zebra-consensus/src/block.rs +++ b/zebra-consensus/src/block.rs @@ -303,6 +303,7 @@ where let mut block_miner_fees = Ok(Amount::zero()); use futures::StreamExt; + let tx_checks_started = std::time::Instant::now(); while let Some(result) = async_checks.next().await { tracing::trace!(?result, remaining = async_checks.len()); let response = result @@ -323,6 +324,18 @@ where } } + // A slow drain here means transaction verification (usually a UTXO + // await on an out-of-order parent) held the block; this stage log + // attributes driver-side commit timeouts to their pipeline stage. + if tx_checks_started.elapsed() > std::time::Duration::from_secs(20) { + tracing::warn!( + ?height, + ?hash, + elapsed = ?tx_checks_started.elapsed(), + "slow transaction verification stage while committing block" + ); + } + // Check the summed block totals if sigops > MAX_BLOCK_SIGOPS { @@ -378,13 +391,25 @@ where }; } - match state_service + let state_commit_started = std::time::Instant::now(); + let state_commit_result = state_service .ready() .await .map_err(|source| VerifyBlockError::StateService { source, hash })? .call(zs::Request::CommitSemanticallyVerifiedBlock(prepared_block)) - .await - { + .await; + // A slow response here means the block sat in the state's + // parent-waiting queue or behind the write task; this stage log + // attributes driver-side commit timeouts to their pipeline stage. + if state_commit_started.elapsed() > std::time::Duration::from_secs(20) { + tracing::warn!( + ?height, + ?hash, + elapsed = ?state_commit_started.elapsed(), + "slow state commit stage while committing block" + ); + } + match state_commit_result { Ok(zs::Response::Committed(committed_hash)) => { assert_eq!(committed_hash, hash, "state must commit correct hash"); Ok(hash) diff --git a/zebra-consensus/src/transaction.rs b/zebra-consensus/src/transaction.rs index 0635c2f9bac..7860110549e 100644 --- a/zebra-consensus/src/transaction.rs +++ b/zebra-consensus/src/transaction.rs @@ -743,14 +743,33 @@ where utxo } else { + let lookup_started = std::time::Instant::now(); let response = state .clone() .oneshot(zebra_state::Request::AwaitUtxo(*outpoint)) .await .map_err(|boxed_error| match boxed_error.downcast::() { - Ok(_) => TransactionError::TransparentInputNotFound, + Ok(_) => { + // A timed-out lookup held this transaction's whole + // block for UTXO_LOOKUP_TIMEOUT; chains of these + // serialize into multi-minute commit stalls that + // are otherwise unattributable from logs. + tracing::warn!( + ?outpoint, + elapsed = ?lookup_started.elapsed(), + "transparent input UTXO lookup timed out" + ); + TransactionError::TransparentInputNotFound + } Err(boxed_error) => TransactionError::from(boxed_error), })?; + if lookup_started.elapsed() > std::time::Duration::from_secs(20) { + tracing::warn!( + ?outpoint, + elapsed = ?lookup_started.elapsed(), + "slow transparent input UTXO lookup delayed transaction verification" + ); + } if let zebra_state::Response::Utxo(utxo) = response { utxo From 3e5b6eb4ca876df7b729aa91868fb066fe99e20f Mon Sep 17 00:00:00 2001 From: roman Date: Mon, 6 Jul 2026 04:49:44 -0600 Subject: [PATCH 09/13] fix(zakura): defer block-sync liveness disconnects during local commit stalls 'No accepted block progress' cannot be blamed on a peer while the node's own apply pipeline is not landing commits: during a ~45-minute verifier-layer commit stall the liveness deadline exiled healthy peers one by one into the 180s no-progress cooldown (observed live: parks +4 on t2, t4 down to 2 peers). When the sequencer view shows bodies in the applying queue, extend the deadline instead of disconnecting; a genuinely dead peer is caught as soon as the pipeline drains, and its timed-out requests were already requeued by expire_due_timeouts. --- CHANGELOG.md | 7 ++ .../src/zakura/block_sync/peer_routine.rs | 102 +++++++++++++++++- 2 files changed, 105 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c706cd9b54..e27339610af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,13 @@ and this project adheres to [Semantic Versioning](https://semver.org). - Fixed a near-tip sync restart loop when a timed-out `AwaitUtxo` lookup in the transaction verifier was converted to `InternalDowncastError` instead of a missing transparent input. +- Zakura block-sync liveness disconnects are now deferred while the local + apply pipeline holds unfinished bodies: "no accepted block progress" cannot + be blamed on a peer while the node's own commits are not landing. During a + ~45-minute local commit stall the old behavior exiled healthy peers one by + one into the no-progress cooldown. A genuinely dead peer is still caught as + soon as the pipeline drains. Deferrals are counted in + `sync.block.liveness.deferred_local_stall`. - Fixed a chain-tip reset (fork-recovery invalidation, `invalidateblock`) being silently discarded by the Zakura chain-tip mirror. The mirror maxed the reset tip against a `latest_chain_tip` watch that may not have observed diff --git a/zebra-network/src/zakura/block_sync/peer_routine.rs b/zebra-network/src/zakura/block_sync/peer_routine.rs index 0b2d44bb229..55aa9380339 100644 --- a/zebra-network/src/zakura/block_sync/peer_routine.rs +++ b/zebra-network/src/zakura/block_sync/peer_routine.rs @@ -1125,14 +1125,34 @@ impl PeerRoutine { // continuous backpressure): plausibly transient local write congestion, not // a dead peer. While outbound is full the select loop does not drain inbound // frames (`if outbound_queue_has_capacity`), so a block the peer already sent - // may be waiting behind our write side. Grant one short, BOUNDED grace. This - // is the *only* liveness extension: a peer that stopped reading holds outbound - // full past `request_timeout`, falls through to the disconnect arm, and is - // disconnected at the liveness deadline — it cannot dodge the timer. + // may be waiting behind our write side. Grant one short, BOUNDED grace: a peer + // that stopped reading holds outbound full past `request_timeout`, falls + // through to the disconnect arm, and is disconnected at the liveness + // deadline — it cannot dodge the timer. self.window .extend_liveness_deadline(now, self.config.request_timeout); Ok(()) } + LivenessOutcome::Disconnect if self.sequencer_view.borrow().applying_len > 0 => { + // Bodies are sitting in the LOCAL apply pipeline, so "no accepted + // block progress" cannot be blamed on this peer: nothing can make + // accepted progress while our own commits are not landing. During a + // 45-minute verifier-layer commit stall this arm's absence exiled + // healthy peers one by one into the no-progress cooldown (observed + // live: t2 parks +4, t4 down to 2 peers). Defer the disconnect; a + // genuinely dead peer is caught as soon as the pipeline drains + // (`applying_len == 0`), and its timed-out requests were already + // requeued to other peers by `expire_due_timeouts`. + self.window + .extend_liveness_deadline(now, self.config.request_timeout); + metrics::counter!("sync.block.liveness.deferred_local_stall").increment(1); + tracing::debug!( + peer = ?self.peer, + applying_len = self.sequencer_view.borrow().applying_len, + "deferring block-sync liveness disconnect: local apply pipeline is stalled" + ); + Ok(()) + } LivenessOutcome::Disconnect => { let error = "block-sync peer made no accepted block progress before liveness deadline"; @@ -2717,4 +2737,78 @@ mod tests { "the peer owes nothing and must not be disconnected" ); } + + /// An expired liveness deadline must not disconnect a peer while the LOCAL + /// apply pipeline holds unfinished bodies: nothing can make "accepted block + /// progress" while our own commits are not landing. During a live + /// verifier-layer commit stall this exiled healthy peers one by one into + /// the no-progress cooldown until the peer set collapsed. Once the + /// pipeline drains, the same expired deadline disconnects as before. + #[tokio::test] + async fn liveness_disconnect_is_deferred_while_local_applies_are_stalled() { + let config = ZakuraBlockSyncConfig::default(); + let budget = ByteBudget::new(1_000_000); + let work = Arc::new(WorkQueue::new(block::Height(0))); + + let cancel = CancellationToken::new(); + let (out_send, _out_recv) = framed_channel(16); + let (_in_send, in_recv) = framed_channel(16); + let peer = ZakuraPeerId::new(vec![9u8; 32]).expect("test peer id is within bounds"); + let session = BlockSyncPeerSession::for_test(peer.clone(), out_send, cancel.clone()); + + let (sequencer_input_tx, _sequencer_input_rx) = mpsc::channel(16); + let (control_tx, _control_rx) = mpsc::unbounded_channel(); + let (actions_tx, _actions_rx) = mpsc::channel(16); + let (routine_to_reactor_tx, _routine_to_reactor_rx) = mpsc::channel(16); + let mut view = initial_view(BlockSyncFrontiers { + finalized_height: block::Height(0), + verified_block_tip: block::Height(100), + verified_block_hash: block::Hash([0; 32]), + }); + view.applying_len = 60; + let (view_tx, view_rx) = watch::channel(view.clone()); + + let mut routine = PeerRoutine::new( + peer, + session, + in_recv, + config, + 0, + budget, + work, + Arc::new(PeerRegistry::new()), + Arc::new(Mutex::new(ThroughputMeter::new(Instant::now()))), + sequencer_input_tx, + Arc::new(AtomicU64::new(0)), + control_tx, + actions_tx, + routine_to_reactor_tx, + view_rx, + cancel, + ZakuraTrace::noop(), + ); + + let now = Instant::now(); + // An armed deadline that expired 10s ago. + routine + .window + .arm_liveness(now - Duration::from_secs(60), Duration::from_secs(50)); + + assert!( + routine.check_block_liveness(now).is_ok(), + "an expired deadline must be deferred while local applies are stalled" + ); + + // The pipeline drains: the same expired deadline now disconnects. + view.applying_len = 0; + view_tx + .send(view) + .expect("view receiver is held by the routine"); + // The deferral extended the deadline by request_timeout; move past it. + let later = now + routine.config.request_timeout + Duration::from_secs(1); + assert!( + routine.check_block_liveness(later).is_err(), + "a drained pipeline must restore the liveness disconnect" + ); + } } From c3734e54e4531da3f3487afedfc7ad9d4b1011ef Mon Sep 17 00:00:00 2001 From: roman Date: Mon, 6 Jul 2026 04:50:27 -0600 Subject: [PATCH 10/13] chore(zakura): fix clippy clone_on_copy in liveness deferral test --- zebra-network/src/zakura/block_sync/peer_routine.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zebra-network/src/zakura/block_sync/peer_routine.rs b/zebra-network/src/zakura/block_sync/peer_routine.rs index 55aa9380339..5408800dd54 100644 --- a/zebra-network/src/zakura/block_sync/peer_routine.rs +++ b/zebra-network/src/zakura/block_sync/peer_routine.rs @@ -2766,7 +2766,7 @@ mod tests { verified_block_hash: block::Hash([0; 32]), }); view.applying_len = 60; - let (view_tx, view_rx) = watch::channel(view.clone()); + let (view_tx, view_rx) = watch::channel(view); let mut routine = PeerRoutine::new( peer, From c9480b5a21cf086053c537f56e40e5956fff1879 Mon Sep 17 00:00:00 2001 From: roman Date: Mon, 6 Jul 2026 08:05:29 -0600 Subject: [PATCH 11/13] debug(zakura): log the state error for scored header-range commit failures InvalidPeerRange rejections score the peer, but the CommitHeaderRangeError variant behind them was only visible at debug level: a stranded-context wedge (every peer's response for the same range rejected identically for hours, observed live at height 4148005) was indistinguishable from real peer misbehavior at default log levels. --- .../start/zakura/header_sync_driver.rs | 31 ++++++++++++++----- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/zebrad/src/commands/start/zakura/header_sync_driver.rs b/zebrad/src/commands/start/zakura/header_sync_driver.rs index b60aef2c34f..73989294e3e 100644 --- a/zebrad/src/commands/start/zakura/header_sync_driver.rs +++ b/zebrad/src/commands/start/zakura/header_sync_driver.rs @@ -789,14 +789,29 @@ pub(crate) async fn drive_zakura_header_sync_actions Date: Mon, 6 Jul 2026 08:19:30 -0600 Subject: [PATCH 12/13] fix(zakura): walk back on repeated header context mismatches instead of scoring peers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A fork-recovery re-commit can leave stale rows in the stored difficulty window; every extension range is then rejected with InvalidDifficultyThreshold, each rejection scored and disconnected the honest serving peer, and the wedge survived restarts (on-disk rows). No link failure ever fires in this state, so the existing walk-back trigger never engages. Classify ValidateContextError commit failures as ContextMismatch: never score the peer, and count them as stale-anchor evidence — the quorum (3 failures, 2 distinct peers) starts the same exponential walk-back, and each deepening re-commits a longer range that rewrites the poisoned window. --- CHANGELOG.md | 10 ++++ .../src/zakura/header_sync/events.rs | 6 ++ .../src/zakura/header_sync/reactor.rs | 21 +++++++ zebra-network/src/zakura/header_sync/tests.rs | 56 +++++++++++++++++++ .../src/zakura/header_sync/validation.rs | 1 + .../start/zakura/header_sync_driver.rs | 19 ++++++- 6 files changed, 110 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e27339610af..b8d9e58ee7c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,16 @@ and this project adheres to [Semantic Versioning](https://semver.org). - Fixed a near-tip sync restart loop when a timed-out `AwaitUtxo` lookup in the transaction verifier was converted to `InternalDowncastError` instead of a missing transparent input. +- Fixed a permanent header-sync wedge where a fork-recovery re-commit left + stale rows in the stored difficulty-validation window: every extension + range was then rejected with `InvalidDifficultyThreshold`, each rejection + scored and disconnected the (honest) serving peer, and the wedge survived + restarts because the stale rows are on disk (observed live at height + 4148005 for hours). Contextual validation failures are now classified as + `ContextMismatch`: they no longer score peers, and repeated identical + failures from independent peers feed the fork-recovery walk-back — each + deepening re-commits a longer range, rewriting the poisoned window until + the chain extends again. - Zakura block-sync liveness disconnects are now deferred while the local apply pipeline holds unfinished bodies: "no accepted block progress" cannot be blamed on a peer while the node's own commits are not landing. During a diff --git a/zebra-network/src/zakura/header_sync/events.rs b/zebra-network/src/zakura/header_sync/events.rs index 78b9b866d84..83d332333cc 100644 --- a/zebra-network/src/zakura/header_sync/events.rs +++ b/zebra-network/src/zakura/header_sync/events.rs @@ -456,6 +456,12 @@ pub enum HeaderSyncMisbehavior { pub enum HeaderSyncCommitFailureKind { /// The supplied headers failed contextual validation or checkpoint consistency. InvalidPeerRange, + /// The range failed contextual validation (difficulty, median-time) against + /// this node's stored header context. When independent peers fail + /// identically, the stored context itself is suspect (a reorg left stale + /// rows in the validation window), so this must never score the peer and + /// instead feeds fork-recovery walk-back evidence. + ContextMismatch, /// Local storage/channel/resource failure; do not score the peer. Local, } diff --git a/zebra-network/src/zakura/header_sync/reactor.rs b/zebra-network/src/zakura/header_sync/reactor.rs index 0219ed1f429..f60fe50a77f 100644 --- a/zebra-network/src/zakura/header_sync/reactor.rs +++ b/zebra-network/src/zakura/header_sync/reactor.rs @@ -687,6 +687,27 @@ impl HeaderSyncReactor { self.report_misbehavior(peer.clone(), HeaderSyncMisbehavior::InvalidRange) .await; } + if kind == HeaderSyncCommitFailureKind::ContextMismatch { + // The range failed contextual validation against OUR stored header + // context. One peer failing could be a bad range; independent peers + // failing identically means the stored validation window itself is + // stale (a reorg left old-branch rows in it) — every extension is + // rejected forever and no link failure ever fires, so this is the + // only signal that can trigger recovery. Reuse the stale-anchor + // quorum and walk the anchor back; each deepening re-commits a + // longer range, rewriting the poisoned window (observed live: + // InvalidDifficultyThreshold at 4148005 from every peer, surviving + // restarts, with honest peers disconnected one by one). + self.state.stale_anchor.record(peer.clone()); + metrics::counter!("sync.header.stale_anchor.context_mismatch").increment(1); + if self.state.stale_anchor.should_reanchor() { + if self.state.best_header_tip > self.state.verified_block_tip { + self.reanchor_to_verified_block_tip().await; + } else { + self.begin_fork_recovery_walk_back(); + } + } + } let key = PendingCommitKey { peer, start_height, diff --git a/zebra-network/src/zakura/header_sync/tests.rs b/zebra-network/src/zakura/header_sync/tests.rs index b40079532a2..6c41e4f6669 100644 --- a/zebra-network/src/zakura/header_sync/tests.rs +++ b/zebra-network/src/zakura/header_sync/tests.rs @@ -2092,6 +2092,62 @@ async fn invalid_async_header_commit_failure_reports_peer_disconnect() { } } +#[tokio::test(flavor = "current_thread")] +async fn context_mismatch_commit_failures_walk_back_without_scoring_peers() { + let network = regtest_network(); + let mut fixture = spawn_test_reactor(startup_for( + network.clone(), + (block::Height(0), network.genesis_hash()), + None, + )); + let peer_a = peer(63); + let peer_b = peer(64); + + connect_peer(&fixture, peer_a.clone()).await; + connect_peer(&fixture, peer_b.clone()).await; + + // Three contextual rejections from two independent peers: the stored + // validation window is suspect, so the reactor must begin a walk-back + // (QueryReanchorTarget) and must NOT score either peer — they are honest; + // the stale rows are ours. + for peer_id in [&peer_a, &peer_b, &peer_a] { + fixture + .handle + .send(HeaderSyncEvent::HeaderRangeCommitFailed { + peer: peer_id.clone(), + start_height: block::Height(1), + count: 1, + kind: HeaderSyncCommitFailureKind::ContextMismatch, + }) + .await + .unwrap(); + } + + let mut saw_reanchor_query = false; + for _ in 0..12 { + match tokio::time::timeout( + std::time::Duration::from_millis(500), + fixture.actions.recv(), + ) + .await + { + Ok(Some(HeaderSyncAction::Misbehavior { peer, reason })) => { + panic!("honest peer {peer:?} scored for a local context mismatch: {reason:?}") + } + Ok(Some(HeaderSyncAction::QueryReanchorTarget { .. })) => { + saw_reanchor_query = true; + break; + } + Ok(Some(_)) => continue, + Ok(None) | Err(_) => break, + } + } + assert!( + saw_reanchor_query, + "repeated context mismatches from independent peers must start a walk-back" + ); +} + #[tokio::test(flavor = "current_thread")] async fn peer_disconnect_removes_outstanding_requests_for_that_peer() { let network = Network::Mainnet; diff --git a/zebra-network/src/zakura/header_sync/validation.rs b/zebra-network/src/zakura/header_sync/validation.rs index 6e16519e27a..a28a21f16a5 100644 --- a/zebra-network/src/zakura/header_sync/validation.rs +++ b/zebra-network/src/zakura/header_sync/validation.rs @@ -101,6 +101,7 @@ pub(super) fn misbehavior_reason_label(reason: HeaderSyncMisbehavior) -> &'stati pub(super) fn commit_failure_reason_label(kind: HeaderSyncCommitFailureKind) -> &'static str { match kind { HeaderSyncCommitFailureKind::InvalidPeerRange => "invalid_peer_range", + HeaderSyncCommitFailureKind::ContextMismatch => "context_mismatch", HeaderSyncCommitFailureKind::Local => "local", } } diff --git a/zebrad/src/commands/start/zakura/header_sync_driver.rs b/zebrad/src/commands/start/zakura/header_sync_driver.rs index 73989294e3e..13cd1e2f2a5 100644 --- a/zebrad/src/commands/start/zakura/header_sync_driver.rs +++ b/zebrad/src/commands/start/zakura/header_sync_driver.rs @@ -794,7 +794,11 @@ pub(crate) async fn drive_zakura_header_sync_actions { + | zebra_state::CommitHeaderRangeError::ConflictingFullBlockHeader { .. } => { HeaderSyncCommitFailureKind::InvalidPeerRange } + // Contextual failures (difficulty, median-time) validate the range + // against OUR stored header context; when the context holds stale + // reorg rows every honest peer fails identically (observed live: + // InvalidDifficultyThreshold at 4148005 from all peers, for hours). + // Never score the peer for these — the reactor treats them as + // stale-anchor evidence and walks back instead. + zebra_state::CommitHeaderRangeError::ValidateContextError(_) => { + HeaderSyncCommitFailureKind::ContextMismatch + } _ => HeaderSyncCommitFailureKind::Local, } } @@ -1951,6 +1963,7 @@ fn trace_state_read_error( fn commit_failure_result_label(kind: HeaderSyncCommitFailureKind) -> &'static str { match kind { HeaderSyncCommitFailureKind::InvalidPeerRange => "invalid_peer_range", + HeaderSyncCommitFailureKind::ContextMismatch => "context_mismatch", HeaderSyncCommitFailureKind::Local => "local_error", } } From 6f8b4ad88c46dbd8bbd97c274d42dfd2f58b4c59 Mon Sep 17 00:00:00 2001 From: roman Date: Mon, 6 Jul 2026 09:29:37 -0600 Subject: [PATCH 13/13] fix(zakura): classify UnknownAnchor header-commit failures as ContextMismatch UnknownAnchor cannot be the peer's fault: the anchor is this node's own request bookkeeping, and the error fires when the reactor's in-memory frontier references a hash the header store does not have (observed live at 4148376 after the frontier advanced over diverged store rows). It was classified InvalidPeerRange, so honest peers were scored and disconnected and no recovery trigger ever fired. As ContextMismatch it feeds the stale-anchor quorum; the walk-back re-anchors from the store, which is exactly the self-correction this state needs. --- .../src/commands/start/zakura/header_sync_driver.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/zebrad/src/commands/start/zakura/header_sync_driver.rs b/zebrad/src/commands/start/zakura/header_sync_driver.rs index 13cd1e2f2a5..f7d1eded447 100644 --- a/zebrad/src/commands/start/zakura/header_sync_driver.rs +++ b/zebrad/src/commands/start/zakura/header_sync_driver.rs @@ -1492,7 +1492,6 @@ pub(crate) fn header_range_commit_failure_kind( | zebra_state::CommitHeaderRangeError::BodySizeCountMismatch { .. } | zebra_state::CommitHeaderRangeError::TreeAuxRootCountMismatch { .. } | zebra_state::CommitHeaderRangeError::TreeAuxRootHeightMismatch { .. } - | zebra_state::CommitHeaderRangeError::UnknownAnchor { .. } | zebra_state::CommitHeaderRangeError::HeightOverflow | zebra_state::CommitHeaderRangeError::ImmutableConflict { .. } | zebra_state::CommitHeaderRangeError::ReorgTooDeep { .. } @@ -1506,7 +1505,15 @@ pub(crate) fn header_range_commit_failure_kind( // InvalidDifficultyThreshold at 4148005 from all peers, for hours). // Never score the peer for these — the reactor treats them as // stale-anchor evidence and walks back instead. - zebra_state::CommitHeaderRangeError::ValidateContextError(_) => { + // UnknownAnchor cannot be the peer's fault at all: the anchor is our + // own request bookkeeping, and it fails when the reactor's in-memory + // frontier references a hash the header store does not have (observed + // live at 4148376 — a mirror-advanced frontier over a store whose + // suffix rows diverged). The walk-back re-anchors from the STORE + // (`QueryReanchorTarget`), which is exactly the self-correction this + // state needs. + zebra_state::CommitHeaderRangeError::ValidateContextError(_) + | zebra_state::CommitHeaderRangeError::UnknownAnchor { .. } => { HeaderSyncCommitFailureKind::ContextMismatch } _ => HeaderSyncCommitFailureKind::Local,