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..aea389ed4 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 @@ -241,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, @@ -1138,10 +1263,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 { @@ -3579,19 +3705,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 @@ -3644,24 +3767,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 @@ -4033,12 +4155,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() => {} + } + } } } @@ -6181,7 +6308,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 +6323,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; } @@ -6315,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 @@ -6497,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 @@ -6799,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, @@ -6847,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, @@ -6863,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(); @@ -10468,6 +10601,157 @@ 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 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 @@ -12526,9 +12810,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 +12856,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(&[]);