diff --git a/changelog.d/overnight-39004835061.md b/changelog.d/overnight-39004835061.md new file mode 100644 index 000000000..0e425d374 --- /dev/null +++ b/changelog.d/overnight-39004835061.md @@ -0,0 +1,3 @@ +Changed + +- Overnight missed mutants from 2026-10-17 are gone or pinned. The startup Class A "filled N" line is gone because the body-queue rehydrate already logs a non-zero fill. diff --git a/crates/rbitcoin-net/src/compact.rs b/crates/rbitcoin-net/src/compact.rs index e2ba6e895..dc8ec0fb9 100644 --- a/crates/rbitcoin-net/src/compact.rs +++ b/crates/rbitcoin-net/src/compact.rs @@ -192,9 +192,8 @@ pub(crate) fn reconstruct>( } *slot = Some(cands[0].borrow().clone()); } - Some(cands) if cands.len() > 1 => { - missing.push(abs as u64); - } + // One candidate fills the slot. Every other count, including a + // short-id collision, is missing for getblocktxn. _ => missing.push(abs as u64), } } @@ -1034,6 +1033,20 @@ mod tests { assert_eq!(classify_v2_cmpct_peer(&ping), CmpctPeerFrame::Ping(7)); let pong = crate::v2::encode_v2_contents(NetworkMessage::Pong(7)).unwrap(); assert_eq!(classify_v2_cmpct_peer(&pong), CmpctPeerFrame::Pong(7)); + let block_hash = BlockHash::from_byte_array([7; 32]); + let get = crate::v2::encode_v2_contents(NetworkMessage::GetBlockTxn( + bitcoin::p2p::message_compact_blocks::GetBlockTxn { + txs_request: bitcoin::bip152::BlockTransactionsRequest { + block_hash, + indexes: vec![1, 4], + }, + }, + )) + .unwrap(); + assert_eq!( + classify_v2_cmpct_peer(&get), + CmpctPeerFrame::GetBlockTxn(vec![1, 4]) + ); assert_eq!(classify_v2_cmpct_peer(&[]), CmpctPeerFrame::Other); } @@ -1262,7 +1275,13 @@ mod tests { assert_eq!(stats.fetched_n, 0); assert_eq!(stats.fetched_bytes, 0); assert_eq!(stats.mempool_n, 2); + assert_eq!( + stats.mempool_bytes, + tx_wire_len(&b1) + tx_wire_len(&b2), + "mempool fill bytes are the sum of the filled transactions" + ); assert_eq!(stats.prefill_n, 1); + assert_eq!(stats.prefill_bytes, tx_wire_len(&block.txdata[0])); assert!(!stats.getdata); let line = stats.to_string(); assert!( @@ -1344,10 +1363,34 @@ mod tests { }; let stats = reconstruct_stats(&hsi, &recon, &fill, &[]); assert_eq!(stats.extra_n, 1); + assert_eq!(stats.extra_bytes, tx_wire_len(&b1)); assert_eq!(stats.mempool_n, 0); assert_eq!(stats.fetched_n, 0); } + #[test] + fn cmpct_reconstruct_stats_orphan_bytes_sum() { + let b1 = spend(15); + let block = sealed_block(vec![coinbase(), b1.clone()]); + let hsi = HeaderAndShortIds::from_block(&block, 8, 2, &[]).unwrap(); + let recon = try_reconstruct( + &hsi, + &shortid_map_from_txs(&block.header, hsi.nonce, 2, [&b1]), + 2, + ) + .expect("orphan fill"); + let fill = CmpctFillSets { + mempool: Default::default(), + extra: Default::default(), + orphan: [b1.compute_wtxid()].into_iter().collect(), + }; + let stats = reconstruct_stats(&hsi, &recon, &fill, &[]); + assert_eq!(stats.orphan_n, 1); + assert_eq!(stats.orphan_bytes, tx_wire_len(&b1)); + assert_eq!(stats.extra_n, 0); + assert_eq!(stats.mempool_n, 0); + } + #[test] fn cmpct_reconstruct_stats_getdata_line() { let hash = BlockHash::from_byte_array([0xab; 32]); @@ -1454,4 +1497,28 @@ mod tests { "missed mempool read is not an empty fill" ); } + + #[test] + fn prefill_indexes_keeps_a_tx_that_fills_the_budget_exactly() { + let mut pad = EXTRA_PREFILL_BUDGET; + let tx = loop { + let candidate = spend_padded(40, pad); + let sz = tx_wire_len(&candidate); + if sz == EXTRA_PREFILL_BUDGET { + break candidate; + } + if sz > EXTRA_PREFILL_BUDGET { + pad = pad.saturating_sub(sz - EXTRA_PREFILL_BUDGET); + } else { + pad += EXTRA_PREFILL_BUDGET - sz; + } + }; + assert_eq!(tx_wire_len(&tx), EXTRA_PREFILL_BUDGET); + let block = sealed_block(vec![coinbase(), tx]); + assert_eq!( + prefill_indexes(&block, &CmpctFillSets::default()), + vec![0, 1], + "a transaction that lands on the extra-prefill budget is packed" + ); + } } diff --git a/crates/rbitcoin-net/src/eviction.rs b/crates/rbitcoin-net/src/eviction.rs index bc6af5232..f342e39f6 100644 --- a/crates/rbitcoin-net/src/eviction.rs +++ b/crates/rbitcoin-net/src/eviction.rs @@ -106,9 +106,6 @@ fn remove_first_k(cands: &mut Vec, k: usize) { /// Protect up to `k` peers from the largest keyed netgroups (Core netgroup protect). fn protect_by_netgroup(cands: &mut Vec, k: usize) { - if k == 0 || cands.is_empty() { - return; - } use std::collections::HashMap; let mut counts: HashMap = HashMap::new(); for c in cands.iter() { @@ -262,6 +259,35 @@ mod tests { assert!(select_inbound_eviction(cands).is_none()); } + #[test] + fn protect_by_netgroup_keeps_one_peer_from_each_largest_group() { + let mut untouched = vec![cand(1, 1, None, 0, 0)]; + protect_by_netgroup(&mut untouched, 0); + assert_eq!(untouched.len(), 1, "k = 0 protects nobody"); + let mut empty = Vec::new(); + protect_by_netgroup(&mut empty, 4); + assert!(empty.is_empty()); + + let group = |id, netgroup| InboundEvictCandidate { + netgroup, + ..cand(id, id, None, 0, 0) + }; + // Group 10 has two peers. Group 1 has one. k = 1 protects the lowest + // id in the larger group and leaves the other two. + let mut cands = vec![group(1, 10), group(2, 10), group(3, 1)]; + protect_by_netgroup(&mut cands, 1); + let ids: Vec<_> = cands.iter().map(|c| c.id).collect(); + assert_eq!(ids, vec![2, 3]); + + let mut capped = vec![group(1, 10), group(3, 1)]; + protect_by_netgroup(&mut capped, 1); + assert_eq!(capped.iter().map(|c| c.id).collect::>(), vec![1]); + + let mut mismatched = vec![group(8, 4), group(9, 5)]; + protect_by_netgroup(&mut mismatched, 1); + assert_eq!(mismatched.iter().map(|c| c.id).collect::>(), vec![9]); + } + #[test] fn eviction_netgroup_ipv4_slash16() { let a: std::net::SocketAddr = "1.2.3.4:1".parse().unwrap(); diff --git a/crates/rbitcoin-net/src/i2p_sam.rs b/crates/rbitcoin-net/src/i2p_sam.rs index c9469bbf0..2eee039bb 100644 --- a/crates/rbitcoin-net/src/i2p_sam.rs +++ b/crates/rbitcoin-net/src/i2p_sam.rs @@ -421,7 +421,8 @@ fn decode_i2p_b64(s: &str) -> Result, NetError> { ))); } }; - acc = (acc << 6) | v; + // `v` is six bits, so it does not overlap the shifted accumulator. + acc = (acc << 6) + v; n += 6; if n >= 8 { n -= 8; @@ -1144,4 +1145,175 @@ mod tests { assert!(i2p_addr_from_sam_b64("!!!").is_err()); assert!(i2p_addr_from_sam_b64(&i2p_b64_encode(&[1, 2, 3])).is_err()); } + + fn enc(msg: &str) -> NetError { + NetError::Encode(msg.to_string()) + } + + #[test] + fn sam_errors_name_the_socket_and_the_reply_that_retry() { + let io = NetError::Io(std::io::Error::other("sam io")); + assert!(session_dead(&io)); + assert!(session_dead(&NetError::Disconnected)); + assert!(sam_retry(&io)); + assert!(sam_retry(&NetError::Disconnected)); + assert!(stream_retry(&io)); + assert!(stream_retry(&NetError::Disconnected)); + assert!(session_dead(&enc("STREAM CONNECT: refused"))); + assert!(!session_dead(&enc("i2p sam dial: connection refused"))); + for needle in [ + "connection reset", + "connection closed", + "i2p sam connect", + "early eof", + "unexpected eof", + "broken pipe", + ] { + assert!(sam_retry(&enc(needle)), "{needle}"); + } + assert!(!sam_retry(&enc("i2p sam stream: RESULT=I2P_ERROR"))); + for needle in [ + "unexpected eof", + "broken pipe", + "connection reset", + "connection closed", + "cant_reach", + "early eof", + "timeout", + ] { + assert!(stream_retry(&enc(needle)), "{needle}"); + } + assert!(!stream_retry(&enc("i2p sam stream: RESULT=I2P_ERROR"))); + } + + #[test] + fn i2p_b64_round_trips_digits_dash_and_tilde() { + for raw in [ + vec![0, 0, 52], + vec![0, 0, 62], + vec![0, 0, 63], + vec![1, 2, 3, 4, 5], + ] { + let text = i2p_b64_encode(&raw); + assert_eq!(decode_i2p_b64(&text).unwrap(), raw, "{text}"); + let plus = text.replace('-', "+"); + let slash = text.replace('~', "/"); + assert_eq!(decode_i2p_b64(&plus).unwrap(), raw, "{plus}"); + assert_eq!(decode_i2p_b64(&slash).unwrap(), raw, "{slash}"); + } + } + + async fn sam_reply(lines: &'static [&'static str]) -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + loop { + let Ok((mut sock, _)) = listener.accept().await else { + break; + }; + tokio::spawn(async move { + for reply in lines { + if read_line(&mut sock).await.is_err() { + return; + } + if write_line(&mut sock, reply).await.is_err() { + return; + } + } + }); + } + }); + addr + } + + #[tokio::test(flavor = "current_thread")] + async fn hello_requires_the_reply_prefix_and_result_ok() { + let ok = sam_reply(&["HELLO REPLY RESULT=OK VERSION=3.1"]).await; + let mut sock = TcpStream::connect(ok).await.unwrap(); + hello(&mut sock).await.unwrap(); + + let missing = sam_reply(&["HELLO REPLY RESULT=NOVERSION"]).await; + let mut sock = TcpStream::connect(missing).await.unwrap(); + let err = hello(&mut sock).await.unwrap_err(); + assert!(err.to_string().contains("i2p sam hello"), "{err}"); + + let bare = sam_reply(&["RESULT=OK"]).await; + let mut sock = TcpStream::connect(bare).await.unwrap(); + let err = hello(&mut sock).await.unwrap_err(); + assert!(err.to_string().contains("i2p sam hello"), "{err}"); + } + + #[tokio::test(flavor = "current_thread")] + async fn read_line_stops_at_16kib() { + async fn once(n: usize) -> Result { + use tokio::io::AsyncWriteExt; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut sock, _) = listener.accept().await.unwrap(); + let body = vec![b'a'; n]; + sock.write_all(&body).await.unwrap(); + sock.write_all(b"\n").await.unwrap(); + }); + let mut sock = TcpStream::connect(addr).await.unwrap(); + let got = read_line(&mut sock).await.map(|s| s.len()); + let _ = server.await; + got + } + assert_eq!(once(16 * 1024).await.unwrap(), 16 * 1024); + assert!(once(16 * 1024 + 1).await.is_err()); + assert_eq!(once(16 + 1024 + 1).await.unwrap(), 16 + 1024 + 1); + } + + #[tokio::test(flavor = "current_thread")] + async fn stream_status_error_is_not_retried() { + let addr = sam_reply(&[ + "HELLO REPLY RESULT=OK VERSION=3.1", + "STREAM STATUS RESULT=I2P_ERROR", + ]) + .await; + let dialer = I2pDialer { + sam_addr: addr, + session_id: "once".into(), + destination: "dest".into(), + forward_port: None, + }; + let started = std::time::Instant::now(); + let err = dialer.stream_connect("peer.b32.i2p").await.unwrap_err(); + assert!( + started.elapsed() < std::time::Duration::from_secs(2), + "a SAM status error is returned, not slept on" + ); + assert!(err.to_string().contains("i2p sam stream"), "{err}"); + + let forward_addr = sam_reply(&[ + "HELLO REPLY RESULT=OK VERSION=3.1", + "STREAM STATUS RESULT=I2P_ERROR", + ]) + .await; + let (control, _held) = { + let hold = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let hold_addr = hold.local_addr().unwrap(); + let accepted = tokio::spawn(async move { + let _ = hold.accept().await; + }); + let control = TcpStream::connect(hold_addr).await.unwrap(); + (control, accepted) + }; + let mut sam = I2pSam { + sam_addr: forward_addr, + session_id: "fwd".into(), + destination: "dest".into(), + forward_port: None, + _control: control, + _forward: None, + }; + let started = std::time::Instant::now(); + let err = sam.stream_forward(9).await.unwrap_err(); + assert!( + started.elapsed() < std::time::Duration::from_secs(2), + "a forward status error is returned, not slept on" + ); + assert!(err.to_string().contains("i2p sam forward"), "{err}"); + } } diff --git a/crates/rbitcoin-net/src/ibd/header_walk.rs b/crates/rbitcoin-net/src/ibd/header_walk.rs index a288163b5..366bcef30 100644 --- a/crates/rbitcoin-net/src/ibd/header_walk.rs +++ b/crates/rbitcoin-net/src/ibd/header_walk.rs @@ -1676,24 +1676,33 @@ fn judge_refill(st: &mut IbdWorkState, hub: &ChainHub, start_h: u32, headers: &[ fn unproved_anchor(st: &IbdWorkState, hub: &ChainHub) -> Option<(BlockHash, u32)> { let front = st.ordered.front().copied()?; let front_h = st.hash_height.get(&front).copied()?; - let reached = st.ordered.iter().any(|hash| { - st.hash_height.get(hash).is_some_and(|ht| { - st.header_walk - .checkpoints + // The highest checkpoint under the queue front is the anchor. A covered + // checkpoint at or above the front means the queue is already proved. + let mut reached = false; + let mut below: Option<&Checkpoint> = None; + for c in &st.header_walk.checkpoints { + if c.height >= front_h { + let covered = st + .ordered .iter() - .any(|c| *ht >= c.height && c.height >= front_h) - }) - }); + .any(|hash| st.hash_height.get(hash).is_some_and(|ht| *ht >= c.height)); + if covered { + reached = true; + } + } else { + let replace = match below { + Some(best) => c.height > best.height, + None => true, + }; + if replace { + below = Some(c); + } + } + } if reached { return None; } - if let Some(cp) = st - .header_walk - .checkpoints - .iter() - .filter(|c| c.height < front_h) - .max_by_key(|c| c.height) - { + if let Some(cp) = below { return Some((cp.hash, cp.height)); } let tip_h = hub.tip_height()?; @@ -6975,4 +6984,450 @@ mod tests { .expect("the confirmed tip prices a header the base cannot reach"); assert_eq!(work, hub.chain_work().unwrap() + next.work()); } + + fn byte_hash(n: u8) -> BlockHash { + let mut raw = [0u8; 32]; + raw[31] = n; + BlockHash::from_byte_array(raw) + } + + fn mark_checkpoint(hash: BlockHash, height: u32) -> Checkpoint { + Checkpoint { + hash, + height, + work: [0; 32], + header: None, + times: Vec::new(), + diff: DiffSnap::default(), + } + } + + fn fresh_walk() -> IbdWorkState { + let (peer, _rx) = slot(0); + IbdWorkState::new(vec![peer], None, None) + } + + fn plant_challenger(st: &mut IbdWorkState, fork: BlockHash) { + st.header_walk.challenger = Some(Challenger { + fork_hash: fork, + fork_height: 1, + tip: WalkTip::default(), + checkpoints: Vec::new(), + }); + } + + #[test] + fn retain_challenger_keeps_a_fork_that_is_still_on_the_candidate() { + let fork = byte_hash(1); + let other = byte_hash(2); + let mut st = fresh_walk(); + st.header_walk.base_hash = Some(fork); + plant_challenger(&mut st, fork); + retain_challenger(&mut st); + assert!( + st.header_walk.challenger.is_some(), + "the walk base still holds the fork" + ); + + st.header_walk.base_hash = Some(other); + st.header_walk.checkpoints.push(mark_checkpoint(fork, 4)); + retain_challenger(&mut st); + assert!( + st.header_walk.challenger.is_some(), + "a checkpoint still holds the fork" + ); + + st.header_walk.checkpoints.clear(); + retain_challenger(&mut st); + assert!( + st.header_walk.challenger.is_none(), + "a fork off the candidate is dropped" + ); + } + + #[test] + fn adopt_heavier_keeps_the_milestone_at_the_fork_height() { + let (_dir, mut hub) = crate::chain::tiny_regtest_hub_labeled("hdr-adopt-pin"); + hub.ensure_genesis().unwrap(); + let prev = byte_hash(3); + let header = Header { + version: bitcoin::block::Version::from_consensus(4), + prev_blockhash: prev, + merkle_root: TxMerkleNode::from_byte_array([4; 32]), + time: 1_600_000_100, + bits: CompactTarget::from_consensus(0x207fffff), + nonce: 1, + }; + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], hub.tip_hash(), hub.tip_height()); + st.hash_height.insert(prev, 5); + hub.milestone.height = 5; + let marker = byte_hash(9); + st.header_walk.tip.milestone_hash = Some(marker); + let mut raw = [0u8; 32]; + raw[31] = 9; + let passed = Work::from_be_bytes(raw); + adopt_heavier(&mut st, &hub, prev, &[header], passed); + assert_eq!(st.header_walk.tip.milestone_hash, Some(marker)); + assert_eq!(st.header_walk.tip.work, passed); + } + + #[test] + fn walk_ahead_of_stored_is_strictly_above_the_recorded_height() { + let mut st = fresh_walk(); + st.header_walk.origin = true; + st.header_walk.tip.height = 4; + st.max_ordered_height = 4; + assert!(!walk_ahead_of_stored(&st)); + st.header_walk.tip.height = 5; + assert!(walk_ahead_of_stored(&st)); + } + + #[test] + fn classify_pins_an_unlinked_batch_a_refill_and_the_first_checkpoint() { + let (_dir, hub) = crate::chain::tiny_regtest_hub_labeled("hdr-classify-pin"); + hub.ensure_genesis().unwrap(); + let gen = hub.tip_hash().unwrap(); + let good = mine(gen, 1); + let mut broken = mine(good.block_hash(), 2); + broken.prev_blockhash = byte_hash(8); + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], Some(gen), Some(0)); + assert_eq!( + classify(&st, &hub, HeaderAsk::InWindow, &[good, broken]), + HeaderClass::Fork, + "an unlinked batch is not a refill of the stored tip" + ); + assert_eq!( + classify(&st, &hub, HeaderAsk::InWindow, &[good]), + HeaderClass::Refill + ); + + st.header_walk.origin = true; + st.header_walk.tip.hash = Some(gen); + st.header_walk.tip.height = 0; + assert_eq!( + classify(&st, &hub, HeaderAsk::InWindow, &[good]), + HeaderClass::WalkAbsorb, + "the first solicited continuation is a checkpoint" + ); + } + + #[test] + fn headers_late_is_logged_only_for_a_late_extension() { + let (_dir, hub) = crate::chain::tiny_regtest_hub_labeled("hdr-late-pin"); + hub.ensure_genesis().unwrap(); + let gen = hub.tip_hash().unwrap(); + let good = mine(gen, 7); + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], Some(gen), Some(0)); + st.header_walk.origin = true; + st.header_walk.tip.hash = Some(gen); + st.header_walk.tip.height = 0; + st.header_walk.tip.header = hub.tip_header(); + rbitcoin_log::capture_logs(true); + assert!(absorb_lookahead( + &mut st, + &hub, + 0, + HeaderAsk::InWindow, + &[good] + )); + let on_time = rbitcoin_log::take_logs(); + rbitcoin_log::capture_logs(false); + assert!( + on_time + .iter() + .all(|(_, line)| !line.contains("ibd: headers late")), + "an in-window extension is not late: {on_time:?}" + ); + + let next = mine(good.block_hash(), 8); + rbitcoin_log::capture_logs(true); + assert!(absorb_lookahead(&mut st, &hub, 0, HeaderAsk::Late, &[next])); + let late = rbitcoin_log::take_logs(); + rbitcoin_log::capture_logs(false); + assert!( + late.iter() + .any(|(_, line)| line.contains("ibd: headers late")), + "a late extension names itself: {late:?}" + ); + } + + fn floor_above(hub: &mut ChainHub) { + hub.milestone = rbitcoin_consensus::Milestone { + height: 100, + anchor: Some(rbitcoin_consensus::MilestoneAnchor { + hash: byte_hash(1), + min_work_be: [0xff; 32], + }), + }; + } + + #[test] + fn ignore_below_floor_pins_the_floor_gate_and_the_queue_cap() { + let (_dir, mut hub) = crate::chain::tiny_regtest_hub_labeled("hdr-floor-pin"); + hub.ensure_genesis().unwrap(); + let gen = hub.tip_hash().unwrap(); + let good = mine(gen, 1); + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], Some(gen), Some(0)); + st.header_walk.origin = true; + st.header_walk.tip.hash = Some(byte_hash(3)); + st.header_walk.tip.work = Work::from_be_bytes([0; 32]); + st.header_walk.dead_ends.insert(gen); + assert!( + !ignore_below_floor(&st, &hub, &[good]), + "above the floor a dead-end parent is not ignored" + ); + + floor_above(&mut hub); + st.header_walk.origin = false; + assert!( + !ignore_below_floor(&st, &hub, &[good]), + "below the floor a walk that has not started is not ignored" + ); + st.header_walk.origin = true; + st.header_walk.dead_ends.clear(); + assert!( + !ignore_below_floor(&st, &hub, &[good]), + "under the cap the stored tip's child is still the download path" + ); + fill_queue(&mut st); + plant_on_queue(&mut st, gen, 0); + assert!( + ignore_below_floor(&st, &hub, &[good]), + "a full queue does not keep writing that child" + ); + } + + fn heavy_proved(st: &mut IbdWorkState, hub: &ChainHub) { + st.header_walk.origin = true; + st.header_walk.proven = true; + st.header_walk.tip.hash = hub.tip_hash(); + st.header_walk.tip.height = hub.tip_height().unwrap_or(0); + st.header_walk.tip.work = Work::from_be_bytes([0xff; 32]); + } + + #[test] + fn reject_refill_miss_pins_each_reason_it_returns() { + let (_dir, hub) = crate::chain::tiny_regtest_hub_labeled("hdr-refill-pin"); + hub.ensure_genesis().unwrap(); + let gen = hub.tip_hash().unwrap(); + let good = mine(gen, 3); + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], Some(gen), Some(0)); + heavy_proved(&mut st, &hub); + st.header_walk + .checkpoints + .push(mark_checkpoint(byte_hash(9), 1)); + st.header_walk.origin = false; + assert!( + !reject_refill_miss(&mut st, &hub, &[good]), + "a walk with no origin does not judge the refill" + ); + + heavy_proved(&mut st, &hub); + st.header_walk.checkpoints.clear(); + let queued = byte_hash(8); + st.ordered.push_back(queued); + st.ordered_set.insert(queued); + st.hash_height.insert(queued, 1); + assert!( + !reject_refill_miss(&mut st, &hub, &[good]), + "no checkpoint means there is nothing to miss" + ); + st.ordered.clear(); + st.ordered_set.clear(); + st.hash_height.remove(&queued); + + heavy_proved(&mut st, &hub); + st.header_walk + .checkpoints + .push(mark_checkpoint(byte_hash(9), 1)); + fill_queue(&mut st); + plant_on_queue(&mut st, gen, 0); + assert!( + !reject_refill_miss(&mut st, &hub, &[good]), + "a full queue is not a refill" + ); + + let stray = mine(byte_hash(7), 4); + let (peer, _rx) = slot(1); + let mut open = IbdWorkState::new(vec![peer], Some(gen), Some(0)); + heavy_proved(&mut open, &hub); + open.header_walk + .checkpoints + .push(mark_checkpoint(byte_hash(9), 1)); + open.header_walk.tip.hash = Some(byte_hash(6)); + assert!( + !reject_refill_miss(&mut open, &hub, &[stray]), + "a batch that does not build on the confirmed tip is not a refill miss" + ); + } + + #[test] + fn proved_above_floor_requires_both() { + let (_dir, hub) = crate::chain::tiny_regtest_hub_labeled("hdr-proved-pin"); + hub.ensure_genesis().unwrap(); + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], hub.tip_hash(), Some(0)); + st.header_walk.proven = false; + st.header_walk.tip.work = Work::from_be_bytes([0xff; 32]); + assert!(!proved_above_floor(&st, &hub)); + st.header_walk.proven = true; + assert!(proved_above_floor(&st, &hub)); + } + + #[test] + fn unproved_anchor_stops_at_a_met_checkpoint_and_keeps_the_one_below() { + let (_dir, hub) = crate::chain::tiny_regtest_hub_labeled("hdr-anchor-pin"); + hub.ensure_genesis().unwrap(); + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], hub.tip_hash(), Some(0)); + let front = byte_hash(5); + st.ordered.push_back(front); + st.ordered_set.insert(front); + st.hash_height.insert(front, 5); + st.header_walk + .checkpoints + .push(mark_checkpoint(byte_hash(3), 3)); + st.header_walk + .checkpoints + .push(mark_checkpoint(byte_hash(4), 5)); + assert!(unproved_anchor(&st, &hub).is_none()); + + st.header_walk.checkpoints.pop(); + assert_eq!(unproved_anchor(&st, &hub), Some((byte_hash(3), 3))); + + st.header_walk.checkpoints.clear(); + st.header_walk + .checkpoints + .push(mark_checkpoint(byte_hash(3), 3)); + st.header_walk + .checkpoints + .push(mark_checkpoint(byte_hash(6), 3)); + assert_eq!( + unproved_anchor(&st, &hub), + Some((byte_hash(3), 3)), + "two checkpoints at one height keep the earlier hash" + ); + } + + #[test] + fn candidate_prev_accepts_base_tip_and_checkpoint() { + let prev = byte_hash(1); + let mut st = fresh_walk(); + st.header_walk.base_hash = Some(prev); + assert!(candidate_prev(&st, prev)); + st.header_walk.base_hash = None; + st.header_walk.tip.hash = Some(prev); + assert!(candidate_prev(&st, prev)); + st.header_walk.tip.hash = None; + st.header_walk.checkpoints.push(mark_checkpoint(prev, 2)); + assert!(candidate_prev(&st, prev)); + st.header_walk.checkpoints.clear(); + assert!(!candidate_prev(&st, prev)); + } + + fn prefix_body(st: &mut IbdWorkState, _hub: &ChainHub, prev: BlockHash) -> Vec
{ + st.header_walk.proven = true; + st.header_walk.origin = true; + st.header_walk.tip.work = Work::from_be_bytes([0xff; 32]); + st.header_walk.base_hash = Some(prev); + st.header_walk + .checkpoints + .push(mark_checkpoint(byte_hash(9), 50)); + vec![mine(prev, 11)] + } + + #[test] + fn proven_header_prefix_writes_the_whole_reply_until_the_candidate_is_proved() { + let (_dir, mut hub) = crate::chain::tiny_regtest_hub_labeled("hdr-prefix-pin"); + hub.ensure_genesis().unwrap(); + let gen = hub.tip_hash().unwrap(); + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], Some(gen), Some(0)); + let prev = byte_hash(4); + let headers = prefix_body(&mut st, &hub, prev); + st.header_walk.proven = false; + assert_eq!(proven_header_prefix(&st, &hub, &headers), headers.len()); + st.header_walk.proven = true; + st.header_walk.origin = false; + assert_eq!(proven_header_prefix(&st, &hub, &headers), headers.len()); + st.header_walk.origin = true; + floor_above(&mut hub); + st.header_walk.tip.work = Work::from_be_bytes([0; 32]); + assert_eq!(proven_header_prefix(&st, &hub, &headers), headers.len()); + hub.milestone.anchor = None; + st.header_walk.tip.work = Work::from_be_bytes([0xff; 32]); + assert_eq!( + proven_header_prefix(&st, &hub, &headers), + 0, + "a proved candidate writes no header that misses the checkpoint" + ); + } + + #[test] + fn refill_misses_checkpoint_only_above_the_queue_tail() { + let tail = byte_hash(1); + let first = mine(tail, 21); + let second = mine(first.block_hash(), 22); + let mut st = fresh_walk(); + st.header_walk.checkpoints.push(mark_checkpoint(tail, 5)); + st.header_walk + .checkpoints + .push(mark_checkpoint(first.block_hash(), 6)); + st.header_walk + .checkpoints + .push(mark_checkpoint(byte_hash(8), 7)); + assert!(refill_misses_checkpoint(&st, 5, &[first, second])); + st.header_walk.checkpoints.clear(); + st.header_walk.checkpoints.push(mark_checkpoint(tail, 5)); + assert!(!refill_misses_checkpoint(&st, 5, &[first])); + } + + #[test] + fn note_empty_rewinds_only_when_every_taller_peer_is_empty() { + let (_dir, mut hub) = crate::chain::tiny_regtest_hub_labeled("hdr-empty-pin"); + hub.ensure_genesis().unwrap(); + floor_above(&mut hub); + let gen = hub.tip_hash().unwrap(); + let (mut tall, _rx0) = slot(0); + tall.peer_height = 10; + let (mut level, _rx1) = slot(1); + level.peer_height = 3; + let mut st = IbdWorkState::new(vec![tall, level], Some(gen), Some(0)); + st.header_walk.origin = true; + st.header_walk.base_hash = Some(gen); + st.header_walk.base_height = 0; + st.header_walk.tip.hash = Some(byte_hash(4)); + st.header_walk.tip.height = 3; + st.header_walk.tip.work = Work::from_be_bytes([0; 32]); + assert!(note_empty(&mut st, &hub, 0)); + assert_eq!(st.header_walk.tip_hash(), Some(gen)); + } + + #[test] + fn note_stored_candidate_ignores_a_dead_end_tip() { + let (_dir, hub) = crate::chain::tiny_regtest_hub_labeled("hdr-stored-pin"); + hub.ensure_genesis().unwrap(); + let gen = hub.tip_hash().unwrap(); + let (peer, _rx) = slot(0); + let mut st = IbdWorkState::new(vec![peer], Some(gen), Some(0)); + st.header_walk.origin = true; + st.header_walk.tip.hash = Some(gen); + st.header_walk.tip.height = 0; + st.header_walk.dead_ends.insert(gen); + note_stored_candidate(&mut st, &hub, &[mine(gen, 30)]); + assert_eq!(st.header_walk.tip_hash(), Some(gen)); + assert!(st.header_walk.checkpoints.is_empty()); + + st.header_walk.dead_ends.clear(); + let mut broken = mine(gen, 31); + broken.bits = CompactTarget::from_consensus(0x1d00ffff); + note_stored_candidate(&mut st, &hub, &[broken]); + assert_eq!(st.header_walk.tip_hash(), Some(gen)); + assert!(st.header_walk.checkpoints.is_empty()); + } } diff --git a/crates/rbitcoin-net/src/ibd/mod.rs b/crates/rbitcoin-net/src/ibd/mod.rs index 9f376c547..5bc845c17 100644 --- a/crates/rbitcoin-net/src/ibd/mod.rs +++ b/crates/rbitcoin-net/src/ibd/mod.rs @@ -445,15 +445,15 @@ pub async fn ibd_cancellable( warn!("ibd: block_queue rehydrate failed (continuing; may re-getdata): {e}"); } } + // A positive count is already logged inside `rehydrate_class_a_into_body_queue` + // (a body-queue row or an archived body above the tip). This match only + // speaks when that walk fills nothing. match rehydrate_class_a_into_body_queue( hub.as_ref(), &mut st, confirm_feed.as_ref(), TIP_HOLE_MAX, ) { - Ok(n) if n > 0 => { - info!("ibd: Class A rehydrate filled {n} body-queue height(s) for tip batch"); - } Ok(0) => { rbitcoin_log::debug!( "ibd: Class A rehydrate filled 0 (tip+1 missing height map, has_block, or no Class A)" diff --git a/crates/rbitcoin-net/src/msg_decode.rs b/crates/rbitcoin-net/src/msg_decode.rs index 9bab0af48..f009810be 100644 --- a/crates/rbitcoin-net/src/msg_decode.rs +++ b/crates/rbitcoin-net/src/msg_decode.rs @@ -152,4 +152,13 @@ mod tests { }); assert!(*done.lock().unwrap()); } + + #[test] + fn decode_permit_pools_are_wider_than_one_slot() { + let blocks = block_decode_permits(); + let ctrl = ctrl_decode_permits(); + assert!(blocks >= 64, "block decodes keep a wide pool, got {blocks}"); + assert!(ctrl >= 8, "header decodes keep their own pool, got {ctrl}"); + assert!(blocks > ctrl); + } } diff --git a/crates/rbitcoin-net/src/net_permissions.rs b/crates/rbitcoin-net/src/net_permissions.rs index a3a17ea6e..e04a081f1 100644 --- a/crates/rbitcoin-net/src/net_permissions.rs +++ b/crates/rbitcoin-net/src/net_permissions.rs @@ -10,21 +10,16 @@ impl NetPermissionFlags { pub const NONE: Self = Self(0); pub const BLOOM: Self = Self(1 << 1); pub const RELAY: Self = Self(1 << 3); - pub const FORCE_RELAY: Self = Self((1 << 2) | (1 << 3)); + /// Bits 2 and 3. ForceRelay implies Relay. + pub const FORCE_RELAY: Self = Self(0b1100); pub const DOWNLOAD: Self = Self(1 << 6); - pub const NOBAN: Self = Self((1 << 4) | (1 << 6)); + /// Bits 4 and 6. NoBan implies Download. + pub const NOBAN: Self = Self(0b0101_0000); pub const MEMPOOL: Self = Self(1 << 5); pub const ADDR: Self = Self(1 << 7); pub const IMPLICIT: Self = Self(1 << 31); - pub const ALL: Self = Self( - Self::BLOOM.0 - | Self::FORCE_RELAY.0 - | Self::RELAY.0 - | Self::NOBAN.0 - | Self::MEMPOOL.0 - | Self::DOWNLOAD.0 - | Self::ADDR.0, - ); + /// Bits 1 through 7. Relay sits inside ForceRelay, and Download sits inside NoBan. + pub const ALL: Self = Self(0b1111_1110); pub const fn has(self, f: Self) -> bool { (self.0 & f.0) == f.0 @@ -590,4 +585,30 @@ mod tests { t.whitelist.push(g); assert!(t.strings_for(ip(), true, bind()).is_empty()); } + + #[test] + fn permission_masks_keep_their_implied_bits() { + assert!(NetPermissionFlags::FORCE_RELAY.has(NetPermissionFlags::RELAY)); + assert_ne!(NetPermissionFlags::FORCE_RELAY, NetPermissionFlags::RELAY); + assert!(NetPermissionFlags::NOBAN.has(NetPermissionFlags::DOWNLOAD)); + assert_ne!(NetPermissionFlags::NOBAN, NetPermissionFlags::DOWNLOAD); + assert_eq!(NetPermissionFlags::NOBAN.0 & (1 << 4), 1 << 4); + for flag in [ + NetPermissionFlags::BLOOM, + NetPermissionFlags::FORCE_RELAY, + NetPermissionFlags::RELAY, + NetPermissionFlags::NOBAN, + NetPermissionFlags::MEMPOOL, + NetPermissionFlags::DOWNLOAD, + NetPermissionFlags::ADDR, + ] { + assert!(NetPermissionFlags::ALL.has(flag), "{flag:?}"); + } + let mut flagged = NetPermissionFlags::ALL; + flagged.add(NetPermissionFlags::IMPLICIT); + assert!(flagged.has(NetPermissionFlags::IMPLICIT)); + flagged.clear_implicit(); + assert!(!flagged.has(NetPermissionFlags::IMPLICIT)); + assert!(flagged.has(NetPermissionFlags::MEMPOOL)); + } } diff --git a/crates/rbitcoin-net/tests/ibd_smoke.rs b/crates/rbitcoin-net/tests/ibd_smoke.rs index c0addc55f..d96c60f67 100644 --- a/crates/rbitcoin-net/tests/ibd_smoke.rs +++ b/crates/rbitcoin-net/tests/ibd_smoke.rs @@ -2,6 +2,7 @@ //! [`P2PNode::sync`]. Hits the main IBD loop, archive pipeline, confirm engine, //! peer_io, and dial/events paths without live datadirs. +use bitcoin::hashes::Hash; use bitcoin::{Block, BlockHash}; use rbitcoin_consensus::{ChainParams, Milestone}; use rbitcoin_net::{IbdConfig, P2PNode}; @@ -110,3 +111,104 @@ async fn ibd_unreachable_peer_errors() { node.shutdown().await; let _ = std::fs::remove_dir_all(&dir); } + +fn logs_from_sync() -> impl Fn() -> Vec { + rbitcoin_log::capture_logs(true); + || { + rbitcoin_log::take_logs() + .into_iter() + .map(|(_, line)| line) + .collect() + } +} + +/// One live seed and a higher target reaches the redial clock. An empty body +/// queue must not claim it noted rows, and a store with no Class A bodies must +/// not announce a seed count of zero. +#[tokio::test(flavor = "current_thread")] +async fn ibd_redials_immediately_and_does_not_invent_a_rehydrate() { + let seed_dir = tmp_dir("seed-redial"); + let peer_dir = tmp_dir("peer-redial"); + let seed = start_node(&seed_dir).await; + seed_chain(&seed, 2); + let peer = start_node(&peer_dir).await; + peer.hub.ensure_genesis().unwrap(); + let archived = peer.hub.query.archived_block_count().unwrap(); + let mut cfg = IbdConfig::for_test(); + cfg.target_peers = 2; + let take = logs_from_sync(); + // This chain is two blocks, so catch-up exits on its own. + let finished = tokio::time::timeout( + Duration::from_secs(20), + peer.sync_cancellable(&[rbitcoin_net::NetAddr::Ip(seed.local_addr)], cfg, None), + ) + .await; + let logs = take(); + rbitcoin_log::capture_logs(false); + assert!(finished.is_ok(), "short-chain ibd did not finish: {logs:?}"); + assert!( + logs.iter().any(|line| line.contains("ibd: redialing")), + "a short peer set redials on the first loop: {logs:?}" + ); + assert!( + logs.iter().all(|line| !line.contains("noted 0")), + "an empty body queue is not reported as noted: {logs:?}" + ); + let seed_line = format!("store has {archived} Class A bodies (seed)"); + if archived > 0 { + assert!( + logs.iter().any(|line| line.contains(&seed_line)), + "Class A bodies already in the store are named: {logs:?}" + ); + } else { + assert!( + logs.iter() + .all(|line| !line.contains("Class A bodies (seed)")), + "an empty store does not announce a Class A seed: {logs:?}" + ); + } + seed.shutdown().await; + peer.shutdown().await; + let _ = std::fs::remove_dir_all(&seed_dir); + let _ = std::fs::remove_dir_all(&peer_dir); +} + +/// A block left in the RAM body queue above the tip is noted into confirm. +#[tokio::test(flavor = "current_thread")] +async fn ibd_notes_a_body_queue_row_above_the_tip() { + let seed_dir = tmp_dir("seed-bq"); + let peer_dir = tmp_dir("peer-bq"); + let seed = start_node(&seed_dir).await; + seed_chain(&seed, 1); + let peer = start_node(&peer_dir).await; + peer.hub.ensure_genesis().unwrap(); + let gen = peer.hub.tip_hash().unwrap(); + let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); + let block = mine(gen, genesis.header.time + 600, 1); + let fk = peer.hub.ensure_header_fk(&block.header).unwrap(); + let raw = bitcoin::consensus::encode::serialize(&block); + peer.hub + .query + .block_queue_enqueue(1, block.block_hash().to_byte_array(), fk.0, &raw) + .unwrap(); + let mut cfg = IbdConfig::for_test(); + cfg.target_peers = 1; + let take = logs_from_sync(); + let finished = tokio::time::timeout( + Duration::from_secs(20), + peer.sync_cancellable(&[rbitcoin_net::NetAddr::Ip(seed.local_addr)], cfg, None), + ) + .await; + let logs = take(); + rbitcoin_log::capture_logs(false); + assert!(finished.is_ok(), "short-chain ibd did not finish: {logs:?}"); + assert!( + logs.iter() + .any(|line| line.contains("ibd: rehydrate: noted 1 in-RAM body queue entries")), + "the queued block is noted: {logs:?}" + ); + seed.shutdown().await; + peer.shutdown().await; + let _ = std::fs::remove_dir_all(&seed_dir); + let _ = std::fs::remove_dir_all(&peer_dir); +}