ibd: start a fresh stall window on the next getdata

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.

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);
     }

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());
     }

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<Self> {
         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<u64> {
+        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<u64> {
+    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<BlockHash> {
         let drained: Vec<BlockHash> = 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.