From b3fa76b0e77cdce385b727b98b70e2a6b784815a Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Tue, 6 Oct 2026 15:16:20 -0600 Subject: [PATCH 1/3] Report blocks the stored-tail drain connects as chain events (#900) (#912) * Report blocks the stored-tail drain connects as chain events (#900) The drain connects a stored block once its parent is the tip: a block that arrived before its parent, or the tail an IBD batch leaves behind. It connected through connect_stored_block, which emits nothing, so -blocknotify, block announcement, Electrum and Esplora subscribers, the streaming API and ZMQ hashblock never heard of that block. The drain now emits BlockConnected after the connect and after the mempool has dropped the block's transactions, as accept_block does. ChainState::emit_chain_event becomes pub(crate) for it; connect_stored_block still emits nothing, since the IBD connector uses it too and stays quiet. The drain keeps its own -stopatheight check: the watcher in main now hears of drained blocks, but only after the walk has moved on. The watcher uses send_if_modified, so when both ask for shutdown the second neither logs nor wakes shutdown waiters again. Co-Authored-By: Claude Opus 5.5 (1M context) * Report a drained block before releasing the accept lock (#900) The stored-tail drain emitted BlockConnected after connect_stored_block had released accept_lock. accept_block reports a connect while it still holds the lock, so a submitblock of the drained block's child could connect and report in between, and subscribers heard of the child before its parent. connect_stored_block_and_report now purges the mempool and emits under the lock, as accept_block does, and emit_chain_event is private again. Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) (cherry picked from commit 6856ee98cdff7a6ec01400b4ca4fe4d749311af4) --- node/src/chain/state.rs | 53 +++++++++++++- node/src/net/manager.rs | 155 +++++++++++++++++++++++++++++++++++++--- satd/src/main.rs | 36 +++++----- satd/tests/regtest.rs | 63 ++++++++++++++++ 4 files changed, 280 insertions(+), 27 deletions(-) diff --git a/node/src/chain/state.rs b/node/src/chain/state.rs index e6160c6b3..c454b903e 100644 --- a/node/src/chain/state.rs +++ b/node/src/chain/state.rs @@ -2779,6 +2779,25 @@ impl ChainState { self.accept_lock.lock() } + /// Whether any thread holds the accept lock, for tests that prove a + /// step happens before it is released. + #[cfg(test)] + pub(crate) fn accept_lock_is_held_for_test(&self) -> bool { + self.accept_lock.is_locked() + } + + /// Hold the chain-event sender, so that an emit blocks until the guard + /// drops, for tests that look at what an emitter holds at that point. + #[cfg(test)] + pub(crate) fn hold_chain_event_sender_for_test( + &self, + ) -> parking_lot::MutexGuard< + '_, + Option>, + > { + self.chain_event_tx.lock() + } + /// Get the highest header height stored (may be ahead of block tip during IBD). pub fn headers_tip_height(&self) -> u32 { self.headers_tip_height.load(Ordering::Relaxed) @@ -4975,8 +4994,30 @@ impl ChainState { /// Connect an already-stored block (DataStored) to the chain tip. /// The block's parent must be the current chain tip. /// - /// Returns the block hash on success. + /// Returns the block hash on success. Emits no chain event: the IBD + /// connector calls this for every block of a sync. pub fn connect_stored_block(&self, hash: &BlockHash) -> Result { + self.connect_stored_block_inner(hash, false) + } + + /// [`Self::connect_stored_block`] for the stored-tail drain, which + /// reports each block as `accept_block` reports a connect (#900): the + /// mempool drops the block's transactions, then `BlockConnected` goes + /// out, both before `accept_lock` is released. Emitting after the + /// release would let a `submitblock` of the child connect and report + /// first, so subscribers would hear of the child before its parent. + pub(crate) fn connect_stored_block_and_report( + &self, + hash: &BlockHash, + ) -> Result { + self.connect_stored_block_inner(hash, true) + } + + fn connect_stored_block_inner( + &self, + hash: &BlockHash, + report: bool, + ) -> Result { use crate::chain::connect_phase::ConnectPhase; let phases = &*self.connect_phases; @@ -5135,6 +5176,16 @@ impl ChainState { // Update MTP cache with this block's timestamp self.push_mtp_cache(entry.height, entry.header.time); + if report { + if let Some(mempool) = self.mempool.get() { + mempool.remove_for_block(&block, entry.height); + } + self.emit_chain_event(crate::chain::events::ChainEvent::BlockConnected { + hash: *hash, + height: entry.height, + }); + } + phases.enter(ConnectPhase::Idle); Ok(*hash) } diff --git a/node/src/net/manager.rs b/node/src/net/manager.rs index bdc7c76d7..61a5ca6b3 100644 --- a/node/src/net/manager.rs +++ b/node/src/net/manager.rs @@ -1138,10 +1138,11 @@ impl PeerManager { /// Core checks the target on every tip it connects /// (`KernelNotifications::blockTip`). Here a connect that emits /// `ChainEvent::BlockConnected` reaches the watcher in `main`; the IBD - /// connector and the stored-tail drain emit no event, so they call this - /// after each connect (#873). Asked for between one connect and the - /// next, shutdown also stops a connector at the target rather than past - /// it, because the connector checks for shutdown before it connects + /// connector emits no event, so it calls this after each connect (#873). + /// The stored-tail drain emits one (#900) and calls this too, because the + /// watcher hears of it asynchronously. Asked for between one connect and + /// the next, shutdown also stops a connector at the target rather than + /// past it, because the connector checks for shutdown before it connects /// another block. fn stop_if_at_height(&self, height: u32) -> bool { let Some((target, shutdown)) = self.stop_at_height.as_ref() else { @@ -6181,7 +6182,12 @@ impl PeerManager { break; }; let fees = Self::compute_block_fee_rates(&block, chain_state); - match chain_state.connect_stored_block(&hash) { + // Connects and reports the block as `accept_block` does. + // `-blocknotify`, block announcement, Electrum and Esplora + // subscribers, the streaming API and ZMQ all learn of new blocks + // from its `BlockConnected`, and without it every one of them + // missed a block that reached the tip here (#900). + match chain_state.connect_stored_block_and_report(&hash) { Ok(_) => { chain_state.bump_connect_heartbeat(); chain_state @@ -6191,8 +6197,10 @@ impl PeerManager { mempool.remove_for_block(&block, block_height); reconsider_orphans_on_block(orphanage, mempool, chain_state, &block); connected += 1; - // `-stopatheight`, which this connect, like the IBD - // connector's, reports to no chain-event subscriber. + // `-stopatheight`. The connect's event reaches the watcher + // in `main` as well, but only after this walk has gone + // on to its next block; checking here stops the walk + // at the target rather than past it (#873). if peer_manager.upgrade().is_some_and(|pm| pm.stop_if_at_height(block_height)) { break; } @@ -12526,9 +12534,10 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } - /// #873: the stored-tail drain, which emits no chain event either, - /// stops at `-stopatheight` and asks for shutdown. The manager only - /// holds the target; the drain walks its own chain state. + /// #873: the stored-tail drain stops at `-stopatheight` and asks for + /// shutdown itself, rather than leaving it to the watcher that hears its + /// chain events after the walk has moved on. The manager only holds the + /// target; the drain walks its own chain state. /// /// Perturbation: without the check, the walk connects all nine. #[test] @@ -12571,6 +12580,132 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// #900: every block the stored-tail drain connects is reported with one + /// `BlockConnected`, in order, as `accept_block` reports its connects. + /// `-blocknotify`, block announcement, Electrum, Esplora SSE, streaming + /// and ZMQ subscribers hear of new blocks only through that event. + /// + /// Perturbation: drop the emit from `connect_stored_tail` and no event + /// arrives. + #[test] + fn the_stored_tail_drain_reports_each_block_it_connects() { + use crate::chain::events::ChainEvent; + use crate::chain::state::tests::{ + build_test_block, make_chain_state, store_block_without_connecting, + }; + + let (cs, dir) = make_chain_state(); + let genesis = bitcoin::constants::genesis_block(Network::Regtest).block_hash(); + let b1 = build_test_block(genesis, 1, 1_707_000_000); + let mut parent = cs.accept_block(&b1).expect("connect block 1").hash(); + let mut headers = Vec::new(); + let mut tail = Vec::new(); + for h in 2..=4u32 { + let b = build_test_block(parent, h, 1_707_000_000 + h); + parent = b.block_hash(); + headers.push(b.header); + tail.push((b, h)); + } + let (accepted, err) = cs.accept_headers(&headers); + assert_eq!(accepted, 3, "fixture: headers must be accepted ({err:?})"); + for (b, h) in &tail { + store_block_without_connecting(&cs, b, *h); + } + let (tx, mut rx) = tokio::sync::broadcast::channel::(16); + cs.set_chain_event_sender(tx); + let chain_state = Arc::new(cs); + + let connected = PeerManager::connect_stored_tail( + &chain_state, + &FeeEstimator::new(), + &Arc::new(Mempool::new(1_000_000, 0)), + &Arc::new(TxOrphanage::with_defaults()), + &std::sync::Weak::new(), + ); + assert_eq!(connected, 3); + assert_eq!(chain_state.tip_height(), 4); + + let events: Vec = std::iter::from_fn(|| rx.try_recv().ok()).collect(); + let expected: Vec<(bitcoin::BlockHash, u32)> = + tail.iter().map(|(b, h)| (b.block_hash(), *h)).collect(); + let got: Vec<(bitcoin::BlockHash, u32)> = events + .iter() + .filter_map(|e| match e { + ChainEvent::BlockConnected { hash, height } => Some((*hash, *height)), + _ => None, + }) + .collect(); + assert_eq!(got, expected, "one BlockConnected per drained block, in order: {events:?}"); + assert_eq!(events.len(), expected.len(), "and nothing else: {events:?}"); + let _ = std::fs::remove_dir_all(&dir); + } + + /// Every other connect at the tip reports its block before it releases + /// `accept_lock`, which is what keeps events in connect order. A drain + /// that released the lock first would let a `submitblock` of the child + /// connect and report in between, so subscribers heard of the child + /// before its parent. The test holds the event sender, so the drain's + /// emit blocks, and checks that the drain still holds the lock there. + #[test] + fn the_stored_tail_drain_reports_a_block_before_releasing_the_accept_lock() { + use crate::chain::events::ChainEvent; + use crate::chain::state::tests::{ + build_test_block, make_chain_state, store_block_without_connecting, + }; + + let (cs, dir) = make_chain_state(); + let genesis = bitcoin::constants::genesis_block(Network::Regtest).block_hash(); + let b1 = build_test_block(genesis, 1, 1_707_000_000); + let parent = cs.accept_block(&b1).expect("connect block 1").hash(); + let b2 = build_test_block(parent, 2, 1_707_000_002); + let (accepted, err) = cs.accept_headers(&[b2.header]); + assert_eq!(accepted, 1, "fixture: header must be accepted ({err:?})"); + store_block_without_connecting(&cs, &b2, 2); + let (tx, mut rx) = tokio::sync::broadcast::channel::(16); + cs.set_chain_event_sender(tx); + let chain_state = Arc::new(cs); + + let connected = std::thread::scope(|scope| { + // Inside the scope, so a failed assertion drops it and lets the + // drain finish before the scope joins it. + let sender = chain_state.hold_chain_event_sender_for_test(); + let drain = scope.spawn(|| { + PeerManager::connect_stored_tail( + &chain_state, + &FeeEstimator::new(), + &Arc::new(Mempool::new(1_000_000, 0)), + &Arc::new(TxOrphanage::with_defaults()), + &std::sync::Weak::new(), + ) + }); + let deadline = Instant::now() + Duration::from_secs(10); + while chain_state.tip_height() < 2 { + assert!(Instant::now() < deadline, "the drain never connected block 2"); + std::thread::sleep(Duration::from_millis(1)); + } + // The tip moves under the lock, so from here the drain either + // still holds it, waiting on the sender, or has let it go. + let watch_until = Instant::now() + Duration::from_millis(500); + while Instant::now() < watch_until { + assert!( + chain_state.accept_lock_is_held_for_test(), + "the drain released accept_lock before reporting block 2" + ); + std::thread::sleep(Duration::from_millis(1)); + } + drop(sender); + drain.join().expect("drain thread") + }); + assert_eq!(connected, 1); + match rx.try_recv() { + Ok(ChainEvent::BlockConnected { hash, height }) => { + assert_eq!((hash, height), (b2.block_hash(), 2)); + } + other => panic!("expected BlockConnected for block 2, got {other:?}"), + } + let _ = std::fs::remove_dir_all(&dir); + } + /// The re-arm gate must refuse to (re)create a scheduler while the /// connect frontier is fork-blocked — the linear connector cannot /// reorg, so arming it there re-wedges on `bad-prevblk`, and with the diff --git a/satd/src/main.rs b/satd/src/main.rs index 8e260dfe4..d8e207475 100644 --- a/satd/src/main.rs +++ b/satd/src/main.rs @@ -1782,10 +1782,12 @@ async fn run() -> Option> // between block-connected and shutdown is bounded by the broadcast // delivery (microseconds) rather than a polling interval. // - // The IBD connector and the stored-tail drain emit no chain event. - // They check the target themselves (the peer manager is built with - // it, below), between one block and the next, so that IBD stops at the - // target rather than past it (#873). + // The IBD connector emits no chain event. It checks the target itself + // (the peer manager is built with it, below), between one block and the + // next, so that IBD stops at the target rather than past it (#873). The + // stored-tail drain emits one (#900) and checks the target as well, so + // by the time its event lands here shutdown may already be asked for; + // the watcher then leaves it alone rather than announcing it twice. if let Some(target_height) = config.stopatheight { let mut rx = chain_event_tx.subscribe(); let stop_tx = shutdown_tx.clone(); @@ -1802,12 +1804,13 @@ async fn run() -> Option> .. }) => { if height >= target_height { - tracing::info!( - target = target_height, - tip = height, - "-stopatheight reached; broadcasting shutdown" - ); - let _ = stop_tx.send(true); + if stop_tx.send_if_modified(|stop| !std::mem::replace(stop, true)) { + tracing::info!( + target = target_height, + tip = height, + "-stopatheight reached; broadcasting shutdown" + ); + } return; } } @@ -1818,12 +1821,13 @@ async fn run() -> Option> // re-check the current tip explicitly so we // don't miss the boundary. if chain_state_for_stop.tip_height() >= target_height { - tracing::info!( - target = target_height, - tip = chain_state_for_stop.tip_height(), - "-stopatheight reached after lag-recovery; broadcasting shutdown" - ); - let _ = stop_tx.send(true); + if stop_tx.send_if_modified(|stop| !std::mem::replace(stop, true)) { + tracing::info!( + target = target_height, + tip = chain_state_for_stop.tip_height(), + "-stopatheight reached after lag-recovery; broadcasting shutdown" + ); + } return; } } diff --git a/satd/tests/regtest.rs b/satd/tests/regtest.rs index 1d2bf5fa7..819d5a593 100644 --- a/satd/tests/regtest.rs +++ b/satd/tests/regtest.rs @@ -458,6 +458,69 @@ fn test_stopatheight_zero_disabled() { node.stop(); } +/// `-blocknotify` runs for a block that reaches the tip through the +/// stored-tail drain (#900). A block that arrives before its parent is +/// stored; once the parent extends the tip, the drain connects it. The drain +/// emitted no chain event, so `-blocknotify`, like every other chain-event +/// subscriber, never heard of that block. +/// +/// Perturbation: drop the emit from `connect_stored_tail` and only the +/// parent is notified. +#[test] +fn blocknotify_runs_for_a_block_the_stored_tail_drain_connects() { + let notify_dir = fresh_test_datadir("satd-blocknotify-drain"); + let notified = notify_dir.join("blocknotify.log"); + let mut node = + TestNode::start(&[&format!("--blocknotify=echo %s >> {}", notified.display())]); + + let template = node.rpc_call("getblocktemplate").unwrap(); + let t = &template["result"]; + let genesis: bitcoin::BlockHash = t["previousblockhash"].as_str().unwrap().parse().unwrap(); + let time = t["curtime"].as_u64().unwrap() as u32; + let parent = build_regtest_block(genesis, 1, time, vec![]); + let child = build_regtest_block(parent.block_hash(), 2, time + 1, vec![]); + let submit = |method: &str, hex: String| { + let r = node + .rpc_call_with_params(method, vec![serde_json::json!(hex)]) + .expect("rpc"); + assert!(r["error"].is_null(), "{method}: {r}"); + }; + + // Both headers first, as a peer announces them before sending blocks; + // then the child, which is stored to wait for its parent, then the parent. + submit("submitheader", hex::encode(bitcoin::consensus::serialize(&parent.header))); + submit("submitheader", hex::encode(bitcoin::consensus::serialize(&child.header))); + submit("submitblock", hex::encode(bitcoin::consensus::serialize(&child))); + let count = node.rpc_call("getblockcount").unwrap(); + assert_eq!(count["result"], 0, "fixture: the child waits for its parent"); + submit("submitblock", hex::encode(bitcoin::consensus::serialize(&parent))); + + // The drain wakes every 500 ms, and the notifier runs each command in turn. + let read = || -> Vec { + std::fs::read_to_string(¬ified) + .unwrap_or_default() + .lines() + .map(str::to_string) + .collect() + }; + let deadline = Instant::now() + test_timeout(30); + while read().len() < 2 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(100)); + } + // A little longer, so a second notification for either block shows up. + std::thread::sleep(Duration::from_secs(1)); + + let count = node.rpc_call("getblockcount").unwrap(); + assert_eq!(count["result"], 2, "the drain connects the child"); + assert_eq!( + read(), + vec![parent.block_hash().to_string(), child.block_hash().to_string()], + "-blocknotify runs once for each block, parent then child" + ); + node.stop(); + let _ = std::fs::remove_dir_all(¬ify_dir); +} + #[test] fn test_sat_cli_integration() { let mut node = TestNode::start(&[]); From 7dc76e436191a8f78f0871a59942d53bac50694c Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Tue, 6 Oct 2026 15:50:46 -0600 Subject: [PATCH 2/3] p2p: run the manager loop's maintenance on time under load (#909) (#917) * p2p: run the manager loop's maintenance on time under load (#909) The manager loop's maintenance pass holds IBD stall detection: the release of a height stuck with one peer (15 s at the connect cursor), the drop of silent peers, and work for idle peers. Its cadences count passes ("every 4 ticks (2s)"), so it has to run every 500 ms. Under initial block download it did not. The fast drain added in #781 set drain_only, so every run of fast drains ended in a wait for the next tick, and the pass after that tick found the queue full again. With the manager storing each arriving block, an fsync apiece, the queue is full at every wake. Measured on a fresh mainnet sync over two minutes, the pass ran 3-4 times instead of ~240, the stale release never fired, and the block at the connect cursor sat with one slow peer for up to 53 s. DrainPacer decides after each drain. Maintenance is due on an interval tick, as before, or once 500 ms have passed since it last started, and then runs after the drain in progress however full the queue is. A full queue otherwise goes straight back to draining (#781's throughput), and does so after maintenance too. A drain_now wake is neither a tick nor late, so it drains and waits, as #776 requires. The interval uses MissedTickBehavior::Delay, so the ticks a busy stretch skipped do not fire as a burst of passes. Co-Authored-By: Claude Opus 5.5 (1M context) * docs(release-notes): cite only runs of the shipped #909 fix Six runs of a release build of this branch against nine of the 0.6.0 release binary, replacing figures that mixed in earlier versions of the fix and an instrumented build. Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) (cherry picked from commit 89ce0e5a4e4a4ae1420c5e61adb4c2a94425319a) --- node/src/net/manager.rs | 247 +++++++++++++++++++++++++++++++++------- 1 file changed, 209 insertions(+), 38 deletions(-) diff --git a/node/src/net/manager.rs b/node/src/net/manager.rs index 61a5ca6b3..46597c4fc 100644 --- a/node/src/net/manager.rs +++ b/node/src/net/manager.rs @@ -196,10 +196,80 @@ const MAX_IN_FLIGHT_BLOCKS_PER_PEER: usize = 256; /// comes straight back for more (see `run`). const EVENTS_PER_DRAIN: usize = 64; -/// Consecutive full drains before the loop takes its normal path anyway. -/// Bounds how long sustained inbound traffic can defer the maintenance -/// section; at `EVENTS_PER_DRAIN` apiece this is 512 messages. -const MAX_FAST_DRAINS: u32 = 8; +/// How often the manager loop runs its periodic maintenance: stall +/// detection and the release of stuck IBD heights, work for idle peers, fee +/// filters, expiries. Every cadence in that section counts these passes +/// ("every 4 ticks (2s)"), so the pass has to come round on time, however +/// busy the event queue is (#909). +const MAINTENANCE_INTERVAL: Duration = Duration::from_millis(500); + +/// What the manager loop does after one drain of the peer event queue. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum AfterDrain { + /// Maintenance is due: run it, then go on draining. + Maintain, + /// The queue was still full and maintenance is not due: go straight + /// back to the queue. + DrainAgain, + /// The queue is drained and maintenance is not due: wait for the next + /// interval tick or a parked pong. + Wait, +} + +/// The manager loop's choice after each drain (#909). +/// +/// Maintenance is due on every interval tick, as before, and also once +/// `MAINTENANCE_INTERVAL` has passed since it last started, so a queue too +/// busy to wait for a tick still gets it. It runs after the drain in +/// progress, however full the queue is. A drain takes at most +/// `EVENTS_PER_DRAIN` events, so maintenance is late by at most one drain. +/// +/// 0.6.0 tied maintenance to an interval wake instead, and let a full queue +/// put it off: a run of fast drains ended in a wait for the next tick, and +/// the pass after that tick found the queue full again. Under initial block +/// download the queue is full at every wake, and with an fsync for every +/// block stored a single drain can take most of a second, so stall detection +/// ran tens of seconds apart and a block held by one slow peer stayed with it +/// for up to a minute. +/// +/// A wake from `drain_now`, for a pong parked behind the queue, is neither a +/// tick nor late, so it drains and waits (#776): maintenance never runs at +/// the rate pings arrive. +#[derive(Debug, Default)] +struct DrainPacer { + /// When maintenance last started; `None` before the first pass. + last_maintenance: Option, + /// An interval tick has woken the loop since maintenance last started. + ticked: bool, +} + +impl DrainPacer { + /// Record that an interval tick woke the loop. + fn ticked(&mut self) { + self.ticked = true; + } + + /// Decide what follows a drain that took `processed` events, at `now`. + fn after_drain(&self, processed: usize, now: Instant) -> AfterDrain { + let due = self.ticked + || self + .last_maintenance + .is_none_or(|t| now.duration_since(t) >= MAINTENANCE_INTERVAL); + if due { + AfterDrain::Maintain + } else if processed >= EVENTS_PER_DRAIN { + AfterDrain::DrainAgain + } else { + AfterDrain::Wait + } + } + + /// Record that maintenance started at `at`. + fn maintained(&mut self, at: Instant) { + self.last_maintenance = Some(at); + self.ticked = false; + } +} /// Token-bucket cap for the promotion-INV drain (§8): at most this many /// reloaded-and-promoted transactions are announced per drain tick, so a @@ -3580,19 +3650,16 @@ impl PeerManager { /// Run the main event loop. Returns when shutdown signal is received. pub async fn run(self: &Arc) { let mut event_rx = self.event_rx.lock().await; - let mut sync_interval = tokio::time::interval(std::time::Duration::from_millis(500)); + let mut sync_interval = tokio::time::interval(MAINTENANCE_INTERVAL); + // A busy queue skips the waits that consume ticks. When it quiets, + // one late tick is enough: a burst of them would run maintenance back + // to back. + sync_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); let mut last_tip: u32 = 0; let mut ticks: u64 = 0; let shutdown = self.shutdown.clone(); - // True when the last wake came from `drain_now` rather than the - // interval: a peer's pong is parked behind the queue, and draining - // it is the whole of what was asked for. The periodic maintenance - // below stays on the interval — running it at the rate pings arrive - // would change every cadence in this loop. - let mut drain_only = false; - // Consecutive iterations that skipped the wait because the event queue - // was still full. Reset as soon as one drain comes up short. - let mut fast_drains: u32 = 0; + // Maintain, drain again, or wait: see `DrainPacer`. + let mut pacer = DrainPacer::default(); loop { // Manager-loop heartbeat: bumped on every iteration so the @@ -3645,24 +3712,23 @@ impl PeerManager { // out the 60-second wait in Core's `p2p_blockfilters`. So when the // queue was still full, come straight back to it. // - // Bounded, because a fast drain skips the maintenance below: - // after `MAX_FAST_DRAINS` in a row we take the normal path even - // with a backlog, so stall detection, fee filters and the rest - // keep their cadence under sustained load. - if processed >= EVENTS_PER_DRAIN && fast_drains < MAX_FAST_DRAINS { - fast_drains += 1; - drain_only = true; - tokio::task::yield_now().await; - continue; - } - fast_drains = 0; - - if drain_only { - drain_only = tokio::select! { - _ = sync_interval.tick() => false, - _ = self.drain_now.notified() => true, - }; - continue; + // The maintenance below runs on time regardless, at most one + // drain late, so stall detection, fee filters and the rest keep + // their cadence under sustained load (#909). + let now = Instant::now(); + match pacer.after_drain(processed, now) { + AfterDrain::DrainAgain => { + tokio::task::yield_now().await; + continue; + } + AfterDrain::Wait => { + tokio::select! { + _ = sync_interval.tick() => pacer.ticked(), + _ = self.drain_now.notified() => {} + } + continue; + } + AfterDrain::Maintain => pacer.maintained(now), } // Check sync progress and request more blocks @@ -4034,12 +4100,17 @@ impl PeerManager { self.expire_compact_state(); } - // Yield to tokio runtime, waking early when a peer is waiting - // on the drain above to answer a ping. - drain_only = tokio::select! { - _ = sync_interval.tick() => false, - _ = self.drain_now.notified() => true, - }; + // Back to a queue the last drain left full. Otherwise yield to + // the tokio runtime, waking early when a peer is waiting on the + // drain above to answer a ping. + if processed >= EVENTS_PER_DRAIN { + tokio::task::yield_now().await; + } else { + tokio::select! { + _ = sync_interval.tick() => pacer.ticked(), + _ = self.drain_now.notified() => {} + } + } } } @@ -10476,6 +10547,106 @@ mod tests { use super::*; use crate::net::peer::{Direction, PeerInfo, PeerState}; + /// Drive the pacer through `secs` of a queue that is full after every + /// drain, each drain taking `drain`. Returns when maintenance started. + fn maintenance_under_a_full_queue(drain: Duration, secs: u64) -> Vec { + let t0 = Instant::now(); + let mut pacer = DrainPacer::default(); + let mut now = t0; + let mut ran = Vec::new(); + while now < t0 + Duration::from_secs(secs) { + match pacer.after_drain(EVENTS_PER_DRAIN, now) { + AfterDrain::Maintain => { + pacer.maintained(now); + ran.push(now); + } + AfterDrain::DrainAgain => {} + AfterDrain::Wait => panic!("a full queue must never wait"), + } + now += drain; + } + ran + } + + /// #909: a queue that stays full, as in initial block download, still + /// gets maintenance every `MAINTENANCE_INTERVAL`, at most one drain late, + /// whether a drain is quick or, with an fsync per stored block, takes + /// most of a second. 0.6.0 let the full queue put maintenance off, and + /// stall detection ran tens of seconds apart. + /// + /// Perturbation: in `after_drain`, check for a full queue before checking + /// whether maintenance is due, and maintenance never runs. + #[test] + fn maintenance_runs_on_time_under_a_queue_that_stays_full() { + for drain in [Duration::from_millis(20), Duration::from_millis(600)] { + let ran = maintenance_under_a_full_queue(drain, 30); + let slowest = ran.windows(2).map(|w| w[1] - w[0]).max().unwrap_or_default(); + assert!( + slowest <= MAINTENANCE_INTERVAL + drain, + "{drain:?} drains: maintenance {slowest:?} apart" + ); + let expected = 30_000 / (MAINTENANCE_INTERVAL + drain).as_millis() as usize; + assert!(ran.len() >= expected, "{drain:?} drains: {} passes, want {expected}", ran.len()); + } + } + + /// #781's fast drains stay: between maintenance passes a full queue is + /// drained again at once rather than once per interval tick. + #[test] + fn a_full_queue_is_drained_again_until_maintenance_is_due() { + let t0 = Instant::now(); + let mut pacer = DrainPacer::default(); + assert_eq!(pacer.after_drain(EVENTS_PER_DRAIN, t0), AfterDrain::Maintain, "first pass"); + pacer.maintained(t0); + for ms in [1, 100, 499] { + let at = t0 + Duration::from_millis(ms); + assert_eq!(pacer.after_drain(EVENTS_PER_DRAIN, at), AfterDrain::DrainAgain, "{ms} ms"); + } + let due = t0 + MAINTENANCE_INTERVAL; + assert_eq!(pacer.after_drain(EVENTS_PER_DRAIN, due), AfterDrain::Maintain); + } + + /// A drained queue waits for the next wake until maintenance is due. + /// That holds for a wake from `drain_now` too (#776): pongs parked + /// behind the queue arrive at the rate peers ping, and maintenance must + /// not run at that rate. + #[test] + fn a_drained_queue_waits_until_maintenance_is_due() { + let t0 = Instant::now(); + let mut pacer = DrainPacer::default(); + pacer.maintained(t0); + for ms in [10, 20, 30, 250, 499] { + let at = t0 + Duration::from_millis(ms); + assert_eq!(pacer.after_drain(1, at), AfterDrain::Wait, "{ms} ms"); + } + assert_eq!(pacer.after_drain(0, t0 + MAINTENANCE_INTERVAL), AfterDrain::Maintain); + } + + /// An idle loop runs maintenance at every interval tick, as before, + /// even when a tick finds it less than `MAINTENANCE_INTERVAL` after the + /// last pass started because that pass began late, behind a slower + /// drain. Skipping such a tick would stretch every cadence counted in + /// ticks. + /// + /// Perturbation: make maintenance due by time alone and the tick after a + /// 40 ms drain is skipped. + #[test] + fn an_idle_loop_maintains_at_every_tick() { + let t0 = Instant::now(); + let mut pacer = DrainPacer::default(); + // Each tick's drain takes this long before the pass decides. + let drains_ms = [0u64, 40, 2, 0, 15, 1, 30, 0]; + for (k, d) in drains_ms.iter().enumerate() { + let tick = t0 + MAINTENANCE_INTERVAL * k as u32; + if k > 0 { + pacer.ticked(); + } + let at = tick + Duration::from_millis(*d); + assert_eq!(pacer.after_drain(2, at), AfterDrain::Maintain, "tick {k}"); + pacer.maintained(at); + } + } + /// A peer relaying a *policy*-rejected tx (fee floor, dust, mempool limits, /// RBF, conflicts, non-standard) must NOT accrue ban score — banning for /// these severs honest peers on low-fee networks (the testnet4 wedge). Only From cb3db3effa6f25f9c23491dcc4d31ea4b90e4d09 Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Tue, 6 Oct 2026 16:23:19 -0600 Subject: [PATCH 3/3] IBD: warn that the connector is stuck only after a 20 s wait (#904) (#911) * IBD: warn that the connector is stuck only after a 10 s wait (#904) The connector logged "Connector stuck waiting for block data" at WARN the first moment it waited on any height. Blocks arrive out of order and the connector takes them in height order, so early in a healthy sync that was nearly one warning per block: 112 of the first 624 log lines on a fresh mainnet sync, 490 of 508 in a fresh Umbrel install. StuckWait tracks which height the connector waits on and since when. The warning fires once that wait reaches 10 s, then every 60 s while it lasts, and carries waited_secs. A new height, or a successful connect, starts over. In the #904 run the waits worth a warning had the block in flight 23-60 s on one peer; 97 of the 112 warnings were under 5 s. Co-Authored-By: Claude Opus 5.5 (1M context) * IBD: raise the stuck-wait warning to 20 s, above the stale-block release (#904) The scheduler takes a block at the connect cursor back from a slow peer after 15 s and asks another. A 10 s threshold warned during waits that release was about to end; 20 s warns only once it has not. Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) (cherry picked from commit 98bca53e8e3d3555fea7c7a1381f4412d0af155d) --- node/src/net/manager.rs | 135 +++++++++++++++++++++++++++++++++++----- 1 file changed, 120 insertions(+), 15 deletions(-) diff --git a/node/src/net/manager.rs b/node/src/net/manager.rs index 46597c4fc..aea389ed4 100644 --- a/node/src/net/manager.rs +++ b/node/src/net/manager.rs @@ -311,6 +311,61 @@ const BG_CATCHUP_PER_PEER_PER_PASS: usize = 16; /// genesis→snapshot validation. const BG_CATCHUP_FLUSH_EVERY: u64 = 2000; +/// How long the IBD connector waits on one height's block data before it +/// warns that it is stuck (#904). Blocks arrive out of order from many peers +/// and the connector takes them in height order, so it waits on some height +/// for a second or two all through a healthy sync. In the run that set this, +/// those waits were all under 5 s; the ones worth a warning had the block in +/// flight on a single peer for 23–60 s. +/// +/// Above the 15 s after which the scheduler takes a block at the connect +/// cursor back from a peer and asks another (`release_stale_inflight`), with +/// room for the maintenance pass that does it: a wait that release ends is +/// the sync handling a slow peer, not a stuck connector. +const STUCK_WAIT_WARN_AFTER: Duration = Duration::from_secs(20); + +/// How often the stuck-wait warning repeats while the wait goes on. +const STUCK_WAIT_WARN_EVERY: Duration = Duration::from_secs(60); + +/// The IBD connector's current wait on one height's block data, for the +/// "Connector stuck waiting for block data" warning. +#[derive(Debug, Clone, Copy)] +struct StuckWait { + height: u32, + since: Instant, + warned_at: Option, +} + +impl StuckWait { + /// Record that the connector is waiting on `height` at `now`, and say + /// whether to warn: once the wait passes `STUCK_WAIT_WARN_AFTER`, then + /// every `STUCK_WAIT_WARN_EVERY` while it lasts. A wait on a different + /// height starts over. + fn should_warn(wait: &mut Option, height: u32, now: Instant) -> bool { + if wait.is_none_or(|w| w.height != height) { + *wait = Some(StuckWait { + height, + since: now, + warned_at: None, + }); + } + let Some(w) = wait.as_mut() else { + return false; + }; + if now.duration_since(w.since) < STUCK_WAIT_WARN_AFTER { + return false; + } + if w + .warned_at + .is_some_and(|t| now.duration_since(t) < STUCK_WAIT_WARN_EVERY) + { + return false; + } + w.warned_at = Some(now); + true + } +} + /// Per-address reconnect backoff state. struct ReconnectState { attempts: u32, @@ -6394,11 +6449,10 @@ impl PeerManager { let mut backpressure_paused = false; let mut backpressure_pause_started: Option = None; - // Track which height we've already logged as "stuck waiting for - // block data" so we don't flood the log when the connector spins - // for 5+ minutes on a missing height. One line per stuck height, - // plus one refresh every 60s while still stuck. - let mut last_stuck_log: Option<(u32, Instant)> = None; + // The height the connector is waiting on for block data, and since + // when, for the "stuck waiting for block data" warning. It fires + // once a wait passes `STUCK_WAIT_WARN_AFTER`, then once a minute. + let mut stuck_wait: Option = None; 'connect: loop { // Shutdown: stop between blocks. Left running, the loop went on @@ -6576,6 +6630,7 @@ impl PeerManager { Ok(_) => { connected_count += 1; retry_count = 0; + stuck_wait = None; // Lock-free progress signal for the stall watchdog. // Must run on every successful connect, before any // subsequent step that takes a lock the watchdog @@ -6878,16 +6933,14 @@ impl PeerManager { chain_state.connect_phases().enter(ConnectPhase::WaitingForBlockData); // Diagnostic for the wedge class where the connector spins // forever on a HeaderOnly entry whose data never arrives. - // Log once per stuck height, then refresh every 60s with a + // Once the wait on this height passes the threshold, log a // scheduler-state snapshot so we can see if the downloader - // is even trying. Cheap: the scheduler read is a single - // RwLock::read() and three HashMap lookups. - let should_log = match last_stuck_log { - None => true, - Some((h, _)) if h != next_height => true, - Some((_, t)) => t.elapsed() >= Duration::from_secs(60), - }; - if should_log { + // is even trying, and refresh it every 60s. A shorter wait is + // out-of-order download working as intended, and warning on + // it filled the log with one line per block (#904). Cheap: + // the scheduler read is a single RwLock::read() and three + // HashMap lookups. + if StuckWait::should_warn(&mut stuck_wait, next_height, Instant::now()) { let ( in_pending, in_flight, @@ -6926,9 +6979,11 @@ impl PeerManager { None:: }); let entry = chain_state.get_block_index(&hash); + let waited_secs = stuck_wait.map(|w| w.since.elapsed().as_secs()); tracing::warn!( height = next_height, %hash, + ?waited_secs, in_pending, in_flight, downloaded, @@ -6942,7 +6997,6 @@ impl PeerManager { data_pos = entry.as_ref().map(|e| e.data_pos), "Connector stuck waiting for block data; scheduler state for this height" ); - last_stuck_log = Some((next_height, Instant::now())); } let (lock, cvar) = &**connect_signal; let mut ready = lock.lock(); @@ -10647,6 +10701,57 @@ mod tests { } } + /// A wait shorter than the threshold is out-of-order download, not a + /// stall: no warning, however many heights it happens at (#904). + #[test] + fn stuck_wait_does_not_warn_on_an_ordinary_wait() { + let t0 = Instant::now(); + let mut wait = None; + assert!(!StuckWait::should_warn(&mut wait, 100, t0)); + assert!(!StuckWait::should_warn(&mut wait, 100, t0 + Duration::from_secs(1))); + // The next height starts its own wait, so waits just under the + // threshold, such as one a stale-block release ends at 15 s, never + // add up. + let under = STUCK_WAIT_WARN_AFTER - Duration::from_secs(1); + for (i, height) in (101..120).enumerate() { + let at = t0 + Duration::from_secs(2) + STUCK_WAIT_WARN_AFTER * i as u32; + assert!(!StuckWait::should_warn(&mut wait, height, at)); + assert!(!StuckWait::should_warn(&mut wait, height, at + under)); + } + } + + /// Past the threshold on one height, the warning fires once, then again + /// a minute later while the same wait goes on. + #[test] + fn stuck_wait_warns_once_past_the_threshold_then_every_minute() { + let t0 = Instant::now(); + let mut wait = None; + assert!(!StuckWait::should_warn(&mut wait, 100, t0)); + let just_before = t0 + STUCK_WAIT_WARN_AFTER - Duration::from_millis(1); + assert!(!StuckWait::should_warn(&mut wait, 100, just_before)); + let first = t0 + STUCK_WAIT_WARN_AFTER; + assert!(StuckWait::should_warn(&mut wait, 100, first)); + assert!(!StuckWait::should_warn(&mut wait, 100, first + Duration::from_secs(1))); + let second = first + STUCK_WAIT_WARN_EVERY; + assert!(!StuckWait::should_warn(&mut wait, 100, second - Duration::from_millis(1))); + assert!(StuckWait::should_warn(&mut wait, 100, second)); + assert!(!StuckWait::should_warn(&mut wait, 100, second + Duration::from_secs(1))); + } + + /// Moving to another height ends the wait, warned or not: the new + /// height gets the full threshold before it can warn. + #[test] + fn stuck_wait_restarts_on_a_new_height() { + let t0 = Instant::now(); + let mut wait = None; + assert!(!StuckWait::should_warn(&mut wait, 100, t0)); + assert!(StuckWait::should_warn(&mut wait, 100, t0 + Duration::from_secs(30))); + let moved = t0 + Duration::from_secs(31); + assert!(!StuckWait::should_warn(&mut wait, 101, moved)); + assert!(!StuckWait::should_warn(&mut wait, 101, moved + Duration::from_secs(5))); + assert!(StuckWait::should_warn(&mut wait, 101, moved + STUCK_WAIT_WARN_AFTER)); + } + /// A peer relaying a *policy*-rejected tx (fee floor, dust, mempool limits, /// RBF, conflicts, non-standard) must NOT accrue ban score — banning for /// these severs honest peers on low-fee networks (the testnet4 wedge). Only