diff --git a/changelog.d/stall-window.md b/changelog.d/stall-window.md new file mode 100644 index 000000000..50ca1ee4c --- /dev/null +++ b/changelog.d/stall-window.md @@ -0,0 +1,3 @@ +Fixed + +- **An IBD stall window starts when its first getdata is written and ends when that peer owes nothing.** Time spent idle between batches no longer counts against the next ask. A later getdata while a hash from that window is still outstanding does not move the deadline. diff --git a/crates/rbitcoin-net/src/ibd/assign.rs b/crates/rbitcoin-net/src/ibd/assign.rs index 4f0f1f6b6..7b99b951d 100644 --- a/crates/rbitcoin-net/src/ibd/assign.rs +++ b/crates/rbitcoin-net/src/ibd/assign.rs @@ -1730,8 +1730,12 @@ pub(in crate::ibd) mod tests { ); st.slots[0].getdata_wire.wrote(1, 9_000); assert!( - super::super::peer_io::getdata_on_wire_stalled(&st.slots[0], 9_000 + 29_999, 30_000), - "last rx at 42ms is the clock; the write at 9s does not refresh it" + !super::super::peer_io::getdata_on_wire_stalled(&st.slots[0], 9_000 + 29_999, 30_000), + "the opening write starts 30s; last rx at 42ms does not" + ); + assert!( + super::super::peer_io::getdata_on_wire_stalled(&st.slots[0], 9_000 + 30_001, 30_000), + "silence after the opening write still stalls" ); let _ = std::fs::remove_dir_all(dir); } @@ -1787,8 +1791,12 @@ pub(in crate::ibd) mod tests { let now = 50_000; st.slots[0].getdata_wire.wrote(1, now); assert!( - getdata_on_wire_stalled(&st.slots[0], now + 1, 30_000), - "once the ask is on the socket, silence is measured from last rx" + !getdata_on_wire_stalled(&st.slots[0], now + 1, 30_000), + "the opening write starts a fresh 30s, ignoring idle time before it" + ); + assert!( + getdata_on_wire_stalled(&st.slots[0], now + 30_001, 30_000), + "silence after the opening write still stalls" ); let _ = std::fs::remove_dir_all(dir); } diff --git a/crates/rbitcoin-net/src/ibd/dial.rs b/crates/rbitcoin-net/src/ibd/dial.rs index 99b0031a3..bb16ba378 100644 --- a/crates/rbitcoin-net/src/ibd/dial.rs +++ b/crates/rbitcoin-net/src/ibd/dial.rs @@ -671,9 +671,12 @@ pub(crate) fn note_misbehavior_dead( /// That peer is left alone while anyone else is delivering blocks. When /// every on-wire peer is withholding, one is disconnected. /// -/// The window opens when getdata is written. Hashes still queued in the -/// writer do not start it. A later getdata write does not refresh either -/// clock. Headers and pings are socket bytes, not block progress. +/// The window opens when the first getdata of an empty window is written. +/// Hashes still queued in the writer do not start it. Idle time before that +/// write does not count. A later getdata write while any of those hashes +/// are still outstanding does not refresh the window. Draining the last +/// outstanding hash disarms it. Headers and pings are socket bytes, not +/// block progress. /// /// Stalled addresses enter a cooldown so redial does not immediately re-open /// the same host under a new peer id (log spam + wasted slots). @@ -1501,8 +1504,10 @@ mod tests { slot.getdata_wire.wrote(1, written_ms); } + /// Idle time before a window does not count. The opening write starts + /// 30s. A later write while that hash is still out does not. #[test] - fn disconnect_stalled_later_getdata_does_not_refresh_silence() { + fn disconnect_stalled_opening_write_starts_a_fresh_window() { let a = addr(14); let mut slot = dummy_slot(8, a, true); let h = hash_byte(0x14); @@ -1522,6 +1527,75 @@ mod tests { Duration::from_secs(30), 50_000, ); + assert!( + !cooldown.contains_key(&a), + "10s after the opening write is inside the new window" + ); + assert!(inflight.contains_key(&h)); + + let later = hash_byte(0x15); + put_on_wire(&mut slot, later, 55_000); + inflight.insert(later, super::super::state::InflightReq::new(8)); + disconnect_stalled_block_peers_at( + std::slice::from_mut(&mut slot), + &mut inflight, + &mut super::super::body::BodyPresence::new(), + &mut cooldown, + &mut strikes, + Instant::now(), + Duration::from_secs(30), + 70_001, + ); + assert!( + cooldown.contains_key(&a), + "the second write must not move the window off t=40s" + ); + assert!(inflight.is_empty()); + } + + /// Draining the last outstanding hash disarms the clock. The next write + /// starts 30s even when the last block is ancient. + #[test] + fn disconnect_stalled_rearms_after_the_window_drains() { + let a = addr(28); + let mut slot = dummy_slot(21, a, true); + let h = hash_byte(0x28); + put_on_wire(&mut slot, h, 0); + slot.rate.note_rx(0); + assert!(slot.track_remove(&h)); + assert!(slot.in_flight.is_empty()); + + let next = hash_byte(0x29); + put_on_wire(&mut slot, next, 100_000); + slot.rate.note_rx(0); + let mut inflight = HashMap::new(); + inflight.insert(next, super::super::state::InflightReq::new(21)); + let mut cooldown = HashMap::new(); + let mut strikes = HashMap::new(); + disconnect_stalled_block_peers_at( + std::slice::from_mut(&mut slot), + &mut inflight, + &mut super::super::body::BodyPresence::new(), + &mut cooldown, + &mut strikes, + Instant::now(), + Duration::from_secs(30), + 110_000, + ); + assert!( + !cooldown.contains_key(&a), + "a drained window does not keep the old deadline" + ); + disconnect_stalled_block_peers_at( + std::slice::from_mut(&mut slot), + &mut inflight, + &mut super::super::body::BodyPresence::new(), + &mut cooldown, + &mut strikes, + Instant::now(), + Duration::from_secs(30), + 130_001, + ); assert!(cooldown.contains_key(&a)); assert!(inflight.is_empty()); } diff --git a/crates/rbitcoin-net/src/ibd/peer_io.rs b/crates/rbitcoin-net/src/ibd/peer_io.rs index 073750e59..b77e49590 100644 --- a/crates/rbitcoin-net/src/ibd/peer_io.rs +++ b/crates/rbitcoin-net/src/ibd/peer_io.rs @@ -21,7 +21,7 @@ use bitcoin::p2p::{Magic, ServiceFlags}; use bitcoin::BlockHash; use std::collections::HashSet; use std::net::SocketAddr; -use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicI64, AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Instant; use tokio::sync::{mpsc, OwnedSemaphorePermit, Semaphore}; @@ -177,22 +177,30 @@ pub(crate) fn solicit_track() -> ( ) } -/// Outstanding getdata that has not left the process, and when the last new -/// hash did leave. +/// Outstanding getdata that has not left the process, and when this stall +/// window opened. /// -/// The stall window opens at [`Self::wrote`], not when the main loop queues -/// the command. A repeat of a hash already in flight is not queued, so it -/// does not move the window. +/// The window opens at the first [`Self::wrote`] after it was disarmed, not +/// when the main loop queues the command. A later write while the window is +/// open does not move it. [`Self::disarm_window`] runs when the peer owes +/// nothing, so the next ask starts a new 30s. pub(crate) struct GetdataWire { unwritten: AtomicUsize, written_ms: AtomicU64, + /// Mono ms the current on-wire window opened. [`WINDOW_DISARMED`] when + /// no getdata of this window is on the socket yet. + window_ms: AtomicI64, } +/// `window_ms` while the peer owes no on-wire block. +const WINDOW_DISARMED: i64 = -1; + impl GetdataWire { pub(crate) fn new() -> Arc { Arc::new(Self { unwritten: AtomicUsize::new(0), written_ms: AtomicU64::new(0), + window_ms: AtomicI64::new(WINDOW_DISARMED), }) } @@ -204,6 +212,9 @@ impl GetdataWire { } /// `n` hashes from one getdata chunk just went out on the socket. + /// + /// The first write of a disarmed window opens the stall clock at + /// `now_ms`. A write into an open window does not. pub(crate) fn wrote(&self, n: usize, now_ms: u64) { if n == 0 { return; @@ -214,6 +225,7 @@ impl GetdataWire { .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |cur| { Some(cur.saturating_sub(n)) }); + self.arm_window_if_closed(now_ms); } pub(crate) fn unwritten(&self) -> usize { @@ -223,6 +235,31 @@ impl GetdataWire { pub(crate) fn written_ms(&self) -> u64 { self.written_ms.load(Ordering::Relaxed) } + + /// Open the stall window at `now_ms` if it is disarmed. + pub(crate) fn arm_window_if_closed(&self, now_ms: u64) { + let opened = i64::try_from(now_ms).unwrap_or(i64::MAX); + let _ = self.window_ms.compare_exchange( + WINDOW_DISARMED, + opened, + Ordering::Relaxed, + Ordering::Relaxed, + ); + } + + /// Drop the stall deadline. The next [`Self::wrote`] starts a new one. + pub(crate) fn disarm_window(&self) { + self.window_ms.store(WINDOW_DISARMED, Ordering::Relaxed); + } + + pub(crate) fn window_ms(&self) -> Option { + let v = self.window_ms.load(Ordering::Relaxed); + if v < 0 { + None + } else { + Some(u64::try_from(v).unwrap_or(0)) + } + } } /// Getdata is already on the socket. An empty window, or a window whose @@ -232,19 +269,58 @@ fn asked_on_wire(slot: &PeerSlot) -> bool { outstanding > 0 && slot.getdata_wire.unwritten() < outstanding } +/// Stall deadline for hashes already on the socket. +/// +/// Nothing on the wire disarms the window, including a full queue that has +/// not been written. An open window stays at its first write. A window that +/// is on the wire but was never armed (tests, or a write the clock missed) +/// opens at [`GetdataWire::written_ms`]. +fn prepare_stall_window(slot: &PeerSlot) -> Option { + if !asked_on_wire(slot) { + slot.getdata_wire.disarm_window(); + return None; + } + if slot.getdata_wire.window_ms().is_none() { + slot.getdata_wire + .arm_window_if_closed(slot.getdata_wire.written_ms()); + } + slot.getdata_wire.window_ms() +} + +/// Copy the rate and lift one clock up to the window open. +/// +/// Idle time before the ask falls behind `open` and drops out. A block or +/// socket byte inside the window stays later than `open` and still counts. +fn rate_in_window(slot: &PeerSlot, open: u64, block_progress: bool) -> PeerRate { + let mut rate = slot.rate; + if block_progress { + if open > rate.progress_ms { + rate.progress_ms = open; + } + } else if open > rate.any_rx_ms { + rate.any_rx_ms = open; + } + rate +} + /// True when this peer owes a block that is already on the wire and no -/// solicited block progress has arrived inside `stall_ms`. +/// solicited block has arrived inside `stall_ms` of this window. /// -/// A getdata write does not move the clock. Socket bytes that are not a -/// requested block do not either. +/// Socket bytes that are not a requested block do not move this clock. pub(crate) fn getdata_on_wire_stalled(slot: &PeerSlot, now_ms: u64, stall_ms: u64) -> bool { - asked_on_wire(slot) && slot.rate.stalled(now_ms, stall_ms, true) + let Some(open) = prepare_stall_window(slot) else { + return false; + }; + rate_in_window(slot, open, true).stalled(now_ms, stall_ms, true) } /// True when getdata is on the wire and this peer has sent no bytes at all -/// inside `stall_ms`. A peer with nothing assigned is not quiet. +/// inside `stall_ms` of this window. A peer with nothing assigned is not quiet. pub(crate) fn getdata_on_wire_quiet(slot: &PeerSlot, now_ms: u64, stall_ms: u64) -> bool { - asked_on_wire(slot) && slot.rate.quiet(now_ms, stall_ms, true) + let Some(open) = prepare_stall_window(slot) else { + return false; + }; + rate_in_window(slot, open, false).quiet(now_ms, stall_ms, true) } /// In-flight light decodes (inv, addr, tx, cmpct) per IBD peer. Not a knob. @@ -263,12 +339,14 @@ impl PeerSlot { pub(crate) fn track_remove(&mut self, hash: &BlockHash) -> bool { let removed = self.in_flight.remove(hash); self.requested_set().remove(hash); + self.disarm_stall_if_idle(); removed } pub(crate) fn track_clear(&mut self) { self.in_flight.clear(); self.requested_set().clear(); + self.disarm_stall_if_idle(); } pub(crate) fn track_retain(&mut self, mut keep: impl FnMut(&BlockHash) -> bool) { @@ -279,13 +357,21 @@ impl PeerSlot { let mut set = self.requested_set(); set.clear(); set.extend(self.in_flight.iter().copied()); + self.disarm_stall_if_idle(); } pub(crate) fn track_drain(&mut self) -> Vec { let drained: Vec = self.in_flight.drain().collect(); self.requested_set().clear(); + self.disarm_stall_if_idle(); drained } + + fn disarm_stall_if_idle(&self) { + if self.in_flight.is_empty() { + self.getdata_wire.disarm_window(); + } + } } /// `true` when `hash` was requested: record `len` and the read time.