diff --git a/docs/manual/src/ibd.md b/docs/manual/src/ibd.md index a98cf5eeb..7306e157b 100644 --- a/docs/manual/src/ibd.md +++ b/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 diff --git a/node/src/net/bg_catchup.rs b/node/src/net/bg_catchup.rs index d03339893..9a31efb44 100644 --- a/node/src/net/bg_catchup.rs +++ b/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, + /// 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,12 +57,28 @@ 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 @@ -58,19 +86,32 @@ impl BgDownloader { /// 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)> { + if peers.is_empty() { + return Vec::new(); + } + let mut room: Vec = peers + .iter() + .map(|p| per_peer.saturating_sub(self.outstanding(*p))) + .collect(); + let mut out: Vec> = 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::*; diff --git a/node/src/net/bg_catchup_ibdwindow_tests.rs b/node/src/net/bg_catchup_ibdwindow_tests.rs new file mode 100644 index 000000000..fa32c7104 --- /dev/null +++ b/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) -> Vec<(u32, BlockHash)> { + range.map(h).collect() +} + +fn given_to(out: &[(PeerId, Vec)], peer: PeerId) -> Vec { + 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 = [&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]); +} diff --git a/node/src/net/ibd.rs b/node/src/net/ibd.rs index 5c13134a5..2a46c1526 100644 --- a/node/src/net/ibd.rs +++ b/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::*; diff --git a/node/src/net/ibd_ibdwindow_tests.rs b/node/src/net/ibd_ibdwindow_tests.rs new file mode 100644 index 000000000..650110313 --- /dev/null +++ b/node/src/net/ibd_ibdwindow_tests.rs @@ -0,0 +1,163 @@ +//! What counts against the download window and the stall clock: only a block +//! that was asked for, above the connect cursor. + +use super::*; +use crate::chain::state::AssumeValid; +use crate::storage::db::InMemoryStore; +use crate::storage::flatfile::FlatFileManager; +use crate::validation::script::NoopVerifier; +use bitcoin::Network; + +fn chain(num_blocks: u32) -> (ChainState, tempfile::TempDir) { + let dir = tempfile::TempDir::new().unwrap(); + let flat_files = FlatFileManager::new(&dir.path().join("blocks")).unwrap(); + let cs = ChainState::new( + Box::new(InMemoryStore::new()), + flat_files, + Network::Regtest, + Box::new(NoopVerifier), + AssumeValid::Disabled, + 450, + 4, + Default::default(), + Default::default(), + Default::default(), + ) + .unwrap(); + let mut parent = bitcoin::constants::genesis_block(Network::Regtest).block_hash(); + for i in 1..=num_blocks { + let block = crate::chain::state::tests::build_test_block(parent, i, 1_300_000_000 + i); + parent = cs.accept_block(&block).unwrap().hash(); + } + (cs, dir) +} + +fn hash_at(cs: &ChainState, height: u32) -> BlockHash { + cs.get_block_hash_by_height(height).unwrap() +} + +/// A block re-sent for a height the connector has already passed is not +/// recorded as downloaded. `downloaded` only drains as the cursor passes a +/// height, so such an entry would sit in the window for the rest of the +/// download. +#[test] +fn a_re_sent_connected_block_does_not_count_as_downloaded() { + let (cs, _dir) = chain(20); + let mut sched = IbdScheduler::new(20, 0, &cs, 50_000); + sched.connect_cursor_advanced(10); + for h in 1..=3 { + sched.block_received(2, h, hash_at(&cs, h)); + } + assert_eq!(sched.progress().0, 0, "connected heights entered the window"); + + // A block that was asked for still counts. + let asked = sched.assign_blocks(1); + assert!(asked.contains(&hash_at(&cs, 11))); + sched.block_received(1, 11, hash_at(&cs, 11)); + assert!(sched.is_downloaded(11), "a requested block counts as downloaded"); + assert_eq!(sched.progress().0, 1); +} + +/// An unrequested block above the cursor is not recorded either: only a +/// height that is in flight is. +#[test] +fn an_unrequested_block_above_the_cursor_does_not_count() { + let (cs, _dir) = chain(20); + let mut sched = IbdScheduler::new(20, 0, &cs, 50_000); + sched.register_peer(2); + sched.block_received(2, 7, hash_at(&cs, 7)); + assert!(!sched.is_downloaded(7)); + assert_eq!(sched.progress().0, 0); +} + +/// The window is full of blocks that are downloaded but not connected, and +/// the height the connector needs next is not in flight. It is still handed +/// out, or nothing would ever fetch it. +#[test] +fn a_full_window_still_hands_out_the_next_height() { + let (cs, _dir) = chain(20); + let mut sched = IbdScheduler::new(20, 0, &cs, 3); + for h in 2..=4 { + sched.mark_downloaded(h); + } + let hashes = sched.assign_blocks(1); + assert!( + hashes.contains(&hash_at(&cs, 1)), + "height 1 not handed out with the window full: {hashes:?}" + ); +} + +/// A peer that withholds the next height and delivers the rest: once that +/// height is released, another peer is given it, even with the window full. +#[test] +fn a_released_next_height_goes_to_another_peer_with_the_window_full() { + let (cs, _dir) = chain(20); + let mut sched = IbdScheduler::new(20, 0, &cs, 3); + sched.register_peer(2); + for h in 2..=4 { + sched.mark_downloaded(h); + } + sched.register_peer(1); + sched.in_flight.insert(1, 1); + sched.in_flight_at.insert(1, Instant::now()); + sched.peer_slots.get_mut(&1).unwrap().assigned.push(1); + assert!(sched.release_height(1, 1)); + assert!(!sched.assign_blocks(1).contains(&hash_at(&cs, 1)), "peer 1 is on cooldown for height 1"); + let hashes = sched.assign_blocks(2); + assert!( + hashes.contains(&hash_at(&cs, 1)), + "height 1 not given to peer 2: {hashes:?}" + ); +} + +/// A released height that waited in the pool while the cursor passed it is +/// not handed out again. +#[test] +fn a_pooled_height_below_the_cursor_is_not_handed_out() { + let (cs, _dir) = chain(20); + let mut sched = IbdScheduler::new(20, 0, &cs, 50_000); + sched.connect_cursor_advanced(10); + let hashes = sched.assign_blocks(1); + let heights: Vec = (1..=20).filter(|h| hashes.contains(&hash_at(&cs, *h))).collect(); + assert!(!heights.is_empty(), "heights above the cursor are handed out"); + assert!(heights.iter().all(|h| *h > 10), "connected heights handed out: {heights:?}"); +} + +/// Only a block asked of the peer resets its stall clock and failure count. +/// A block it was not asked for (connected already, or held by another peer) +/// says nothing about the heights it still owes. +#[test] +fn only_a_requested_block_resets_the_stall_clock() { + let (cs, _dir) = chain(40); + let mut sched = IbdScheduler::new(40, 0, &cs, 50_000); + sched.assign_blocks(1); + sched.assign_blocks(2); + let theirs = sched.peer_inflight_heights(2)[0]; + let old = Instant::now() - Duration::from_secs(100); + { + let slots = sched.peer_slots.get_mut(&1).unwrap(); + slots.last_activity = old; + slots.consecutive_failures = 2; + } + + // Peer 1 sends a block peer 2 was asked for. + sched.block_received(1, theirs, hash_at(&cs, theirs)); + let slots = &sched.peer_slots[&1]; + assert_eq!(slots.consecutive_failures, 2, "an unrequested block reset the failure count"); + assert_eq!(slots.last_activity, old, "an unrequested block reset the stall clock"); + assert_eq!(sched.detect_stalls(Duration::from_secs(60)), vec![1], "peer 1 still stalls"); + + // Its own block does reset it. + let mut sched = IbdScheduler::new(40, 0, &cs, 50_000); + sched.assign_blocks(1); + { + let slots = sched.peer_slots.get_mut(&1).unwrap(); + slots.last_activity = old; + slots.consecutive_failures = 2; + } + let h = sched.peer_inflight_heights(1)[0]; + sched.block_received(1, h, hash_at(&cs, h)); + let slots = &sched.peer_slots[&1]; + assert_eq!(slots.consecutive_failures, 0); + assert!(slots.last_activity > old); +} diff --git a/node/src/net/manager.rs b/node/src/net/manager.rs index 42ba23abc..1c0c652fb 100644 --- a/node/src/net/manager.rs +++ b/node/src/net/manager.rs @@ -190,6 +190,14 @@ const UNREADABLE_REFETCH_INTERVAL: Duration = Duration::from_secs(10); /// refreshes the stamp; an entry this old belongs to a request nobody is /// making any more. const BLOCK_IN_FLIGHT_TTL: Duration = Duration::from_secs(120); +/// How long the tip-following sweep leaves a block it asked one peer for +/// before asking again. Core asks one peer per block (`mapBlocksInFlight`) +/// and moves on only when that request fails; the sweep asked every peer +/// for every missing block on each pass, every few seconds. +const SWEEP_REREQUEST_AFTER: Duration = Duration::from_secs(30); +/// Most blocks the tip-following sweep asks of one peer in a pass, Core's +/// `MAX_BLOCKS_IN_TRANSIT_PER_PEER`. +const SWEEP_BLOCKS_PER_PEER: usize = 16; /// How long a compact block being reconstructed suppresses an ordinary fetch /// of the same block. Core marks a block it is reconstructing as in flight /// and does not download it twice; satd's tip-following sweep would otherwise @@ -333,10 +341,15 @@ const BG_CATCHUP_WINDOW: u32 = 1024; /// A background-range `getdata` older than this is assumed lost and /// becomes eligible for re-request on the next download pass. const BG_CATCHUP_STALE_SECS: u64 = 30; -/// Cap on the number of fresh background-range `getdata` heights issued -/// per peer per download pass, so the catch-up downloader does not flood -/// a single peer. -const BG_CATCHUP_PER_PEER_PER_PASS: usize = 16; +/// Most background-range heights outstanding to one peer at a time, as +/// Core caps `MAX_BLOCKS_IN_TRANSIT_PER_PEER` (16, `net_processing.cpp`). +/// A peer that has not answered gets no more until it does or its requests +/// time out. +const BG_CATCHUP_PER_PEER: usize = 16; +/// Core serves a NODE_NETWORK_LIMITED peer's blocks only this far below its +/// tip: `NODE_NETWORK_LIMITED_MIN_BLOCKS` (288) less Core's two-block margin +/// (`net_processing.cpp:154`, `:1533`). +const LIMITED_PEER_BLOCK_DEPTH: u32 = 288 - 2; /// Flush the background coin cache to disk every N connected blocks so a /// crash resumes from a recent private tip instead of redoing the whole /// genesis→snapshot validation. @@ -639,6 +652,10 @@ mod peerpunish_tests; #[path = "manager_addrdials_tests.rs"] mod addrdials_tests; +#[cfg(test)] +#[path = "manager_ibdwindow_tests.rs"] +mod ibdwindow_tests; + /// How a transaction is announced to a peer: `MSG_WTX` carrying the wtxid /// to a peer that negotiated BIP 339 wtxid relay, the txid to the rest. fn tx_announcement(wtxid_relay: bool, txid: bitcoin::Txid, wtxid: bitcoin::Wtxid) -> Inventory { @@ -4558,14 +4575,14 @@ impl PeerManager { let peers = self.peers.read(); peers.iter() .filter(|(_, h)| { - h.info.state == PeerState::Connected && h.info.serves_blocks() + h.info.state == PeerState::Connected + && h.info.serves_blocks() + && h.info.advertises_blocks() }) .map(|(id, _)| *id) .collect() }; - for pid in &peer_ids { - self.request_missing_blocks(*pid); - } + self.sweep_missing_blocks(&peer_ids); } } @@ -5089,6 +5106,26 @@ impl PeerManager { if block_hashes.is_empty() { tracing::debug!(id, count = inventory.len(), "Peer sent notfound (non-block)"); } else { + // A tip-following request it cannot answer no longer + // holds the block back from the sweep's next pass. + if let Some(asked) = self.in_flight_blocks.write().get_mut(&id) { + for h in &block_hashes { + asked.remove(h); + } + } + // The background download's heights too, so the next + // pass asks a different peer instead of waiting out + // the stale timeout. + if self.bg_downloader.read().in_flight_len() > 0 { + let mut dl = self.bg_downloader.write(); + for h in &block_hashes { + if let Some(entry) = self.chain_state.get_block_index(h) + && dl.note_not_found(entry.height, id) + { + tracing::debug!(id, height = entry.height, "background download: notfound, released"); + } + } + } let mut released_heights: Vec = Vec::new(); { let mut ibd = self.ibd.write(); @@ -5614,9 +5651,19 @@ impl PeerManager { if !handle.info.serves_blocks() { return; } + // Nor does Core download from a peer that advertises neither + // NODE_NETWORK nor NODE_NETWORK_LIMITED (`CanServeBlocks`), such + // as a light client: it cannot serve what it is asked for. + if !handle.info.advertises_blocks() { + return; + } // What the peer claimed at connect, or has announced since. - (handle.info.best_height as i64).max(handle.info.best_known_height.map_or(-1, i64::from)) + ( + (handle.info.best_height as i64).max(handle.info.best_known_height.map_or(-1, i64::from)), + handle.info.is_limited(), + ) }; + let (peer_height, limited) = peer_height; // i32::saturating_sub avoids underflow for early-IBD targets. if peer_height < (target_height as i64).saturating_sub(1000) { // Peer is not synced enough to be a useful IBD source. Skip. @@ -5629,10 +5676,19 @@ impl PeerManager { let Ok(max_height) = u32::try_from(peer_height) else { return }; let mut ibd = self.ibd.write(); if let Some(scheduler) = ibd.as_mut() { - scheduler.register_peer(peer_id); if max_height <= scheduler.connect_cursor() { return; } + // A pruned (NODE_NETWORK_LIMITED) peer keeps only its recent + // blocks. Core asks it for a block only within + // [`LIMITED_PEER_BLOCK_DEPTH`] of its tip (`net_processing.cpp:1533`) + // and not at all during IBD (`:6164`). The scheduler can hand out + // any height above the cursor, so use such a peer only when every + // one of those is that recent. + if limited && max_height - scheduler.connect_cursor() >= LIMITED_PEER_BLOCK_DEPTH { + return; + } + scheduler.register_peer(peer_id); let hashes = scheduler.assign_blocks_up_to(peer_id, max_height); if !hashes.is_empty() { drop(ibd); @@ -6093,10 +6149,13 @@ impl PeerManager { // Don't penalize the peer; the block may become valid later. tracing::debug!(%hash, "IBD block store: parent unknown, skipping"); } - Err(e) => { + Err(e) if Self::store_error_is_the_blocks(&e) => { tracing::debug!(%hash, "IBD block store failed: {}", e); self.add_ban_score(id, 10, &format!("block rejected: {}", e)); } + Err(e) => { + tracing::warn!(%hash, id, "IBD block could not be stored: {}", e); + } } } @@ -6121,16 +6180,32 @@ impl PeerManager { | Err(crate::chain::state::ChainError::BadPrevBlock) => { tracing::debug!(%hash, "bg catch-up block store: parent unknown, skipping"); } - Err(e) => { + Err(e) if Self::store_error_is_the_blocks(&e) => { tracing::debug!(%hash, "bg catch-up block store failed: {}", e); // Smaller penalty than forward IBD: a malformed historical // block is suspicious but the catch-up download is best- // effort and will re-request from another peer. self.add_ban_score(id, 5, &format!("bg block rejected: {}", e)); } + Err(e) => { + tracing::warn!(%hash, id, "bg catch-up block could not be stored: {}", e); + } } } + /// Whether a `store_block` failure was the block's own fault: it failed a + /// check, or contradicts a checkpoint. Everything else (a block-file or + /// database write, a snapshot load in progress) is this node's, and Core + /// never charges the sender for it: a failed write ends in `FatalError` + /// (`validation.cpp`, `AcceptBlock`), not in `Misbehaving`. + fn store_error_is_the_blocks(e: &crate::chain::state::ChainError) -> bool { + matches!( + e, + crate::chain::state::ChainError::Validation(_) + | crate::chain::state::ChainError::CheckpointMismatch(_) + ) + } + /// A block's data was stored — clear it from the background download /// tracker (no-op for forward-range heights) and wake the background /// connect loop so it can attempt the next in-order connect. @@ -6191,46 +6266,29 @@ impl PeerManager { return; } + // Core downloads historical blocks only from a peer that keeps them + // all: NODE_NETWORK, not a limited one (`net_processing.cpp:6176`). let peer_ids: Vec = { let peers = self.peers.read(); - peers + let mut ids: Vec = peers .iter() - .filter(|(_, h)| h.info.state == PeerState::Connected) + .filter(|(_, h)| { + h.info.state == PeerState::Connected + && h.info.serves_blocks() + && h.info.advertises_blocks() + && !h.info.is_limited() + }) .map(|(id, _)| *id) - .collect() + .collect(); + ids.sort_unstable(); + ids }; if peer_ids.is_empty() { return; } - // Round-robin the needed heights across peers, capping how many we - // assign to any one peer this pass. - let mut assignments: HashMap> = HashMap::new(); - { - let mut dl = self.bg_downloader.write(); - let mut next_peer = 0usize; - let mut per_peer: HashMap = HashMap::new(); - for (h, hash) in needed { - // Find a peer under the per-pass cap. - let mut placed = false; - for _ in 0..peer_ids.len() { - let pid = peer_ids[next_peer % peer_ids.len()]; - next_peer += 1; - let count = per_peer.entry(pid).or_insert(0); - if *count < BG_CATCHUP_PER_PEER_PER_PASS { - *count += 1; - dl.mark_in_flight(h, pid, now); - assignments.entry(pid).or_default().push(hash); - placed = true; - break; - } - } - if !placed { - // Every peer is at the per-pass cap; defer the rest. - break; - } - } - } + let assignments = + self.bg_downloader.write().assign(&needed, &peer_ids, BG_CATCHUP_PER_PEER, now); for (pid, hashes) in assignments { for chunk in hashes.chunks(128) { self.send_to_peer(pid, sync::make_getdata_blocks(chunk)); @@ -9528,6 +9586,86 @@ impl PeerManager { self.request_full_blocks(id, &to_request); } + /// The tip-following sweep: ask `peers` for the blocks the best header + /// chain lacks, each block of one peer. A block asked of anyone (in full) + /// less than [`SWEEP_REREQUEST_AFTER`] ago is left alone; an older one + /// goes to a peer not asked before when there is one. At most + /// [`SWEEP_BLOCKS_PER_PEER`] per peer per pass; the rest wait for the + /// next. Core keeps one request per block (`mapBlocksInFlight`); the + /// sweep used to send every missing block to every peer on each pass. + /// + /// A block goes to a peer known to have it, as Core downloads from a peer + /// only up to its `pindexBestKnownBlock` and from a limited peer only + /// within [`LIMITED_PEER_BLOCK_DEPTH`] of it (`net_processing.cpp:1533`). + /// When no peer qualifies (heights not yet known), any peer is asked. + fn sweep_missing_blocks(&self, peers: &[PeerId]) { + if peers.is_empty() { + return; + } + let missing = self.missing_blocks_to_fetch(); + if missing.is_empty() { + return; + } + // Each peer's best known height and whether it is limited. + let reach: Vec<(i64, bool)> = { + let handles = self.peers.read(); + peers + .iter() + .map(|id| { + handles.get(id).map_or((-1, false), |h| { + let height = (h.info.best_height as i64) + .max(h.info.best_known_height.map_or(-1, i64::from)); + (height, h.info.is_limited()) + }) + }) + .collect() + }; + let can_serve = |i: usize, height: Option| { + let Some(height) = height.map(i64::from) else { return true }; + let (best, limited) = reach[i]; + best >= height && !(limited && best - height >= i64::from(LIMITED_PEER_BLOCK_DEPTH)) + }; + // Who has been asked for each block in full, and the newest ask. + let mut asked: HashMap)> = HashMap::new(); + for (peer, requests) in self.in_flight_blocks.read().iter() { + for (hash, r) in requests { + let age = r.at.elapsed(); + if r.compact || age >= BLOCK_IN_FLIGHT_TTL { + continue; + } + let entry = asked.entry(*hash).or_insert((age, Vec::new())); + entry.0 = entry.0.min(age); + entry.1.push(*peer); + } + } + let mut given: Vec> = vec![Vec::new(); peers.len()]; + let mut next = 0usize; + for hash in missing { + let before: &[PeerId] = match asked.get(&hash) { + Some((age, _)) if *age < SWEEP_REREQUEST_AFTER => continue, + Some((_, who)) => who, + None => &[], + }; + let height = self.chain_state.get_block_index(&hash).map(|e| e.height); + let any_has_it = (0..peers.len()).any(|i| can_serve(i, height)); + let pick = |fresh_only: bool| { + (0..peers.len()).map(|k| (next + k) % peers.len()).find(|&i| { + given[i].len() < SWEEP_BLOCKS_PER_PEER + && !(fresh_only && before.contains(&peers[i])) + && (!any_has_it || can_serve(i, height)) + }) + }; + let Some(i) = pick(true).or_else(|| pick(false)) else { + continue; + }; + given[i].push(hash); + next = i + 1; + } + for (peer, hashes) in peers.iter().zip(given) { + self.request_full_blocks(*peer, &hashes); + } + } + /// Fetch what a `headers` message from `id`, ending at `last`, left /// missing: Bitcoin Core's `HeadersDirectFetchBlocks` (v31.1 /// `net_processing.cpp:2844`), on satd's tip-following fetch. @@ -9680,7 +9818,7 @@ impl PeerManager { /// on its way in compactly from this moment, so the tip-following sweep /// holds off for [`COMPACT_RECONSTRUCT_SUPPRESSION`], as it does for a /// reconstruction; if nothing usable comes by then, the sweep fetches it - /// in full from every peer. + /// in full. fn note_compact_requested(&self, id: PeerId, hash: bitcoin::BlockHash) { self.record_block_requests(id, &[hash], true); self.compact_in_progress.write().insert(hash, Instant::now()); diff --git a/node/src/net/manager_ibdwindow_tests.rs b/node/src/net/manager_ibdwindow_tests.rs new file mode 100644 index 000000000..31d65360b --- /dev/null +++ b/node/src/net/manager_ibdwindow_tests.rs @@ -0,0 +1,265 @@ +//! Block download bookkeeping on a live `PeerManager`: who is charged for a +//! failed store, which peers are asked for blocks, and how the tip-following +//! sweep shares missing blocks out. + +use super::*; +use crate::net::flow::InFlight; +use bitcoin::p2p::message_blockdata::Inventory; + +fn mk_pm() -> (Arc, tempfile::TempDir) { + use crate::chain::state::AssumeValid; + use crate::storage::db::InMemoryStore; + use crate::storage::flatfile::FlatFileManager; + use crate::validation::script::NoopVerifier; + + let dir = tempfile::TempDir::new().unwrap(); + let flat_files = FlatFileManager::new(&dir.path().join("blocks")).unwrap(); + let chain_state = Arc::new( + ChainState::new( + Box::new(InMemoryStore::new()), + flat_files, + Network::Regtest, + Box::new(NoopVerifier), + AssumeValid::Disabled, + 450, + 4, + Default::default(), + Default::default(), + Default::default(), + ) + .unwrap(), + ); + let mempool = Arc::new(Mempool::new(1_000_000, 0)); + let fee_estimator = Arc::new(FeeEstimator::new()); + let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); + let pm = PeerManager::new(chain_state, mempool, fee_estimator, Network::Regtest, shutdown_rx); + (pm, dir) +} + +fn add_peer(pm: &PeerManager, id: PeerId, services: ServiceFlags, best_height: i32) -> mpsc::Receiver { + let addr = SocketAddr::from(([10, 62, 0, id as u8], 8333)); + let mut info = PeerInfo::new(id, addr, Direction::Outbound); + info.state = PeerState::Connected; + info.services = services; + info.best_height = best_height; + let (tx, rx) = mpsc::channel::(64); + pm.peers.write().insert( + id, + PeerHandle { + info, + msg_tx: tx.into(), + disconnect: Arc::new(tokio::sync::Notify::new()), + flow: Arc::new(crate::net::flow::PeerFlow::new()), + last_getheaders_sent: None, + last_mempool_served: None, + fee_filter_sent: None, + stats: PeerStats::new(NetTotals::new()), + }, + ); + rx +} + +fn ban_score(pm: &PeerManager, id: PeerId) -> u32 { + pm.peers.read()[&id].info.ban_score +} + +/// `n` blocks on top of genesis, not stored. +fn blocks(n: u32) -> Vec { + let mut parent = bitcoin::constants::genesis_block(Network::Regtest).block_hash(); + (1..=n) + .map(|i| { + let block = crate::chain::state::tests::build_test_block(parent, i, 1_300_000_000 + i); + parent = block.block_hash(); + block + }) + .collect() +} + +fn blocks_asked(msgs: &[NetworkMessage]) -> Vec { + msgs.iter() + .filter_map(|m| match m { + NetworkMessage::GetData(inv) => Some(inv), + _ => None, + }) + .flatten() + .filter_map(|i| match i { + Inventory::Block(h) | Inventory::WitnessBlock(h) => Some(*h), + _ => None, + }) + .collect() +} + +fn drain(rx: &mut mpsc::Receiver) -> Vec { + std::iter::from_fn(|| rx.try_recv().ok()).collect() +} + +/// Make the block files unwritable: the append file read-only and its cached +/// handle dropped. `false` when the process can write it anyway (root). +fn break_block_files(pm: &PeerManager, dir: &tempfile::TempDir) -> bool { + use std::os::unix::fs::PermissionsExt; + let blocks = dir.path().join("blocks"); + for entry in std::fs::read_dir(&blocks).unwrap() { + let path = entry.unwrap().path(); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o444)).unwrap(); + } + std::fs::set_permissions(&blocks, std::fs::Permissions::from_mode(0o555)).unwrap(); + pm.chain_state.resync_block_append_pos().unwrap(); + let probe = std::fs::read_dir(&blocks).unwrap().next().unwrap().unwrap().path(); + std::fs::OpenOptions::new().append(true).open(probe).is_err() +} + +fn restore_block_files(dir: &tempfile::TempDir) { + use std::os::unix::fs::PermissionsExt; + let blocks = dir.path().join("blocks"); + let _ = std::fs::set_permissions(&blocks, std::fs::Permissions::from_mode(0o755)); +} + +/// A block that could not be written to this node's own disk is not the +/// sender's fault. Core ends a failed write in `FatalError`, never +/// `Misbehaving`. A block that fails its checks still is. +#[test] +fn a_local_write_failure_does_not_charge_the_sender() { + type Store = fn(&PeerManager, PeerId, bitcoin::Block); + let paths: [(&str, Store, u32); 2] = [ + ("forward IBD", |pm, id, b| pm.handle_block_ibd(id, b), 10), + ("background catch-up", |pm, id, b| pm.store_bg_block(id, b), 5), + ]; + for (path, store, penalty) in paths { + // The disk refuses the write. + let (pm, dir) = mk_pm(); + let _rx = add_peer(&pm, 1, ServiceFlags::NETWORK, 1); + if !break_block_files(&pm, &dir) { + eprintln!("skipping: block files are writable despite their mode (running as root?)"); + restore_block_files(&dir); + return; + } + let block = blocks(1).remove(0); + let hash = block.block_hash(); + store(&pm, 1, block); + restore_block_files(&dir); + assert!(!pm.chain_state.has_block_data(&hash), "{path}: the write did not fail"); + assert_eq!(ban_score(&pm, 1), 0, "{path}: the sender was charged for our disk"); + + // The block itself is bad. + let (pm, _dir) = mk_pm(); + let _rx = add_peer(&pm, 1, ServiceFlags::NETWORK, 1); + let mut block = blocks(1).remove(0); + block.txdata[0].output[0].value = bitcoin::Amount::from_sat(1); + store(&pm, 1, block); + assert_eq!(ban_score(&pm, 1), penalty, "{path}: a bad block went uncharged"); + } +} + +/// Core downloads only from a peer that advertises NODE_NETWORK or +/// NODE_NETWORK_LIMITED, and from a limited one only blocks within 286 of +/// its tip. +#[test] +fn ibd_work_goes_only_to_peers_that_can_serve_it() { + let chain = blocks(5); + let cases = [ + ("NODE_NETWORK", ServiceFlags::NETWORK, 5, true), + ("no block service", ServiceFlags::WITNESS, 5, false), + ("limited, every height recent", ServiceFlags::NETWORK_LIMITED, 5, true), + ("limited, heights too deep", ServiceFlags::NETWORK_LIMITED, 300, false), + ]; + for (what, services, best_height, expect) in cases { + let (pm, _dir) = mk_pm(); + for b in &chain { + pm.chain_state.accept_header(&b.header).unwrap(); + } + *pm.ibd.write() = Some(IbdScheduler::new(5, 0, &pm.chain_state, 1024)); + let mut rx = add_peer(&pm, 1, services, best_height); + pm.assign_peer_work(1); + let asked = blocks_asked(&drain(&mut rx)); + assert_eq!(!asked.is_empty(), expect, "{what}: asked for {asked:?}"); + } +} + +/// The tip-following sweep asks one peer per block. Across two passes, each +/// missing block is asked for exactly once; once that request is old, it is +/// asked of a different peer. +#[test] +fn the_sweep_asks_one_peer_per_block() { + let (pm, _dir) = mk_pm(); + let chain = blocks(5); + for b in &chain { + pm.chain_state.accept_header(&b.header).unwrap(); + } + let mut rxs: Vec<_> = (1..=3).map(|id| add_peer(&pm, id, ServiceFlags::NETWORK, 5)).collect(); + let peers = [1, 2, 3]; + + pm.sweep_missing_blocks(&peers); + pm.sweep_missing_blocks(&peers); + let first: Vec> = rxs.iter_mut().map(|rx| blocks_asked(&drain(rx))).collect(); + for b in &chain { + let hash = b.block_hash(); + let times = first.iter().flatten().filter(|h| **h == hash).count(); + assert_eq!(times, 1, "block {hash} asked for {times} times: {first:?}"); + } + + // The requests age past the re-request delay. + for asked in pm.in_flight_blocks.write().values_mut() { + for r in asked.values_mut() { + r.at -= SWEEP_REREQUEST_AFTER + Duration::from_secs(1); + } + } + pm.sweep_missing_blocks(&peers); + let second: Vec> = rxs.iter_mut().map(|rx| blocks_asked(&drain(rx))).collect(); + for b in &chain { + let hash = b.block_hash(); + let before: Vec = (0..3).filter(|i| first[*i].contains(&hash)).collect(); + let now: Vec = (0..3).filter(|i| second[*i].contains(&hash)).collect(); + assert_eq!(now.len(), 1, "block {hash} re-asked of {now:?}"); + assert_ne!(now, before, "block {hash} re-asked of the same peer"); + } +} + +/// `notfound` for a background-download height releases it at once. +#[test] +fn notfound_releases_a_background_height() { + let (pm, _dir) = mk_pm(); + let chain = blocks(2); + for b in &chain { + pm.chain_state.accept_header(&b.header).unwrap(); + } + let _rx1 = add_peer(&pm, 1, ServiceFlags::NETWORK, 2); + let _rx2 = add_peer(&pm, 2, ServiceFlags::NETWORK, 2); + pm.bg_downloader.write().mark_in_flight(1, 1, Instant::now()); + pm.bg_downloader.write().mark_in_flight(2, 1, Instant::now()); + + let notfound = |h: &bitcoin::Block| NetworkMessage::NotFound(vec![Inventory::WitnessBlock(h.block_hash())]); + pm.handle_message(2, notfound(&chain[0]), InFlight::adopt(None)); + assert!(pm.bg_downloader.read().is_in_flight(1), "peer 2 was not asked for it"); + pm.handle_message(1, notfound(&chain[0]), InFlight::adopt(None)); + assert!(!pm.bg_downloader.read().is_in_flight(1), "notfound from the peer asked releases it"); + assert!(pm.bg_downloader.read().is_in_flight(2)); +} + +/// The sweep asks for a block only a peer known to have it: not one whose +/// best block is lower, and not a limited peer for a block deeper than it +/// keeps. A `notfound` frees the block for the next pass. +#[test] +fn the_sweep_asks_a_peer_that_has_the_block() { + let (pm, _dir) = mk_pm(); + let chain = blocks(300); + for b in &chain { + pm.chain_state.accept_header(&b.header).unwrap(); + } + let mut behind = add_peer(&pm, 1, ServiceFlags::NETWORK, 0); + let mut limited = add_peer(&pm, 2, ServiceFlags::NETWORK_LIMITED, 300); + let mut full = add_peer(&pm, 3, ServiceFlags::NETWORK, 300); + pm.sweep_missing_blocks(&[1, 2, 3]); + assert!(blocks_asked(&drain(&mut behind)).is_empty(), "a peer without the blocks was asked"); + // The limited peer keeps blocks 15..=300 (within 286 of its tip). + let deep: Vec = chain[..14].iter().map(|b| b.block_hash()).collect(); + let from_limited = blocks_asked(&drain(&mut limited)); + assert!(!from_limited.is_empty(), "the limited peer is asked for the blocks it keeps"); + assert!(from_limited.iter().all(|h| !deep.contains(h)), "a limited peer was asked for deep blocks"); + let asked = blocks_asked(&drain(&mut full)); + assert!(asked.contains(&chain[0].block_hash()), "the peer that has them was not asked: {asked:?}"); + + // It answers notfound: the next pass may ask again at once. + pm.handle_message(3, NetworkMessage::NotFound(vec![Inventory::WitnessBlock(chain[0].block_hash())]), InFlight::adopt(None)); + pm.sweep_missing_blocks(&[1, 2, 3]); + assert!(blocks_asked(&drain(&mut full)).contains(&chain[0].block_hash())); +} diff --git a/node/src/net/peer.rs b/node/src/net/peer.rs index 64f8256ed..ecead6a24 100644 --- a/node/src/net/peer.rs +++ b/node/src/net/peer.rs @@ -370,6 +370,20 @@ impl PeerInfo { self.conn_type != ConnType::AddrFetch } + /// Whether the peer's services say it can serve blocks at all: Core's + /// `CanServeBlocks` (`net_processing.cpp:1152-1155`), NODE_NETWORK or + /// NODE_NETWORK_LIMITED. A light client sets neither. + pub fn advertises_blocks(&self) -> bool { + self.services.has(ServiceFlags::NETWORK) || self.services.has(ServiceFlags::NETWORK_LIMITED) + } + + /// Whether the peer serves only recent blocks: Core's `IsLimitedPeer` + /// (`net_processing.cpp:1159-1163`), NODE_NETWORK_LIMITED without + /// NODE_NETWORK. + pub fn is_limited(&self) -> bool { + !self.services.has(ServiceFlags::NETWORK) && self.services.has(ServiceFlags::NETWORK_LIMITED) + } + /// Whether this peer participates in tx relay — the BIP 37 `fRelay` /// flag from its version message. A peer that set `relay = false` /// (e.g. Bitcoin Core under `-blocksonly`, or a block-relay-only diff --git a/satd/tests/compact_blocks.rs b/satd/tests/compact_blocks.rs index 9c5b4e191..0deba7813 100644 --- a/satd/tests/compact_blocks.rs +++ b/satd/tests/compact_blocks.rs @@ -684,12 +684,12 @@ fn a_second_compact_peer_for_the_same_hash_is_capped_at_three() { let (b, compact) = partial_block(&node, 3, 5); let hash = b.block_hash(); // Only three peers can be high-bandwidth, so the four take the other road - // past the unsolicited-push gate: the node asks each of them for the - // block after the header is announced -- the announcer for a - // `cmpctblock`, the others in full once the tip-following sweep reaches - // them. - peers[0].send(NetworkMessage::Headers(vec![b.header])); + // past the unsolicited-push gate: each announces the block by `inv` + // before the node knows its header, and the node asks each of them for + // it. (The tip-following sweep asks one peer per block, so it would not + // ask all four.) for (i, p) in peers.iter_mut().enumerate() { + p.send(NetworkMessage::Inv(vec![Inventory::Block(hash)])); assert!( p.recv_until(is_block_request_for(hash), test_timeout(20)).is_some(), "the node must request the block from peer {i}"