diff --git a/crates/rbitcoin-net/src/ibd/assign.rs b/crates/rbitcoin-net/src/ibd/assign.rs index a2874bb2..da14d26a 100644 --- a/crates/rbitcoin-net/src/ibd/assign.rs +++ b/crates/rbitcoin-net/src/ibd/assign.rs @@ -13,11 +13,20 @@ //! - BQ payload **≥ assign-stop** (default 1 GiB) → holes only within the //! ~1 min tip-rate window **and** not past fetched_hi (do not grow past //! fetched; do not densify far holes outside the window) +//! - **Getdata owners:** tip-hole adds up to [`TIP_HOLE_MAX_PEERS`] immediately, +//! then drops at most one owner of that hash when a sibling has stream rx or +//! that owner is a relative-slow outlier among owners — never a wall-clock +//! whole-set abort. Densify is single-peer, issued by recent-bps rank. +//! Default cap **8**; **16** only if recent bps ≥ 2× pack median; `hole>0` +//! drips **2** for tip-race headroom. //! - Never request beyond densify horizon; events refuse far bodies too. //! - One body-queue copy per height (receive path drops duplicates). -use super::assign_plan::far_slots_per_peer; -use super::peer_io::{touch_block_progress, PeerCmd, PeerSlot}; +use super::assign_plan::densify_slots_for_peer; +use super::dial::{ + median_u64, relative_slow_pick, RelativeSlowSample, RELATIVE_SLOW_CLUSTER_SPREAD, +}; +use super::peer_io::{ibd_mono_ms, snapshot_peer_rx, touch_block_progress, PeerCmd, PeerSlot}; use super::state::{self, IbdWorkState}; use super::status::LoopStats; use super::{ @@ -134,6 +143,7 @@ pub(crate) fn assign_work_ordered( if alive.is_empty() { return; } + snapshot_peer_rx(&st.slots); prune_satisfied_inflight(&mut st.slots, &mut st.inflight, hub); prune_off_path_inflight(st); @@ -210,15 +220,34 @@ pub(crate) fn assign_work_ordered( return; } + // Leave per-peer headroom for tip races while hole>0. Extra densify slots + // only for a recent-bps outlier (≥ 2× pack median) when the pack is spread. + let tip_hole = !tip_holes.is_empty(); + let (pack_median, pack_tight) = pack_recent_bps(&st.slots, &alive); + let caps: HashMap = alive + .iter() + .map(|&pid| { + ( + pid, + densify_cap_for( + &st.slots, + pid, + cfg.per_peer, + tip_hole, + pack_median, + pack_tight, + ), + ) + }) + .collect(); + issued += steal_hung_densify(st, hub, &alive, tip_batch_hi, &caps); + let mut room = cfg.window.saturating_sub(st.inflight.len()); if room == 0 { finish_assign(loop_stats, t0, issued); return; } - // Leave per-peer headroom for tip races while hole>0. - let densify_per_peer = far_slots_per_peer(cfg.per_peer, !tip_holes.is_empty()); - let densify_hi = path_lo.saturating_add(CONTIG_DENSIFY_AHEAD); let depth_bytes = hub.query.block_queue_stats().1; let fetched_hi = hub @@ -241,6 +270,13 @@ pub(crate) fn assign_work_ordered( } st.assign_path_lo = path_lo; st.densify_scan_lo = st.densify_scan_lo.max(path_lo); + if !alive + .iter() + .any(|&pid| peer_has_slot(st, pid, caps.get(&pid).copied().unwrap_or(1))) + { + finish_assign(loop_stats, t0, issued); + return; + } let densify_lo = path_lo.max(st.densify_scan_lo); let densify = collect_height_band(st, hub, densify_lo, band_hi, room.max(1)); if densify.is_empty() { @@ -248,30 +284,24 @@ pub(crate) fn assign_work_ordered( return; } - let mut peer_i = st.assign_rot; - st.assign_rot = st.assign_rot.wrapping_add(1); + let ranked = rank_peers_by_speed(&st.slots, &alive, &HashSet::new()); let mut densify_q = densify; - while room > 0 && !densify_q.is_empty() { - let mut any = false; - for _ in 0..alive.len() { - if room == 0 || densify_q.is_empty() { + for &pid in &ranked { + if room == 0 || densify_q.is_empty() { + break; + } + let cap = caps.get(&pid).copied().unwrap_or(1); + while room > 0 && !densify_q.is_empty() { + if !peer_has_slot(st, pid, cap) { break; } - let pid = alive[peer_i % alive.len()]; - peer_i += 1; - if !peer_has_slot(st, pid, densify_per_peer) { - continue; - } let Some(h) = pop_need(&mut densify_q, st, hub) else { break; }; - if issue_one(st, pid, h, &mut room, &mut issued) { - any = true; + if !issue_one(st, pid, h, &mut room, &mut issued) { + break; } } - if !any { - break; - } } finish_assign(loop_stats, t0, issued); @@ -562,21 +592,219 @@ fn demote_zombie_pending_for_fetch( body.mark_missing(hash); } -/// Tip-hole getdata older than this with no claimable wire is cleared and re-issued -/// (mainnet freeze: inflight stuck, soft frozen, hole=1 forever). +/// Stream rx older than this is not “recent” for tip-hole owner eviction. +/// Matches the absolute stall floor so slow-but-steady 64 KiB ticks stay live. +const TIP_HOLE_RX_STALE: Duration = Duration::from_secs(30); + +fn peer_has_recent_rx(slot: &PeerSlot, now_ms: u64) -> bool { + let p = slot.last_rx_progress_ms.load(Ordering::Relaxed); + p != 0 && now_ms.saturating_sub(p) <= TIP_HOLE_RX_STALE.as_millis() as u64 +} + +/// Which current owner of a tip-hole hash to drop from **this hash** (not disconnect). /// -/// Short on purpose: confirm claim waits ~5s per tick while tip is blocked; 20s -/// left the same slow race set holding hole=1 while densify progressed. -const TIP_HOLE_INFLIGHT_STALE: Duration = Duration::from_secs(6); - -/// Rank alive peer ids for tip-hole getdata: prefer peers not in `avoid`, then -/// higher live `speed_sample` bps, then lower id. Unsampled peers sort last -/// among non-avoided (bps=0). -pub(crate) fn rank_tip_hole_peers( +/// - No owner has recent rx → none (too early / first 64 KiB still in flight). +/// - Some have recent rx, some do not → drop a no-rx owner (quick dead-racer). +/// - All have recent rx → [`relative_slow_pick`] among those owners (`min_samples` = +/// owner count). Tight cluster → none. +/// - Solo owner: drop only when no recent rx and `started_at` is ≥ [`TIP_HOLE_RX_STALE`]. +pub(crate) fn tip_hole_owner_to_drop( + owners: &[usize], + slots: &[PeerSlot], + started_at: Instant, +) -> Option { + if owners.is_empty() { + return None; + } + let now_ms = ibd_mono_ms(); + let mut recent = Vec::new(); + let mut stale = Vec::new(); + for &id in owners { + let Some(slot) = slots.iter().find(|s| s.id == id && s.alive) else { + stale.push(id); + continue; + }; + if peer_has_recent_rx(slot, now_ms) { + recent.push(id); + } else { + stale.push(id); + } + } + if owners.len() == 1 { + if !recent.is_empty() { + return None; + } + if Instant::now().duration_since(started_at) >= TIP_HOLE_RX_STALE { + return Some(owners[0]); + } + return None; + } + if recent.is_empty() { + return None; + } + if let Some(&id) = stale.iter().min() { + return Some(id); + } + let samples: Vec = owners + .iter() + .filter_map(|&id| { + let s = slots.iter().find(|s| s.id == id && s.alive)?; + Some(RelativeSlowSample { + peer_id: id, + bps: s.recent_bps().unwrap_or(0), + has_inflight: true, + }) + }) + .collect(); + relative_slow_pick(&samples, samples.len()) +} + +fn drop_hash_owner(st: &mut IbdWorkState, hash: BlockHash, pid: usize) { + if let Some(s) = st.slots.iter_mut().find(|s| s.id == pid) { + s.in_flight.remove(&hash); + } + if let Some(req) = st.inflight.get_mut(&hash) { + if req.remove_peer(pid) { + st.inflight.remove(&hash); + } + } +} + +fn peer_bps(slots: &[PeerSlot], pid: usize) -> u64 { + slots + .iter() + .find(|s| s.id == pid && s.alive) + .and_then(|s| s.recent_bps()) + .unwrap_or(0) +} + +fn pack_recent_bps(slots: &[PeerSlot], alive: &[usize]) -> (Option, bool) { + let mut samples: Vec = alive + .iter() + .filter_map(|&pid| { + slots + .iter() + .find(|s| s.id == pid && s.alive) + .and_then(|s| s.recent_bps()) + }) + .collect(); + if samples.is_empty() { + return (None, true); + } + samples.sort_unstable(); + let lo = samples[0]; + let hi = samples[samples.len() - 1]; + let tight = if lo == 0 { + hi == 0 + } else { + hi <= lo.saturating_mul(RELATIVE_SLOW_CLUSTER_SPREAD) + }; + (Some(median_u64(&samples)), tight) +} + +fn densify_cap_for( + slots: &[PeerSlot], + pid: usize, + per_peer: usize, + tip_hole: bool, + pack_median: Option, + pack_tight: bool, +) -> usize { + let bps = slots + .iter() + .find(|s| s.id == pid && s.alive) + .and_then(|s| s.recent_bps()); + densify_slots_for_peer(per_peer, tip_hole, bps, pack_median, pack_tight) +} + +/// Move hung single-peer densify getdata to a faster peer with a free slot. +/// +/// Hung = no recent stream rx and inflight age ≥ [`TIP_HOLE_RX_STALE`]. Does not +/// steal while the owner is still pulling, or when no faster peer exists. +/// Faster-but-full: drop the hung hash and rewind `densify_scan_lo`. +fn steal_hung_densify( + st: &mut IbdWorkState, + hub: &ChainHub, + alive: &[usize], + tip_batch_hi: u32, + densify_caps: &HashMap, +) -> u64 { + let now = Instant::now(); + let now_ms = ibd_mono_ms(); + let candidates: Vec<(BlockHash, u32, usize, Instant)> = st + .inflight + .iter() + .filter_map(|(h, req)| { + if req.len() != 1 { + return None; + } + let &ht = st.hash_height.get(h)?; + if ht <= tip_batch_hi { + return None; + } + let &pid = req.peers.iter().next()?; + Some((*h, ht, pid, req.started_at)) + }) + .collect(); + let hung: Vec = candidates + .into_iter() + .filter(|(h, ht, pid, started)| { + if super::progress::claim_ready(hub, &mut st.body, *ht, h) { + return false; + } + let Some(slot) = st.slots.iter().find(|s| s.id == *pid && s.alive) else { + return false; + }; + if peer_has_recent_rx(slot, now_ms) { + return false; + } + now.duration_since(*started) >= TIP_HOLE_RX_STALE + }) + .map(|(h, _, _, _)| h) + .collect(); + let mut issued = 0u64; + for h in hung { + let Some(owner) = st + .inflight + .get(&h) + .and_then(|req| req.peers.iter().copied().next()) + else { + continue; + }; + let owner_bps = peer_bps(&st.slots, owner); + let mut faster: Vec = alive + .iter() + .copied() + .filter(|&pid| pid != owner && peer_bps(&st.slots, pid) > owner_bps) + .collect(); + if faster.is_empty() { + continue; + } + faster.sort_by(|&a, &b| peer_bps(&st.slots, b).cmp(&peer_bps(&st.slots, a))); + let dest = faster + .into_iter() + .find(|&pid| peer_has_slot(st, pid, densify_caps.get(&pid).copied().unwrap_or(1))); + let ht = st.hash_height.get(&h).copied(); + drop_hash_owner(st, h, owner); + if let Some(pid) = dest { + let mut room = 1usize; + let _ = issue_one(st, pid, h, &mut room, &mut issued); + } else if let Some(ht) = ht { + st.densify_scan_lo = st.densify_scan_lo.min(ht); + } + } + issued +} + +/// Rank alive peer ids for getdata: prefer peers not in `avoid`, then +/// higher live [`PeerSlot::recent_bps`], then lower id. Unsampled peers sort +/// last among non-avoided (bps=0). +pub(crate) fn rank_peers_by_speed( slots: &[PeerSlot], alive: &[usize], avoid: &std::collections::HashSet, ) -> Vec { + snapshot_peer_rx(slots); let mut ranked: Vec = alive.to_vec(); ranked.sort_by(|&a, &b| { let avoided_a = avoid.contains(&a) as u8; @@ -586,8 +814,7 @@ pub(crate) fn rank_tip_hole_peers( slots .iter() .find(|s| s.id == pid && s.alive) - .and_then(|s| s.speed_sample()) - .map(|(_, bps)| bps) + .and_then(|s| s.recent_bps()) .unwrap_or(0) }; bps(b).cmp(&bps(a)).then_with(|| a.cmp(&b)) @@ -598,8 +825,10 @@ pub(crate) fn rank_tip_hole_peers( /// Cover each tip-hole hash with multi-peer getdata, preferring faster peers. /// -/// Stale tip-batch inflight is cleared after [`TIP_HOLE_INFLIGHT_STALE`] and -/// re-raced, preferring peers that were **not** in the cleared set. +/// While the hole is open, at most one current owner of **this hash** is dropped +/// per call when a sibling is pulling or that owner is a relative-slow outlier +/// among owners ([`tip_hole_owner_to_drop`]). The whole race set is never +/// cleared on request age. pub(crate) fn cover_tip_holes( st: &mut IbdWorkState, hub: &ChainHub, @@ -636,10 +865,10 @@ pub(crate) fn cover_tip_holes( demote_zombie_pending_for_fetch(&mut st.body, hub, h, ht); let mut avoid: HashSet = HashSet::new(); if let Some(req) = st.inflight.get(&h) { - if now.duration_since(req.started_at) >= TIP_HOLE_INFLIGHT_STALE { - avoid = req.peers.clone(); - clear_hash_inflight(&mut st.slots, &mut st.inflight, h); - st.body.mark_missing(h); + let owners: Vec = req.peers.iter().copied().collect(); + if let Some(pid) = tip_hole_owner_to_drop(&owners, &st.slots, req.started_at) { + drop_hash_owner(st, h, pid); + avoid.insert(pid); } } let (already, second_at) = st @@ -653,11 +882,14 @@ pub(crate) fn cover_tip_holes( } let mut need = want - already; let mut placed_any = false; - let ranked = rank_tip_hole_peers(&st.slots, alive, &avoid); + let ranked = rank_peers_by_speed(&st.slots, alive, &avoid); for &pid in &ranked { if need == 0 { break; } + if avoid.contains(&pid) { + continue; + } let Some(idx) = st.slots.iter().position(|s| s.id == pid && s.alive) else { continue; }; @@ -759,23 +991,28 @@ mod tests { cmd_tx, in_flight: HashSet::new(), block_progress_ms: Arc::new(AtomicU64::new(0)), + last_rx_progress_ms: Arc::new(AtomicU64::new(0)), peer_height: 100, connected_ms: 1, first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive: true, task, } } fn tmp_hub() -> (std::path::PathBuf, ChainHub) { + static N: AtomicU64 = AtomicU64::new(0); let dir = std::env::temp_dir().join(format!( - "rbitcoin-assign-{}-{}", + "rbitcoin-assign-{}-{}-{}", std::process::id(), std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap() - .as_nanos() + .as_nanos(), + N.fetch_add(1, Ordering::Relaxed) )); let _ = std::fs::create_dir_all(&dir); let q = Query::open_or_create(dir.join("store")).unwrap(); @@ -785,6 +1022,36 @@ mod tests { ) } + /// Serialize env mutators — parallel suite races `bq_assign_stop_bytes`. + static BQ_ASSIGN_STOP_ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + + struct AssignStopEnvRestore(Option, Option); + impl Drop for AssignStopEnvRestore { + fn drop(&mut self) { + match self.0.take() { + Some(v) => std::env::set_var("RBITCOIN_BLOCK_QUEUE_BYTES", v), + None => std::env::remove_var("RBITCOIN_BLOCK_QUEUE_BYTES"), + } + match self.1.take() { + Some(v) => std::env::set_var("RBITCOIN_BLOCK_QUEUE_GB", v), + None => std::env::remove_var("RBITCOIN_BLOCK_QUEUE_GB"), + } + } + } + + fn lock_default_assign_stop() -> (std::sync::MutexGuard<'static, ()>, AssignStopEnvRestore) { + let g = BQ_ASSIGN_STOP_ENV_LOCK + .lock() + .unwrap_or_else(|e| e.into_inner()); + let restore = AssignStopEnvRestore( + std::env::var_os("RBITCOIN_BLOCK_QUEUE_BYTES"), + std::env::var_os("RBITCOIN_BLOCK_QUEUE_GB"), + ); + std::env::remove_var("RBITCOIN_BLOCK_QUEUE_BYTES"); + std::env::remove_var("RBITCOIN_BLOCK_QUEUE_GB"); + (g, restore) + } + #[test] fn clear_inflight_add_peer_pop_need_and_tip_holes() { let (dir, hub) = tmp_hub(); @@ -916,6 +1183,45 @@ mod tests { assert!(!archive_pipeline_saturated(0, 32, true)); } + #[test] + fn tip_hole_owner_to_drop_too_early_dead_racer_and_solo() { + use super::super::peer_io::ibd_mono_ms; + let slots = vec![dummy_slot(0), dummy_slot(1)]; + let started = Instant::now(); + assert_eq!( + tip_hole_owner_to_drop(&[0, 1], &slots, started), + None, + "no rx yet is too early" + ); + slots[1] + .last_rx_progress_ms + .store(ibd_mono_ms().max(1), Ordering::Relaxed); + assert_eq!( + tip_hole_owner_to_drop(&[0, 1], &slots, started), + Some(0), + "silent owner drops when sibling has rx" + ); + slots[0] + .last_rx_progress_ms + .store(ibd_mono_ms().max(1), Ordering::Relaxed); + assert_eq!( + tip_hole_owner_to_drop(&[0], &slots, Instant::now() - Duration::from_secs(7)), + None, + "solo with live rx is kept" + ); + assert_eq!( + tip_hole_owner_to_drop(&[0], &slots, Instant::now() - Duration::from_secs(31)), + None, + "solo with live rx kept even if started_at is old" + ); + slots[0].last_rx_progress_ms.store(0, Ordering::Relaxed); + assert_eq!( + tip_hole_owner_to_drop(&[0], &slots, Instant::now() - Duration::from_secs(31)), + Some(0), + "solo hung with no rx after 30s is replaced" + ); + } + #[test] fn densify_yields_peer_slots_while_tip_hole_open() { use super::super::assign_plan::far_slots_per_peer; @@ -924,6 +1230,377 @@ mod tests { assert_eq!(far_slots_per_peer(16, false), 8); } + fn plant_work_path(st: &mut IbdWorkState, lo: u32, hi: u32) { + for ht in lo..=hi { + let hash = h(ht); + st.record_height(hash, ht); + st.height_to_hash.insert(ht, hash); + st.ordered_set.insert(hash); + st.ordered.push_back(hash); + st.max_ordered_height = ht; + st.body.mark_missing(hash); + } + } + + fn inject_bps(slot: &mut PeerSlot, now: u64, bytes: u64) { + slot.connected_ms = now.saturating_sub(2_000); + slot.first_data_ms + .store(now.saturating_sub(1_000), Ordering::Relaxed); + slot.bytes_rx.store(bytes, Ordering::Relaxed); + } + + #[test] + fn densify_hung_owner_stolen_to_faster_peer() { + use super::super::peer_io::ibd_mono_ms; + use super::super::state::InflightReq; + use bitcoin::hashes::Hash as _; + let (dir, hub) = tmp_hub(); + hub.ensure_genesis().unwrap(); + let mut st = IbdWorkState::new( + vec![dummy_slot(0), dummy_slot(1)], + hub.tip_hash(), + hub.tip_height(), + ); + let stats = LoopStats::default(); + let mut cfg = IbdConfig::for_test(); + cfg.window = 64; + cfg.per_peer = 16; + let path_lo = hub.tip_height().unwrap_or(0).saturating_add(1); + plant_work_path(&mut st, path_lo, 40); + hub.query + .block_queue_offer(path_lo, h(path_lo).to_byte_array(), 1, &[0u8; 80]) + .unwrap(); + st.body.mark_pending(h(path_lo)); + let hung = h(40); + let mut req = InflightReq::new(0); + req.started_at = Instant::now() - Duration::from_secs(31); + st.inflight.insert(hung, req); + st.slots[0].in_flight.insert(hung); + let now = ibd_mono_ms().max(2_000); + inject_bps(&mut st.slots[1], now, 1_000_000); + assign_work_ordered( + &mut st, + &hub, + &cfg, + &stats, + path_lo, + AssignDepth::Full, + None, + ); + let peers = &st.inflight[&hung].peers; + assert!( + peers.contains(&1) && !peers.contains(&0), + "hung densify must move to faster peer; peers={peers:?}" + ); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn densify_slow_but_rx_live_not_stolen() { + use super::super::peer_io::ibd_mono_ms; + use super::super::state::InflightReq; + use bitcoin::hashes::Hash as _; + let (dir, hub) = tmp_hub(); + hub.ensure_genesis().unwrap(); + let mut st = IbdWorkState::new( + vec![dummy_slot(0), dummy_slot(1)], + hub.tip_hash(), + hub.tip_height(), + ); + let stats = LoopStats::default(); + let mut cfg = IbdConfig::for_test(); + cfg.window = 64; + cfg.per_peer = 16; + let path_lo = hub.tip_height().unwrap_or(0).saturating_add(1); + plant_work_path(&mut st, path_lo, 40); + hub.query + .block_queue_offer(path_lo, h(path_lo).to_byte_array(), 1, &[0u8; 80]) + .unwrap(); + st.body.mark_pending(h(path_lo)); + let hung = h(40); + let mut req = InflightReq::new(0); + req.started_at = Instant::now() - Duration::from_secs(31); + st.inflight.insert(hung, req); + st.slots[0].in_flight.insert(hung); + let now = ibd_mono_ms().max(2_000); + st.slots[0] + .last_rx_progress_ms + .store(now, Ordering::Relaxed); + inject_bps(&mut st.slots[1], now, 1_000_000); + assign_work_ordered( + &mut st, + &hub, + &cfg, + &stats, + path_lo, + AssignDepth::Full, + None, + ); + assert!( + st.inflight[&hung].contains_peer(0), + "live rx must not be stolen; peers={:?}", + st.inflight[&hung].peers + ); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn densify_hung_no_faster_peer_does_not_steal() { + use super::super::state::InflightReq; + use bitcoin::hashes::Hash as _; + let (dir, hub) = tmp_hub(); + hub.ensure_genesis().unwrap(); + let mut st = IbdWorkState::new(vec![dummy_slot(0)], hub.tip_hash(), hub.tip_height()); + let stats = LoopStats::default(); + let mut cfg = IbdConfig::for_test(); + cfg.window = 64; + cfg.per_peer = 16; + let path_lo = hub.tip_height().unwrap_or(0).saturating_add(1); + plant_work_path(&mut st, path_lo, 40); + hub.query + .block_queue_offer(path_lo, h(path_lo).to_byte_array(), 1, &[0u8; 80]) + .unwrap(); + st.body.mark_pending(h(path_lo)); + let hung = h(40); + let mut req = InflightReq::new(0); + req.started_at = Instant::now() - Duration::from_secs(31); + st.inflight.insert(hung, req); + st.slots[0].in_flight.insert(hung); + assign_work_ordered( + &mut st, + &hub, + &cfg, + &stats, + path_lo, + AssignDepth::Full, + None, + ); + assert!( + st.inflight[&hung].contains_peer(0), + "solo hung densify has no faster peer to steal to" + ); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn densify_hung_no_slot_rewinds_scan_lo() { + use super::super::peer_io::ibd_mono_ms; + use super::super::state::InflightReq; + use bitcoin::hashes::Hash as _; + let (dir, hub) = tmp_hub(); + hub.ensure_genesis().unwrap(); + let mut st = IbdWorkState::new( + vec![dummy_slot(0), dummy_slot(1)], + hub.tip_hash(), + hub.tip_height(), + ); + let stats = LoopStats::default(); + let mut cfg = IbdConfig::for_test(); + cfg.window = 1; + cfg.per_peer = 2; + let path_lo = hub.tip_height().unwrap_or(0).saturating_add(1); + plant_work_path(&mut st, path_lo, 41); + hub.query + .block_queue_offer(path_lo, h(path_lo).to_byte_array(), 1, &[0u8; 80]) + .unwrap(); + st.body.mark_pending(h(path_lo)); + let hung = h(40); + let other = h(41); + let mut req = InflightReq::new(0); + req.started_at = Instant::now() - Duration::from_secs(31); + st.inflight.insert(hung, req); + st.slots[0].in_flight.insert(hung); + st.inflight.insert(other, InflightReq::new(1)); + st.slots[1].in_flight.insert(other); + st.slots[1] + .last_rx_progress_ms + .store(ibd_mono_ms().max(1), Ordering::Relaxed); + let now = ibd_mono_ms().max(2_000); + inject_bps(&mut st.slots[1], now, 1_000_000); + st.densify_scan_lo = 90; + assign_work_ordered( + &mut st, + &hub, + &cfg, + &stats, + path_lo, + AssignDepth::Full, + None, + ); + assert!( + !st.inflight.contains_key(&hung), + "hung hash cleared when faster peer has no slot" + ); + assert!( + st.densify_scan_lo <= 40, + "scan_lo must rewind to hung height; scan_lo={}", + st.densify_scan_lo + ); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn densify_issues_to_fastest_peer_first() { + let _env = lock_default_assign_stop(); + use super::super::peer_io::ibd_mono_ms; + use bitcoin::hashes::Hash as _; + let (dir, hub) = tmp_hub(); + hub.ensure_genesis().unwrap(); + let mut st = IbdWorkState::new( + vec![dummy_slot(0), dummy_slot(1), dummy_slot(2)], + hub.tip_hash(), + hub.tip_height(), + ); + let stats = LoopStats::default(); + let mut cfg = IbdConfig::for_test(); + cfg.window = 128; + cfg.per_peer = 16; + let path_lo = hub.tip_height().unwrap_or(0).saturating_add(1); + plant_work_path(&mut st, path_lo, 40); + for ht in path_lo..=32 { + hub.query + .block_queue_offer(ht, h(ht).to_byte_array(), 1, &[0u8; 80]) + .unwrap(); + st.body.mark_pending(h(ht)); + } + let now = ibd_mono_ms().max(2_000); + inject_bps(&mut st.slots[0], now, 100_000); + inject_bps(&mut st.slots[1], now, 10_000_000); + inject_bps(&mut st.slots[2], now, 1_000_000); + st.densify_scan_lo = 40; + assign_work_ordered( + &mut st, + &hub, + &cfg, + &stats, + path_lo, + AssignDepth::Full, + None, + ); + let want = h(40); + assert!( + st.inflight.get(&want).is_some_and(|r| r.contains_peer(1)), + "first densify hash must go to fastest peer; inflight={:?}", + st.inflight.get(&want).map(|r| &r.peers) + ); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn densify_skips_band_walk_when_peers_at_cap() { + use super::super::state::InflightReq; + use bitcoin::hashes::Hash as _; + let (dir, hub) = tmp_hub(); + hub.ensure_genesis().unwrap(); + let mut st = IbdWorkState::new( + vec![dummy_slot(0), dummy_slot(1)], + hub.tip_hash(), + hub.tip_height(), + ); + let stats = LoopStats::default(); + let mut cfg = IbdConfig::for_test(); + cfg.window = 128; + cfg.per_peer = 16; + let path_lo = hub.tip_height().unwrap_or(0).saturating_add(1); + plant_work_path(&mut st, path_lo, 70); + for ht in path_lo..=32 { + hub.query + .block_queue_offer(ht, h(ht).to_byte_array(), 1, &[0u8; 80]) + .unwrap(); + st.body.mark_pending(h(ht)); + } + for i in 0..16u32 { + let ht = 40 + i; + let hash = h(ht); + let pid = (i % 2) as usize; + st.inflight.insert(hash, InflightReq::new(pid)); + st.slots[pid].in_flight.insert(hash); + } + st.densify_scan_lo = 40; + let before_keys: HashSet<_> = st.inflight.keys().copied().collect(); + assign_work_ordered( + &mut st, + &hub, + &cfg, + &stats, + path_lo, + AssignDepth::Full, + None, + ); + let after_keys: HashSet<_> = st.inflight.keys().copied().collect(); + assert_eq!(after_keys, before_keys, "no new densify when peers at cap"); + assert_eq!( + st.densify_scan_lo, 40, + "band walk must not advance scan_lo; scan_lo={}", + st.densify_scan_lo + ); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn densify_fast_peer_receives_more_than_eight() { + let _env = lock_default_assign_stop(); + use super::super::peer_io::ibd_mono_ms; + use bitcoin::hashes::Hash as _; + let (dir, hub) = tmp_hub(); + hub.ensure_genesis().unwrap(); + let mut st = IbdWorkState::new( + vec![dummy_slot(0), dummy_slot(1), dummy_slot(2)], + hub.tip_hash(), + hub.tip_height(), + ); + let stats = LoopStats::default(); + let mut cfg = IbdConfig::for_test(); + cfg.window = 128; + cfg.per_peer = 16; + let path_lo = hub.tip_height().unwrap_or(0).saturating_add(1); + plant_work_path(&mut st, path_lo, 52); + for ht in path_lo..=32 { + hub.query + .block_queue_offer(ht, h(ht).to_byte_array(), 1, &[0u8; 80]) + .unwrap(); + st.body.mark_pending(h(ht)); + } + while ibd_mono_ms() < 1_200 { + std::thread::sleep(Duration::from_millis(5)); + } + let now = ibd_mono_ms(); + let win = now.saturating_sub(5_000).max(1); + inject_bps(&mut st.slots[0], now, 5_000_000); + inject_bps(&mut st.slots[1], now, 15_000_000); + inject_bps(&mut st.slots[2], now, 5_000_000); + for s in &st.slots { + s.window_start_ms.store(win, Ordering::Relaxed); + s.window_start_bytes.store(0, Ordering::Relaxed); + } + st.densify_scan_lo = 33; + assign_work_ordered( + &mut st, + &hub, + &cfg, + &stats, + path_lo, + AssignDepth::Full, + None, + ); + assert_eq!( + st.slots[1].in_flight.len(), + 16, + "2×-median outlier must get full densify cap" + ); + assert!( + st.slots[0].in_flight.len() <= 8, + "non-outlier stays at half cap; n={}", + st.slots[0].in_flight.len() + ); + assert!( + st.slots[2].in_flight.len() <= 8, + "non-outlier stays at half cap; n={}", + st.slots[2].in_flight.len() + ); + let _ = std::fs::remove_dir_all(dir); + } + /// Wrong first-wins body at tip+1 is not claim-ready; cover must dequeue and /// re-get the work-path hash (general hole=1 with bq soft growing ahead). #[test] @@ -1070,10 +1747,10 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } - /// Stale tip-hole inflight (≥6s) with no claimable wire must clear and re-race. - /// Mainnet freeze: hole=1, inflight stuck forever, soft frozen, conf=0. + /// Aged started_at does not clear the whole race set when owners have rx. #[test] - fn cover_tip_holes_re_races_stale_inflight() { + fn cover_tip_holes_does_not_clear_whole_set_on_started_at() { + use super::super::peer_io::ibd_mono_ms; use super::super::state::InflightReq; let (dir, hub) = tmp_hub(); hub.ensure_genesis().unwrap(); @@ -1088,50 +1765,182 @@ mod tests { st.record_height(hole, ht); st.height_to_hash.insert(ht, hole); st.body.mark_missing(hole); - // Fresh inflight (<6s) must not re-race yet. - let mut fresh = InflightReq::new(0); - fresh.started_at = Instant::now() - Duration::from_secs(3); - st.inflight.insert(hole, fresh); + let mut req = InflightReq::new(0); + req.add_peer(1); + req.started_at = Instant::now() - Duration::from_secs(7); + st.inflight.insert(hole, req); st.slots[0].in_flight.insert(hole); + st.slots[1].in_flight.insert(hole); + let now = ibd_mono_ms().max(1); + st.slots[0] + .last_rx_progress_ms + .store(now, Ordering::Relaxed); + st.slots[1] + .last_rx_progress_ms + .store(now, Ordering::Relaxed); + // Tight bps cluster so relative-slow among owners does not fire. + for i in 0..2 { + st.slots[i].connected_ms = now.saturating_sub(2_000); + st.slots[i] + .first_data_ms + .store(now.saturating_sub(1_000), Ordering::Relaxed); + st.slots[i].bytes_rx.store(500_000, Ordering::Relaxed); + } + let cfg = IbdConfig::for_test(); + let alive: Vec = st.slots.iter().filter(|s| s.alive).map(|s| s.id).collect(); let holes = contiguous_tip_holes(&mut st, &hub, 8); - assert_eq!(holes, vec![hole]); + let _ = cover_tip_holes(&mut st, &hub, &cfg, &alive, &holes); + let peers = &st.inflight[&hole].peers; + assert!( + peers.contains(&0) && peers.contains(&1), + "aged started_at must not drop live racers; peers={peers:?}" + ); + let _ = std::fs::remove_dir_all(dir); + } + + /// Sibling pulling the block → silent owner is dropped from this hash. + #[test] + fn cover_tip_holes_drops_owner_with_no_rx_when_sibling_progresses() { + use super::super::peer_io::ibd_mono_ms; + use super::super::state::InflightReq; + let (dir, hub) = tmp_hub(); + hub.ensure_genesis().unwrap(); + let mut st = IbdWorkState::new( + vec![dummy_slot(0), dummy_slot(1), dummy_slot(2)], + hub.tip_hash(), + hub.tip_height(), + ); + let hole = h(0x52); + let tip = hub.tip_height().unwrap_or(0); + let ht = tip.saturating_add(1); + st.record_height(hole, ht); + st.height_to_hash.insert(ht, hole); + st.body.mark_missing(hole); + let mut req = InflightReq::new(0); + req.add_peer(1); + st.inflight.insert(hole, req); + st.slots[0].in_flight.insert(hole); + st.slots[1].in_flight.insert(hole); + st.slots[1] + .last_rx_progress_ms + .store(ibd_mono_ms().max(1), Ordering::Relaxed); let cfg = IbdConfig::for_test(); let alive: Vec = st.slots.iter().filter(|s| s.alive).map(|s| s.id).collect(); - let issued_fresh = cover_tip_holes(&mut st, &hub, &cfg, &alive, &holes); - // Still at want peers (race fills), but started_at not cleared as stale. - let age_fresh = Instant::now().duration_since(st.inflight[&hole].started_at); + let holes = contiguous_tip_holes(&mut st, &hub, 8); + let _ = cover_tip_holes(&mut st, &hub, &cfg, &alive, &holes); + let peers = &st.inflight[&hole].peers; + assert!( + !peers.contains(&0), + "no-rx owner must leave the race when a sibling has rx; peers={peers:?}" + ); assert!( - age_fresh >= Duration::from_secs(2), - "fresh inflight must keep original started_at; age={age_fresh:?} issued={issued_fresh}" + peers.contains(&1), + "progressing sibling must stay; peers={peers:?}" ); + let _ = std::fs::remove_dir_all(dir); + } - // Frozen inflight from a prior race that never delivered wire (≥6s). + /// All owners have rx: drop half-median outlier; keep a tight 2× pack. + #[test] + fn cover_tip_holes_drops_relative_slow_owner_among_progressing() { + use super::super::peer_io::ibd_mono_ms; + use super::super::state::InflightReq; + let (dir, hub) = tmp_hub(); + hub.ensure_genesis().unwrap(); + let mut st = IbdWorkState::new( + vec![dummy_slot(0), dummy_slot(1), dummy_slot(2)], + hub.tip_hash(), + hub.tip_height(), + ); + let hole = h(0x53); + let tip = hub.tip_height().unwrap_or(0); + let ht = tip.saturating_add(1); + st.record_height(hole, ht); + st.height_to_hash.insert(ht, hole); + st.body.mark_missing(hole); + let now = ibd_mono_ms().max(2_000); + let mut req = InflightReq::new(0); + req.add_peer(1); + req.add_peer(2); + st.inflight.insert(hole, req); + for i in 0..3 { + st.slots[i].in_flight.insert(hole); + st.slots[i] + .last_rx_progress_ms + .store(now, Ordering::Relaxed); + st.slots[i].connected_ms = now.saturating_sub(2_000); + st.slots[i] + .first_data_ms + .store(now.saturating_sub(1_000), Ordering::Relaxed); + } + st.slots[0].bytes_rx.store(100_000, Ordering::Relaxed); + st.slots[1].bytes_rx.store(1_000_000, Ordering::Relaxed); + st.slots[2].bytes_rx.store(1_000_000, Ordering::Relaxed); + let cfg = IbdConfig::for_test(); + let alive: Vec = st.slots.iter().filter(|s| s.alive).map(|s| s.id).collect(); + let holes = contiguous_tip_holes(&mut st, &hub, 8); + let _ = cover_tip_holes(&mut st, &hub, &cfg, &alive, &holes); + let peers = &st.inflight[&hole].peers; + assert!( + !peers.contains(&0), + "half-median owner must drop; peers={peers:?}" + ); + + // Tight cluster: rebuild with similar bps. st.inflight.clear(); for s in st.slots.iter_mut() { s.in_flight.clear(); } - let mut frozen = InflightReq::new(0); - frozen.started_at = Instant::now() - Duration::from_secs(7); - st.inflight.insert(hole, frozen); - st.slots[0].in_flight.insert(hole); - assert!( - !super::super::progress::claim_ready(&hub, &mut st.body, ht, &hole), - "no wire → not claim-ready" - ); - let issued = cover_tip_holes(&mut st, &hub, &cfg, &alive, &holes); + let mut tight = InflightReq::new(0); + tight.add_peer(1); + tight.add_peer(2); + st.inflight.insert(hole, tight); + for i in 0..3 { + st.slots[i].in_flight.insert(hole); + st.slots[i].bytes_rx.store(500_000, Ordering::Relaxed); + } + let _ = cover_tip_holes(&mut st, &hub, &cfg, &alive, &holes); + let peers = &st.inflight[&hole].peers; assert!( - issued >= 1, - "stale inflight must re-race getdata; issued={issued}" + peers.contains(&0) && peers.contains(&1) && peers.contains(&2), + "tight 2× pack must keep all owners; peers={peers:?}" ); - assert!( - st.inflight.contains_key(&hole), - "hash remains inflight after re-race" + let _ = std::fs::remove_dir_all(dir); + } + + /// Solo owner with live rx is not “slowest of one” even if started_at is old. + #[test] + fn cover_tip_holes_solo_slow_but_rx_live_kept() { + use super::super::peer_io::ibd_mono_ms; + use super::super::state::InflightReq; + let (dir, hub) = tmp_hub(); + hub.ensure_genesis().unwrap(); + let mut st = IbdWorkState::new( + vec![dummy_slot(0), dummy_slot(1)], + hub.tip_hash(), + hub.tip_height(), ); - // Fresh started_at (not still the 7s-old stamp). - let age = Instant::now().duration_since(st.inflight[&hole].started_at); + let hole = h(0x54); + let tip = hub.tip_height().unwrap_or(0); + let ht = tip.saturating_add(1); + st.record_height(hole, ht); + st.height_to_hash.insert(ht, hole); + st.body.mark_missing(hole); + let mut req = InflightReq::new(0); + req.started_at = Instant::now() - Duration::from_secs(7); + st.inflight.insert(hole, req); + st.slots[0].in_flight.insert(hole); + st.slots[0] + .last_rx_progress_ms + .store(ibd_mono_ms().max(1), Ordering::Relaxed); + let cfg = IbdConfig::for_test(); + let alive: Vec = st.slots.iter().filter(|s| s.alive).map(|s| s.id).collect(); + let holes = contiguous_tip_holes(&mut st, &hub, 8); + let _ = cover_tip_holes(&mut st, &hub, &cfg, &alive, &holes); + let peers = &st.inflight[&hole].peers; assert!( - age < Duration::from_secs(2), - "re-race must reset started_at; age={age:?}" + peers.contains(&0), + "solo slow-but-steady owner must stay; peers={peers:?}" ); let _ = std::fs::remove_dir_all(dir); } @@ -1172,7 +1981,7 @@ mod tests { let cfg = IbdConfig::for_test(); let alive: Vec = st.slots.iter().filter(|s| s.alive).map(|s| s.id).collect(); let avoid = HashSet::new(); - let ranked = rank_tip_hole_peers(&st.slots, &alive, &avoid); + let ranked = rank_peers_by_speed(&st.slots, &alive, &avoid); assert_eq!(ranked[0], 1, "fastest peer first: ranked={ranked:?}"); assert_eq!(ranked[1], 2, "medium second: ranked={ranked:?}"); assert_eq!(ranked[2], 0, "slow last: ranked={ranked:?}"); @@ -1188,9 +1997,44 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } - /// After stale clear, re-race prefers peers not in the cleared set. + /// Historically-fast / currently-idle sorts behind a currently-fast peer. + #[test] + fn rank_peers_prefers_recent_window_over_lifetime() { + use super::super::peer_io::ibd_mono_ms; + let mut st = IbdWorkState::new( + vec![dummy_slot(0), dummy_slot(1), dummy_slot(2)], + None, + Some(0), + ); + while ibd_mono_ms() < 1_200 { + std::thread::sleep(Duration::from_millis(5)); + } + let now = ibd_mono_ms(); + let win = now.saturating_sub(5_000).max(1); + inject_bps(&mut st.slots[0], now, 10_000_000); + st.slots[0].window_start_ms.store(win, Ordering::Relaxed); + st.slots[0] + .window_start_bytes + .store(10_000_000, Ordering::Relaxed); + inject_bps(&mut st.slots[1], now, 200_000); + st.slots[1].window_start_ms.store(win, Ordering::Relaxed); + st.slots[1].window_start_bytes.store(0, Ordering::Relaxed); + let alive = vec![0usize, 1, 2]; + let ranked = rank_peers_by_speed(&st.slots, &alive, &HashSet::new()); + assert_eq!( + ranked[0], 1, + "currently-fast must rank first; ranked={ranked:?}" + ); + assert!( + st.slots[0].speed_sample().expect("lifetime").1 > st.slots[1].recent_bps().unwrap_or(0), + "peer 0 lifetime still exceeds peer 1 recent" + ); + } + + /// After dropping a silent owner, replacement prefers peers not in avoid. #[test] fn cover_tip_holes_rerace_avoids_prior_peers() { + use super::super::peer_io::ibd_mono_ms; use super::super::state::InflightReq; let (dir, hub) = tmp_hub(); hub.ensure_genesis().unwrap(); @@ -1208,10 +2052,12 @@ mod tests { let mut frozen = InflightReq::new(0); frozen.add_peer(1); - frozen.started_at = Instant::now() - Duration::from_secs(7); st.inflight.insert(hole, frozen); st.slots[0].in_flight.insert(hole); st.slots[1].in_flight.insert(hole); + st.slots[1] + .last_rx_progress_ms + .store(ibd_mono_ms().max(1), Ordering::Relaxed); let cfg = IbdConfig::for_test(); let alive: Vec = st.slots.iter().filter(|s| s.alive).map(|s| s.id).collect(); @@ -1219,11 +2065,14 @@ mod tests { let issued = cover_tip_holes(&mut st, &hub, &cfg, &alive, &holes); assert!(issued >= 1, "issued={issued}"); let peers = &st.inflight[&hole].peers; - // Prefer 2,3,4 over reusing 0,1 first — at least one new peer in race. + assert!( + !peers.contains(&0), + "silent owner 0 must be dropped; peers={peers:?}" + ); let new = peers.iter().any(|&p| p >= 2); assert!( new, - "re-race should include peers outside cleared set; peers={peers:?}" + "replacement should include peers outside avoid; peers={peers:?}" ); let _ = std::fs::remove_dir_all(dir); } @@ -1659,31 +2508,16 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } - /// Serialize env mutators — parallel suite races `bq_assign_stop_bytes`. - static BQ_ASSIGN_STOP_ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); - /// Over assign-stop: densify within confirm window ∩ fetched; not past window. #[test] fn densify_over_assign_stop_clamps_window_and_fetched() { let _g = BQ_ASSIGN_STOP_ENV_LOCK .lock() .unwrap_or_else(|e| e.into_inner()); - let prev_b = std::env::var_os("RBITCOIN_BLOCK_QUEUE_BYTES"); - let prev_g = std::env::var_os("RBITCOIN_BLOCK_QUEUE_GB"); - struct Restore(Option, Option); - impl Drop for Restore { - fn drop(&mut self) { - match self.0.take() { - Some(v) => std::env::set_var("RBITCOIN_BLOCK_QUEUE_BYTES", v), - None => std::env::remove_var("RBITCOIN_BLOCK_QUEUE_BYTES"), - } - match self.1.take() { - Some(v) => std::env::set_var("RBITCOIN_BLOCK_QUEUE_GB", v), - None => std::env::remove_var("RBITCOIN_BLOCK_QUEUE_GB"), - } - } - } - let _restore = Restore(prev_b, prev_g); + let _restore = AssignStopEnvRestore( + std::env::var_os("RBITCOIN_BLOCK_QUEUE_BYTES"), + std::env::var_os("RBITCOIN_BLOCK_QUEUE_GB"), + ); std::env::remove_var("RBITCOIN_BLOCK_QUEUE_GB"); std::env::set_var("RBITCOIN_BLOCK_QUEUE_BYTES", "2048"); diff --git a/crates/rbitcoin-net/src/ibd/assign_plan.rs b/crates/rbitcoin-net/src/ibd/assign_plan.rs index ccb572c4..f6c715d7 100644 --- a/crates/rbitcoin-net/src/ibd/assign_plan.rs +++ b/crates/rbitcoin-net/src/ibd/assign_plan.rs @@ -63,6 +63,33 @@ pub(crate) fn far_slots_per_peer(per_peer: usize, tip_hole: bool) -> usize { } } +/// Per-peer densify cap: drip 2 while a tip hole is open; otherwise half of +/// `per_peer`, or `per_peer` when this peer's recent bps is ≥ 2× pack median +/// and the pack is not a tight cluster. +pub(crate) fn densify_slots_for_peer( + per_peer: usize, + tip_hole: bool, + peer_bps: Option, + pack_median: Option, + pack_tight: bool, +) -> usize { + let base = far_slots_per_peer(per_peer, tip_hole); + if tip_hole || pack_tight { + return base; + } + let Some(bps) = peer_bps else { + return base; + }; + let Some(med) = pack_median else { + return base; + }; + if med > 0 && bps >= med.saturating_mul(2) { + per_peer + } else { + base + } +} + /// Whether to request more headers past the soft cap. /// /// Only when the ordered path is **mostly claim-ready** (dense body-queue / @@ -279,4 +306,45 @@ mod tests { assert_eq!(far_slots_per_peer(1, false), 1); // max(0,1)=1 assert_eq!(far_slots_per_peer(3, false), 1); } + + #[test] + fn densify_slots_tip_hole_is_two_for_all() { + assert_eq!( + densify_slots_for_peer(16, true, Some(10_000_000), Some(1_000_000), false), + 2 + ); + assert_eq!(densify_slots_for_peer(16, true, None, None, true), 2); + } + + #[test] + fn densify_slots_tight_pack_stays_half() { + assert_eq!( + densify_slots_for_peer(16, false, Some(1_500_000), Some(1_000_000), true), + 8 + ); + assert_eq!( + densify_slots_for_peer(16, false, Some(2_000_000), Some(1_000_000), true), + 8 + ); + } + + #[test] + fn densify_slots_fast_outlier_gets_full() { + assert_eq!( + densify_slots_for_peer(16, false, Some(2_000_000), Some(1_000_000), false), + 16 + ); + assert_eq!( + densify_slots_for_peer(16, false, Some(1_500_000), Some(1_000_000), false), + 8 + ); + assert_eq!( + densify_slots_for_peer(16, false, None, Some(1_000_000), false), + 8 + ); + assert_eq!( + densify_slots_for_peer(16, false, Some(2_000_000), None, false), + 8 + ); + } } diff --git a/crates/rbitcoin-net/src/ibd/dial.rs b/crates/rbitcoin-net/src/ibd/dial.rs index 0f3230fe..a5ad93b5 100644 --- a/crates/rbitcoin-net/src/ibd/dial.rs +++ b/crates/rbitcoin-net/src/ibd/dial.rs @@ -1,6 +1,8 @@ //! Peer dial, header request, stall disconnect / cooldown. -use super::peer_io::{ibd_mono_ms, spawn_peer, PeerCmd, PeerEventSinks, PeerSlot}; +use super::peer_io::{ + ibd_mono_ms, snapshot_peer_rx, spawn_peer, PeerCmd, PeerEventSinks, PeerSlot, +}; use crate::chain::ChainHub; use crate::error::NetError; use crate::peers::{trying_connection_log, PeerConnType}; @@ -133,6 +135,7 @@ pub(crate) fn relative_slow_with_hysteresis( /// Build mature relative-slow samples from live slots (age + bytes floors). pub(crate) fn mature_relative_slow_samples(slots: &[PeerSlot]) -> Vec { + snapshot_peer_rx(slots); let now = ibd_mono_ms(); let mut out = Vec::new(); for s in slots { @@ -150,7 +153,7 @@ pub(crate) fn mature_relative_slow_samples(slots: &[PeerSlot]) -> Vec SocketAddr { SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, o)), 8333) @@ -540,10 +543,13 @@ mod tests { cmd_tx, in_flight: HashSet::new(), block_progress_ms: Arc::new(AtomicU64::new(0)), + last_rx_progress_ms: Arc::new(AtomicU64::new(0)), peer_height: 0, connected_ms: 0, first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive, task, } @@ -994,4 +1000,49 @@ mod tests { assert!(suspect.is_none()); assert!(cooldown.is_empty()); } + + #[test] + fn mature_relative_slow_samples_uses_recent_bps() { + let now = { + while ibd_mono_ms() < 1_200 { + std::thread::sleep(std::time::Duration::from_millis(5)); + } + ibd_mono_ms() + }; + let win = now.saturating_sub(5_000).max(1); + let mut slots: Vec<_> = (0..8) + .map(|i| dummy_slot(i, addr(50 + i as u8), true)) + .collect(); + for (i, s) in slots.iter_mut().enumerate() { + s.first_data_ms.store(1, Ordering::Relaxed); + s.bytes_rx + .store(RELATIVE_SLOW_MIN_BYTES * 4, Ordering::Relaxed); + s.window_start_ms.store(win, Ordering::Relaxed); + if i == 0 { + s.window_start_bytes + .store(RELATIVE_SLOW_MIN_BYTES * 4, Ordering::Relaxed); + s.in_flight.insert(BlockHash::from_byte_array([1u8; 32])); + } else { + s.window_start_bytes.store(0, Ordering::Relaxed); + } + } + assert_eq!(slots[0].recent_bps().unwrap(), 0); + assert!(slots[1].recent_bps().unwrap() > 0); + assert!( + slots[0].speed_sample().expect("lifetime").1 > 0, + "lifetime sample kept for AddrMan" + ); + let samples = mature_relative_slow_samples(&slots); + if now.saturating_sub(1) >= RELATIVE_SLOW_MIN_AGE_MS { + let idle = samples + .iter() + .find(|s| s.peer_id == 0) + .expect("idle peer still mature"); + assert_eq!( + idle.bps, 0, + "mature sample must use recent window, not lifetime" + ); + assert!(samples.iter().filter(|s| s.peer_id != 0).all(|s| s.bps > 0)); + } + } } diff --git a/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs b/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs index 095769a2..51b22481 100644 --- a/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs +++ b/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs @@ -882,10 +882,13 @@ fn confirmed_height_mids_blocked_while_densify_ahead_leaves_tip_hole() { cmd_tx, in_flight: HashSet::new(), block_progress_ms: Arc::new(AtomicU64::new(0)), + last_rx_progress_ms: Arc::new(AtomicU64::new(0)), peer_height: 100, connected_ms: 1, first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive: true, task, }; @@ -1079,10 +1082,13 @@ fn zombie_pending_mid_at_confirmed_height_never_reget() { cmd_tx, in_flight: HashSet::new(), block_progress_ms: Arc::new(AtomicU64::new(0)), + last_rx_progress_ms: Arc::new(AtomicU64::new(0)), peer_height: 100, connected_ms: 1, first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive: true, task, }; @@ -1652,10 +1658,13 @@ fn apply_peer_event_body_and_control_surface() { cmd_tx, in_flight: HashSet::new(), block_progress_ms: Arc::new(AtomicU64::new(0)), + last_rx_progress_ms: Arc::new(AtomicU64::new(0)), peer_height: 10, connected_ms: 1, first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive: true, task, } @@ -1903,10 +1912,13 @@ fn apply_peer_event_block_framed_bq_horizon_and_headers_done() { cmd_tx, in_flight: HashSet::new(), block_progress_ms: Arc::new(AtomicU64::new(0)), + last_rx_progress_ms: Arc::new(AtomicU64::new(0)), peer_height: 5, connected_ms: 1, first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive: true, task, } @@ -2169,10 +2181,13 @@ fn block_framed_raw_offers_body_queue_with_confirm_feed() { cmd_tx, in_flight: HashSet::new(), block_progress_ms: Arc::new(AtomicU64::new(0)), + last_rx_progress_ms: Arc::new(AtomicU64::new(0)), peer_height: 5, connected_ms: 1, first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive: true, task, } @@ -2316,10 +2331,13 @@ fn known_headers_re_admit_to_ordered_after_tip_drain() { cmd_tx, in_flight: HashSet::new(), block_progress_ms: Arc::new(AtomicU64::new(0)), + last_rx_progress_ms: Arc::new(AtomicU64::new(0)), peer_height: 5, connected_ms: 1, first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive: true, task, } @@ -2476,10 +2494,13 @@ fn path_slot_first_wins_chained_via_headers() { cmd_tx, in_flight: HashSet::new(), block_progress_ms: Arc::new(AtomicU64::new(0)), + last_rx_progress_ms: Arc::new(AtomicU64::new(0)), peer_height: 10, connected_ms: 1, first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive: true, task, } diff --git a/crates/rbitcoin-net/src/ibd/peer_io.rs b/crates/rbitcoin-net/src/ibd/peer_io.rs index 1193092e..0a82c1cb 100644 --- a/crates/rbitcoin-net/src/ibd/peer_io.rs +++ b/crates/rbitcoin-net/src/ibd/peer_io.rs @@ -93,7 +93,15 @@ pub(crate) struct PeerSlot { /// Hashes currently requested from this peer. pub in_flight: HashSet, /// Last block-download progress as [`ibd_mono_ms`]. + /// + /// Touched on getdata issue (empty in-flight), stream bytes, block, notfound. + /// Stall-disconnect uses this clock. pub block_progress_ms: Arc, + /// Last **receive** progress as [`ibd_mono_ms`] (stream 64 KiB / block / notfound). + /// + /// Not touched when issuing getdata. `0` = no rx yet. Tip-hole owner eviction + /// uses this so a slow download is not treated as a hung racer. + pub last_rx_progress_ms: Arc, /// Peer's `version.start_height` (best-effort network tip signal). pub peer_height: u32, /// Mono ms when the slot became live (post-handshake). @@ -102,6 +110,10 @@ pub(crate) struct PeerSlot { pub first_data_ms: AtomicU64, /// Cumulative block payload bytes (speed sample). pub bytes_rx: AtomicU64, + /// Main-thread recent-bps window origin ([`ibd_mono_ms`]). `0` = unset. + pub window_start_ms: AtomicU64, + /// `bytes_rx` at [`window_start_ms`]. + pub window_start_bytes: AtomicU64, pub alive: bool, pub task: JoinHandle<()>, } @@ -134,6 +146,58 @@ impl PeerSlot { let bps = bytes.saturating_mul(1000) / elapsed_ms; Some((latency_ms, bps)) } + + /// Rotate the recent-bps window at [`RX_BPS_WINDOW_MS`]. IBD main thread only. + pub fn snapshot_rx(&self) { + self.snapshot_rx_at(ibd_mono_ms()); + } + + pub(crate) fn snapshot_rx_at(&self, now: u64) { + let now = now.max(1); + let bytes = self.bytes_rx.load(Ordering::Relaxed); + let start = self.window_start_ms.load(Ordering::Relaxed); + if start == 0 { + self.window_start_ms.store(now, Ordering::Relaxed); + self.window_start_bytes.store(bytes, Ordering::Relaxed); + return; + } + if now.saturating_sub(start) >= RX_BPS_WINDOW_MS { + self.window_start_ms.store(now, Ordering::Relaxed); + self.window_start_bytes.store(bytes, Ordering::Relaxed); + } + } + + /// Complete-block bytes/sec over the current ~60s window. + /// + /// Unset or too-young windows fall back to lifetime [`speed_sample`] bps. + pub fn recent_bps(&self) -> Option { + self.recent_bps_at(ibd_mono_ms()) + } + + pub(crate) fn recent_bps_at(&self, now: u64) -> Option { + let start = self.window_start_ms.load(Ordering::Relaxed); + if start == 0 { + return self.speed_sample().map(|(_, b)| b); + } + let elapsed = now.saturating_sub(start).max(1); + if elapsed < RX_BPS_MIN_ELAPSED_MS { + return self.speed_sample().map(|(_, b)| b); + } + let bytes = self.bytes_rx.load(Ordering::Relaxed); + let start_b = self.window_start_bytes.load(Ordering::Relaxed); + let delta = bytes.saturating_sub(start_b); + Some(delta.saturating_mul(1000) / elapsed) + } +} + +/// Recent complete-block ranking window (ms). +pub(crate) const RX_BPS_WINDOW_MS: u64 = 60_000; +const RX_BPS_MIN_ELAPSED_MS: u64 = 1_000; + +pub(crate) fn snapshot_peer_rx(slots: &[PeerSlot]) { + for s in slots { + s.snapshot_rx(); + } } impl Drop for PeerSlot { @@ -156,12 +220,14 @@ pub(crate) fn touch_block_progress(ms: &AtomicU64) { pub(crate) fn note_block_progress(slots: &mut [PeerSlot], peer: usize) { if let Some(s) = slots.iter_mut().find(|s| s.id == peer) { touch_block_progress(&s.block_progress_ms); + touch_block_progress(&s.last_rx_progress_ms); } } pub(crate) fn note_block_rx(slots: &mut [PeerSlot], peer: usize, wire_bytes: usize) { if let Some(s) = slots.iter_mut().find(|s| s.id == peer) { touch_block_progress(&s.block_progress_ms); + touch_block_progress(&s.last_rx_progress_ms); s.note_rx_bytes(wire_bytes as u64); } } @@ -196,7 +262,9 @@ pub(crate) async fn spawn_peer( // stall the receive half and look like a peer stall). let (out_tx, mut out_rx) = mpsc::unbounded_channel::(); let block_progress_ms = Arc::new(AtomicU64::new(ibd_mono_ms())); + let last_rx_progress_ms = Arc::new(AtomicU64::new(0)); let progress_io = Arc::clone(&block_progress_ms); + let rx_io = Arc::clone(&last_rx_progress_ms); // Parent owns concurrent read + write tasks. Aborting the parent (PeerSlot // Drop / stall disconnect) must abort both children — plain JoinHandle drop @@ -228,6 +296,7 @@ pub(crate) async fn spawn_peer( if buffered >= prog_mark + STEP || buffered <= STEP { prog_mark = buffered; touch_block_progress(&progress_io); + touch_block_progress(&rx_io); } }) .await; @@ -243,6 +312,7 @@ pub(crate) async fn spawn_peer( if frame.is_block() || frame.is_notfound() { touch_block_progress(&progress_io); + touch_block_progress(&rx_io); } if frame.is_block() { @@ -268,6 +338,7 @@ pub(crate) async fn spawn_peer( } let progress = Arc::clone(&progress_io); + let rx = Arc::clone(&rx_io); let sinks_d = sinks_r.clone(); // Non-block: decode off-thread. Never await a decode permit // on the reader (stalls TCP). Soft budgets gate *requests* only. @@ -283,6 +354,7 @@ pub(crate) async fn spawn_peer( } NetworkMessage::NotFound(inv) => { touch_block_progress(&progress); + touch_block_progress(&rx); let hashes: Vec = inv .iter() .filter_map(|i| match i { @@ -448,10 +520,13 @@ pub(crate) async fn spawn_peer( cmd_tx, in_flight: HashSet::new(), block_progress_ms, + last_rx_progress_ms, peer_height, connected_ms: ibd_mono_ms(), first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive: true, task, }) @@ -529,10 +604,13 @@ mod tests { cmd_tx, in_flight: HashSet::new(), block_progress_ms: Arc::new(AtomicU64::new(0)), + last_rx_progress_ms: Arc::new(AtomicU64::new(0)), peer_height: 100, connected_ms: 1, first_data_ms: AtomicU64::new(0), bytes_rx: AtomicU64::new(0), + window_start_ms: AtomicU64::new(0), + window_start_bytes: AtomicU64::new(0), alive: true, task, } @@ -592,6 +670,36 @@ mod tests { assert!(ibd_mono_ms() > 0); } + #[test] + fn recent_bps_prefers_current_window_over_lifetime() { + let mut idle = dummy_slot(0); + idle.connected_ms = 1; + idle.first_data_ms.store(1, Ordering::Relaxed); + idle.bytes_rx.store(10_000_000, Ordering::Relaxed); + idle.window_start_ms.store(115_000, Ordering::Relaxed); + idle.window_start_bytes.store(10_000_000, Ordering::Relaxed); + assert_eq!(idle.recent_bps_at(120_000), Some(0)); + assert!( + idle.speed_sample().expect("lifetime").1 > 0, + "AddrMan lifetime sample must remain" + ); + + let mut live = dummy_slot(1); + live.connected_ms = 110_000; + live.first_data_ms.store(114_000, Ordering::Relaxed); + live.bytes_rx.store(500_000, Ordering::Relaxed); + live.window_start_ms.store(115_000, Ordering::Relaxed); + live.window_start_bytes.store(0, Ordering::Relaxed); + assert_eq!(live.recent_bps_at(120_000), Some(100_000)); + + idle.snapshot_rx_at(115_000 + RX_BPS_WINDOW_MS); + assert_eq!( + idle.window_start_ms.load(Ordering::Relaxed), + 115_000 + RX_BPS_WINDOW_MS + ); + assert_eq!(idle.window_start_bytes.load(Ordering::Relaxed), 10_000_000); + } + #[test] fn event_sinks_send_body_and_ctrl() { let (body_tx, mut body_rx) = mpsc::unbounded_channel(); diff --git a/docs/ibd-memory.md b/docs/ibd-memory.md index 024f3c92..1a124844 100644 --- a/docs/ibd-memory.md +++ b/docs/ibd-memory.md @@ -82,7 +82,8 @@ is over target. window, outstanding requests remain finite (per-peer in-flight window). Enqueueing those bodies cannot create a truly unbounded leak; the backlog drains as confirm dequeues. Bound queue size by **not requesting**, not by -**not reading**. +**not reading**. A tight slow pack (local bottleneck) keeps 8 densify getdata +per peer and is not stolen or relative-slow disconnected. Historical regression (do not reintroduce): bounded arch_job Full-drop and reader-side decode-permit wait before the next frame made peers look dead while