fix(zakura): single-peer sync wedge and block-sync backpressure busy-spin

CHANGELOG.md

@@ -18,6 +18,25 @@ and this project adheres to [Semantic Versioning](https://semver.org).
 
 ### Fixed
 
+- Fixed an intermittent Zakura sync wedge on a node with a single block-sync
+  peer. When a node makes early block progress via another source — inbound
+  gossip or the legacy `BlocksByHash` path on a dual-stack node — its Zakura
+  block-sync peer stays "unproven" and pinned at the initial one-probe cap,
+  because that progress does not flow through the peer's own body handler. If it
+  is the node's only peer, the no-progress liveness reaper then disconnects it
+  and the node wedges below the tip with no way to pull the remaining backfill.
+  A non-destructive committed-view advance now credits the peer's no-progress
+  probe budget when the verified tip advances via any source, so a progressing
+  node's sole peer keeps probing. In addition, on regtest the body-sync stall
+  watchdog now falls back to the legacy downloader after 60s (rather than the
+  mainnet 10 minutes) so a stalled node recovers within the regtest e2e budget.
+- Fixed a block-sync busy-spin under sustained byte-budget backpressure. The
+  sequencer re-published its progress view (waking the reactor and every per-peer
+  routine) even when no schedulable field had changed, which combined with the
+  per-attempt floor-funding request to spin a routine's refill loop with no timer
+  while the budget was pinned — wasting CPU and, under load, starving commit
+  progress. The sequencer now wakes watchers only when a scheduling-relevant field
+  actually changes.
 - Fixed an out-of-memory crash during Zakura block sync when the header chain
   runs far ahead of the commit tip. The block-sync applying buffer holds decoded
   block bodies ahead of the in-order committer; its look-ahead budget counted

zebra-network/src/zakura/block_sync/admission.rs

@@ -1,4 +1,5 @@
-use std::time::{Duration, Instant};
+use std::time::Duration;
+use tokio::time::Instant;
 
 use zebra_chain::block;
 

zebra-network/src/zakura/block_sync/bbr.rs

@@ -1,4 +1,5 @@
-use std::time::{Duration, Instant};
+use std::time::Duration;
+use tokio::time::Instant;
 
 use super::{
     config::{CwndUnit, ZakuraBlockSyncConfig},

zebra-network/src/zakura/block_sync/bench.rs

@@ -23,13 +23,14 @@ use std::{
         atomic::{AtomicU64, Ordering},
         Arc,
     },
-    time::{Duration, Instant},
+    time::Duration,
 };
 
 use serde_json::Value;
 use tokio::{
     sync::{mpsc, watch},
     task::JoinHandle,
+    time::Instant,
 };
 use zebra_chain::block::{self, Block};
 use zebra_jsonl_trace::{JsonlTraceGuard, JsonlTracer};

zebra-network/src/zakura/block_sync/mod.rs

@@ -8,7 +8,7 @@ use std::{
     collections::{BTreeMap, HashMap, HashSet, VecDeque},
     io::{self, Cursor, Read, Write},
     sync::{Arc, Mutex as StdMutex},
-    time::{Duration, Instant},
+    time::Duration,
 };
 
 use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt};
