test(zakura): make block-sync fuzzers deterministic

CHANGELOG.md

@@ -18,6 +18,12 @@ and this project adheres to [Semantic Versioning](https://semver.org).
 
 ### Fixed
 
+- 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. 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
@@ -2273,10 +2273,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,97 @@ 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.
+        publish_sequencer_view(&self.view_tx, next);
+    }
+}
+
+fn publish_sequencer_view(view_tx: &watch::Sender<SequencerView>, next: SequencerView) {
+    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)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn test_view() -> SequencerView {
+        SequencerView {
+            verified_tip: block::Height(1),
+            verified_hash: block::Hash([1; 32]),
+            download_floor: block::Height(1),
+            finalized: block::Height(1),
+            reset_epoch: 0,
+            reaction_epoch: 0,
+            reorder_len: 0,
+            applying_len: 0,
+            reorder_buffered_bytes: 0,
+            applying_buffered_bytes: 0,
+            unsubmitted_applying_count: 0,
+            submitted_applying_count: 0,
+            submitted_applying_bytes: 0,
+            committed_bytes_per_sec: 0,
+            committed_blocks_per_sec: 0,
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn sequencer_view_rate_refresh_does_not_wake_watchers() {
+        let initial = test_view();
+        let (view_tx, mut view_rx) = watch::channel(initial);
+
+        let rate_only = SequencerView {
+            committed_bytes_per_sec: 1024,
+            committed_blocks_per_sec: 3,
+            ..initial
+        };
+        publish_sequencer_view(&view_tx, rate_only);
+
+        assert_eq!(*view_rx.borrow(), rate_only);
+        assert!(
+            time::timeout(Duration::from_millis(1), view_rx.changed())
+                .await
+                .is_err(),
+            "throughput-only view refresh must not wake watchers"
+        );
+
+        let schedulable = SequencerView {
+            reaction_epoch: 1,
+            ..rate_only
+        };
+        publish_sequencer_view(&view_tx, schedulable);
+
+        view_rx
+            .changed()
+            .await
+            .expect("sequencer view sender is still live");
+        assert_eq!(*view_rx.borrow(), schedulable);
     }
 }

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/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),
 }