IBD: count only requested blocks in the download window

docs/manual/src/ibd.md

@@ -17,7 +17,11 @@ pipeline that keeps the network, the disk, and the CPU busy at the same time.
 - **Parallel block download.** satd fetches blocks from many peers at once, the
   way BitTorrent downloads pieces from a swarm. Download throughput scales with
   aggregate peer bandwidth instead of one peer's round-trip time. `-maxahead`
-  bounds how far ahead of the connect tip downloaded blocks may be staged.
+  bounds how far ahead of the connect tip downloaded blocks may be staged. The
+  256 blocks just above the connect tip are always fetched, even with the
+  window full, so the block the connector needs next is never held back by it.
+  As in Bitcoin Core, blocks are asked only of peers that advertise
+  `NODE_NETWORK`, or `NODE_NETWORK_LIMITED` for blocks within 286 of their tip.
 - **Background prefetch workers.** `-prefetchworkers` threads pre-read blocks
   from the flat files without holding the chainstate lock. They deserialize
   each block, compute txids, run the context-free transaction validation, and

node/src/net/bg_catchup.rs

@@ -15,11 +15,16 @@
 //! sequential requests above the background connect cursor is sufficient.
 //! Robustness against a peer that accepts a `getdata` but never delivers
 //! comes from a stale-request timeout that returns the height to the
-//! request pool on the next pass.
+//! request pool on the next pass, and from releasing a height the peer
+//! answers with `notfound`. Each peer has at most a fixed number of heights
+//! outstanding, as Core caps `MAX_BLOCKS_IN_TRANSIT_PER_PEER`, and a released
+//! height goes to a different peer when there is one.
 
 use std::collections::HashMap;
 use std::time::{Duration, Instant};
 
+use bitcoin::BlockHash;
+
 use crate::net::peer::PeerId;
 
 /// Tracks which historical block heights have an outstanding `getdata`