@@ -17,7 +17,7 @@ use thiserror::Error;
 use tokio::{
     sync::{mpsc, oneshot, watch},
     task::JoinHandle,
-    time,
+    time::{self, Instant},
 };
 use tokio_util::sync::CancellationToken;
 use zebra_chain::{

zebra-network/src/zakura/block_sync/peer_registry.rs

@@ -23,9 +23,9 @@
 use std::{
     collections::{BTreeMap, HashMap},
     sync::Mutex as StdMutex,
-    time::Instant,
 };
 
+use tokio::time::Instant;
 use zebra_chain::block;
 
 use super::{

zebra-network/src/zakura/block_sync/peer_routine.rs

@@ -53,8 +53,8 @@ use crate::zakura::{
     trace::{block_sync_trace as bs_trace, BLOCK_SYNC_TABLE},
     Admit, FramedRecv, OrderedSendError, SinkReject,
 };
-use std::{sync::Arc, time::Duration, time::Instant};
-use tokio::time;
+use std::{sync::Arc, time::Duration};
+use tokio::time::{self, Instant};
 use zebra_chain::{block, serialization::ZcashSerialize};
 
 /// How long a routine avoids re-taking a height it just returned on a failure
@@ -225,6 +225,10 @@ pub(super) struct PeerRoutine {
     /// Last `reset_epoch` this routine reacted to, so a `view.changed()` can tell
     /// a destructive reset (in-place clear of outstanding) from a plain advance.
     last_reset_epoch: u64,
+    /// Verified tip observed at the last committed-view change, so a non-destructive
+    /// advance (the node making block progress via any source) can credit an unproven
+    /// peer's probe streak instead of letting a sole peer wedge at its one-probe cap.
+    last_seen_verified_tip: block::Height,
     /// When our outbound queue to this peer *first* filled in the current continuous full
     /// stretch (`None` while it has capacity). Lets the liveness check tell transient local
     /// write congestion (just filled) from a peer that stopped reading for `request_timeout`
@@ -264,6 +268,7 @@ impl PeerRoutine {
     ) -> Self {
         let window = DownloadWindow::new(&config);
         let last_reset_epoch = sequencer_view.borrow().reset_epoch;
+        let last_seen_verified_tip = sequencer_view.borrow().verified_tip;
         let status_reply_meter = super::state::RateMeter::new(config.status_refresh_interval);
         let inbound_status_meter = super::state::RateMeter::new(
             config.status_refresh_interval.min(Duration::from_secs(1)),
@@ -301,6 +306,7 @@ impl PeerRoutine {
             routine_to_reactor,
             sequencer_view,
             last_reset_epoch,
+            last_seen_verified_tip,
             outbound_full_since: None,
             cancel,
             trace,
@@ -551,15 +557,27 @@ impl PeerRoutine {
     /// the post-`reset_above` `WorkQueue`. The transport is never torn down:
     /// reset clears outstanding work in place instead of respawning the routine.
     fn on_view_changed(&mut self) {
-        let reset_epoch = self.sequencer_view.borrow().reset_epoch;
+        let view = *self.sequencer_view.borrow();
+        let reset_epoch = view.reset_epoch;
         if reset_epoch == self.last_reset_epoch {
             // A non-destructive advance: the floor/tip the routine reads come
             // straight from the live `view` each time they are needed, so nothing
             // to do but let the want-work loop re-run at the top (a committed
             // floor advance may GC our fully-committed outstanding).
+            //
+            // If the committed verified tip advanced — via this peer *or* any other
+            // source (gossip, a dual-stack node's legacy `BlocksByHash` path, another
+            // block-sync peer) — credit an unproven peer's no-progress probe streak so a
+            // node that is progressing does not wedge its sole peer at the one-probe cap
+            // while its bodies arrive elsewhere.
+            if view.verified_tip > self.last_seen_verified_tip {
+                self.last_seen_verified_tip = view.verified_tip;
+                self.window.clear_no_progress_probe_streak();
+            }
             return;
         }
         self.last_reset_epoch = reset_epoch;
+        self.last_seen_verified_tip = view.verified_tip;
         self.trace_wake("view_reset");
         // The Sequencer already pinned its floor/tip and `work.reset_above`'d the
         // dropped successor heights. Return our unreceived outstanding to
@@ -2273,10 +2291,10 @@ impl Drop for PeerRoutine {
 mod tests {
     use std::sync::atomic::AtomicU64;
     use std::sync::{Arc, Mutex};
-    use std::time::{Duration, Instant};
+    use std::time::Duration;
 
     use tokio::sync::{mpsc, watch};
-    use tokio::time::timeout;
+    use tokio::time::{timeout, Instant};
     use tokio_util::sync::CancellationToken;
     use zebra_chain::block;
 

zebra-network/src/zakura/block_sync/sequencer_task.rs

@@ -148,7 +148,7 @@ pub(super) enum SequencerControlInput {
 /// The progress view the reactor reacts to. A `watch` (latest-wins) send never
 /// blocks, so the task never blocks on the reactor and the bounded input channel
 /// cannot deadlock against it.
-#[derive(Copy, Clone, Debug)]
+#[derive(Copy, Clone, Debug, PartialEq)]
 pub(super) struct SequencerView {
     pub(super) verified_tip: block::Height,
     pub(super) verified_hash: block::Hash,
@@ -749,7 +749,7 @@ impl SequencerTask {
             .saturating_add(body_input_bytes);
         self.budget
             .audit(expected_budget, "block-sync sequencer view");
-        let _ = self.view_tx.send_replace(SequencerView {
+        let next = SequencerView {
             verified_tip: self.sequencer.verified_tip(),
             verified_hash: self.verified_block_hash,
             download_floor: self.sequencer.floor(),
@@ -765,6 +765,35 @@ impl SequencerTask {
             submitted_applying_bytes: self.sequencer.submitted_applying_bytes(),
             committed_bytes_per_sec: self.committed_throughput.bytes_per_sec(),
             committed_blocks_per_sec: self.committed_throughput.blocks_per_sec(),
+        };
+        // Only wake watchers (the reactor + every per-peer routine) when a field
+        // they schedule against actually changed. The two committed_*_per_sec rates
+        // are observability-only; without this guard a no-op control input — e.g. a
+        // `FundFloorReservation` that shed nothing while the byte budget is pinned —
+        // still publishes an otherwise-identical view and re-wakes the requesting
+        // routine's `sequencer_view.changed()` arm into an immediate refill retry.
+        // That is a timer-free reactor<->sequencer<->routine busy-spin: it wastes a
+        // core (and starves progress under CI load) on a real clock and fully wedges
+        // a `start_paused` test clock, which auto-advances only once every task
+        // parks. Keep the stored rates fresh, but notify only on a schedulable change.
+        self.view_tx.send_if_modified(|current| {
+            let schedulable_changed = view_schedulable_ne(current, &next);
+            *current = next;
+            schedulable_changed
         });
     }
 }
+
+/// True when two views differ in any field the reactor or per-peer routines
+/// schedule against. Ignores the observability-only committed throughput rates,
+/// which move on nearly every sample and must not, on their own, wake — or under a
+/// paused test clock, spin — the whole fleet of watchers.
+fn view_schedulable_ne(a: &SequencerView, b: &SequencerView) -> bool {
+    let strip_rates = |v: &SequencerView| {
+        let mut v = *v;
+        v.committed_bytes_per_sec = 0;
+        v.committed_blocks_per_sec = 0;
+        v
+    };
+    strip_rates(a) != strip_rates(b)
+}

zebra-network/src/zakura/block_sync/service.rs

@@ -3,10 +3,8 @@ use crate::zakura::{
     handle_pipe_exit, spawn_supervised_pipe, FramedRecv, FramedSend, OrderedSendError, Peer,
     PeerStreamSession, Service, SinkReject, Stream, StreamMode, ZakuraPeerId, FRAME_HEADER_BYTES,
 };
-use std::{
-    sync::atomic::{AtomicU64, Ordering},
-    time::Instant,
-};
+use std::sync::atomic::{AtomicU64, Ordering};
+use tokio::time::Instant;
 
 /// Maximum frame bytes for one stream-6 body frame plus protocol framing.
 ///

zebra-network/src/zakura/block_sync/state.rs

@@ -689,6 +689,22 @@ impl DownloadWindow {
         self.clear_liveness_if_idle();
     }
 
+    /// Clear only the no-progress *probe* streak, so an unproven peer may probe again up to
+    /// its [`no_progress_request_cap`](Self::no_progress_request_cap). `last_block_at` (proof)
+    /// and the liveness deadline are untouched — a peer that never serves a body is still
+    /// governed by the liveness deadline like any other unproven peer.
+    ///
+    /// Used when we do *not* want a sole unproven peer to stay wedged at the one-probe cap:
+    /// (1) the local verified tip advanced via another source (gossip / a dual-stack node's
+    /// legacy `BlocksByHash` path / another peer) — the node is progressing, so this peer's
+    /// probe budget should not stay charged; and (2) when the liveness reaper would otherwise
+    /// disconnect our *only* peer — we keep it and let it re-probe instead of wedging with no
+    /// way to pull the remaining backfill. Mirrors [`note_view_reset`](Self::note_view_reset)
+    /// for the non-reset cases.
+    pub(super) fn clear_no_progress_probe_streak(&mut self) {
+        self.requests_without_block_progress = 0;
+    }
+
     /// Push the block-liveness deadline out by `timeout` when a would-be disconnect is
     /// attributable to *local* outbound backpressure, not the peer: while our outbound queue
     /// is full the routine stops draining inbound, so a useful body may be sitting unread.

zebra-network/src/zakura/block_sync/tests.rs

@@ -640,6 +640,54 @@ fn block_liveness_uses_probe_cap_until_first_accepted_body() {
     assert_eq!(window.no_progress_request_cap(), 8);
 }
 
+/// Regression for the dual-stack single-peer wedge: an unproven peer that has spent its one
+/// initial probe is pinned at `no_progress_request_cap() == initial_block_probe_requests`. When
+/// the node's verified tip advances via *another* source (inbound gossip, a dual-stack node's
+/// legacy `BlocksByHash` path, or another peer), `clear_no_progress_probe_streak` must let the
+/// peer probe again *without* marking it proven. Without this, a sole-peer node that syncs its
+/// early blocks over legacy holds its only Zakura peer at the one-probe cap, the no-progress
+/// reaper disconnects it, and the node wedges below the tip with no way to pull the backfill
+/// (gossip only pushes new blocks). The `on_view_changed` non-destructive-advance path and the
+/// `check_block_liveness` sole-peer guard both call this.
+#[test]
+fn block_liveness_external_progress_reopens_unproven_probe_budget() {
+    let config = ZakuraBlockSyncConfig {
+        initial_block_probe_requests: 1,
+        max_requests_without_block_progress: 8,
+        ..ZakuraBlockSyncConfig::default()
+    };
+    let timeout = config.effective_liveness_timeout();
+    let now = Instant::now();
+    let mut window = DownloadWindow::new(&config);
+
+    // Spend the single initial probe: the unproven peer is now capped out.
+    window.outstanding.push(window_request(1));
+    window.arm_liveness(now, timeout);
+    assert_eq!(window.requests_without_block_progress, 1);
+    assert!(
+        window.requests_without_block_progress >= window.no_progress_request_cap(),
+        "the unproven peer is pinned at its one-probe cap"
+    );
+
+    // The node made block progress via another source — no body arrived through THIS peer.
+    window.clear_no_progress_probe_streak();
+
+    assert_eq!(window.requests_without_block_progress, 0);
+    assert!(
+        window.requests_without_block_progress < window.no_progress_request_cap(),
+        "the peer may probe again instead of wedging at the cap"
+    );
+    assert!(
+        !window.has_block_progress(),
+        "external progress must not mark the peer proven"
+    );
+    assert_eq!(
+        window.no_progress_request_cap(),
+        1,
+        "still unproven: the cap stays at the initial probe budget"
+    );
+}
+
 #[test]
 fn block_liveness_resuming_after_idle_gets_fresh_deadline() {
     let timeout = ZakuraBlockSyncConfig::default().effective_liveness_timeout();

zebra-network/src/zakura/testkit/blocksync_fuzz/scenario.rs

@@ -140,6 +140,9 @@ impl ServeProfile {
     }
 
     /// A slow peer: a fixed RTT before the first block plus per-block serve latency.
+    // Retained scaffolding for the deferred `fuzz_reorg` scenario (awaiting the
+    // high-fidelity `Committer<MockVerifier>` tier); not currently constructed.
+    #[allow(dead_code)]
     pub(crate) fn slow(rtt: Duration, per_block: Duration) -> Self {
         Self {
             first_block_latency: LatencyDist::Fixed(rtt),
@@ -271,6 +274,9 @@ pub(crate) enum TipEventKind {
     /// Move the best-header target down to `height` (`HeaderReanchored`).
     HeaderReanchor(block::Height),
     /// Reset the verified-body tip down to `height` (`VerifiedReset`) — a reorg/rollback.
+    // Retained scaffolding for the deferred `fuzz_reorg` scenario (awaiting the
+    // high-fidelity `Committer<MockVerifier>` tier); not currently constructed.
+    #[allow(dead_code)]
     VerifiedReset(block::Height),
 }