diff --git a/changelog.d/ibd-stall-recovery.md b/changelog.d/ibd-stall-recovery.md new file mode 100644 index 0000000000..a413b2a8b2 --- /dev/null +++ b/changelog.d/ibd-stall-recovery.md @@ -0,0 +1,6 @@ +Fixed + +- **Header-cache cleanup keeps the height and parent of stored path headers still waiting for the IBD download queue.** The queue can refill from them. +- **A new lookahead header batch does not jump ahead of stored headers waiting for queue space.** Those slots stay for the next contiguous headers. +- **Only an empty reconnect round with no live IBD peer spends the reconnect budget.** A replacement connection that already started is allowed to finish. A spent budget still stops IBD without entering tip mode. +- **A follow-session peer that goes over the message or byte rate budget is disconnected on that frame.** Its unanswered block asks are released however the session ends, including `disconnectnode` and eviction, so another peer can be asked right away. A second rate-limit disconnect from the same address within an hour refuses that address for a day. Noban, manual, and local peers leave no strike. diff --git a/crates/rbitcoin-net/src/ibd/events/mod.rs b/crates/rbitcoin-net/src/ibd/events/mod.rs index fe4c1ab928..b22d0b6d2b 100644 --- a/crates/rbitcoin-net/src/ibd/events/mod.rs +++ b/crates/rbitcoin-net/src/ibd/events/mod.rs @@ -213,6 +213,9 @@ fn try_enqueue_ordered_header( if !st.is_on_path(&hash, ht) || st.ordered.len() >= MAX_ORDERED_HEADERS { return false; } + if super::path::next_path_height(st, hub).is_some_and(|next| ht > next) { + return false; + } // The queue is a window over headers already written. Stop at the soft cap. if st.header_walk.has_checkpoints() && st.ordered.len() >= ORDERED_HEADERS_SOFT_CAP { return false; diff --git a/crates/rbitcoin-net/src/ibd/exit.rs b/crates/rbitcoin-net/src/ibd/exit.rs index 9dfcd22868..a8ed367947 100644 --- a/crates/rbitcoin-net/src/ibd/exit.rs +++ b/crates/rbitcoin-net/src/ibd/exit.rs @@ -6,7 +6,7 @@ use super::state::IbdWorkState; /// Max empty redial rounds while fully dark before giving up mid catch-up. -pub const MAX_DARK_EMPTY_REDIALS: u32 = 4; +const MAX_DARK_EMPTY_REDIALS: u32 = 4; /// Highest connected best-chain height we already occupy (not competing `hash_height`). pub(crate) fn path_high_water(st: &IbdWorkState, tip_h: u32) -> u32 { @@ -173,17 +173,32 @@ pub enum AllPeersDead { GiveUpMidCatchup, } +/// Empty redial rounds in a row with no live peer. A round that adds a peer, +/// or an empty top-up while a peer is still up, starts the count over. +pub(crate) fn dark_empty_after(dark: u32, added: usize, any_alive: bool) -> u32 { + if added > 0 || any_alive { + 0 + } else { + dark.saturating_add(1) + } +} + +/// A live peer can always be topped up. With none, dialing stops after +/// [`MAX_DARK_EMPTY_REDIALS`] dark empty rounds. +pub(crate) fn may_spawn_redial(alive_n: usize, dark: u32) -> bool { + alive_n > 0 || dark < MAX_DARK_EMPTY_REDIALS +} + /// Decide what to do when `slots` has no live peers. pub fn all_peers_dead_action( st: &IbdWorkState, tip_h: u32, redial_in_flight: bool, - dark_redial_empty: u32, ) -> AllPeersDead { if ibd_caught_up(st, tip_h) { return AllPeersDead::CatchupComplete; } - if !redial_in_flight || dark_redial_empty >= MAX_DARK_EMPTY_REDIALS { + if !redial_in_flight { return AllPeersDead::GiveUpMidCatchup; } AllPeersDead::WaitRedial @@ -193,6 +208,39 @@ pub fn all_peers_dead_action( mod tests { use super::*; + #[test] + fn redial_budget_spends_only_dark_empty_rounds_and_waits_out_a_started_dial() { + assert_eq!( + dark_empty_after(MAX_DARK_EMPTY_REDIALS - 1, 0, true), + 0, + "an empty top-up while a peer is up does not spend the dark budget" + ); + assert_eq!(dark_empty_after(MAX_DARK_EMPTY_REDIALS, 1, false), 0); + + let mut dark = 0; + for _ in 0..MAX_DARK_EMPTY_REDIALS { + assert!(may_spawn_redial(0, dark)); + dark = dark_empty_after(dark, 0, false); + } + assert!( + !may_spawn_redial(0, dark), + "dark empty rounds stop the next dial" + ); + assert!(may_spawn_redial(1, dark), "a live peer still tops up"); + + let mut st = IbdWorkState::new(Vec::new(), None, Some(21_792)); + st.max_peer_height = 326_676; + assert_eq!( + all_peers_dead_action(&st, 21_792, true), + AllPeersDead::WaitRedial, + "a dial already started is waited out" + ); + assert_eq!( + all_peers_dead_action(&st, 21_792, false), + AllPeersDead::GiveUpMidCatchup + ); + } + /// all_peers_dead + catchup_complete edge matrix (mid-chain / caught-up / tip-0). #[allow(clippy::cognitive_complexity)] // one fixture, many exit arms #[test] @@ -202,11 +250,11 @@ mod tests { mid.max_peer_height = 958_820; mid.max_ready_height = 161_000; assert_eq!( - all_peers_dead_action(&mid, 161_249, false, 0), + all_peers_dead_action(&mid, 161_249, false), AllPeersDead::GiveUpMidCatchup ); assert_eq!( - all_peers_dead_action(&mid, 161_249, true, 0), + all_peers_dead_action(&mid, 161_249, true), AllPeersDead::WaitRedial ); @@ -216,7 +264,7 @@ mod tests { done.max_ready_height = 100; done.headers_done = true; assert_eq!( - all_peers_dead_action(&done, 100, false, 0), + all_peers_dead_action(&done, 100, false), AllPeersDead::CatchupComplete ); @@ -239,13 +287,7 @@ mod tests { assert!(!ibd_caught_up(&zero, 0)); assert!(!ibd_caught_up(&zero, 0)); assert_eq!( - all_peers_dead_action(&zero, 0, false, 0), - AllPeersDead::GiveUpMidCatchup - ); - - // Dark redial budget exhausted while redial still marked in-flight. - assert_eq!( - all_peers_dead_action(&mid, 161_249, true, MAX_DARK_EMPTY_REDIALS), + all_peers_dead_action(&zero, 0, false), AllPeersDead::GiveUpMidCatchup ); @@ -487,7 +529,7 @@ mod tests { zero.max_ready_height = 0; assert!(!ibd_caught_up(&zero, 0)); assert_eq!( - all_peers_dead_action(&zero, 0, false, 0), + all_peers_dead_action(&zero, 0, false), AllPeersDead::GiveUpMidCatchup ); @@ -497,7 +539,7 @@ mod tests { done.max_ready_height = 100; done.headers_done = true; assert_eq!( - all_peers_dead_action(&done, 100, false, 0), + all_peers_dead_action(&done, 100, false), AllPeersDead::CatchupComplete ); diff --git a/crates/rbitcoin-net/src/ibd/header_walk.rs b/crates/rbitcoin-net/src/ibd/header_walk.rs index e26fd8ff9b..06817f77a5 100644 --- a/crates/rbitcoin-net/src/ibd/header_walk.rs +++ b/crates/rbitcoin-net/src/ibd/header_walk.rs @@ -5752,6 +5752,66 @@ mod tests { } } + #[test] + fn new_lookahead_batch_does_not_skip_stored_queue_successors() { + let (_dir, mut hub) = crate::chain::tiny_regtest_hub_labeled("header-walk-no-queue-gap"); + hub.ensure_genesis().unwrap(); + let genesis = hub.tip_hash().unwrap(); + floor_unreachable(&mut hub); + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], Some(genesis), Some(0)); + let tail = mine(genesis, 1); + queue_at_cap_ending_on(&mut st, &hub, tail); + assert!(send_getheaders(&mut st, &hub).unwrap()); + let next = mine(tail.block_hash(), 2); + let third = mine(next.block_hash(), 3); + apply(&mut st, &hub, 0, vec![next, third]); + for _ in 0..2 { + let dropped = st.ordered.pop_front().unwrap(); + st.ordered_set.remove(&dropped); + } + let fourth = mine(third.block_hash(), 4); + apply(&mut st, &hub, 0, vec![fourth]); + super::super::path::append_stored_headers(&mut st, &hub); + let queued: Vec<_> = st.ordered.iter().rev().take(2).copied().collect(); + assert_eq!( + queued, + vec![third.block_hash(), next.block_hash()], + "new lookahead must leave the free queue slots for already stored successors" + ); + assert_eq!(st.header_walk.tip_hash(), Some(fourth.block_hash())); + } + + #[test] + fn drained_queue_refills_stored_headers_after_known_header_pruning() { + let (_dir, mut hub) = crate::chain::tiny_regtest_hub_labeled("header-walk-pruned-refill"); + hub.ensure_genesis().unwrap(); + let genesis = hub.tip_hash().unwrap(); + floor_unreachable(&mut hub); + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], Some(genesis), Some(0)); + let tail = mine(genesis, 1); + queue_at_cap_ending_on(&mut st, &hub, tail); + st.known_headers.extend(st.ordered.iter().copied()); + assert!(send_getheaders(&mut st, &hub).unwrap()); + let next = mine(tail.block_hash(), 2); + let last = mine(next.block_hash(), 3); + apply(&mut st, &hub, 0, vec![next, last]); + assert_eq!(st.header_walk.tip_hash(), Some(last.block_hash())); + while st.ordered.len() > 1 { + let dropped = st.ordered.pop_front().unwrap(); + st.ordered_set.remove(&dropped); + } + assert_eq!(st.ordered.back(), Some(&tail.block_hash())); + st.hygiene(); + assert_eq!( + super::super::path::append_stored_headers(&mut st, &hub), + 2, + "both stored successors must remain available to the download queue after pruning" + ); + assert_eq!(path_top(&st, &hub), Some((last.block_hash(), 3))); + } + #[test] fn a_heavier_fork_below_the_floor_is_written_when_the_queue_has_room() { let (_dir, mut hub) = crate::chain::tiny_regtest_hub_labeled("header-walk-stray-write"); diff --git a/crates/rbitcoin-net/src/ibd/mod.rs b/crates/rbitcoin-net/src/ibd/mod.rs index 786ffd83c6..9f376c5472 100644 --- a/crates/rbitcoin-net/src/ibd/mod.rs +++ b/crates/rbitcoin-net/src/ibd/mod.rs @@ -57,8 +57,9 @@ use events::{apply_confirm_events, apply_peer_event, disconnect_all_peers, updat #[cfg(test)] pub(crate) use events::{apply_confirm_reject, apply_headers_event}; use exit::{ - all_peers_dead_action, best_chain_remainder, empty_path_header_fan, header_lag_behind_peers, - ibd_caught_up, path_drained, should_unlatch_headers_done, AllPeersDead, + all_peers_dead_action, best_chain_remainder, dark_empty_after, empty_path_header_fan, + header_lag_behind_peers, ibd_caught_up, may_spawn_redial, path_drained, + should_unlatch_headers_done, AllPeersDead, }; #[cfg(test)] pub(crate) use header_walk::{ @@ -785,18 +786,18 @@ pub async fn ibd_cancellable( st.slots.push(s); n += 1; } + dark_redial_empty = + dark_empty_after(dark_redial_empty, n, st.slots.iter().any(|s| s.alive)); if n > 0 { st.slots.sort_by_key(|s| s.id); info!( "ibd: redial added {n} peer(s); live={}", st.slots.iter().filter(|s| s.alive).count() ); - dark_redial_empty = 0; if st.ordered.is_empty() && !st.headers_done { enqueue_header_cmd(st, HeaderCmd::Ask); } } else { - dark_redial_empty = dark_redial_empty.saturating_add(1); warn!("ibd: redial returned 0 peers (empty_rounds={dark_redial_empty})"); } } @@ -818,6 +819,7 @@ pub async fn ibd_cancellable( }; if redial_handle.is_none() && alive_n < target + && may_spawn_redial(alive_n, dark_redial_empty) && !peer_sess.book().is_empty() && last_redial.elapsed() >= redial_interval { @@ -980,7 +982,7 @@ pub async fn ibd_cancellable( // All peers dead — never treat mid-chain peer death as catch-up complete. if st.slots.iter().all(|s| !s.alive) { let tip_h = hub.tip_height().unwrap_or(0); - match all_peers_dead_action(st, tip_h, redial_handle.is_some(), dark_redial_empty) { + match all_peers_dead_action(st, tip_h, redial_handle.is_some()) { AllPeersDead::CatchupComplete => { info!( "ibd: catch-up complete (no live peers) tip={tip_h} max_peer_height={} max_ready={} — exiting IBD", diff --git a/crates/rbitcoin-net/src/ibd/state.rs b/crates/rbitcoin-net/src/ibd/state.rs index 73b09e50ed..b20e280558 100644 --- a/crates/rbitcoin-net/src/ibd/state.rs +++ b/crates/rbitcoin-net/src/ibd/state.rs @@ -437,8 +437,14 @@ impl IbdWorkState { || (*ht > queue_tail && self.header_walk.indexes_through(*ht))) }); if self.known_headers.len() > live.len().saturating_add(4096) { - self.known_headers - .retain(|h| live.contains(h) || inflight.contains_key(h)); + self.known_headers.retain(|h| { + live.contains(h) + || inflight.contains_key(h) + || self + .hash_height + .get(h) + .is_some_and(|ht| self.height_to_hash.get(ht) == Some(h)) + }); self.header_fks.retain(|h, _| { live.contains(h) || inflight.contains_key(h) || self.known_headers.contains(h) }); diff --git a/crates/rbitcoin-net/src/peer.rs b/crates/rbitcoin-net/src/peer.rs index 7c9ae5c26f..e94c6c2fe4 100644 --- a/crates/rbitcoin-net/src/peer.rs +++ b/crates/rbitcoin-net/src/peer.rs @@ -145,18 +145,24 @@ async fn accept_received_from_peer( }) } -/// One-day in-memory refusal. Noban and manual peers are not recorded. -/// A loopback address is every local connection, so Core disconnects that -/// peer and does not discourage the address. -fn note_threshold_refusal(session: Option<&crate::peers::LivePeer>) { - let Some(s) = session.filter(|s| !misbehavior_exempt(s)) else { - return; - }; - if local_addr_skips_discourage(s.addr.ip()) { - return; +/// The hub and address a misbehavior exit may record. Noban and manual peers +/// are not recorded. A loopback address is every local connection, so Core +/// disconnects that peer and does not discourage the address. +fn refusable_addr( + session: Option<&crate::peers::LivePeer>, +) -> Option<(Arc, std::net::IpAddr)> { + let s = session.filter(|s| !misbehavior_exempt(s))?; + let ip = s.addr.ip(); + if local_addr_skips_discourage(ip) { + return None; } - if let Some(hub) = s.peer_hub() { - hub.note_misbehavior_addr(s.addr.ip()); + Some((s.peer_hub()?, ip)) +} + +/// One-day in-memory refusal. +fn note_threshold_refusal(session: Option<&crate::peers::LivePeer>) { + if let Some((hub, ip)) = refusable_addr(session) { + hub.note_misbehavior_addr(ip); } } @@ -174,6 +180,16 @@ fn threshold_disconnect(session: Option<&crate::peers::LivePeer>) -> NetError { NetError::Protocol("peer misbehavior threshold") } +/// The first rate-limit exit from a recordable address leaves a strike. A +/// second one within [`crate::peers::PeerHub::RATE_STRIKE_SECS`] is a +/// threshold exit, as two overflows in one session were. +fn rate_limit_disconnect(session: Option<&crate::peers::LivePeer>) -> NetError { + if refusable_addr(session).is_some_and(|(hub, ip)| hub.note_rate_strike(ip)) { + return threshold_disconnect(session); + } + NetError::Protocol("peer rate limit") +} + /// Core `MaybeDiscourageAndDisconnect`: misbehavior never disconnects or /// discourages a noban peer or a manual one (`--connect`, `addnode`). fn misbehavior_exempt(s: &crate::peers::LivePeer) -> bool { @@ -1668,7 +1684,11 @@ pub async fn peer_session_with( rbitcoin_log::warn!("p2p: {peer_s} initial getheaders queue failed: {e}"); } - let mut follow = PeerFollowState::new(); + let mut owned_follow = SessionFollow { + hub: hub.clone(), + state: PeerFollowState::new(), + }; + let follow = &mut owned_follow.state; if let Some(s) = meta.session.as_ref() { follow.wtxid_relay = s.wtxid_relay(); } @@ -1788,7 +1808,8 @@ pub async fn peer_session_with( if follow.ban_score >= BAN_SCORE_THRESHOLD { return Err(threshold_disconnect(session.as_deref())); } - continue; + // Close to free this session's asks. A second strike refuses the address. + return Err(rate_limit_disconnect(session.as_deref())); } // Ping/pong: cheap 8-byte path — never leave the I/O task for decode. if frame.is_ping() { @@ -1810,7 +1831,7 @@ pub async fn peer_session_with( frame, hub.as_ref(), &out_tx, - &mut follow, + follow, session.as_deref(), ) .await?; @@ -1851,7 +1872,7 @@ pub async fn peer_session_with( } else if on_tip_event( hub.as_ref(), &out_tx, - &mut follow, + follow, session.as_deref(), tip, ) @@ -1871,7 +1892,7 @@ pub async fn peer_session_with( if on_tip_event( hub.as_ref(), &out_tx, - &mut follow, + follow, session.as_deref(), tip, ) @@ -1895,7 +1916,7 @@ pub async fn peer_session_with( on_tx_announce( hub.as_ref(), &out_tx, - &follow, + follow, session.as_deref(), ann, )?; @@ -1911,7 +1932,7 @@ pub async fn peer_session_with( on_inv_flush( hub.as_ref(), &out_tx, - &follow, + follow, session.as_deref(), flush, ); @@ -1921,7 +1942,7 @@ pub async fn peer_session_with( on_heartbeat( hub.as_ref(), &out_tx, - &mut follow, + follow, session.as_deref(), &mut requested_since, &mut tx_announce_rx, @@ -1948,14 +1969,14 @@ pub async fn peer_session_with( return Ok(()); } if !follow.getdata_tail.is_empty() { - serve_getdata_tail(hub.as_ref(), &out_tx, &mut follow, session.as_deref()) + serve_getdata_tail(hub.as_ref(), &out_tx, follow, session.as_deref()) .await?; } if !follow.getblocktxn_tail.is_empty() { serve_getblocktxn_tail( hub.as_ref(), &out_tx, - &mut follow, + follow, session.as_deref(), ) .await?; @@ -2582,6 +2603,23 @@ struct PeerFollowState { getblocktxn_tail: VecDeque, } +/// A live session's follow state. Dropping it, including when +/// `PeerHub::disconnect_id` aborts the session task, releases the body asks +/// the peer never answered: `asked_blocks` is hub-wide, so a held ask would +/// keep every other peer from fetching that body. +struct SessionFollow { + hub: Arc, + state: PeerFollowState, +} + +impl Drop for SessionFollow { + fn drop(&mut self) { + for h in self.state.requested_blocks.drain() { + self.hub.forget_asked_block(&h); + } + } +} + impl PeerFollowState { fn new() -> Self { Self { diff --git a/crates/rbitcoin-net/src/peer_dos.rs b/crates/rbitcoin-net/src/peer_dos.rs index fdbc85dcc7..8e6b07de10 100644 --- a/crates/rbitcoin-net/src/peer_dos.rs +++ b/crates/rbitcoin-net/src/peer_dos.rs @@ -21,7 +21,10 @@ pub const DEFAULT_MAX_MSGS_PER_SEC: u32 = 4_000; /// Max framed payload bytes per peer per window (BIP324 contents size). /// ~16 MiB/s: enough for concurrent block + tx relay; still caps multi-peer floods. pub const DEFAULT_MAX_BYTES_PER_SEC: u64 = 16_000_000; -/// Disconnect score added when a peer exceeds rate limits (disconnect at 100). +/// Misbehavior score added when a peer exceeds rate limits. Decoy and +/// unknown-type frames disconnect at 100. An application message disconnects +/// on its first overflow; a repeat from that address within an hour is a +/// threshold exit. pub const RATE_LIMIT_BAN_SCORE: u32 = 50; /// Disconnect score for oversized protocol messages already rejected as MessageTooLarge. pub const OVERSIZE_BAN_SCORE: u32 = 100; diff --git a/crates/rbitcoin-net/src/peer_tests.rs b/crates/rbitcoin-net/src/peer_tests.rs index 48739252e7..c89d22fc3c 100644 --- a/crates/rbitcoin-net/src/peer_tests.rs +++ b/crates/rbitcoin-net/src/peer_tests.rs @@ -2990,6 +2990,105 @@ async fn write_peer_msg(raw: &mut V2PlainSession, msg: NetworkMessage) { .unwrap(); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn rate_limited_block_request_disconnects_without_waiting_for_another_message() { + let mut t = paused_serve("rate-request", 1, &["/rbitcoin:test(rate-request)/"]).await; + let (raw, _) = &mut t.clients[0]; + // The handshake is outside the rate window; send no follow-session replies. + for _ in 0..crate::peer_dos::DEFAULT_MAX_MSGS_PER_SEC { + write_peer_msg(raw, NetworkMessage::Ping(7)).await; + } + write_peer_msg(raw, getdata_msg(&t.hashes)).await; + let closed = tokio::time::timeout(Duration::from_secs(2), async { + loop { + match next_peer_msg(raw).await { + Err(_) => return true, + Ok(NetworkMessage::Block(_)) => return false, + Ok(_) => {} + } + } + }) + .await; + raw.close(); + t.node.shutdown().await; + let _ = std::fs::remove_dir_all(&t.dir); + assert!( + matches!(closed, Ok(true)), + "an over-budget block request must close the session, not disappear: {closed:?}" + ); +} + +/// A follow session that has asked for one block body it will never get. +async fn serve_with_open_ask(label: &str) -> (PausedServe, BlockHash) { + let mut t = paused_serve(label, 1, &["/rbitcoin:test(open-ask)/"]).await; + let tip = t.node.hub.tip_header().unwrap(); + let mut child = bitcoin::block::Header { + version: bitcoin::block::Version::from_consensus(4), + prev_blockhash: tip.block_hash(), + merkle_root: bitcoin::TxMerkleNode::from_byte_array([0x33; 32]), + time: tip.time + 1, + bits: tip.bits, + nonce: 0, + }; + rbitcoin_consensus::grind_regtest_pow(&mut child); + let hash = child.block_hash(); + let (raw, _) = &mut t.clients[0]; + write_peer_msg(raw, NetworkMessage::Headers(vec![child])).await; + let asked = tokio::time::timeout(Duration::from_secs(2), async { + loop { + if let NetworkMessage::GetData(inv) = next_peer_msg(raw).await.unwrap() { + if inv.iter().any(|i| { + matches!(i, Inventory::WitnessBlock(h) | Inventory::CompactBlock(h) if *h == hash) + }) { + return; + } + } + } + }) + .await; + assert!(asked.is_ok(), "the header announce must ask for its body"); + assert!(t.node.hub.already_have_or_asked_block(&hash)); + (t, hash) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn closed_follow_session_releases_its_block_asks() { + let (mut t, hash) = serve_with_open_ask("closed-asks").await; + t.clients[0].0.close(); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while !t.node.peers.live_peers().is_empty() { + assert!(std::time::Instant::now() < deadline, "closed session"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + let still_asked = t.node.hub.already_have_or_asked_block(&hash); + t.node.shutdown().await; + let _ = std::fs::remove_dir_all(&t.dir); + assert!( + !still_asked, + "a disconnect must release the session's body asks for other peers" + ); +} + +/// `disconnectnode`, inbound eviction, and stale-tip replacement abort the +/// session task instead of letting it return. +#[tokio::test] +async fn aborted_follow_session_releases_its_block_asks() { + let (mut t, hash) = serve_with_open_ask("abort-asks").await; + assert!(t.node.peers.disconnect_id(t.clients[0].1.id)); + let deadline = std::time::Instant::now() + Duration::from_secs(2); + while t.node.hub.already_have_or_asked_block(&hash) && std::time::Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(10)).await; + } + let still_asked = t.node.hub.already_have_or_asked_block(&hash); + t.clients[0].0.close(); + t.node.shutdown().await; + let _ = std::fs::remove_dir_all(&t.dir); + assert!( + !still_asked, + "an aborted session must release its body asks for other peers" + ); +} + /// Unknown BIP324 short id on a live follow session counts as `*other*` /// and the peer stays up (Core `test_msgtype`). #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -4021,6 +4120,106 @@ fn threshold_exit_refuses_the_address_without_punish_disconnect() { ); } +#[test] +fn second_rate_limit_disconnect_from_an_address_discourages_it() { + let peers = crate::peers::PeerHub::new(); + let now = 1_700_000_200u64; + peers.set_mock_now(now); + let ver = bitcoin::p2p::message_network::VersionMessage { + version: 70016, + services: bitcoin::p2p::ServiceFlags::NETWORK, + timestamp: 0, + receiver: bitcoin::p2p::address::Address::new( + &std::net::SocketAddr::from(([0, 0, 0, 0], 0)), + bitcoin::p2p::ServiceFlags::NONE, + ), + sender: bitcoin::p2p::address::Address::new( + &std::net::SocketAddr::from(([0, 0, 0, 0], 0)), + bitcoin::p2p::ServiceFlags::NONE, + ), + nonce: 5, + user_agent: "/rbitcoin:test/".into(), + start_height: 0, + relay: true, + }; + let flood = |addr: std::net::SocketAddr| { + let peer = peers.register(addr, addr, &ver, true, crate::peers::PeerConnType::Inbound); + rate_limit_disconnect(Some(peer.as_ref())) + }; + + let first = std::net::SocketAddr::from(([7, 7, 7, 7], 8333)); + assert!(matches!( + flood(first), + crate::error::NetError::Protocol("peer rate limit") + )); + assert!( + !peers.inbound_discouraged(first), + "one rate-limit disconnect does not refuse the address" + ); + assert!(matches!( + flood(std::net::SocketAddr::from(([7, 7, 7, 7], 9999))), + crate::error::NetError::Protocol("peer misbehavior threshold") + )); + assert!( + peers.inbound_discouraged(first), + "a second rate-limit disconnect from that address refuses it" + ); + + let slow = std::net::SocketAddr::from(([7, 7, 7, 8], 8333)); + let _ = flood(slow); + peers.set_mock_now(now + crate::peers::PeerHub::RATE_STRIKE_SECS); + let _ = flood(slow); + assert!( + !peers.inbound_discouraged(slow), + "a strike older than the window does not count" + ); +} + +#[test] +fn exempt_peer_rate_limit_leaves_no_strike_on_its_address() { + let peers = crate::peers::PeerHub::new(); + peers.set_mock_now(1_700_000_300); + let ver = bitcoin::p2p::message_network::VersionMessage { + version: 70016, + services: bitcoin::p2p::ServiceFlags::NETWORK, + timestamp: 0, + receiver: bitcoin::p2p::address::Address::new( + &std::net::SocketAddr::from(([0, 0, 0, 0], 0)), + bitcoin::p2p::ServiceFlags::NONE, + ), + sender: bitcoin::p2p::address::Address::new( + &std::net::SocketAddr::from(([0, 0, 0, 0], 0)), + bitcoin::p2p::ServiceFlags::NONE, + ), + nonce: 6, + user_agent: "/rbitcoin:test/".into(), + start_height: 0, + relay: true, + }; + let flood = |addr: std::net::SocketAddr, conn: crate::peers::PeerConnType| { + let peer = peers.register(addr, addr, &ver, false, conn); + rate_limit_disconnect(Some(peer.as_ref())) + }; + + let manual = std::net::SocketAddr::from(([6, 6, 6, 6], 8333)); + let _ = flood(manual, crate::peers::PeerConnType::Manual); + let noban = std::net::SocketAddr::from(([6, 6, 6, 7], 8333)); + peers.set_noban(true); + let _ = flood(noban, crate::peers::PeerConnType::Inbound); + peers.set_noban(false); + for exempt in [manual, noban] { + let again = std::net::SocketAddr::from((exempt.ip(), 9999)); + assert!( + matches!( + flood(again, crate::peers::PeerConnType::Inbound), + crate::error::NetError::Protocol("peer rate limit") + ), + "{exempt}: an exempt overflow leaves no strike for the next peer" + ); + assert!(!peers.inbound_discouraged(again)); + } +} + #[test] fn evicted_netgroup_waits_less_than_a_day_and_the_set_is_capped() { let peers = crate::peers::PeerHub::new(); diff --git a/crates/rbitcoin-net/src/peers.rs b/crates/rbitcoin-net/src/peers.rs index db524ac932..b7636defd7 100644 --- a/crates/rbitcoin-net/src/peers.rs +++ b/crates/rbitcoin-net/src/peers.rs @@ -1449,6 +1449,8 @@ pub struct PeerHub { discouraged_addrs: Mutex>, /// Netgroup → unix second a slot-loss refusal ends. discouraged_groups: Mutex>, + /// Rate-limit disconnect address → unix second its strike ends. + rate_strikes: Mutex>, } fn canonical_bind(addr: SocketAddr) -> SocketAddr { @@ -1544,6 +1546,7 @@ impl PeerHub { net_perms: Mutex::new(crate::net_permissions::NetPermTable::default()), discouraged_addrs: Mutex::new(HashMap::new()), discouraged_groups: Mutex::new(HashMap::new()), + rate_strikes: Mutex::new(HashMap::new()), }) } @@ -2031,6 +2034,8 @@ impl PeerHub { pub(crate) const NETGROUP_SLOT_WAIT_SECS: u64 = 10 * 60; /// The in-memory set does not grow past this many addresses. pub(crate) const DISCOURAGE_CAP: usize = 10_000; + /// A rate-limit strike counts toward the next one for this long. + pub(crate) const RATE_STRIKE_SECS: u64 = 60 * 60; fn sweep_discouraged(&self, now: u64) { self.discouraged_addrs @@ -2058,6 +2063,23 @@ impl PeerHub { g.insert(ip, until); } + /// True when `ip` still has a strike from an earlier rate-limit + /// disconnect; that strike is spent. Otherwise records one for + /// [`Self::RATE_STRIKE_SECS`]. Expired strikes are dropped here. At + /// [`Self::DISCOURAGE_CAP`] a new address is not recorded. + pub(crate) fn note_rate_strike(&self, ip: IpAddr) -> bool { + let now = self.now_secs(); + let mut g = self.rate_strikes.lock().unwrap_or_else(|e| e.into_inner()); + g.retain(|_, until| *until > now); + if g.remove(&ip).is_some() { + return true; + } + if g.len() < Self::DISCOURAGE_CAP { + g.insert(ip, now.saturating_add(Self::RATE_STRIKE_SECS)); + } + false + } + /// The netgroup that just lost an inbound slot waits /// [`Self::NETGROUP_SLOT_WAIT_SECS`]. pub(crate) fn note_slot_evict(&self, group: u64) {