@@ -37,6 +42,13 @@ pub struct BgDownloader {
     /// A request older than this is assumed lost and becomes eligible for
     /// re-request (to a possibly different peer) on the next pass.
     stale_after: Duration,
+    /// The peer a released height was last asked of (timed out or answered
+    /// `notfound`). The next pass gives the height to someone else when it
+    /// can, rather than back to the peer that did not deliver it.
+    released_from: HashMap<u32, PeerId>,
+    /// Where the next pass starts its round-robin over the peer list, so the
+    /// lowest needed height does not go to the same peer every pass.
+    next_start: usize,
 }
 
 impl BgDownloader {
@@ -45,32 +57,61 @@ impl BgDownloader {
             in_flight: HashMap::new(),
             window: window.max(1),
             stale_after,
+            released_from: HashMap::new(),
+            next_start: 0,
         }
     }
 
     /// A height's data was stored — it no longer needs requesting.
     pub fn note_stored(&mut self, height: u32) {
         self.in_flight.remove(&height);
+        self.released_from.remove(&height);
+    }
+
+    /// `peer` answered `notfound` for `height`. Release it if it was asked of
+    /// that peer, so another peer can be asked on the next pass. Returns
+    /// whether it was released.
+    pub fn note_not_found(&mut self, height: u32, peer: PeerId) -> bool {
+        if self.in_flight.get(&height).is_some_and(|(p, _)| *p == peer) {
+            self.in_flight.remove(&height);
+            self.released_from.insert(height, peer);
+            true
+        } else {
+            false
+        }
     }
 
     /// Drop every in-flight request assigned to a departed peer so the
     /// heights are re-requested promptly rather than waiting out the
     /// stale timeout.
     pub fn note_peer_gone(&mut self, peer: PeerId) {
         self.in_flight.retain(|_, (p, _)| *p != peer);
+        self.released_from.retain(|_, p| *p != peer);
     }
 
     /// Forget all outstanding requests (handoff completed / background
     /// detached — the tracker is about to go idle).
     pub fn reset(&mut self) {
         self.in_flight.clear();
+        self.released_from.clear();
     }
 
     /// Drop requests older than `stale_after` so they re-enter the pool.
     pub fn release_stale(&mut self, now: Instant) {
         let stale = self.stale_after;
-        self.in_flight
-            .retain(|_, (_, at)| now.duration_since(*at) < stale);
+        let released_from = &mut self.released_from;
+        self.in_flight.retain(|&h, (p, at)| {
+            let keep = now.duration_since(*at) < stale;
+            if !keep {
+                released_from.insert(h, *p);
+            }
+            keep
+        });
+    }
+
+    /// How many heights are outstanding to `peer`.
+    pub fn outstanding(&self, peer: PeerId) -> usize {
+        self.in_flight.values().filter(|(p, _)| *p == peer).count()
     }
 
     pub fn in_flight_len(&self) -> usize {
@@ -104,8 +145,56 @@ impl BgDownloader {
     pub fn mark_in_flight(&mut self, height: u32, peer: PeerId, now: Instant) {
         self.in_flight.insert(height, (peer, now));
     }
+
+    /// Share `needed` (ascending, none in flight) out over `peers`, at most
+    /// `per_peer` outstanding to each, and record every request. Round-robin
+    /// from a start that moves on each pass; a released height skips the
+    /// peer it was released from unless that peer is the only one. Heights
+    /// that find no peer with room wait for the next pass. Returns the hashes
+    /// to ask of each peer.
+    pub fn assign(
+        &mut self,
+        needed: &[(u32, BlockHash)],
+        peers: &[PeerId],
+        per_peer: usize,
+        now: Instant,
+    ) -> Vec<(PeerId, Vec<BlockHash>)> {
+        if peers.is_empty() {
+            return Vec::new();
+        }
+        let mut room: Vec<usize> = peers
+            .iter()
+            .map(|p| per_peer.saturating_sub(self.outstanding(*p)))
+            .collect();
+        let mut out: Vec<Vec<BlockHash>> = vec![Vec::new(); peers.len()];
+        let mut next = self.next_start % peers.len();
+        self.next_start = self.next_start.wrapping_add(1);
+        for &(height, hash) in needed {
+            let avoid = if peers.len() > 1 { self.released_from.get(&height).copied() } else { None };
+            let Some(i) = (0..peers.len())
+                .map(|k| (next + k) % peers.len())
+                .find(|&i| room[i] > 0 && Some(peers[i]) != avoid)
+            else {
+                continue;
+            };
+            room[i] -= 1;
+            out[i].push(hash);
+            self.in_flight.insert(height, (peers[i], now));
+            next = i + 1;
+        }
+        peers
+            .iter()
+            .copied()
+            .zip(out)
+            .filter(|(_, hashes)| !hashes.is_empty())
+            .collect()
+    }
 }
 
+#[cfg(test)]
+#[path = "bg_catchup_ibdwindow_tests.rs"]
+mod ibdwindow_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;

node/src/net/bg_catchup_ibdwindow_tests.rs

@@ -0,0 +1,92 @@
+//! How the background download shares heights out: a per-peer cap on what is
+//! outstanding, a released height going to a different peer, and `notfound`.
+
+use super::*;
+use bitcoin::hashes::Hash as _;
+
+fn h(n: u32) -> (u32, BlockHash) {
+    let mut bytes = [0u8; 32];
+    bytes[..4].copy_from_slice(&n.to_le_bytes());
+    (n, BlockHash::from_byte_array(bytes))
+}
+
+fn heights(range: std::ops::RangeInclusive<u32>) -> Vec<(u32, BlockHash)> {
+    range.map(h).collect()
+}
+
+fn given_to(out: &[(PeerId, Vec<BlockHash>)], peer: PeerId) -> Vec<u32> {
+    out.iter()
+        .filter(|(p, _)| *p == peer)
+        .flat_map(|(_, hashes)| hashes.iter().map(|x| u32::from_le_bytes(x.to_byte_array()[..4].try_into().unwrap())))
+        .collect()
+}
+
+/// The cap is on what a peer has outstanding, not on what one pass adds:
+/// a peer that has not answered gets nothing more on the next pass.
+#[test]
+fn a_peer_at_the_cap_gets_nothing_more_on_the_next_pass() {
+    let mut d = BgDownloader::new(1024, Duration::from_secs(30));
+    let now = Instant::now();
+    let first = d.assign(&heights(1..=40), &[1], 16, now);
+    assert_eq!(given_to(&first, 1).len(), 16);
+    let second = d.assign(&heights(17..=40), &[1], 16, now);
+    assert!(given_to(&second, 1).is_empty(), "peer 1 given more with 16 outstanding: {second:?}");
+
+    // As it delivers, it has room again.
+    d.note_stored(1);
+    d.note_stored(2);
+    let third = d.assign(&heights(17..=40), &[1], 16, now);
+    assert_eq!(given_to(&third, 1), vec![17, 18]);
+}
+
+/// A height that timed out goes to another peer, not back to the one that
+/// held it; with only that peer, it goes back to it.
+#[test]
+fn a_released_height_goes_to_a_different_peer() {
+    let t0 = Instant::now();
+    // From either round-robin start.
+    for start in 0..2 {
+        let mut d = BgDownloader::new(1024, Duration::from_secs(30));
+        for _ in 0..start {
+            d.assign(&[], &[1, 2], 16, t0);
+        }
+        d.mark_in_flight(5, 1, t0);
+        d.release_stale(t0 + Duration::from_secs(31));
+        assert!(!d.is_in_flight(5));
+        let out = d.assign(&[h(5)], &[1, 2], 16, t0);
+        assert_eq!(given_to(&out, 2), vec![5], "start {start}: height 5 went back to peer 1: {out:?}");
+    }
+
+    let mut alone = BgDownloader::new(1024, Duration::from_secs(30));
+    alone.mark_in_flight(5, 1, t0);
+    alone.release_stale(t0 + Duration::from_secs(31));
+    assert_eq!(given_to(&alone.assign(&[h(5)], &[1], 16, t0), 1), vec![5]);
+}
+
+/// The round-robin start moves on each pass, so the lowest needed height is
+/// not asked of the same peer every time.
+#[test]
+fn the_round_robin_start_moves_each_pass() {
+    let mut d = BgDownloader::new(1024, Duration::from_secs(30));
+    let now = Instant::now();
+    let first = d.assign(&[h(1)], &[1, 2], 16, now);
+    d.note_stored(1);
+    let second = d.assign(&[h(2)], &[1, 2], 16, now);
+    let firsts: Vec<PeerId> = [&first, &second].iter().map(|o| o[0].0).collect();
+    assert_ne!(firsts[0], firsts[1], "both passes started at the same peer");
+}
+
+/// `notfound` from the peer asked releases the height at once, for a
+/// different peer; `notfound` from anyone else releases nothing.
+#[test]
+fn notfound_releases_only_the_askers_height() {
+    let mut d = BgDownloader::new(1024, Duration::from_secs(30));
+    let now = Instant::now();
+    d.mark_in_flight(5, 1, now);
+    assert!(!d.note_not_found(5, 2), "peer 2 was not asked");
+    assert!(d.is_in_flight(5));
+    assert!(d.note_not_found(5, 1));
+    assert!(!d.is_in_flight(5));
+    let out = d.assign(&[h(5)], &[1, 2], 16, now);
+    assert_eq!(given_to(&out, 2), vec![5]);
+}

node/src/net/ibd.rs

@@ -248,16 +248,22 @@ impl IbdScheduler {
         // This prevents stalls where random blocks are downloaded far ahead but the
         // connect thread is blocked waiting for the next sequential block.
         // 256 blocks ensures the near-cursor region is well-covered across multiple peers.
+        //
+        // The zone is exempt from `max_ahead`. The window counts heights that
+        // are downloaded but not yet connected, and those only drain as the
+        // cursor advances; if the window is full while the height the
+        // connector needs next is not in flight (released from a peer that
+        // did not deliver it, or after a restart with a smaller window), a
+        // gated zone would never hand that height out again and the download
+        // would stop for good. Core's window is measured from the last common
+        // block, so the blocks just above it are always inside
+        // (`nWindowEnd`, `net_processing.cpp:1436-1441`). The overshoot is
+        // bounded by the zone's 256 heights.
         let priority_end = (self.connect_cursor + 256).min(self.target_height).min(max_height);
         for h in (self.connect_cursor + 1)..=priority_end {
             if hashes.len() >= budget {
                 break;
             }
-            let total_ahead =
-                self.downloaded.len() as u32 + self.in_flight.len() as u32 + hashes.len() as u32;
-            if total_ahead >= self.max_ahead {
-                break;
-            }
             if self.downloaded.contains(&h) || self.in_flight.contains_key(&h) {
                 continue;
             }
@@ -292,7 +298,9 @@ impl IbdScheduler {
                 None => break,
             };
 
-            if self.downloaded.contains(&height) {
+            // Already connected: a released height can sit in the pool while
+            // its block arrives late and is connected from disk.
+            if height <= self.connect_cursor || self.downloaded.contains(&height) {
                 continue;
             }
             if self.in_flight.contains_key(&height) {
@@ -340,35 +348,46 @@ impl IbdScheduler {
     /// good: the very wedge the best-header rows exist to prevent, reached
     /// through the delivery door instead of the index.
     ///
+    /// Only a height that is in flight counts, and only one above the connect
+    /// cursor is recorded as downloaded. `downloaded` drains only as the
+    /// cursor passes it, so an already-connected height recorded there would
+    /// stay in the window for the rest of the download. The peer's activity
+    /// clock and failure count move only for a block that was asked of that
+    /// peer, as Core clears `m_stalling_since` only when a requested block
+    /// arrives (`RemoveBlockRequest`, `net_processing.cpp:1199-1230`); a block nobody
+    /// asked it for says nothing about the heights it still owes.
+    ///
     /// Returns `true` if the peer has capacity for more work.
     pub fn block_received(&mut self, peer_id: PeerId, height: u32, hash: BlockHash) -> bool {
-        let wanted = self.height_to_hash.get(&height) == Some(&hash);
-        if wanted {
+        let holder = if self.height_to_hash.get(&height) == Some(&hash) {
+            self.in_flight.get(&height).copied()
+        } else {
+            None
+        };
+        if let Some(holder) = holder {
             // Clear the height from whichever peer actually holds it. After a
             // re-key and reassignment that need not be the peer delivering it,
             // and leaving the holder's slot occupied would leak its capacity.
-            if let Some(holder) = self.in_flight.remove(&height)
-                && holder != peer_id
-                && let Some(slots) = self.peer_slots.get_mut(&holder)
-            {
+            self.in_flight.remove(&height);
+            self.in_flight_at.remove(&height);
+            if let Some(slots) = self.peer_slots.get_mut(&holder) {
                 slots.assigned.retain(|&h| h != height);
             }
-            self.in_flight_at.remove(&height);
-            self.downloaded.insert(height);
+            if height > self.connect_cursor {
+                self.downloaded.insert(height);
+            }
             // Drop any cooldown entries for this height.
             self.height_peer_cooldown.remove(&height);
         }
 
         if let Some(slots) = self.peer_slots.get_mut(&peer_id) {
-            // The peer proved it is alive and delivering, whether or not the
-            // block is still one we want.
-            if wanted {
-                slots.assigned.retain(|&h| h != height);
+            if holder == Some(peer_id) {
+                // The peer delivered what it was asked for: it is alive.
+                slots.blocks_received += 1;
+                slots.last_activity = Instant::now();
+                slots.consecutive_failures = 0;
+                slots.last_failure_at = None;
             }
-            slots.blocks_received += 1;
-            slots.last_activity = Instant::now();
-            slots.consecutive_failures = 0;
-            slots.last_failure_at = None;
             return (slots.assigned.len() as u32) < self.per_peer_limit / 2;
         }
         false
@@ -821,6 +840,10 @@ impl IbdScheduler {
     }
 }
 
+#[cfg(test)]
+#[path = "ibd_ibdwindow_tests.rs"]
+mod ibdwindow_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;