diff --git a/CHANGELOG.md b/CHANGELOG.md index b38ae2c84c1..b8d9e58ee7c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,66 @@ 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 + ~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 + 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 + 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 + 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 + 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/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 diff --git a/zebra-network/src/zakura/block_sync/peer_routine.rs b/zebra-network/src/zakura/block_sync/peer_routine.rs index 3625d439246..5408800dd54 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"; @@ -1141,7 +1161,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 +1205,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 +2355,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 +2625,190 @@ 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" + ); + } + + /// 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); + + 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" + ); + } } 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/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/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 = ""; diff --git a/zebrad/src/commands/start.rs b/zebrad/src/commands/start.rs index 6e20001c166..9e6784ec79c 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}; @@ -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(), @@ -980,19 +985,16 @@ 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()), + zakura_apply_gate.clone(), ) .in_current_span(), ) @@ -2130,11 +2132,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 { @@ -2910,6 +2912,130 @@ 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). + /// 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}; + + 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); @@ -3086,6 +3212,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; }, @@ -3228,6 +3355,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; }, @@ -3377,6 +3505,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; }, @@ -3490,6 +3619,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; }, @@ -3591,6 +3721,7 @@ mod zakura_header_sync_driver_tests { 1, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -3693,6 +3824,7 @@ mod zakura_header_sync_driver_tests { 1, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -3804,6 +3936,7 @@ mod zakura_header_sync_driver_tests { 1, zebra_network::zakura::ZakuraTrace::noop(), None, + super::zakura::ZakuraApplyGate::new(), async move { let _ = shutdown_rx.await; }, @@ -4205,6 +4338,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; }, @@ -4298,6 +4432,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; }, @@ -4418,6 +4553,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; }, @@ -4541,6 +4677,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; }, @@ -4677,6 +4814,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; }, @@ -4806,6 +4944,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; }, @@ -4960,6 +5099,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/header_sync_driver.rs b/zebrad/src/commands/start/zakura/header_sync_driver.rs index 3f69c5f641a..f7d1eded447 100644 --- a/zebrad/src/commands/start/zakura/header_sync_driver.rs +++ b/zebrad/src/commands/start/zakura/header_sync_driver.rs @@ -789,14 +789,33 @@ pub(crate) async fn drive_zakura_header_sync_actions { publish_header_frontier( @@ -1210,6 +1235,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 @@ -1349,15 +1492,30 @@ 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 { .. } | zebra_state::CommitHeaderRangeError::CheckpointConflict { .. } - | zebra_state::CommitHeaderRangeError::ConflictingFullBlockHeader { .. } - | zebra_state::CommitHeaderRangeError::ValidateContextError(_) => { + | 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. + // 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, } } @@ -1429,13 +1587,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( @@ -1580,6 +1736,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, @@ -1788,6 +1970,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", } } diff --git a/zebrad/src/commands/start/zakura/mod.rs b/zebrad/src/commands/start/zakura/mod.rs index bff8f791fd8..750ba4242a0 100644 --- a/zebrad/src/commands/start/zakura/mod.rs +++ b/zebrad/src/commands/start/zakura/mod.rs @@ -7,6 +7,111 @@ 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(); + } + } + + /// 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). + 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; @@ -25,9 +130,9 @@ 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, root_covered_query_best_header_tip, - tree_aux_roots_for_served_header_range, + 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::{ drive_zakura_header_sync_actions, mirror_zakura_full_block_commits, 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..fa9a1a79304 100644 --- a/zebrad/src/components/sync.rs +++ b/zebrad/src/components/sync.rs @@ -516,24 +516,81 @@ 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(); - } +/// 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. +/// 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, +} - if let Some(endpoint) = zakura_endpoint { - endpoint.shutdown_sync_tasks().await; +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, +) { + 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. #[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)] #[serde(deny_unknown_fields, default)] @@ -953,26 +1010,26 @@ 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))] - 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, - zakura_endpoint: Option, - zakura_shutdown: Option, + apply_gate: std::sync::Arc, ) -> Result<(), Report> where RS: Service @@ -992,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; stopping \ - Zakura sync drivers and falling back to legacy ChainSync so legacy peers \ - can drive body sync" - ); - stop_zakura_sync(zakura_endpoint.as_ref(), &zakura_shutdown).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, + 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; + 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 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" + "Zakura watchdog probed legacy peers for connected blocks ahead" ); - stop_zakura_sync(zakura_endpoint.as_ref(), &zakura_shutdown).await; - return self.sync().await; + 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 ae67059bf04..15ee7f8457f 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,23 @@ 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; + // 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!( - driver_view.is_cancelled(), - "falling back to legacy must cancel the Zakura sync drivers' shutdown token" + 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] 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 +286,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,10 +320,11 @@ 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; + let gate = crate::commands::start::zakura::ZakuraApplyGate::new(); + futures::executor::block_on(super::super::engage_legacy_fallback_alongside_zakura(&gate)); assert!( - driver_view.is_cancelled(), - "legacy-informed fallback must cancel the Zakura sync drivers' shutdown token" + gate.is_yielded(), + "fallback must yield the Zakura apply gate" ); } @@ -405,33 +408,81 @@ 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(); +/// Locks in the fallback behavior: engaging legacy fallback must not signal any +/// 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!( - !token.is_cancelled(), - "precondition: a fresh token is not cancelled" + !drain.is_finished(), + "the drain waits for in-flight applies" ); - - // 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" + 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()); +} + +/// 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)); } -/// On a Zakura-only node there is no endpoint shutdown token, so the hand-off helper must be a -/// no-op rather than panic. +/// The apply gate round-trips: yielded blocks new applies, unyield restores them. #[tokio::test] -async fn stop_zakura_sync_is_a_noop_without_a_token() { - // Must not panic. - stop_zakura_sync(None, &None).await; +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" + ); }