diff --git a/TESTING.md b/TESTING.md index 482774d232..d42f2d05b1 100644 --- a/TESTING.md +++ b/TESTING.md @@ -377,6 +377,8 @@ A cell that names a crate-local test describes the witness that exists today. It | `electrum_idle_timeout_disconnects_quiet_client` | Electrum | Idle timeout closes a quiet socket | | `esplora_broadcast_visible_in_rpc_and_electrum` | Node + Electrum + Esplora + RPC | One `run_p2p` datadir: HTTP `sendrawtransaction` / `testmempoolaccept` (allowed, missing-or-spent, exact 100 sat/kvB min-relay accept + one-sat-under reject, RBF one-sat-short incremental reject + exact incremental accept); Esplora `POST /tx` parent and mempool child appear in `getrawmempool` and Electrum mempool/history (`fee` on unconfirmed, including child `height = -1`); Electrum `listunspent` of that child is `height=-1` and the parent UTXO drops; process `gettxout` / `getchaintips`; Esplora `POST /txs/package` 1p1c (including parent-alone below min-relay + paying child), 25-tx accept, 26-tx and over-weight `package too large`; serving-only `submitpackage` refuses (relay off); live `GET /mempool` / `/mempool/txids` / `/mempool/recent` / `/fee-estimates` answers 503 while flow is cold and the chain holds too little fee history (a thin live pool alone sets no rate); process `getmempoolancestors` / descendants / cluster / `gettxspendingprevout` / feerate diagram / verbose `getrawmempool` on that 1p1c; `waitforblockheight` timeout=0 while behind returns the live tip; GBT stale `longpollid` is immediate; current id / `waitfornewblock` / `waitforblockheight` wake on the pad `generate`; `getblockhash` tip ok / tip+1 `-8`; unknown `getblock` `-5`; verbosity 0 hex and 2 vin/vout; `GET /blocks` 10 newest, `/blocks/0` and `/blocks/:tip` (start past tip clamps); `/block/:hash/txs/:start` last page shorter than 25, one-past last page 404, unknown hash 404; `/block/:hash/txids` + coinbase merkle-proof + unspent `outspend/0`; `/tx/:id/outspends`; `/block` JSON/raw/status/`txid/0` (OOB 404); `/tx/:id/raw` vs hex; merkleblock-proof; `/block-height` (missing 404); `/block/:hash/header` 160 hex; `/tx/:id/status` + full JSON (`unknown` OP_TRUE type, coinbase vin) and missing-tx 404s; OP_TRUE scripthash info/summary/utxo/`txs/chain` cursor and combined `/txs`. No-hub: mempool and fee routes, and `POST /tx` 503. Header bytes match the wire header. `tx_status_json` matches the status route. Esplora `/tx/:id/status` confirms the package parent on generate; `generate` includes those txs (parent before child) then leaves IBD (relay on); `submitpackage` maxfeerate reject, 1p1c success, already-in-mempool continue, below-min-relay parent + paying child success, 26-tx / over-weight `package too large`; immature coinbase sendraw rejects. Package JSON errors. `testmempoolaccept` dry-run reports the orphan count. `gettxout` with `include_mempool`, on a disconnected tip, and on a leftover. `generate` selects the chained mempool parent first. A `submitpackage` child failure keeps the parent. Maxburn `submitpackage` rejects. Waiters return when the node stops. Unix `--rpc-socket` (mode 0660, no datadir `rpc.sock`) `getblockcount` without Authorization; TCP `GET`/`POST /internal/mempool/txs`; `GET /internal/block/:hash/txs` full list vs public 25/page; `POST /internal/txs/outspends/by-txid` same-length unknown `[]` slot; `GET /address-prefix/bc1` **404**; unauthenticated Core REST on the RPC listener (`chaininfo`, block hash/headers/block/tx, mempool info/contents, `getutxos`, `deploymentinfo`, basic `blockfilter` bin/hex/json) and `getblockfilter` with `--block-filter-index` | | `fee_history_backfills_from_the_chain_when_relay_starts` | Node + RPC | `run_p2p` on a mature regtest datadir, started twice; `generate` leaves IBD and turns relay on; the preload writes the fee history file and the restart extends it; with flow cold and too little history, `estimatesmartfee` answers Core's insufficient-data shape. Rates from a ready history are the hub's `far_horizon_follows_block_history_not_pool_tail`; the success object is `smart_fee_json` | +| `asmap_loopback_peers_report_their_asn` | Node + RPC | One regtest `run_p2p`, no seeds and no outbound peer. `{datadir}/ip_asn.dat` is the loopback map: three BIP324 inbounds bound to `127.0.0.1`, `.2`, and `.3` show `getpeerinfo.mapped_as` 1, 2, and 9, and the startup line names that file's length and digest. A second start with relative `--asmap` of a lone return maps `127.0.0.1` to ASN 1 (length 3, digest `709e80c8`). | +| `asmap_refused_file_uses_prefix_netgroups` | Node | One exit per bad file (`max_run_secs` 0, no peer): a second default, a trailing one bit, eight trailing zeros, two short matches, a match past the last IP bit, a jump off the file, a jump that leaves too few bits, the smallest jump alone, an empty file, and a one-byte file each log `Sanity check of asmap file` and do not log `Opened asmap file`. A missing `--asmap` path logs `Failed to open asmap file`. | | `node_listen_and_exit` | Node + Electrum + Esplora + RPC | One `run_p2p` datadir, restarted with its one `--connect` refusing (a pinned connect at genesis still enters tip mode): a taken `--health-listen` port stops the start before the store opens; a junk `peers` file and a missing `--asmap` start an empty book and exit, and the saved book records the refused connect; the next start loads that book and a valid `ip_asn.dat`, and Esplora, Electrum, and RPC answer at genesis until `stop`. On that start `/healthz` is 200 and `/progress` is 200 JSON while genesis `/readyz` is initial block download and `/metrics` reports the same; `generate` makes `/readyz` 200 and the scrape ready at height 1. An Electrum port and an RPC port another process holds warn, `/readyz` names both, and `/metrics` is absent without `--metrics`. Without `--connect` and with seeds on, regtest resolves none and the node exits short of tip mode; after a `--prune-seqsigwit` start, an unpruned start refuses. Live peers are `node_run_p2p_short`. The `/readyz` phase, lag, and tip-age table stays `readiness_reports_the_first_failing_gate`. A timed-out probe keeping its permit, the full-gate 503, and the routes answering again are `health_gate_caps` | | `tor_control_onion_lifecycle` | Node + RPC | One `run_p2p` datadir against a fake Tor control port and SAM bridge (live Tor and i2pd are overlay-functional). A cookie from another Tor (SAFECOOKIE server hash mismatch), a 2-byte cookie, and a Tor that offers only plain COOKIE each refuse the start, and none sends `AUTHENTICATE`. Password auth with `--listen-onion`, `--i2p-accept-incoming`, Electrum, and Esplora: `ADD_ONION NEW` per service with the P2P virtual port on the loopback bind, each key saved `0600` under `onion/`, each SAM destination under `i2p/`, `STREAM FORWARD` to each port, and `getnetworkinfo.localaddresses` lists the three onions and the I2P address. A SAFECOOKIE restart reuses every saved key and destination | | `enter_tip_mode_indexes` | Node + Electrum + RPC | One `run_p2p` datadir restarted with `--sh-index` off, on, off, on. Off with no index yet: RPC, tip follow, and Electrum listen, `blockchain.scripthash.get_history` fails closed, and `generateblock` mines three OP_TRUE coinbases. First start on: the index is collected from Class A before history answers, and the OP_TRUE history has three rows. Off again: Electrum still listens and the leftover watermark keeps those three history rows. On after a crash that left a collect run and a lagging include high-water mark: the durable index resumes under write-behind, so the run is discarded (not merged) and history answers, and the next block lands in history | diff --git a/changelog.d/overnight-38686116770.md b/changelog.d/overnight-38686116770.md new file mode 100644 index 0000000000..c74250dfdb --- /dev/null +++ b/changelog.d/overnight-38686116770.md @@ -0,0 +1,5 @@ +Changed + +- Overnight missed mutants are gone or pinned on a peer message, an RPC result, or a reopened asmap or mempool file. Dead helpers (asmap byte views, the block-cache locator, the dial warm locator, and the witness wtxid-count reject) are deleted instead of skipped. +- A headers ask sends the walk locator, which stops at eight hashes, and still appends the confirmed tip. It no longer drops walk hashes once eight names were already queued. +- The bound P2P port is written to `{datadir}/run/p2p.addr`, the same way the RPC and wallet listeners publish their addresses. diff --git a/crates/rbitcoin-consensus/src/block/mod.rs b/crates/rbitcoin-consensus/src/block/mod.rs index 95feeaf531..00f97d2f9e 100644 --- a/crates/rbitcoin-consensus/src/block/mod.rs +++ b/crates/rbitcoin-consensus/src/block/mod.rs @@ -555,10 +555,8 @@ fn check_witness_commitment_with_wtxids( let mut committed = [0u8; 32]; committed.copy_from_slice(&b[6..38]); - if precomputed_non_cb.len() != block.txdata.len().saturating_sub(1) { - return Err(ConsensusError::BadBlock("wtxid count mismatch")); - } - // Witness merkle: coinbase wtxid is 32 zero bytes. + // Witness merkle: coinbase wtxid is 32 zero bytes. The caller builds + // `precomputed_non_cb` as one wtxid per non-coinbase tx. let mut leaves = Vec::with_capacity(block.txdata.len()); leaves.push([0u8; 32]); leaves.extend_from_slice(precomputed_non_cb); diff --git a/crates/rbitcoin-mempool/src/store.rs b/crates/rbitcoin-mempool/src/store.rs index 7d19718001..8d60bc8332 100644 --- a/crates/rbitcoin-mempool/src/store.rs +++ b/crates/rbitcoin-mempool/src/store.rs @@ -724,8 +724,7 @@ impl Mempool { let logical = body_logical_len(&self.body)?; let end = logical.saturating_add(need); if end > self.body.len() { - let new_len = (self.body.len().max(4096) * 2).max(end + 4096); - self.body.resize(new_len, 0); + self.body.resize(end, 0); } self.body[8..16].copy_from_slice(&(end as u64).to_le_bytes()); Ok(logical as u64) @@ -740,31 +739,20 @@ impl Mempool { let body_path = self.dir.join("tx.body"); let logical = body_logical_len(&self.body)? as u64; let start = self.body_persisted_len.min(logical); - if start < BODY_HEADER as u64 { + if logical != start { self.body_file - .seek(SeekFrom::Start(0)) + .seek(SeekFrom::Start(8)) .map_err(|e| MempoolError::io(&body_path, e))?; self.body_file - .write_all(&self.body[..logical as usize]) + .write_all(&self.body[8..16]) .map_err(|e| MempoolError::io(&body_path, e))?; - } else { - if start > BODY_HEADER as u64 || logical != start { - self.body_file - .seek(SeekFrom::Start(8)) - .map_err(|e| MempoolError::io(&body_path, e))?; - self.body_file - .write_all(&self.body[8..16]) - .map_err(|e| MempoolError::io(&body_path, e))?; - } - if logical > start { - self.body_file - .seek(SeekFrom::Start(start)) - .map_err(|e| MempoolError::io(&body_path, e))?; - self.body_file - .write_all(&self.body[start as usize..logical as usize]) - .map_err(|e| MempoolError::io(&body_path, e))?; - } } + self.body_file + .seek(SeekFrom::Start(start)) + .map_err(|e| MempoolError::io(&body_path, e))?; + self.body_file + .write_all(&self.body[start as usize..logical as usize]) + .map_err(|e| MempoolError::io(&body_path, e))?; self.body_file .set_len(logical) .map_err(|e| MempoolError::io(&body_path, e))?; @@ -1089,7 +1077,6 @@ fn open_or_init_slots(path: &Path, slot_cap: u32) -> Result<(File, Vec), Mem } fn open_or_init_body(path: &Path) -> Result<(File, Vec), MempoolError> { - let initial = BODY_HEADER + 64; if path.exists() { let mut file = OpenOptions::new() .read(true) @@ -1116,10 +1103,6 @@ fn open_or_init_body(path: &Path) -> Result<(File, Vec), MempoolError> { if logical > len { return Err(MempoolError::Corrupt("body logical past file")); } - // Spare capacity in process for appends (not on disk until persist). - if buf.len() < initial { - buf.resize(initial, 0); - } Ok((file, buf)) } else { let mut file = OpenOptions::new() @@ -1128,11 +1111,11 @@ fn open_or_init_body(path: &Path) -> Result<(File, Vec), MempoolError> { .create_new(true) .open(path) .map_err(|e| MempoolError::io(path, e))?; - let mut buf = vec![0u8; initial]; + let mut buf = vec![0u8; BODY_HEADER]; buf[0..4].copy_from_slice(&MEM_MAGIC); buf[4..6].copy_from_slice(&MEM_SCHEMA.to_le_bytes()); buf[8..16].copy_from_slice(&(BODY_HEADER as u64).to_le_bytes()); - file.write_all(&buf[..BODY_HEADER]) + file.write_all(&buf) .map_err(|e| MempoolError::io(path, e))?; file.flush().map_err(|e| MempoolError::io(path, e))?; Ok((file, buf)) @@ -1314,6 +1297,39 @@ pub(crate) mod tests { assert!(mp.dir().join("slots").exists()); assert!(mp.dir().join("tx.body").exists()); } + assert_eq!( + fs::metadata(dir.join("tx.body")).unwrap().len(), + BODY_HEADER as u64, + "an empty body file is the 16-byte header" + ); + let slots_path = dir.join("slots"); + let slots = fs::read(&slots_path).unwrap(); + let mut longer = slots.clone(); + longer.push(0xab); + fs::write(&slots_path, &longer).unwrap(); + Mempool::open_or_create(&dir).expect("slots longer than the table still open"); + fs::write(&slots_path, &slots[..slots.len() - 1]).unwrap(); + let err = match Mempool::open_or_create(&dir) { + Err(e) => e, + Ok(_) => panic!("a short slots file must not open"), + }; + assert!( + matches!(err, MempoolError::Corrupt("slots file short")), + "{err}" + ); + fs::write(&slots_path, &slots).unwrap(); + let body_path = dir.join("tx.body"); + let body = fs::read(&body_path).unwrap(); + assert_eq!(body.len(), BODY_HEADER); + fs::write(&body_path, &body[..BODY_HEADER - 1]).unwrap(); + let err = match Mempool::open_or_create(&dir) { + Err(e) => e, + Ok(_) => panic!("a short body file must not open"), + }; + assert!( + matches!(err, MempoolError::Corrupt("body too short")), + "{err}" + ); let _ = fs::remove_dir_all(&dir); } @@ -1461,6 +1477,16 @@ pub(crate) mod tests { .unwrap(); let persisted = mp.body_persisted_len(); mp.persist_slots_without_body().unwrap(); + let on_disk = fs::read(dir.join("slots")).unwrap(); + assert_eq!( + on_disk[SLOTS_HEADER], SLOT_FREE, + "a live row past the durable body is free on disk" + ); + assert_eq!( + &on_disk[SLOTS_HEADER + 16..SLOTS_HEADER + 48], + t1.as_byte_array(), + "demotion keeps the txid and only clears LIVE" + ); drop(mp); let mp = Mempool::open_or_create(&dir).unwrap(); let live = mp @@ -1738,9 +1764,19 @@ pub(crate) mod tests { "expected grow on full table, got {:?}", r.err().map(|e| e.to_string()) ); - assert!(mp.meta().slot_cap > tiny); + assert_eq!(mp.meta().slot_cap, 16_388); assert_eq!(mp.live_count(), tiny + 1); - // Must not be the old Corrupt message. + drop(mp); + assert_eq!( + fs::metadata(dir.join("slots")).unwrap().len(), + 786_640, + "grown slots file is the 16-byte header plus 16388 records" + ); + let mp = Mempool::open_or_create(&dir).unwrap(); + assert_eq!(mp.meta().slot_cap, 16_388); + // Grow persists the four rows already in the table. The fifth slot is + // RAM until the next sidecar write, so a reopen keeps those four. + assert_eq!(mp.load_live_txs().unwrap().len(), 4); assert!(!format!("{:?}", MempoolError::Full).contains("corrupt")); let _ = fs::remove_dir_all(&dir); } diff --git a/crates/rbitcoin-net/src/asmap.rs b/crates/rbitcoin-net/src/asmap.rs index f6608a8eeb..23848213d5 100644 --- a/crates/rbitcoin-net/src/asmap.rs +++ b/crates/rbitcoin-net/src/asmap.rs @@ -42,18 +42,12 @@ impl AsMap { Ok(Self::from_bytes(bytes)) } + // An empty file fails sanity, so a loaded map is never empty. + #[allow(clippy::len_without_is_empty)] pub fn len(&self) -> usize { self.bytes.len() } - pub fn is_empty(&self) -> bool { - self.bytes.is_empty() - } - - pub fn as_bytes(&self) -> &[u8] { - &self.bytes - } - pub fn interpret_ip16(&self, ip16: &[u8; 16]) -> u32 { interpret(&self.bytes, ip16) } @@ -76,6 +70,53 @@ pub const TWO_PREFIX_ASMAP: &[u8] = &[ 0x00, 0x00, 0x01, ]; +/// `::ffff:127.0.0.1` → ASN 1, `.2` → ASN 2, `.3` → ASN 9. +/// +/// Default ASN 9, a one-bit match then an eight-bit match, the shared +/// `::ffff:127.0.0.0/120` prefix, then a jump on the bit that separates `.1` +/// from `.2` and `.3`. The fall-through consumes the last IP bit and returns 1. +/// The jump target returns 2, or the default when that bit does not match (`.3`). +pub const LOOPBACK_ASMAP: &[u8] = &[ + 0x07, 0x80, 0x18, 0xfb, 0x03, 0xec, 0x0f, 0xb0, 0x3f, 0xc0, 0xfe, 0x00, 0xfb, 0x03, 0xec, 0x0f, + 0xb0, 0x3f, 0xc0, 0xfe, 0x00, 0xfb, 0x03, 0xec, 0x07, 0xd8, 0xff, 0x7f, 0xff, 0xff, 0xfd, 0xfd, + 0xf7, 0x07, 0xd8, 0x1f, 0x60, 0x1f, 0x10, 0x3a, 0x01, 0x00, 0x0c, 0x00, 0x80, +]; + +/// Every address returns ASN 1. Length 3, digest `709e80c8`. +pub const LONE_RETURN_ASMAP: &[u8] = &[0x00, 0x00, 0x00]; + +/// A second `DEFAULT`. Sanity rejects it. +pub const ASMAP_SECOND_DEFAULT: &[u8] = &[0x07, 0x00, 0x38, 0x00, 0xc0, 0x00, 0x00, 0x08]; + +/// Return ASN 1 plus a trailing one bit. +pub const ASMAP_TRAILING_ONE: &[u8] = &[0x00, 0x00, 0x02]; + +/// Return ASN 1 plus eight trailing zero bits. +pub const ASMAP_EIGHT_TRAILING_ZEROS: &[u8] = &[0x00, 0x00, 0x00, 0x00]; + +/// Two one-bit matches in a row, then a return. +pub const ASMAP_TWO_SHORT_MATCHES: &[u8] = &[0x63, 0x00, 0x00, 0x00]; + +/// Sixteen full-byte matches, then one more match bit. +pub const ASMAP_MATCH_PAST_END: &[u8] = &[ + 0xfb, 0x03, 0xec, 0x0f, 0xb0, 0x3f, 0xc0, 0xfe, 0x00, 0xfb, 0x03, 0xec, 0x0f, 0xb0, 0x3f, 0xc0, + 0xfe, 0x00, 0xfb, 0x03, 0xec, 0x0f, 0xb0, 0x3f, 0xc0, 0xfe, 0x00, 0xfb, 0x03, 0xec, 0x0f, 0xb0, + 0x3f, 0xc0, 0xfe, 0x00, 0x03, 0x00, 0x00, +]; + +/// A jump of 5000 bits, then a return. +pub const ASMAP_JUMP_PAST_FILE: &[u8] = &[0xfd, 0x71, 0x3a, 0x00, 0x00]; + +/// A jump that leaves too few IP bits for the match after it. +pub const ASMAP_JUMP_SHORT_TAIL: &[u8] = &[ + 0xfb, 0x03, 0xec, 0x0f, 0xb0, 0x3f, 0xc0, 0xfe, 0x00, 0xfb, 0x03, 0xec, 0x0f, 0xb0, 0x3f, 0xc0, + 0xfe, 0x00, 0xfb, 0x03, 0xec, 0x0f, 0xb0, 0x3f, 0xc0, 0xfe, 0x00, 0xfb, 0x03, 0xec, 0x0f, 0xb0, + 0x3f, 0x40, 0x00, 0x00, 0x80, 0xfd, 0x01, 0x00, 0x00, 0x02, +]; + +/// Smallest jump, with nothing after it. +pub const ASMAP_JUMP_OFF_END: &[u8] = &[0x01]; + /// IPv4 as `::ffff:0:0/96` + 32 bits; IPv6 as 16 raw bytes. pub fn ip16_for_lookup(ip: IpAddr) -> [u8; 16] { match ip { @@ -158,13 +199,16 @@ fn bit_width(x: u32) -> u32 { } /// Core `Interpret`. Aborts return `0` (never panic). +/// +/// There is no `pos` bound. At or past the last bit, [`decode_type`] yields +/// an invalid opcode and this returns 0, the same result as stopping when +/// `pos` reaches the bit length. pub fn interpret(asmap: &[u8], ip16: &[u8; 16]) -> u32 { let mut pos = 0usize; - let endpos = asmap.len().saturating_mul(8); let mut ip_bit = 0usize; let ip_bits_end = 128usize; let mut default_asn = 0u32; - while pos < endpos { + loop { let opcode = decode_type(&mut pos, asmap); if opcode == RETURN { let asn = decode_asn(&mut pos, asmap); @@ -180,9 +224,6 @@ pub fn interpret(asmap: &[u8], ip16: &[u8; 16]) -> u32 { if ip_bit == ip_bits_end { break; } - if i64::from(jump) >= (endpos - pos) as i64 { - break; - } let Some(bit) = consume_bit_be(&mut ip_bit, ip16) else { break; }; @@ -274,7 +315,6 @@ fn sanity_return( fn sanity_jump( pos: &mut usize, asmap: &[u8], - endpos: usize, bits: &mut i32, jumps: &mut Vec<(usize, i32)>, ) -> bool { @@ -282,9 +322,8 @@ fn sanity_jump( if jump == INVALID { return false; } - if i64::from(jump) > (endpos - *pos) as i64 { - return false; - } + // A jump to EOF or past it cannot be popped onto a finishing RETURN + // (the target compare fails, or the walker falls off and returns false). if *bits == 0 { return false; } @@ -344,7 +383,7 @@ fn sanity_check_bits(asmap: &[u8], mut bits: i32) -> bool { None => prevopcode = JUMP, } } else if opcode == JUMP { - if !sanity_jump(&mut pos, asmap, endpos, &mut bits, &mut jumps) { + if !sanity_jump(&mut pos, asmap, &mut bits, &mut jumps) { return false; } prevopcode = JUMP; @@ -476,38 +515,7 @@ fn encode_two_prefix_asmap() -> Vec { #[cfg(test)] mod tests { use super::*; - use std::net::{Ipv4Addr, Ipv6Addr}; - - #[test] - fn asmap_two_prefix_maps_distinct_asns() { - let bytes = two_prefix_asmap_bytes(); - let m = AsMap::from_bytes(bytes).expect("fixture sanity"); - assert_eq!(m.mapped_as(IpAddr::V4(Ipv4Addr::new(1, 2, 0, 0))), 1); - assert_eq!(m.mapped_as(IpAddr::V4(Ipv4Addr::new(3, 4, 0, 0))), 2); - assert_eq!(m.mapped_as(IpAddr::V4(Ipv4Addr::new(1, 2, 9, 9))), 1); - let mapped = IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0xffff, 0x0102, 0x0304)); - assert_eq!(m.mapped_as(mapped), 1); - assert_eq!(m.mapped_as(IpAddr::V6(Ipv6Addr::LOCALHOST)), 0); - } - - #[test] - fn asmap_truncated_file_is_none() { - assert!(AsMap::from_bytes(Vec::new()).is_none()); - assert!(AsMap::from_bytes(vec![0]).is_none()); - let dir = std::env::temp_dir().join(format!( - "rbitcoin-asmap-trunc-{}-{}", - std::process::id(), - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_nanos() - )); - std::fs::create_dir_all(&dir).unwrap(); - let path = dir.join("bad.dat"); - std::fs::write(&path, [0u8]).unwrap(); - assert!(AsMap::from_path(&path).unwrap().is_none()); - let _ = std::fs::remove_dir_all(&dir); - } + use std::net::{IpAddr, Ipv4Addr}; #[test] fn asmap_interpret_garbage_returns_zero_no_panic() { @@ -516,15 +524,6 @@ mod tests { assert_eq!(interpret(&[0], &[0u8; 16]), 0); } - #[test] - fn asmap_lone_return_asn1() { - let bits = instr_return(1); - let bytes = pack_le(&bits); - let m = AsMap::from_bytes(bytes).expect("RETURN ASN1"); - assert_eq!(m.interpret_ip16(&[0u8; 16]), 1); - assert_eq!(m.interpret_ip16(&[0xff; 16]), 1); - } - #[test] fn asmap_two_prefix_const_matches_encoder() { assert_eq!(two_prefix_asmap_bytes(), TWO_PREFIX_ASMAP); @@ -542,18 +541,4 @@ mod tests { assert_eq!(m.interpret_ip16(&ip16), 1); assert_eq!(ip16, ip16_for_lookup(IpAddr::V4(Ipv4Addr::new(1, 2, 0, 0)))); } - - #[test] - fn sanity_check_bits_rejects_jump_at_zero_bits_and_double_default() { - assert!(!sanity_check_bits(&[], 128)); - let mut bits = instr_jump(17); - bits.extend(instr_return(1)); - assert!(!sanity_check_bits(&pack_le(&bits), 0)); - let mut bits = instr_type(DEFAULT); - bits.extend(encode_bits(1, 1, &ASN_BIT_SIZES)); - bits.extend(instr_type(DEFAULT)); - bits.extend(encode_bits(1, 1, &ASN_BIT_SIZES)); - bits.extend(instr_return(1)); - assert!(AsMap::from_bytes(pack_le(&bits)).is_none()); - } } diff --git a/crates/rbitcoin-net/src/cache.rs b/crates/rbitcoin-net/src/cache.rs index 77e4befbab..fc5c0b9768 100644 --- a/crates/rbitcoin-net/src/cache.rs +++ b/crates/rbitcoin-net/src/cache.rs @@ -59,15 +59,11 @@ impl BlockCache { self.inner.read().unwrap().chain.last().copied() } - pub fn len(&self) -> usize { - self.inner.read().unwrap().chain.len() - } - pub fn is_empty(&self) -> bool { self.inner.read().unwrap().chain.is_empty() } - pub fn body_count(&self) -> usize { + pub(crate) fn stored_bodies(&self) -> usize { self.inner.read().unwrap().by_hash.len() } @@ -137,10 +133,9 @@ impl BlockCache { } g.by_hash.insert(hash, block); g.chain.push(hash); - let depth = self.body_depth; - if g.chain.len() > depth { - let drop_to = g.chain.len() - depth; - let stale: Vec = g.chain[..drop_to].to_vec(); + let extra = g.chain.len().saturating_sub(self.body_depth); + if extra != 0 { + let stale: Vec = g.chain[..extra].to_vec(); for h in stale { g.by_hash.remove(&h); } @@ -148,31 +143,6 @@ impl BlockCache { Ok(()) } - /// Locator hashes newest-first for getheaders. - pub fn locator(&self) -> Vec { - let g = self.inner.read().unwrap(); - if g.chain.is_empty() { - return vec![BlockHash::from_byte_array([0u8; 32])]; - } - let mut out = Vec::new(); - let mut i = g.chain.len() as i64 - 1; - let mut step = 1i64; - while i >= 0 { - out.push(g.chain[i as usize]); - if out.len() >= 10 { - step *= 2; - } - i -= step; - } - // Always include genesis - if let Some(g0) = g.chain.first() { - if out.last() != Some(g0) { - out.push(*g0); - } - } - out - } - /// Headers after the best common locator hash, up to Core `MAX_HEADERS_RESULTS` (2000). /// /// Only returns headers still present as full bodies in the tip window; callers @@ -184,10 +154,6 @@ impl BlockCache { } let mut start = 0usize; 'outer: for loc in locator { - if loc.to_byte_array() == [0u8; 32] { - start = 0; - break; - } for (i, h) in g.chain.iter().enumerate() { if h == loc { start = i + 1; diff --git a/crates/rbitcoin-net/src/chain.rs b/crates/rbitcoin-net/src/chain.rs index 12eeca8954..45f4cd9a3d 100644 --- a/crates/rbitcoin-net/src/chain.rs +++ b/crates/rbitcoin-net/src/chain.rs @@ -274,8 +274,6 @@ pub(crate) fn reject_is_mutated(reason: &str) -> bool { || reason.contains("bad-txns-duplicate") || reason.contains("witness commitment") || reason.contains("bad-witness-nonce") - || reason.contains("missing witness commitment") - || reason.contains("wtxid count") || reason.contains("unexpected witness") } @@ -488,13 +486,13 @@ impl ChainHub { self.mining.block_version.store(v, Ordering::Relaxed); } - /// GBT `version`: override or versionbits TOP_BITS | testdummy (bit 28). + /// GBT `version`: override, or versionbits top bits with testdummy (`0x30000000`). pub fn gbt_block_version(&self) -> i32 { let v = self.mining.block_version.load(Ordering::Relaxed); if v != 0 { v } else { - 0x2000_0000 | (1 << 28) + 0x3000_0000 } } @@ -2817,7 +2815,7 @@ impl ChainHub { } pub fn cache_body_count(&self) -> usize { - self.cache.body_count() + self.cache.stored_bodies() } pub fn held_body_count(&self) -> usize { diff --git a/crates/rbitcoin-net/src/ibd/dial.rs b/crates/rbitcoin-net/src/ibd/dial.rs index bb16ba378e..344c556040 100644 --- a/crates/rbitcoin-net/src/ibd/dial.rs +++ b/crates/rbitcoin-net/src/ibd/dial.rs @@ -426,19 +426,6 @@ pub(crate) fn request_headers_from( Ok(s.cmd_tx.send(PeerCmd::GetHeaders { locator }).is_ok()) } -thread_local! { - static WARM_LOCATOR: std::cell::RefCell>> = - const { std::cell::RefCell::new(None) }; -} - -pub(crate) fn set_warm_locator(locator: Option>) { - WARM_LOCATOR.with(|slot| *slot.borrow_mut() = locator); -} - -fn warm_locator() -> Option> { - WARM_LOCATOR.with(|slot| slot.borrow().clone()) -} - /// Locator for IBD getheaders: prefer the **work-path tip** (highest ordered / /// archived hash) ahead of the confirmed store tip. /// @@ -451,26 +438,26 @@ pub(crate) fn ibd_header_locator( work_tips: &[BlockHash], ) -> Result, NetError> { let mut locator = Vec::with_capacity(32); + // Eight hashes is the walk's own stop. This only enforces the protocol + // maximum, so a stored tip prepended in front does not drop the oldest + // checkpoint. for h in work_tips { + if locator.len() >= crate::codec::MAX_LOCATOR_SZ { + break; + } if !locator.contains(h) { locator.push(*h); } - if locator.len() >= 8 { - break; - } } if let Some(t) = hub.tip_hash() { if !locator.contains(&t) { locator.push(t); } } - let rest = if let Some(warm) = warm_locator() { - warm - } else { - hub.query - .locator_hashes() - .map_err(|e| NetError::Consensus(e.to_string()))? - }; + let rest = hub + .query + .locator_hashes() + .map_err(|e| NetError::Consensus(e.to_string()))?; for h in rest { if !locator.contains(&h) { locator.push(h); diff --git a/crates/rbitcoin-net/src/ibd/events/mod.rs b/crates/rbitcoin-net/src/ibd/events/mod.rs index fb1593fe0e..f189582518 100644 --- a/crates/rbitcoin-net/src/ibd/events/mod.rs +++ b/crates/rbitcoin-net/src/ibd/events/mod.rs @@ -726,9 +726,7 @@ fn complete_stored_headers( && live < MAX_ORDERED_HEADERS && (live < ORDERED_HEADERS_SOFT_CAP || need_ready_headroom); } - if batch_len == 0 { - return false; - } + // A stored plan has at least one header. An empty batch never reaches here. on_known_headers_batch(st, hub, 0, batch_len); false } @@ -814,21 +812,19 @@ fn apply_block_framed( Ok(_offer) => { let _ = try_complete_awaiting_reorg(st, hub); } - Err(rbitcoin_store::StoreError::Rejected(why)) - if why == rbitcoin_query::Query::UNDECODABLE_WIRE_MSG => - { - warn!("ibd: peer[{peer}] block {hash} h={height}: {why}; dropping peer"); - st.body.mark_missing(hash); - st.reopen_for_densify(&[hash]); - disconnect_peer( - &mut st.slots, - &mut st.addr_cooldown, - &mut st.addr_strikes, - peer, - ); - return; - } Err(e) => { + if store_reject_drops_peer(&e) { + warn!("ibd: peer[{peer}] block {hash} h={height}: {e}; dropping peer"); + st.body.mark_missing(hash); + st.reopen_for_densify(&[hash]); + disconnect_peer( + &mut st.slots, + &mut st.addr_cooldown, + &mut st.addr_strikes, + peer, + ); + return; + } rbitcoin_log::warn!("ibd: body queue offer failed ({e}) h={height}"); st.body.mark_missing(hash); return; @@ -840,6 +836,17 @@ fn apply_block_framed( } } +/// Undecodable wire is the sender's block. Any other store refusal stays a +/// re-get: the peer may still have an honest copy. +fn store_reject_drops_peer(err: &rbitcoin_store::StoreError) -> bool { + match err { + rbitcoin_store::StoreError::Rejected(why) => { + *why == rbitcoin_query::Query::UNDECODABLE_WIRE_MSG + } + _ => false, + } +} + fn apply_block_decode_failed(st: &mut IbdWorkState, peer: usize, hash: BlockHash) { note_block_progress(&mut st.slots, peer); clear_hash_inflight(&mut st.slots, &mut st.inflight, hash); diff --git a/crates/rbitcoin-net/src/ibd/header_walk.rs b/crates/rbitcoin-net/src/ibd/header_walk.rs index 21fd5b3f5e..5c9e9211b6 100644 --- a/crates/rbitcoin-net/src/ibd/header_walk.rs +++ b/crates/rbitcoin-net/src/ibd/header_walk.rs @@ -186,25 +186,29 @@ impl HeaderWalk { if !out.contains(&h) { out.push(h); } - if i == 0 { - break; - } step = step.saturating_mul(2); } out } /// Height of the header after the candidate, when the walk has a tip. + /// + /// Callers run after the origin is installed, so a missing tip hash is the + /// only stop. A height of zero is the parent, not a later header. pub(crate) fn next_height(&self) -> Option { - if !self.origin || self.tip.hash.is_none() { - return None; - } + self.tip.hash?; Some(self.tip.height.saturating_add(1)) } /// `height` is on the candidate and past the queue, so hygiene keeps it. + /// + /// The origin and the tip hash are installed together. Hygiene then only + /// needs the height to sit on that tip, above the genesis parent. pub(crate) fn indexes_through(&self, height: u32) -> bool { - self.origin && self.tip.hash.is_some() && height > 0 && height <= self.tip.height + if self.tip.hash.is_none() || !self.origin || height == 0 { + return false; + } + height <= self.tip.height } /// Test stand-in for a walk that already proved the confirmed tip. @@ -296,9 +300,6 @@ pub(crate) fn load_origin(hub: &ChainHub, input: &OriginInput) -> OriginLoaded { let mut latch = WalkTip { height: *next_h, hash: Some(*next), - milestone_hash, - milestone_prev, - milestone_header, ..WalkTip::default() }; remember_milestone(&mut latch, hub, hdr.prev_blockhash, &hdr); @@ -387,7 +388,9 @@ pub(crate) struct HeaderStatusView { pub(crate) fn header_status_view(st: &IbdWorkState, hub: &ChainHub) -> HeaderStatusView { let walk = wants_lookahead(st); - let quiet = st.header_walk.origin && st.header_walk.proven && !walk; + // Proven is set only after the origin is installed, so origin is not a + // second input to the caught-up line. + let quiet = st.header_walk.proven && !walk; HeaderStatusView { origin: st.header_walk.origin, height: st.header_walk.tip_height(), @@ -427,16 +430,13 @@ pub(crate) fn log_status(st: &mut IbdWorkState, hub: &ChainHub, horizon: u32) { } } -/// A live peer is still ahead of the walk and has not been retired from it. +/// A connected peer is still ahead of the walk and has not been retired from it. /// -/// Before the origin is known, the advertised high-water mark is enough to -/// send the first ask. After that, advertised height alone does not keep the -/// lane on. +/// The advertised high-water mark is not a second signal. A disconnected +/// peer leaves that mark behind, and the ask goes out only when someone +/// still connected can serve the next header. pub(crate) fn wants_lookahead(st: &IbdWorkState) -> bool { let tip = st.header_walk.tip_height(); - if !st.header_walk.origin { - return st.max_peer_height > tip; - } st.slots .iter() .any(|s| s.alive && s.peer_height > tip && !st.header_walk.walk_quiet.contains(&s.id)) @@ -711,15 +711,16 @@ pub(crate) fn suppress_competing_chain( if st.header_walk.tip_hash() == Some(prev) || batch_ends_on_checkpoint(st, headers) { return false; } - if st.ordered.len() < ORDERED_HEADERS_SOFT_CAP && stored_top_hash(st, hub) == Some(prev) { - return false; - } + // A short queue whose batch continues the stored tip is a refill. Classify + // returns that before this function runs, so the length comparison is not + // a second gate. if challenger_extends(st, prev) { return extend_challenger(st, hub, Some(peer), headers); } // A batch that has not met the next checkpoint is not a competing chain. // Treating it as one would evict a real fork that has less work so far. - if st.header_walk.proven && !below && candidate_prev(st, prev) { + // `!proven && !below` already returned, so `proven` is not a second gate. + if !below && candidate_prev(st, prev) { if let Some(start_h) = height_of(st, hub, prev) { if matches!(path_agree(st, start_h, headers, true), PathAgree::Short) { return true; @@ -3212,6 +3213,10 @@ mod tests { let (s0, mut rx) = slot(0); let mut st = IbdWorkState::new(vec![s0], Some(gen), Some(0)); fill_queue(&mut st); + assert!( + !st.header_walk.has_checkpoints(), + "a walk with no replies has no checkpoints" + ); let before = hub.query.store().header_count(); log_status(&mut st, &hub, 50_000); assert!( diff --git a/crates/rbitcoin-net/src/ibd/mod.rs b/crates/rbitcoin-net/src/ibd/mod.rs index f2f6e65ef4..786ffd83c6 100644 --- a/crates/rbitcoin-net/src/ibd/mod.rs +++ b/crates/rbitcoin-net/src/ibd/mod.rs @@ -38,33 +38,46 @@ pub use dial::connect_timeout_for; pub use perf_log::{read_platform_rss, tip_perf_json, ProcessRss, TipPerfLog}; use archive::{rehydrate_block_queue_into_confirm, rehydrate_class_a_into_body_queue}; -use assign_plan::{need_ready_headroom, want_headers_beyond_soft_cap}; +pub(crate) use assign_plan::need_ready_headroom; +use assign_plan::want_headers_beyond_soft_cap; use confirm::{offer_confirm_ready, spawn_confirm_engine, ConfirmEvent, ConfirmFeed}; +#[cfg(test)] +pub(crate) use confirm::{ConfirmRejectClass, RejectParent}; -use assign::{assign_work_ordered, bq_pipeline_saturated, AssignDepth}; +use assign::bq_pipeline_saturated; +pub(crate) use assign::{assign_work_ordered, AssignDepth}; use cadence::IbdLoopCadence; use dial::{ admit_cooldown_fallback, alive_dial_addrs, apply_dial_result, dial_batch, dial_blocked_addrs, disconnect_relative_slow_block_peers, disconnect_stalled_block_peers, expire_addr_cooldown, redial_want, release_peer_block_work, }; -use events::{ - apply_confirm_events, apply_peer_event, disconnect_all_peers, drain_ready_peer_and_body_events, - update_confirm_lag, -}; +pub(crate) use events::drain_ready_peer_and_body_events; +use events::{apply_confirm_events, apply_peer_event, disconnect_all_peers, update_confirm_lag}; +#[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, }; -use peer_io::{empty_header_fk_bus, sample_peer_rates, PeerCmd, PeerEvent, PeerEventSinks}; -use progress::{ - read_progress_published, spawn_status_thread, work_chain_progress_at, StatusBoard, StatusSnap, - TipRateTracker, +#[cfg(test)] +pub(crate) use header_walk::{ + candidate_tip, candidate_work, forget_walk_peer, header_poll, header_status_view, + install_origin, load_origin, note_block_inv, note_reservation, origin_input, send_getheaders, + settle_walk_peer, take_header_ask, HeaderPoll, HeaderPollIn, }; -use published::spawn_header_thread; -use state::IbdWorkState; +pub(crate) use peer_io::{empty_header_fk_bus, PeerCmd, PeerEvent}; +use peer_io::{sample_peer_rates, PeerEventSinks}; +#[cfg(test)] +pub(crate) use peer_io::{solicit_track, GetdataWire, PeerSlot}; +#[cfg(test)] +pub(crate) use progress::{log_progress_tick, StatusClock}; +pub(crate) use progress::{read_progress_published, StatusSnap}; +use progress::{spawn_status_thread, work_chain_progress_at, StatusBoard, TipRateTracker}; +pub(crate) use published::spawn_header_thread; +pub(crate) use state::IbdWorkState; use state::{enqueue_header_cmd, HeaderCmd}; -use status::LoopStats; +pub(crate) use status::LoopStats; use crate::chain::ChainHub; use crate::codec::MAX_HEADERS_RESULTS; diff --git a/crates/rbitcoin-net/src/ibd/published.rs b/crates/rbitcoin-net/src/ibd/published.rs index 19b1f779cd..8a573b5edb 100644 --- a/crates/rbitcoin-net/src/ibd/published.rs +++ b/crates/rbitcoin-net/src/ibd/published.rs @@ -1,7 +1,6 @@ //! Header thread. It persists batches, loads the walk origin, and reseeds. //! The coordination loop only queues those commands. -use super::dial::set_warm_locator; use super::events::run_header_batch; use super::header_walk::{install_origin, load_origin, origin_input, send_getheaders}; use super::path::{load_resume_entries, stage_resumed_path}; @@ -35,7 +34,6 @@ pub(crate) fn spawn_header_thread( } else { Some(load_origin(hub.as_ref(), &input)) }; - set_warm_locator(hub.query.locator_hashes().ok()); { let mut st = shared.lock().unwrap_or_else(|err| err.into_inner()); if let Some(loaded) = loaded { @@ -43,7 +41,6 @@ pub(crate) fn spawn_header_thread( } let _ = send_getheaders(&mut st, hub.as_ref()); } - set_warm_locator(None); } HeaderCmd::Reseed => { let exclude: Vec<[u8; 32]> = { diff --git a/crates/rbitcoin-net/src/lib.rs b/crates/rbitcoin-net/src/lib.rs index 387cc51c79..24fa6739bf 100644 --- a/crates/rbitcoin-net/src/lib.rs +++ b/crates/rbitcoin-net/src/lib.rs @@ -31,7 +31,11 @@ mod tx_relay; mod v2; mod versionbits_warn; -pub use asmap::{AsMap, TWO_PREFIX_ASMAP}; +pub use asmap::{ + AsMap, ASMAP_EIGHT_TRAILING_ZEROS, ASMAP_JUMP_OFF_END, ASMAP_JUMP_PAST_FILE, + ASMAP_JUMP_SHORT_TAIL, ASMAP_MATCH_PAST_END, ASMAP_SECOND_DEFAULT, ASMAP_TRAILING_ONE, + ASMAP_TWO_SHORT_MATCHES, LONE_RETURN_ASMAP, LOOPBACK_ASMAP, TWO_PREFIX_ASMAP, +}; pub use cache::BlockCache; pub use chain::{check_block_proposal_with, AcceptOutcome, ChainHub, ChainTipInfo, TipEvent}; pub use compact::{ diff --git a/crates/rbitcoin-net/src/peer_header_dos_journey.rs b/crates/rbitcoin-net/src/peer_header_dos_journey.rs index d03e4be4f3..592fd5352e 100644 --- a/crates/rbitcoin-net/src/peer_header_dos_journey.rs +++ b/crates/rbitcoin-net/src/peer_header_dos_journey.rs @@ -1775,3 +1775,108 @@ async fn below_floor_full_batch_does_not_page() { ); let _ = std::fs::remove_dir_all(dir); } + +/// A side block three days behind the best header is still served. Every +/// `+` and `/` rewrite of the 30-day limit is shorter than three days. +#[test] +fn three_day_side_block_is_served() { + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("stale-serve"); + hub.ensure_genesis().unwrap(); + let gen = hub.tip_hash().unwrap(); + let t0 = 1_700_000_000u32; + hub.clock.set_mock(i64::from(t0)); + let old = rbitcoin_consensus::mine_regtest_paying(gen, t0, 1, op_true(), vec![]); + hub.accept_block(old.clone()).unwrap(); + let old_hash = old.block_hash(); + hub.invalidate_block(old_hash).unwrap(); + let t1 = t0 + 3 * 24 * 60 * 60; + hub.clock.set_mock(i64::from(t1)); + let tip = rbitcoin_consensus::mine_regtest_paying(gen, t1, 1, op_true(), vec![]); + hub.accept_block(tip).unwrap(); + assert!(!hub.is_connected(&old_hash)); + let served = + headers_reply_for_getheaders(&hub, &GetHeadersMessage::new(vec![], old_hash)).unwrap(); + assert_eq!( + served.len(), + 1, + "a side block three days behind the tip is served" + ); + assert_eq!(served[0].block_hash(), old_hash); + let _ = std::fs::remove_dir_all(dir); +} + +/// The store has no tip. `getheaders` is answered from the recent body cache. +#[test] +fn getheaders_serves_cached_bodies_when_the_store_has_no_tip() { + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("cache-serve"); + assert!(hub.tip_height().is_none()); + let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); + hub.cache.push_best(genesis.clone()).unwrap(); + let mut prev = genesis.block_hash(); + let mut time = genesis.header.time; + let mut hashes = vec![prev]; + for h in 1..=3u32 { + time = time.saturating_add(600); + let block = rbitcoin_consensus::mine_regtest_paying(prev, time, h, op_true(), vec![]); + prev = block.block_hash(); + hashes.push(prev); + hub.cache.push_best(block).unwrap(); + } + let stop = hashes[2]; + let served = headers_reply_for_getheaders( + &hub, + &GetHeadersMessage::new(vec![hashes[0]], stop), + ) + .unwrap(); + assert_eq!(served.len(), 2, "headers start after the locator and stop at the hash"); + assert_eq!(served[0].block_hash(), hashes[1]); + assert_eq!(served[1].block_hash(), stop); + let _ = std::fs::remove_dir_all(dir); +} + +/// Noting a later block must not forget the earlier ask. A second peer +/// announcing the same headers does not fetch the first body again. +#[tokio::test] +async fn a_later_ask_keeps_the_earlier_block() { + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("asked-keep"); + hub.ensure_genesis().unwrap(); + let peers = crate::peers::PeerHub::new(); + let gen = hub.tip_hash().unwrap(); + let t = hub.tip_header().unwrap().time.saturating_add(1); + let a = rbitcoin_consensus::mine_regtest_paying(gen, t, 1, op_true(), vec![]); + let b = rbitcoin_consensus::mine_regtest_paying(a.block_hash(), t.saturating_add(1), 2, op_true(), vec![]); + let peer = live_peer(&peers, 18601, 40, false); + let (out_tx, mut out_rx) = mpsc::unbounded_channel(); + let mut follow = PeerFollowState::new(); + push( + &hub, + &out_tx, + &mut follow, + Some(peer.as_ref()), + NetworkMessage::Headers(vec![a.header, b.header]), + ) + .await; + let first = asked_blocks(&mut out_rx); + assert!( + first.contains(&a.block_hash()) && first.contains(&b.block_hash()), + "both bodies are fetched once, got {first:?}" + ); + + let peer2 = live_peer(&peers, 18602, 41, false); + let (out2, mut rx2) = mpsc::unbounded_channel(); + let mut follow2 = PeerFollowState::new(); + push( + &hub, + &out2, + &mut follow2, + Some(peer2.as_ref()), + NetworkMessage::Headers(vec![a.header, b.header]), + ) + .await; + let second = asked_blocks(&mut rx2); + assert!( + !second.contains(&a.block_hash()), + "the first body stays asked after the later one is noted, got {second:?}" + ); + let _ = std::fs::remove_dir_all(dir); +} diff --git a/crates/rbitcoin-net/src/peer_header_sync_journey.rs b/crates/rbitcoin-net/src/peer_header_sync_journey.rs new file mode 100644 index 0000000000..a0f5690dd4 --- /dev/null +++ b/crates/rbitcoin-net/src/peer_header_sync_journey.rs @@ -0,0 +1,1228 @@ +// Header-sync pins. Each chapter drives the IBD loop's own functions and +// checks the getheaders, getdata, shutdown, datadir header, or status line +// a peer or operator would see. + +use crate::ibd::{ + apply_confirm_reject, apply_headers_event, assign_work_ordered, candidate_tip, candidate_work, + drain_ready_peer_and_body_events, empty_header_fk_bus, forget_walk_peer, header_poll, + header_status_view, install_origin, load_origin, log_progress_tick, need_ready_headroom, + note_block_inv, note_reservation, origin_input, read_progress_published, + send_getheaders, settle_walk_peer, take_header_ask, + solicit_track, spawn_header_thread, AssignDepth, ConfirmRejectClass, GetdataWire, HeaderPoll, + HeaderPollIn, IbdWorkState, LoopStats, PeerCmd, PeerEvent, PeerSlot, RejectParent, StatusClock, + StatusSnap, ORDERED_HEADERS_SOFT_CAP, +}; +use crate::seeds::AddrMan; +use bitcoin::block::Header; +use bitcoin::{CompactTarget, TxMerkleNode}; +use std::sync::atomic::{AtomicU32, AtomicU64}; +use std::sync::{Arc, Mutex}; +use std::time::Instant; +use tokio::sync::mpsc; + +fn mine(prev: BlockHash, n: u32) -> Header { + let mut h = Header { + version: bitcoin::block::Version::from_consensus(4), + prev_blockhash: prev, + merkle_root: TxMerkleNode::from_byte_array([n as u8; 32]), + time: 1_600_000_000 + n, + bits: CompactTarget::from_consensus(0x207f_ffff), + nonce: n, + }; + rbitcoin_consensus::grind_regtest_pow(&mut h); + h +} + +fn chain_of(mut prev: BlockHash, n: u32) -> Vec
{ + let mut out = Vec::with_capacity(n as usize); + for i in 1..=n { + let h = mine(prev, i); + prev = h.block_hash(); + out.push(h); + } + out +} + +fn slot_at(id: usize, ip: Ipv4Addr, height: u32) -> (PeerSlot, mpsc::UnboundedReceiver) { + let (cmd_tx, rx) = mpsc::unbounded_channel(); + let task = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .spawn(async {}); + let addr = SocketAddr::new(IpAddr::V4(ip), 8333); + let (requested, solicited_bytes, solicited_ms) = solicit_track(); + ( + PeerSlot { + id, + addr, + net: crate::NetAddr::from_socket(addr), + cmd_tx, + in_flight: Default::default(), + requested, + solicited_bytes, + solicited_ms, + any_rx_ms: Arc::new(AtomicU64::new(0)), + getdata_wire: GetdataWire::new(), + peer_height: height, + connected_ms: 1, + first_data_ms: 0, + bytes_rx_total: Arc::new(AtomicU64::new(0)), + rate: Default::default(), + alive: true, + task, + }, + rx, + ) +} + +fn slot(id: usize, height: u32) -> (PeerSlot, mpsc::UnboundedReceiver) { + slot_at(id, Ipv4Addr::new(10, 9, 0, id as u8), height) +} + +fn hub_pair(label: &str) -> (rbitcoin_query::testutil::TempDir, crate::chain::ChainHub) { + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled(label); + hub.ensure_genesis().unwrap(); + (dir, hub) +} + +fn fresh( + label: &str, + height: u32, +) -> ( + rbitcoin_query::testutil::TempDir, + crate::chain::ChainHub, + IbdWorkState, + mpsc::UnboundedReceiver, +) { + let (dir, hub) = hub_pair(label); + let (s0, rx) = slot(0, height); + let st = IbdWorkState::new(vec![s0], hub.tip_hash(), hub.tip_height()); + (dir, hub, st, rx) +} + +fn poll_in(live: usize, headroom: bool, lag: u32, min_cache: usize, fan: usize) -> HeaderPollIn { + HeaderPollIn { + live, + need_ready_headroom: headroom, + lag, + min_cache, + fan, + } +} + +/// The coordination loop's header arm: ask, latch, or leave the peer alone. +fn drive_poll(st: &mut IbdWorkState, hub: &crate::chain::ChainHub, poll: HeaderPollIn) { + match header_poll(st, poll) { + HeaderPoll::Ask => { + let _ = send_getheaders(st, hub); + } + HeaderPoll::LatchDone => { + st.headers_done = true; + } + HeaderPoll::Wait => {} + } +} + +fn took_getheaders(rx: &mut mpsc::UnboundedReceiver) -> bool { + matches!(rx.try_recv(), Ok(PeerCmd::GetHeaders { .. })) +} + +fn locator_of(rx: &mut mpsc::UnboundedReceiver) -> Vec { + match rx.try_recv() { + Ok(PeerCmd::GetHeaders { locator }) => locator, + Ok(_) => panic!("expected getheaders, got another command"), + Err(_) => panic!("expected getheaders, channel was empty"), + } +} + +fn status_line(st: &IbdWorkState, hub: &crate::chain::ChainHub) -> String { + let view = header_status_view(st, hub); + let snap = StatusSnap { + peers: st.slots.iter().filter(|s| s.alive).count(), + horizon: view.horizon, + hole: 0, + header_origin: view.origin, + header_height: view.height, + header_stored: view.stored, + header_queue: view.queue, + header_work_ok: view.work_ok, + header_walk: view.walk, + header_quiet: view.quiet, + }; + rbitcoin_log::capture_logs(true); + let published = read_progress_published(hub.query.as_ref()); + let mut clock = StatusClock::new(published.tip.unwrap_or(0), Instant::now()); + log_progress_tick( + &mut clock, + &snap, + &published, + (0, 0, 0), + 0, + 0, + Instant::now(), + ); + let logs = rbitcoin_log::take_logs(); + rbitcoin_log::capture_logs(false); + logs.into_iter().map(|(_, msg)| msg).collect::>().join("\n") +} + +fn plant_origin(st: &mut IbdWorkState, hub: &crate::chain::ChainHub) { + let input = origin_input(st, hub); + let loaded = load_origin(hub, &input); + install_origin(st, hub, loaded); +} + +fn fill_soft_queue(st: &mut IbdWorkState) { + st.ordered.clear(); + st.ordered_set.clear(); + for i in 0..ORDERED_HEADERS_SOFT_CAP { + let mut raw = [0xab; 32]; + raw[..4].copy_from_slice(&(i as u32).to_le_bytes()); + let placeholder = BlockHash::from_byte_array(raw); + st.ordered.push_back(placeholder); + st.ordered_set.insert(placeholder); + } +} + +/// Retarget every 20 headers. Look-ahead batches are not in the header store, +/// so a later fork has to keep the period header it saw. +fn arm_retarget(hub: &mut crate::chain::ChainHub) { + hub.params.btc.no_pow_retargeting = false; + hub.params.btc.allow_min_difficulty_blocks = false; + let spacing = hub.params.btc.pow_target_spacing; + hub.params.btc.pow_target_timespan = spacing.saturating_mul(20); +} + +fn mine_bits(prev: BlockHash, n: u32, time: u32, bits: CompactTarget, salt: u8) -> Header { + let mut merkle = [salt; 32]; + merkle[28..32].copy_from_slice(&n.to_le_bytes()); + let mut h = Header { + version: bitcoin::block::Version::from_consensus(4), + prev_blockhash: prev, + merkle_root: TxMerkleNode::from_byte_array(merkle), + time, + bits, + nonce: 0, + }; + rbitcoin_consensus::grind_regtest_pow(&mut h); + h +} + +fn chain_time(genesis_time: u32, spacing: u32, height: u32) -> u32 { + genesis_time.saturating_add(spacing.saturating_mul(height)) +} + +fn bits_for( + hub: &crate::chain::ChainHub, + genesis: &Header, + prior: &[Header], + height: u32, + time: u32, + parent: &Header, +) -> CompactTarget { + let interval = hub.params.difficulty_adjustment_interval(); + let period_first = if interval > 0 && height.is_multiple_of(interval) { + let start = height - interval; + if start == 0 { + Some(genesis.time) + } else { + prior.get((start as usize) - 1).map(|h| h.time) + } + } else { + None + }; + rbitcoin_consensus::next_work_bits( + &hub.params, + height, + parent.bits, + parent.time, + time, + period_first, + |_| None, + ) + .expect("difficulty bits") +} + +fn extend_chain( + hub: &crate::chain::ChainHub, + genesis: &Header, + prior: &[Header], + extra: u32, + salt: u8, +) -> Vec
{ + let spacing = hub.params.btc.pow_target_spacing as u32; + let mut all = prior.to_vec(); + let mut parent = prior.last().copied().unwrap_or(*genesis); + let start = prior.len() as u32; + let mut out = Vec::with_capacity(extra as usize); + for n in 1..=extra { + let height = start + n; + let time = chain_time(genesis.time, spacing, height); + let bits = bits_for(hub, genesis, &all, height, time, &parent); + let hdr = mine_bits(parent.block_hash(), height, time, bits, salt); + all.push(hdr); + out.push(hdr); + parent = hdr; + } + out +} + +/// One accepted header, then an empty download queue, so the walk sits above +/// the stored tip while the ordered-height mark says the path is already there. +/// One stored header, recorded before the walk has an origin, so the origin +/// load lifts the candidate without a checkpoint. The queue is then a full +/// window of placeholders whose tail is the confirmed tip: a later ask is +/// ahead of that path and is not filled from the header index. +fn walk_ahead_of_store(st: &mut IbdWorkState, hub: &crate::chain::ChainHub) -> BlockHash { + let gen = hub.tip_hash().unwrap(); + let hdr = mine(gen, 1); + let hash = hdr.block_hash(); + apply_headers_event(st, hub, 0, vec![hdr]); + plant_origin(st, hub); + st.ordered.clear(); + st.ordered_set.clear(); + for i in 0..ORDERED_HEADERS_SOFT_CAP { + let mut raw = [0xab; 32]; + raw[..4].copy_from_slice(&(i as u32).to_le_bytes()); + let placeholder = BlockHash::from_byte_array(raw); + st.ordered.push_back(placeholder); + st.ordered_set.insert(placeholder); + } + let tail = st.ordered.back().copied().expect("soft-cap queue"); + st.hash_height.insert(tail, 0); + st.max_ordered_height = u32::MAX; + hash +} + +#[test] +fn caught_up_empty_path_does_not_ask_again() { + let (dir, hub, mut st, mut rx) = fresh("sync-latch", 0); + drive_poll(&mut st, &hub, poll_in(0, false, 0, 10, 0)); + st.slots[0].peer_height = 100_000; + drive_poll(&mut st, &hub, poll_in(0, false, 0, 10, 1)); + assert!( + !took_getheaders(&mut rx), + "an empty path with no peer ahead latches; a later advertisement waits" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn soft_cap_with_lag_does_not_open_header_sync() { + let (dir, hub, mut st, mut rx) = fresh("sync-le", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + drive_poll( + &mut st, + &hub, + poll_in(ORDERED_HEADERS_SOFT_CAP, false, 1, 1, 0), + ); + assert!( + !took_getheaders(&mut rx), + "at the soft cap a lagging queue does not ask unless headroom, a taller peer, or stored look-ahead says so" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn headroom_past_the_soft_cap_still_asks() { + let (dir, hub, mut st, mut rx) = fresh("sync-head", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + drive_poll( + &mut st, + &hub, + poll_in(ORDERED_HEADERS_SOFT_CAP, true, 0, 1, 0), + ); + assert!( + took_getheaders(&mut rx), + "ready headroom past the soft cap asks for the next header" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_taller_peer_past_the_soft_cap_still_asks() { + let (dir, hub, mut st, mut rx) = fresh("sync-wants", 100_000); + walk_ahead_of_store(&mut st, &hub); + st.max_ordered_height = u32::MAX; + drive_poll( + &mut st, + &hub, + poll_in(ORDERED_HEADERS_SOFT_CAP, false, 0, 1, 0), + ); + assert!( + took_getheaders(&mut rx), + "a connected peer above the walk asks past the soft cap" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn look_ahead_past_the_soft_cap_still_asks() { + let (dir, hub, mut st, mut rx) = fresh("sync-walk", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + st.max_ordered_height = 0; + drive_poll( + &mut st, + &hub, + poll_in(ORDERED_HEADERS_SOFT_CAP, false, 1, 1, 0), + ); + assert!( + took_getheaders(&mut rx), + "a walk tip past the stored path asks past the soft cap" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_short_queue_asks_until_the_cache_is_full() { + let (dir, hub, mut st, mut rx) = fresh("ask-lt", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + drive_poll(&mut st, &hub, poll_in(5, false, 0, 10, 0)); + assert!( + took_getheaders(&mut rx), + "five queued headers against a ten-header cache still asks" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_full_cache_without_lag_does_not_ask() { + let (dir, hub, mut st, mut rx) = fresh("ask-le", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + drive_poll(&mut st, &hub, poll_in(5, false, 0, 5, 0)); + assert!( + !took_getheaders(&mut rx), + "a queue that already fills the cache does not ask" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn lag_asks_even_when_the_cache_is_full() { + let (dir, hub, mut st, mut rx) = fresh("ask-lag", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + drive_poll(&mut st, &hub, poll_in(10, false, 3, 5, 0)); + assert!( + took_getheaders(&mut rx), + "lag behind the walk asks even when the cache is full" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn header_lag_is_its_own_reason_to_ask() { + let (dir, hub, mut st, mut rx) = fresh("ask-or-lag", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + drive_poll(&mut st, &hub, poll_in(10, false, 3, 5, 0)); + assert!(took_getheaders(&mut rx), "lag alone asks"); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn ready_headroom_asks_when_the_cache_is_full() { + let (dir, hub, mut st, mut rx) = fresh("ask-or-head", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + drive_poll(&mut st, &hub, poll_in(10, true, 0, 5, 0)); + assert!(took_getheaders(&mut rx), "headroom alone asks"); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_taller_peer_asks_when_the_cache_is_full() { + let (dir, hub, mut st, mut rx) = fresh("ask-or-wants", 100_000); + walk_ahead_of_store(&mut st, &hub); + st.max_ordered_height = u32::MAX; + drive_poll(&mut st, &hub, poll_in(10, false, 0, 5, 0)); + assert!(took_getheaders(&mut rx), "a taller peer alone asks"); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn stored_look_ahead_asks_when_the_cache_is_full() { + let (dir, hub, mut st, mut rx) = fresh("ask-or-walk", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + st.max_ordered_height = 0; + drive_poll(&mut st, &hub, poll_in(10, false, 0, 5, 0)); + assert!( + took_getheaders(&mut rx), + "look-ahead past the stored path alone asks" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn headers_already_done_does_not_ask() { + let (dir, hub, mut st, mut rx) = fresh("poll-done", 100_000); + st.headers_done = true; + drive_poll(&mut st, &hub, poll_in(5, false, 3, 10, 1)); + assert!( + !took_getheaders(&mut rx), + "a latched header sync does not ask again" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn an_empty_queue_with_no_fan_latches_without_asking() { + let (dir, hub, mut st, mut rx) = fresh("poll-zero", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + drive_poll(&mut st, &hub, poll_in(0, false, 0, 10, 0)); + assert!( + !took_getheaders(&mut rx), + "no live headers and no fan latches instead of asking" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_block_announcement_asks_immediately() { + let (dir, hub, mut st, mut rx) = fresh("poll-inv", 100_000); + // No challenge is pending. The poll must not treat an idle lane as an ask. + drive_poll(&mut st, &hub, poll_in(0, false, 0, 10, 0)); + assert!( + !took_getheaders(&mut rx), + "without a block announcement an idle lane does not ask" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn locator_names_the_earlier_checkpoint() { + let (dir, hub, mut st, mut rx) = fresh("loc-ck", 100_000); + plant_origin(&mut st, &hub); + assert!(send_getheaders(&mut st, &hub).unwrap()); + let _ = rx.try_recv(); + let gen = hub.tip_hash().unwrap(); + let first = mine(gen, 1); + let first_hash = first.block_hash(); + apply_headers_event(&mut st, &hub, 0, vec![first]); + while rx.try_recv().is_ok() {} + assert!(send_getheaders(&mut st, &hub).unwrap()); + while rx.try_recv().is_ok() {} + let second = mine(first_hash, 2); + apply_headers_event(&mut st, &hub, 0, vec![second]); + while rx.try_recv().is_ok() {} + assert!(send_getheaders(&mut st, &hub).unwrap()); + let locator = locator_of(&mut rx); + assert!( + locator.contains(&first_hash), + "the getheaders locator names the earlier checkpoint, got {locator:?}" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn getheaders_locator_stops_at_eight_hashes() { + let (dir, hub, mut st, mut rx) = fresh("loc-eight", 100_000); + plant_origin(&mut st, &hub); + let gen = hub.tip_hash().unwrap(); + // One header per batch: each accepted batch is one checkpoint. 256 keeps + // the eighth hash off checkpoint 0, so a ninth comparison still names one. + let headers = chain_of(gen, 256); + let early = headers[1].block_hash(); + for hdr in headers { + apply_headers_event(&mut st, &hub, 0, vec![hdr]); + } + while rx.try_recv().is_ok() {} + assert!( + send_getheaders(&mut st, &hub).unwrap(), + "a peer still ahead of a long walk is asked" + ); + let locator = locator_of(&mut rx); + assert!( + locator.contains(&early), + "getheaders names an early checkpoint, got {locator:?}" + ); + assert_eq!( + locator.len(), + 9, + "getheaders is eight walk hashes plus the confirmed tip, got {locator:?}" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn successor_at_the_walk_tip_is_what_gets_asked() { + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("next-h"); + let (s0, mut rx) = slot(0, 50_000); + let mut st = IbdWorkState::new(vec![s0], None, None); + let parent = BlockHash::from_byte_array([0x11; 32]); + st.record_height(parent, 0); + assert!( + send_getheaders(&mut st, &hub).unwrap(), + "nothing is indexed at the next height, so the peer is asked" + ); + assert!( + took_getheaders(&mut rx), + "the ask is a getheaders, not a skipped successor" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn full_header_window_asks_again_under_the_soft_cap() { + let (dir, hub, mut st, mut rx) = fresh("batch-under", 500_000); + let gen = hub.tip_hash().unwrap(); + let headers = chain_of(gen, crate::codec::MAX_HEADERS_RESULTS as u32); + apply_headers_event(&mut st, &hub, 0, headers); + assert!( + took_getheaders(&mut rx), + "a full header window under the soft cap asks for the next window" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn full_header_window_stops_at_the_soft_cap() { + let (dir, hub, mut st, mut rx) = fresh("batch-cap", 500_000); + let pad = ORDERED_HEADERS_SOFT_CAP - crate::codec::MAX_HEADERS_RESULTS; + for i in 0..pad { + let mut raw = [0u8; 32]; + raw[..4].copy_from_slice(&(i as u32).to_le_bytes()); + st.ordered_set.insert(BlockHash::from_byte_array(raw)); + } + let gen = hub.tip_hash().unwrap(); + let headers = chain_of(gen, crate::codec::MAX_HEADERS_RESULTS as u32); + apply_headers_event(&mut st, &hub, 0, headers); + assert!( + !took_getheaders(&mut rx), + "a full window that lands on the soft cap does not ask for more" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn ctrl_headers_are_applied_while_the_header_thread_has_room() { + let (dir, hub) = hub_pair("drain-room"); + let hub = Arc::new(hub); + let (s0, _rx) = slot(0, 10); + let mut st = IbdWorkState::new(vec![s0], hub.tip_hash(), hub.tip_height()); + let (tx, rx) = mpsc::channel(1); + st.header_cmd = Some(tx.clone()); + let shared = Arc::new(Mutex::new(st)); + let join = spawn_header_thread( + Arc::clone(&shared), + Arc::clone(&hub), + rx, + empty_header_fk_bus(), + ); + let gen = hub.tip_hash().unwrap(); + let hdr = mine(gen, 7); + let hash = hdr.block_hash(); + let (ctrl_tx, mut ctrl_rx) = mpsc::unbounded_channel(); + ctrl_tx + .send(PeerEvent::Headers { + peer: 0, + headers: vec![hdr], + }) + .unwrap(); + let (body_tx, mut body_rx) = mpsc::unbounded_channel(); + let mut book = AddrMan::new(); + let local = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1); + { + let mut st = shared.lock().unwrap(); + drain_ready_peer_and_body_events( + &mut st, + hub.as_ref(), + &mut body_rx, + &mut ctrl_rx, + &AtomicU32::new(0), + &LoopStats::default(), + &mut book, + local, + None, + ) + .unwrap(); + st.header_cmd = None; + } + drop(tx); + drop(ctrl_tx); + drop(body_tx); + join.join().unwrap(); + assert!( + hub.header_of(&hash).is_some(), + "a headers message is written while the header thread still has room" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_mutated_block_drops_the_sender_not_a_noban_neighbor() { + let (dir, hub) = hub_pair("punish-or"); + let (noban_slot, _noban_rx) = slot_at(1, Ipv4Addr::new(127, 0, 0, 1), 10); + let (sender, mut sender_rx) = slot_at(2, Ipv4Addr::new(10, 1, 1, 2), 10); + let mut st = IbdWorkState::new(vec![noban_slot, sender], hub.tip_hash(), hub.tip_height()); + let peers = crate::peers::PeerHub::new(); + let mut table = crate::net_permissions::NetPermTable::default(); + table + .whitelist + .push(crate::net_permissions::parse_whitelist("noban,out@127.0.0.1").unwrap()); + peers.set_net_perms(table); + st.perms = Some(peers); + let hash = BlockHash::from_byte_array([0x44; 32]); + apply_confirm_reject( + &mut st, + 1, + hash, + ConfirmRejectClass::SoftWire, + "bad-txnmrklroot", + None, + Some(&hub), + 1, + None, + Some(2), + RejectParent::default(), + ); + assert!( + matches!(sender_rx.try_recv(), Ok(PeerCmd::Shutdown)), + "the sender of a mutated block is dropped" + ); + assert!(st.slots[0].alive, "the noban neighbor stays connected"); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_batched_consensus_reject_does_not_reget_the_first_hash() { + let (dir, hub, mut st, mut rx) = fresh("cascade-and", 20); + let hash = BlockHash::from_byte_array([0x55; 32]); + st.record_height(hash, 50); + st.densify_scan_lo = 100; + st.body.mark_pending(hash); + apply_confirm_reject( + &mut st, + 50, + hash, + ConfirmRejectClass::ConsensusInvalid, + "consensus: bad-txns", + Some(hub.query.as_ref()), + Some(&hub), + 2, + None, + None, + RejectParent { + hash: Some(BlockHash::from_byte_array([0x01; 32])), + unreadable: false, + }, + ); + let stats = LoopStats::default(); + let cfg = crate::ibd::IbdConfig::for_test(); + assign_work_ordered(&mut st, &hub, &cfg, &stats, AssignDepth::Full, None); + assert!( + !took_getdata(&mut rx), + "a two-block consensus wave does not ask again for the first hash" + ); + let _ = std::fs::remove_dir_all(dir); +} + +fn took_getdata(rx: &mut mpsc::UnboundedReceiver) -> bool { + matches!(rx.try_recv(), Ok(PeerCmd::GetData { .. })) +} + +#[test] +fn growing_the_queue_keeps_the_peer_on_the_walk() { + let (dir, hub, mut fast, mut fast_rx) = { + let (dir, hub) = hub_pair("settle-grow"); + let (mut a, ra) = slot(0, 100_000); + a.first_data_ms = 2; + let (b, _rb) = slot(1, 100_000); + let mut st = IbdWorkState::new(vec![a, b], hub.tip_hash(), hub.tip_height()); + plant_origin(&mut st, &hub); + (dir, hub, st, ra) + }; + let before = fast.ordered.len(); + let tip = candidate_tip(&fast); + let dummy = BlockHash::from_byte_array([0x61; 32]); + fast.ordered.push_back(dummy); + fast.ordered_set.insert(dummy); + settle_walk_peer(&mut fast, &hub, 0, tip, before, 1); + assert!(send_getheaders(&mut fast, &hub).unwrap()); + assert!( + took_getheaders(&mut fast_rx), + "a reply that queues a header keeps the fast peer as the ask" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn before_origin_a_short_reply_does_not_retire_the_peer() { + let (dir, hub) = hub_pair("settle-origin"); + let (mut a, mut ra) = slot(0, 100_000); + a.first_data_ms = 2; + let (b, _rb) = slot(1, 100_000); + let mut st = IbdWorkState::new(vec![a, b], hub.tip_hash(), hub.tip_height()); + settle_walk_peer(&mut st, &hub, 0, None, 0, 1); + assert!(send_getheaders(&mut st, &hub).unwrap()); + assert!( + took_getheaders(&mut ra), + "before the origin is known a short reply does not move the ask" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn equal_work_is_not_below_the_floor() { + let (dir, mut hub) = hub_pair("floor-eq"); + let (mut fast, mut fast_rx) = slot(0, 100_000); + fast.first_data_ms = 2; + fast.connected_ms = 1; + let (mut slow, mut slow_rx) = slot(1, 100_000); + slow.first_data_ms = 80; + slow.connected_ms = 1; + let mut st = IbdWorkState::new(vec![fast, slow], hub.tip_hash(), hub.tip_height()); + plant_origin(&mut st, &hub); + let work = candidate_work(&st); + hub.milestone = rbitcoin_consensus::Milestone { + height: 1, + anchor: Some(rbitcoin_consensus::MilestoneAnchor { + hash: hub.tip_hash().unwrap(), + min_work_be: work, + }), + }; + let tip = candidate_tip(&st); + let queued = st.ordered.len(); + settle_walk_peer(&mut st, &hub, 0, tip, queued, 1); + assert!(send_getheaders(&mut st, &hub).unwrap()); + assert!( + took_getheaders(&mut slow_rx), + "work equal to the floor still retires the fast peer" + ); + assert!( + fast_rx.try_recv().is_err(), + "the fast peer is not asked again" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn forgetting_a_different_peer_keeps_the_challenge() { + let (dir, hub) = hub_pair("forget-ch"); + let (mut challenged, mut challenged_rx) = slot(0, 100_000); + challenged.first_data_ms = 80; + challenged.connected_ms = 1; + let (mut other, mut other_rx) = slot(1, 100_000); + other.first_data_ms = 2; + other.connected_ms = 1; + let mut st = IbdWorkState::new( + vec![challenged, other], + hub.tip_hash(), + hub.tip_height(), + ); + plant_origin(&mut st, &hub); + note_block_inv(&mut st, 0); + forget_walk_peer(&mut st, 1); + assert!(send_getheaders(&mut st, &hub).unwrap()); + assert!( + took_getheaders(&mut challenged_rx), + "dropping another peer keeps the block-inv challenge on its sender" + ); + assert!(other_rx.try_recv().is_err(), "the other peer is not asked"); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn an_unproven_origin_logs_headers_in_progress() { + let (dir, mut hub, mut st, _rx) = fresh("quiet-and", 0); + hub.milestone = rbitcoin_consensus::Milestone { + height: 1, + anchor: Some(rbitcoin_consensus::MilestoneAnchor { + hash: hub.tip_hash().unwrap(), + min_work_be: [0xff; 32], + }), + }; + plant_origin(&mut st, &hub); + let line = status_line(&st, &hub); + assert!( + line.contains("phase=write") || line.contains("phase=walk"), + "an unproved origin is still in progress: {line}" + ); + assert!( + !line.contains("headers done"), + "an unproved origin is not the caught-up line: {line}" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_proved_quiet_walk_logs_headers_done() { + let (dir, hub, mut st, _rx) = fresh("quiet-not", 0); + st.header_walk.prove_through(0); + let line = status_line(&st, &hub); + assert!( + line.contains("headers done"), + "a proved walk with no peer ahead logs headers done: {line}" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn block_getdata_skips_the_reserved_header_peer() { + let (dir, hub) = hub_pair("peers-le"); + let (a, mut ra) = slot(0, 10); + let (b, mut rb) = slot(1, 10); + let mut st = IbdWorkState::new(vec![a, b], hub.tip_hash(), hub.tip_height()); + st.slots[0].peer_height = 100_000; + assert!(send_getheaders(&mut st, &hub).unwrap()); + let _ = ra.try_recv(); + let hash = BlockHash::from_byte_array([0x72; 32]); + st.record_height(hash, 40); + st.body.mark_missing(hash); + st.reorg.register_explore([hash], None); + let stats = LoopStats::default(); + let cfg = crate::ibd::IbdConfig::for_test(); + assign_work_ordered(&mut st, &hub, &cfg, &stats, AssignDepth::Full, None); + assert!( + matches!(rb.try_recv(), Ok(PeerCmd::GetData { .. })), + "block getdata goes to the peer that is not serving headers" + ); + assert!( + ra.try_recv().is_err(), + "the reserved header peer does not also get the block" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_peer_level_with_the_walk_does_not_ask() { + let (dir, hub, mut st, mut rx) = fresh("wants-eq", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = st.header_walk.tip_height(); + st.max_ordered_height = u32::MAX; + drive_poll( + &mut st, + &hub, + poll_in(ORDERED_HEADERS_SOFT_CAP, false, 0, 1, 0), + ); + assert!( + !took_getheaders(&mut rx), + "a peer at the walk tip is not ahead, so the soft cap stays closed" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn above_the_soft_cap_with_lag_does_not_ask() { + let (dir, hub, mut st, mut rx) = fresh("sync-gt", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + drive_poll( + &mut st, + &hub, + poll_in(ORDERED_HEADERS_SOFT_CAP + 1, false, 1, 1, 0), + ); + assert!( + !took_getheaders(&mut rx), + "one past the soft cap does not open header sync" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn no_checkpoint_still_asks_for_body_headroom() { + let (dir, hub, mut st, mut rx) = fresh("ck-head", 0); + walk_ahead_of_store(&mut st, &hub); + st.slots[0].peer_height = 0; + st.max_ordered_height = u32::MAX; + let headroom = need_ready_headroom(true, st.header_walk.has_checkpoints()); + drive_poll( + &mut st, + &hub, + poll_in(ORDERED_HEADERS_SOFT_CAP, headroom, 0, 1, 0), + ); + assert!( + took_getheaders(&mut rx), + "with no checkpoint yet, a full queue still asks when bodies want headroom" + ); + let _ = std::fs::remove_dir_all(dir); +} + +/// A block inv opened a challenge and the reply has arrived, so the lane is +/// clear and the challenge is waiting to be judged. +fn challenge_answered( + st: &mut IbdWorkState, + hub: &crate::chain::ChainHub, + peer: usize, + rx: &mut mpsc::UnboundedReceiver, +) { + note_block_inv(st, peer); + assert!(send_getheaders(st, hub).unwrap()); + assert!(took_getheaders(rx), "the block inv asks that peer"); + let _ = take_header_ask(st, peer); +} + +#[test] +fn forgetting_another_peer_keeps_the_reservation() { + let (dir, hub) = hub_pair("forget-res"); + let (mut slow, mut slow_rx) = slot(0, 100_000); + slow.first_data_ms = 80; + slow.connected_ms = 1; + let (mut fast, mut fast_rx) = slot(1, 0); + fast.first_data_ms = 2; + fast.connected_ms = 1; + let mut st = IbdWorkState::new(vec![slow, fast], hub.tip_hash(), hub.tip_height()); + plant_origin(&mut st, &hub); + assert!(send_getheaders(&mut st, &hub).unwrap()); + assert!(took_getheaders(&mut slow_rx), "the tall peer takes the ask"); + let gen = hub.tip_hash().unwrap(); + apply_headers_event(&mut st, &hub, 0, vec![mine(gen, 1)]); + while slow_rx.try_recv().is_ok() {} + st.slots[1].peer_height = 100_000; + forget_walk_peer(&mut st, 1); + assert!(send_getheaders(&mut st, &hub).unwrap()); + assert!( + took_getheaders(&mut slow_rx), + "forgetting a different peer keeps the reserved header peer" + ); + assert!(fast_rx.try_recv().is_err(), "the faster peer is not asked"); + let _ = std::fs::remove_dir_all(dir); +} + +/// Judge one answered challenge, then the next getheaders shows who is still +/// on the walk. +fn judge_challenge( + label: &str, + tip_before: Option, + work_before: [u8; 32], +) { + let (dir, hub) = hub_pair(label); + let (mut fast, mut fast_rx) = slot(0, 100_000); + fast.first_data_ms = 2; + fast.connected_ms = 1; + let (mut slow, mut slow_rx) = slot(1, 100_000); + slow.first_data_ms = 80; + slow.connected_ms = 1; + let mut st = IbdWorkState::new(vec![fast, slow], hub.tip_hash(), hub.tip_height()); + plant_origin(&mut st, &hub); + challenge_answered(&mut st, &hub, 0, &mut fast_rx); + note_reservation(&mut st, &hub, 0, tip_before, work_before); + assert!(send_getheaders(&mut st, &hub).unwrap()); + assert!( + took_getheaders(&mut slow_rx), + "a challenge reply that does not beat the candidate asks the other peer" + ); + assert!(fast_rx.try_recv().is_err(), "the challenged peer is retired"); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn equal_work_does_not_beat_the_candidate() { + let (dir, hub) = hub_pair("beat-ge"); + let (fast, _fast_rx) = slot(0, 100_000); + let (slow, _slow_rx) = slot(1, 100_000); + let mut st = IbdWorkState::new(vec![fast, slow], hub.tip_hash(), hub.tip_height()); + plant_origin(&mut st, &hub); + let work = candidate_work(&st); + drop(st); + let _ = std::fs::remove_dir_all(dir); + judge_challenge( + "beat-ge2", + Some(BlockHash::from_byte_array([0x22; 32])), + work, + ); +} + +#[test] +fn same_tip_does_not_beat_the_candidate() { + let (dir, hub) = hub_pair("beat-or"); + let (fast, _rx) = slot(0, 100_000); + let (slow, _rx2) = slot(1, 100_000); + let mut st = IbdWorkState::new(vec![fast, slow], hub.tip_hash(), hub.tip_height()); + plant_origin(&mut st, &hub); + let tip = candidate_tip(&st); + drop(st); + let _ = std::fs::remove_dir_all(dir); + judge_challenge("beat-or2", tip, [0u8; 32]); +} + +#[test] +fn a_reply_at_the_same_work_retires_the_challenger() { + let (dir, hub) = hub_pair("beat-not"); + let (fast, _rx) = slot(0, 100_000); + let (slow, _rx2) = slot(1, 100_000); + let mut st = IbdWorkState::new(vec![fast, slow], hub.tip_hash(), hub.tip_height()); + plant_origin(&mut st, &hub); + let tip = candidate_tip(&st); + let work = candidate_work(&st); + drop(st); + let _ = std::fs::remove_dir_all(dir); + judge_challenge("beat-not2", tip, work); +} + +#[test] +fn forgetting_another_peer_keeps_the_open_challenge() { + let (dir, hub) = hub_pair("forget-open"); + let (mut fast, mut fast_rx) = slot(0, 100_000); + fast.first_data_ms = 2; + fast.connected_ms = 1; + let (mut slow, mut slow_rx) = slot(1, 100_000); + slow.first_data_ms = 80; + slow.connected_ms = 1; + let mut st = IbdWorkState::new(vec![fast, slow], hub.tip_hash(), hub.tip_height()); + plant_origin(&mut st, &hub); + challenge_answered(&mut st, &hub, 0, &mut fast_rx); + forget_walk_peer(&mut st, 1); + let tip = candidate_tip(&st); + let work = candidate_work(&st); + note_reservation(&mut st, &hub, 0, tip, work); + assert!(send_getheaders(&mut st, &hub).unwrap()); + assert!( + took_getheaders(&mut slow_rx), + "forgetting a different peer keeps the open challenge on its sender" + ); + assert!(fast_rx.try_recv().is_err(), "the challenged peer is retired"); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn an_unlinked_batch_stays_on_the_wire_peer() { + let (dir, hub, mut st, _rx) = fresh("sup-or", 100_000); + plant_origin(&mut st, &hub); + let gen = hub.tip_hash().unwrap(); + let side = mine(gen, 7); + hub.ensure_headers_batch(&[side]).unwrap(); + let side_hash = side.block_hash(); + st.record_height(side_hash, 1); + let first = mine(side_hash, 8); + let second = mine(side_hash, 9); + let before = hub.query.store().header_count(); + apply_headers_event(&mut st, &hub, 0, vec![first, second]); + assert!( + st.slots[0].alive, + "headers that do not link are not a reason to drop the peer" + ); + assert_eq!( + hub.query.store().header_count(), + before + 2, + "an unlinked batch off a stored side header is still written" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_short_fork_off_a_known_header_is_not_stored() { + let (dir, hub, mut st, _rx) = fresh("sup-and", 100_000); + plant_origin(&mut st, &hub); + let gen = hub.tip_hash().unwrap(); + let side = mine(gen, 4); + hub.ensure_headers_batch(&[side]).unwrap(); + let side_hash = side.block_hash(); + st.record_height(side_hash, 1); + let hdr = mine(side_hash, 5); + let before = hub.query.store().header_count(); + apply_headers_event(&mut st, &hub, 0, vec![hdr]); + assert_eq!( + hub.query.store().header_count(), + before, + "a short fork off a side header is not written over the walk" + ); + assert!(st.slots[0].alive, "that fork does not drop the peer"); + let _ = std::fs::remove_dir_all(dir); +} + +/// A lighter fork off an unstored look-ahead checkpoint retargets from that +/// checkpoint's period header. Dropping the snap makes the next period's +/// `nBits` uncomputable, and the peer is disconnected. +#[test] +fn a_fork_retarget_uses_the_challenger_period() { + let (dir, mut hub, mut st, mut rx) = fresh("chal-diff", 100_000); + arm_retarget(&mut hub); + let gen_hash = hub.tip_hash().unwrap(); + let genesis = hub.header_of(&gen_hash).unwrap(); + plant_origin(&mut st, &hub); + fill_soft_queue(&mut st); + let main_a = extend_chain(&hub, &genesis, &[], 20, 1); + let main_b = extend_chain(&hub, &genesis, &main_a, 10, 1); + assert!(send_getheaders(&mut st, &hub).unwrap()); + assert!(took_getheaders(&mut rx), "the look-ahead ask goes to the peer"); + apply_headers_event(&mut st, &hub, 0, main_a.clone()); + assert!(st.slots[0].alive, "the first look-ahead window is accepted"); + assert!(send_getheaders(&mut st, &hub).unwrap()); + assert!(took_getheaders(&mut rx), "the next look-ahead ask goes to the peer"); + apply_headers_event(&mut st, &hub, 0, main_b); + assert!(st.slots[0].alive, "the second look-ahead window is accepted"); + // The download path names a different header at the period height than the + // checkpoint that opened that period. The store fallback would retarget + // from this decoy; the challenger snap must keep the checkpoint header. + let period = main_a[19]; + let parent = main_a[18]; + let decoy = mine_bits( + parent.block_hash(), + 20, + period.time.saturating_add(2_000_000), + period.bits, + 9, + ); + hub.ensure_headers_batch(&[decoy]).unwrap(); + st.record_height(decoy.block_hash(), 20); + let fork_a = extend_chain(&hub, &genesis, &main_a, 10, 2); + apply_headers_event(&mut st, &hub, 0, fork_a.clone()); + assert!( + st.slots[0].alive, + "a lighter fork off the first checkpoint stays on that peer" + ); + let fork_b = extend_chain(&hub, &genesis, &[&main_a[..], &fork_a[..]].concat(), 10, 2); + apply_headers_event(&mut st, &hub, 0, fork_b); + assert!( + st.slots[0].alive, + "the retarget on that fork uses the checkpoint period, so the peer stays" + ); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_header_only_parent_outside_the_walk_is_extended() { + let (dir, hub, mut st, _rx) = fresh("sup-or2", 100_000); + plant_origin(&mut st, &hub); + let gen = hub.tip_hash().unwrap(); + let side = mine(gen, 6); + hub.ensure_headers_batch(&[side]).unwrap(); + let side_hash = side.block_hash(); + // Height is known, but the path slot is not this header, so it is not a + // candidate parent. The short-fork gate must not swallow the write. + st.record_height(side_hash, 1); + st.height_to_hash.remove(&1); + let hdr = mine(side_hash, 11); + let before = hub.query.store().header_count(); + apply_headers_event(&mut st, &hub, 0, vec![hdr]); + assert_eq!( + hub.query.store().header_count(), + before + 1, + "a header on a stored parent that is not the walk is written" + ); + assert!(st.slots[0].alive, "that header does not drop the peer"); + let _ = std::fs::remove_dir_all(dir); +} + +#[test] +fn a_local_challenge_still_retires_a_short_reply() { + let (dir, hub) = hub_pair("settle-ch"); + let (mut fast, mut fast_rx) = slot(0, 100_000); + fast.first_data_ms = 2; + fast.connected_ms = 1; + let (mut slow, mut slow_rx) = slot(1, 100_000); + slow.first_data_ms = 80; + slow.connected_ms = 1; + let mut st = IbdWorkState::new(vec![fast, slow], hub.tip_hash(), hub.tip_height()); + plant_origin(&mut st, &hub); + challenge_answered(&mut st, &hub, 0, &mut fast_rx); + let tip = candidate_tip(&st); + let queued = st.ordered.len(); + settle_walk_peer(&mut st, &hub, 0, tip, queued, 1); + assert!(send_getheaders(&mut st, &hub).unwrap()); + assert!( + took_getheaders(&mut slow_rx), + "a short reply to our own block inv retires that peer" + ); + assert!(fast_rx.try_recv().is_err(), "the challenged peer is retired"); + let _ = std::fs::remove_dir_all(dir); +} diff --git a/crates/rbitcoin-net/src/peer_tests.rs b/crates/rbitcoin-net/src/peer_tests.rs index 618cc5b244..48739252e7 100644 --- a/crates/rbitcoin-net/src/peer_tests.rs +++ b/crates/rbitcoin-net/src/peer_tests.rs @@ -4496,6 +4496,7 @@ include!("peer_catchup_journey.rs"); include!("peer_hostile_journey.rs"); include!("peer_tip_announce_journey.rs"); include!("peer_header_dos_journey.rs"); +include!("peer_header_sync_journey.rs"); include!("peer_blocksonly_journey.rs"); /// Core `PrepareBlockFilterRequest`: start past stop, or a range of 1000+ diff --git a/crates/rbitcoin-node/src/run.rs b/crates/rbitcoin-node/src/run.rs index 00c7f9c817..d0b4c7855c 100644 --- a/crates/rbitcoin-node/src/run.rs +++ b/crates/rbitcoin-node/src/run.rs @@ -371,6 +371,7 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { } } .map_err(|e| NodeError::Config(format!("p2p start: {e}")))?; + publish_listener_addr(config.datadir.path(), "p2p", node.local_addr); let spend_sync = SpendSync::spawn(std::sync::Arc::clone(&node.hub.query)); status.attach_p2p(&node.hub, &node.peers); for extra in &config.listen.p2p_extra { @@ -2635,59 +2636,6 @@ mod tests { ); } - #[test] - fn load_asmap_missing_configured_is_none() { - let dir = std::env::temp_dir().join(format!( - "rbitcoin-asmap-miss-{}", - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_nanos() - )); - std::fs::create_dir_all(&dir).unwrap(); - assert!(load_asmap(&dir, Some(Path::new("no-such-asmap"))).is_none()); - assert!(load_asmap(&dir, None).is_none()); - let _ = std::fs::remove_dir_all(&dir); - } - - #[test] - fn load_asmap_valid_tiny_file() { - let dir = std::env::temp_dir().join(format!( - "rbitcoin-asmap-ok-{}", - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_nanos() - )); - std::fs::create_dir_all(&dir).unwrap(); - let path = dir.join("ip_asn.dat"); - std::fs::write(&path, rbitcoin_net::TWO_PREFIX_ASMAP).unwrap(); - let m = load_asmap(&dir, None).expect("default ip_asn.dat"); - assert_eq!( - m.mapped_as(std::net::IpAddr::V4(std::net::Ipv4Addr::new(1, 2, 0, 0))), - 1 - ); - let rel = load_asmap(&dir, Some(Path::new("ip_asn.dat"))).expect("relative asmap"); - assert_eq!(rel.len(), m.len()); - let _ = std::fs::remove_dir_all(&dir); - } - - #[test] - fn load_asmap_truncated_is_none() { - let dir = std::env::temp_dir().join(format!( - "rbitcoin-asmap-bad-{}", - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_nanos() - )); - std::fs::create_dir_all(&dir).unwrap(); - let path = dir.join("bad.dat"); - std::fs::write(&path, [0u8]).unwrap(); - assert!(load_asmap(&dir, Some(path.as_path())).is_none()); - let _ = std::fs::remove_dir_all(&dir); - } - #[test] fn tip_follow_checks_stale_on_perf_and_rpc_stop() { assert!( diff --git a/crates/rbitcoin-rpc/src/methods_tests.rs b/crates/rbitcoin-rpc/src/methods_tests.rs index 94b0492774..8af02d1496 100644 --- a/crates/rbitcoin-rpc/src/methods_tests.rs +++ b/crates/rbitcoin-rpc/src/methods_tests.rs @@ -3209,40 +3209,6 @@ fn getpeerinfo_sent_pingwait_and_limited_services() { let _ = std::fs::remove_dir_all(&dir); } -#[test] -fn getpeerinfo_mapped_as_when_asmap() { - use bitcoin::p2p::address::Address; - use bitcoin::p2p::message_network::VersionMessage; - use bitcoin::p2p::ServiceFlags; - use rbitcoin_net::{AsMap, PeerConnType, PeerHub, TWO_PREFIX_ASMAP}; - use std::net::{IpAddr, Ipv4Addr, SocketAddr}; - use std::sync::Arc; - - let (mut ctx, dir) = ctx_empty(); - let hub = PeerHub::new(); - hub.set_asmap(Some(Arc::new( - AsMap::from_bytes(TWO_PREFIX_ASMAP.to_vec()).expect("fixture"), - ))); - let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 18444); - let bind = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 18445); - let ver = VersionMessage { - version: 70016, - services: ServiceFlags::NETWORK | ServiceFlags::WITNESS | ServiceFlags::P2P_V2, - timestamp: 0, - receiver: Address::new(&addr, ServiceFlags::NONE), - sender: Address::new(&bind, ServiceFlags::NONE), - nonce: 1, - user_agent: "/rbitcoin:0.1.0(testnode0)/".into(), - start_height: 0, - relay: true, - }; - let _live = hub.register(addr, bind, &ver, false, PeerConnType::OutboundFullRelay); - ctx.peers = Some(hub); - let r = dispatch(&ctx, "getpeerinfo", vec![]).unwrap(); - assert_eq!(r.as_array().unwrap()[0]["mapped_as"], json!(1)); - let _ = std::fs::remove_dir_all(&dir); -} - /// Every `P2PNode` attaches a dialer, so no session reaches this arm. /// `node_run_p2p_short` owns the connected `addnode` / `disconnectnode` results. #[test] diff --git a/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs b/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs index ea27824afa..22f21c4d90 100644 --- a/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs +++ b/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs @@ -582,6 +582,61 @@ fn chain_ops_validate_address(ctx: &RpcContext) { assert_eq!(still_never["code"], ERR_METHOD_NOT_FOUND); } +/// A second genesis hub. Tip age, the mining-info template latch, and the +/// header-tip cap are visible here before this chain has assembled a template +/// or left IBD. The long regtest journey has already done both. +#[test] +fn chain_ops_fresh_tip_age_template_and_header_cap() { + let (ctx, dir, hub) = ctx_regtest_hub(); + let gen_time = hub.tip_header().unwrap().time; + // 10_000s is inside a day and outside either `+` rewrite of `24 * 60 * 60`. + dispatch( + &ctx, + "setmocktime", + vec![json!(i64::from(gen_time) + 10_000)], + ) + .unwrap(); + let info = dispatch(&ctx, "getblockchaininfo", vec![]).unwrap(); + assert_eq!( + info["initialblockdownload"], false, + "a tip 10000s old is still inside the 24h IBD window" + ); + + let prev = hub.tip_hash().unwrap(); + for i in 0..129u32 { + let mut hdr = hub.tip_header().unwrap(); + hdr.version = bitcoin::block::Version::from_consensus(4); + hdr.prev_blockhash = prev; + hdr.time = gen_time.saturating_add(2).saturating_add(i); + hdr.nonce = 0; + let mut merkle = [0u8; 32]; + merkle[..4].copy_from_slice(&i.to_le_bytes()); + hdr.merkle_root = bitcoin::TxMerkleNode::from_byte_array(merkle); + rbitcoin_consensus::grind_regtest_pow(&mut hdr); + let hex = hex_encode(bitcoin::consensus::encode::serialize(&hdr)); + dispatch(&ctx, "submitheader", vec![json!(hex)]).unwrap(); + } + let tips = dispatch(&ctx, "getchaintips", vec![]).unwrap(); + let n = tips + .as_array() + .unwrap() + .iter() + .filter(|t| t["status"] == "headers-only") + .count(); + assert_eq!(n, 128, "header-only tips stay at the 128 cap"); + + let before = dispatch(&ctx, "getmininginfo", vec![]).unwrap(); + assert!( + before.get("currentblockweight").is_none(), + "getmininginfo names a template weight only after one is built: {before}" + ); + let _tmpl = dispatch(&ctx, "getblocktemplate", vec![json!({"rules": ["segwit"]})]).unwrap(); + let after = dispatch(&ctx, "getmininginfo", vec![]).unwrap(); + assert_eq!(after["currentblockweight"], 8_000); + assert_eq!(after["currentblocktx"], 0); + let _ = std::fs::remove_dir_all(&dir); +} + fn chain_ops_echo_overlap(ctx: &RpcContext) { let empty = dispatch(ctx, "echo", vec![]).unwrap(); assert_eq!(empty, json!([])); diff --git a/crates/rbitcoin-test/Cargo.toml b/crates/rbitcoin-test/Cargo.toml index 3a841eaaf9..69f44dc173 100644 --- a/crates/rbitcoin-test/Cargo.toml +++ b/crates/rbitcoin-test/Cargo.toml @@ -55,6 +55,10 @@ path = "tests/cross_surface.rs" name = "end_of_ibd" path = "tests/end_of_ibd.rs" +[[test]] +name = "asmap_journey" +path = "tests/asmap_journey.rs" + [features] # Live private-mesh journeys. Default `cargo test` stays fake SOCKS/SAM/control. overlay = [] diff --git a/crates/rbitcoin-test/tests/asmap_journey.rs b/crates/rbitcoin-test/tests/asmap_journey.rs new file mode 100644 index 0000000000..f735e2c2f0 --- /dev/null +++ b/crates/rbitcoin-test/tests/asmap_journey.rs @@ -0,0 +1,328 @@ +//! Asmap files an operator drops in the datadir, checked from one node. +//! +//! Inbound clients are raw BIP324 sockets bound on `127.0.0.0/8`. No second +//! node, no DNS seeds, no outbound dial. + +use bitcoin::p2p::address::Address; +use bitcoin::p2p::message::NetworkMessage; +use bitcoin::p2p::message_network::VersionMessage; +use bitcoin::p2p::ServiceFlags; +use rbitcoin_net::V2PlainSession; +use rbitcoin_node::{run_p2p, NodeConfig}; +use rbitcoin_primitives::Network; +use rbitcoin_test::TestDatadir; +use serde_json::{json, Value}; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +fn llvm_cov_wall(default_secs: u64, llvm_secs: u64) -> Duration { + if std::env::var_os("CARGO_LLVM_COV").is_some() { + Duration::from_secs(llvm_secs) + } else { + Duration::from_secs(default_secs) + } +} + +type StartupLogs = Vec<(rbitcoin_log::Level, String)>; + +fn spawn_run_p2p_logs( + cfg: NodeConfig, +) -> tokio::task::JoinHandle<(Result<(), rbitcoin_node::NodeError>, StartupLogs)> { + match std::fs::remove_dir_all(cfg.datadir.path().join("run")) { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => panic!("clear published listeners: {e}"), + } + tokio::task::spawn_blocking(move || { + let _block = rbitcoin_net::BlockingRegion::enter(); + rbitcoin_log::capture_logs(true); + let result = tokio::runtime::Handle::current().block_on(run_p2p(cfg)); + let logs = rbitcoin_log::take_logs(); + rbitcoin_log::capture_logs(false); + (result, logs) + }) +} + +async fn stop_logged( + rpc: SocketAddr, + node: tokio::task::JoinHandle<(Result<(), rbitcoin_node::NodeError>, StartupLogs)>, +) -> StartupLogs { + let _ = jsonrpc(rpc, "stop", json!([])).await; + match tokio::time::timeout(Duration::from_secs(20), node).await { + Ok(Ok((Ok(()), logs))) => logs, + Ok(Ok((Err(e), _))) => panic!("run_p2p error after stop: {e}"), + Ok(Err(e)) => panic!("run_p2p join: {e}"), + Err(_) => panic!("run_p2p did not exit after stop"), + } +} + +async fn published_listener(datadir: &Path, name: &str) -> SocketAddr { + let path = datadir.join("run").join(format!("{name}.addr")); + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if let Ok(text) = std::fs::read_to_string(&path) { + if let Ok(addr) = text.trim().parse::() { + if addr.port() != 0 { + return addr; + } + } + } + if Instant::now() >= deadline { + panic!( + "bound {name} address was not published at {}", + path.display() + ); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +async fn jsonrpc(addr: SocketAddr, method: &str, params: Value) -> Value { + rbitcoin_test::jsonrpc::jsonrpc(addr, method, params).await +} + +fn base_cfg(dir: &Path) -> NodeConfig { + let mut cfg = NodeConfig::default() + .with_datadir(dir) + .with_network(Network::Regtest) + .with_tiny_heads(); + cfg.listen.use_seeds = false; + cfg +} + +fn write_token(dir: &Path) { + std::fs::write(dir.join("rpc.token"), "pass\n").unwrap(); +} + +fn log_text(logs: &StartupLogs) -> String { + logs.iter() + .map(|(_, line)| line.as_str()) + .collect::>() + .join("\n") +} + +async fn handshake(bind_ip: Ipv4Addr, to: SocketAddr, agent: &str) -> V2PlainSession { + let sock = tokio::net::TcpSocket::new_v4().expect("v4 socket"); + sock.bind(SocketAddr::new(IpAddr::V4(bind_ip), 0)) + .unwrap_or_else(|e| panic!("bind {bind_ip}: {e}")); + let stream = tokio::time::timeout(Duration::from_secs(5), sock.connect(to)) + .await + .unwrap_or_else(|_| panic!("connect {bind_ip} to {to} timed out")) + .unwrap_or_else(|e| panic!("connect {bind_ip} to {to}: {e}")); + let mut raw = V2PlainSession::outbound_bip324(stream) + .await + .expect("bip324"); + let ver = VersionMessage { + version: 70016, + services: ServiceFlags::NETWORK | ServiceFlags::WITNESS, + timestamp: 0, + receiver: Address::new(&to, ServiceFlags::NONE), + sender: Address::new(&to, ServiceFlags::NONE), + nonce: u64::from(bind_ip.octets()[3]), + user_agent: agent.into(), + start_height: 0, + relay: true, + }; + for msg in [NetworkMessage::Version(ver), NetworkMessage::Verack] { + raw.write_contents(&rbitcoin_net::encode_v2_contents(msg).expect("encode")) + .await + .expect("write handshake"); + } + raw +} + +fn mapped_as(rows: &Value, ip: Ipv4Addr) -> u64 { + let prefix = format!("{ip}:"); + let row = rows + .as_array() + .unwrap_or_else(|| panic!("getpeerinfo is not a list: {rows}")) + .iter() + .find(|row| { + row["addr"] + .as_str() + .is_some_and(|addr| addr.starts_with(&prefix)) + }) + .unwrap_or_else(|| panic!("no peer {ip} in {rows}")); + row["mapped_as"] + .as_u64() + .unwrap_or_else(|| panic!("peer {ip} has no mapped_as: {row}")) +} + +async fn wait_peers(rpc: SocketAddr, n: usize) -> Value { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let info = jsonrpc(rpc, "getpeerinfo", json!([])).await; + let count = info["result"].as_array().map(|a| a.len()).unwrap_or(0); + if count >= n { + return info["result"].clone(); + } + if Instant::now() >= deadline { + panic!("expected {n} peers, got {info}"); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +fn assert_opened(logs: &StartupLogs, len: usize, digest: &str) { + let text = log_text(logs); + assert!( + text.contains(&format!("({len} bytes, digest {digest})")), + "opened line missing from {text}" + ); + assert!( + !text.contains("Sanity check of asmap file"), + "a valid asmap was refused: {text}" + ); +} + +/// Loopback peers report the ASN from `{datadir}/ip_asn.dat`, then a relative +/// `--asmap` of a lone return maps the same peer to ASN 1. +#[tokio::test(flavor = "multi_thread")] +async fn asmap_loopback_peers_report_their_asn() { + let wall = llvm_cov_wall(20, 60); + let fut = async { + let td = TestDatadir::new().unwrap(); + let dir = td.path(); + std::fs::write(dir.join("ip_asn.dat"), rbitcoin_net::LOOPBACK_ASMAP).unwrap(); + write_token(&dir); + let mut cfg = base_cfg(&dir).with_p2p_listen("127.0.0.1:0".parse().unwrap()); + cfg.rpc.listen = Some("127.0.0.1:0".parse().unwrap()); + let node = spawn_run_p2p_logs(cfg); + let rpc = published_listener(&dir, "rpc").await; + let p2p = published_listener(&dir, "p2p").await; + let _a = handshake(Ipv4Addr::new(127, 0, 0, 1), p2p, "/asmap-1/").await; + let _b = handshake(Ipv4Addr::new(127, 0, 0, 2), p2p, "/asmap-2/").await; + let _c = handshake(Ipv4Addr::new(127, 0, 0, 3), p2p, "/asmap-3/").await; + let rows = wait_peers(rpc, 3).await; + assert_eq!(mapped_as(&rows, Ipv4Addr::new(127, 0, 0, 1)), 1, "{rows}"); + assert_eq!(mapped_as(&rows, Ipv4Addr::new(127, 0, 0, 2)), 2, "{rows}"); + assert_eq!(mapped_as(&rows, Ipv4Addr::new(127, 0, 0, 3)), 9, "{rows}"); + let logs = stop_logged(rpc, node).await; + assert_opened(&logs, rbitcoin_net::LOOPBACK_ASMAP.len(), "0d2045c8"); + + let td = TestDatadir::new().unwrap(); + let dir = td.path(); + std::fs::write(dir.join("asn1.dat"), rbitcoin_net::LONE_RETURN_ASMAP).unwrap(); + write_token(&dir); + let mut cfg = base_cfg(&dir).with_p2p_listen("127.0.0.1:0".parse().unwrap()); + cfg.asmap = Some(PathBuf::from("asn1.dat")); + cfg.rpc.listen = Some("127.0.0.1:0".parse().unwrap()); + let node = spawn_run_p2p_logs(cfg); + let rpc = published_listener(&dir, "rpc").await; + let p2p = published_listener(&dir, "p2p").await; + let _peer = handshake(Ipv4Addr::new(127, 0, 0, 1), p2p, "/asmap-lone/").await; + let rows = wait_peers(rpc, 1).await; + assert_eq!(mapped_as(&rows, Ipv4Addr::new(127, 0, 0, 1)), 1, "{rows}"); + let logs = stop_logged(rpc, node).await; + assert_opened(&logs, 3, "709e80c8"); + }; + tokio::time::timeout(wall, fut).await.unwrap_or_else(|_| { + panic!("asmap_loopback_peers_report_their_asn wall timeout ({wall:?})") + }); +} + +fn assert_refused(logs: &StartupLogs, phrase: &str) { + let text = log_text(logs); + assert!(text.contains(phrase), "missing {phrase:?} in {text}"); + assert!( + !text.contains("Opened asmap file"), + "a refused asmap loaded: {text}" + ); +} + +async fn boot_refused(cfg: NodeConfig, phrase: &str) { + let node = spawn_run_p2p_logs(cfg); + match tokio::time::timeout(Duration::from_secs(20), node).await { + Ok(Ok((Ok(()), logs))) => assert_refused(&logs, phrase), + Ok(Ok((Err(e), logs))) => panic!("run_p2p error: {e}; logs={}", log_text(&logs)), + Ok(Err(e)) => panic!("run_p2p join: {e}"), + Err(_) => panic!("refused asmap boot did not exit"), + } +} + +/// Each bad asmap is refused at startup. The node exits once tip entry is +/// ready (`max_run_secs` 0) and never dials a peer. +#[tokio::test(flavor = "multi_thread")] +async fn asmap_refused_file_uses_prefix_netgroups() { + let wall = llvm_cov_wall(20, 60); + let cases: &[(&str, &[u8], &str)] = &[ + ( + "second-default.dat", + rbitcoin_net::ASMAP_SECOND_DEFAULT, + "Sanity check of asmap file", + ), + ( + "trailing-one.dat", + rbitcoin_net::ASMAP_TRAILING_ONE, + "Sanity check of asmap file", + ), + ( + "eight-zeros.dat", + rbitcoin_net::ASMAP_EIGHT_TRAILING_ZEROS, + "Sanity check of asmap file", + ), + ( + "two-short.dat", + rbitcoin_net::ASMAP_TWO_SHORT_MATCHES, + "Sanity check of asmap file", + ), + ( + "match-past.dat", + rbitcoin_net::ASMAP_MATCH_PAST_END, + "Sanity check of asmap file", + ), + ( + "jump-past.dat", + rbitcoin_net::ASMAP_JUMP_PAST_FILE, + "Sanity check of asmap file", + ), + ( + "jump-short.dat", + rbitcoin_net::ASMAP_JUMP_SHORT_TAIL, + "Sanity check of asmap file", + ), + ( + "jump-off.dat", + rbitcoin_net::ASMAP_JUMP_OFF_END, + "Sanity check of asmap file", + ), + ]; + let fut = async { + for (name, bytes, phrase) in cases { + let td = TestDatadir::new().unwrap(); + let dir = td.path(); + std::fs::write(dir.join(name), bytes).unwrap(); + let mut cfg = base_cfg(&dir).with_p2p_listen("127.0.0.1:0".parse().unwrap()); + cfg.asmap = Some(PathBuf::from(name)); + cfg.max_run_secs = Some(0); + boot_refused(cfg, phrase).await; + } + + let td = TestDatadir::new().unwrap(); + let dir = td.path(); + std::fs::write(dir.join("empty.dat"), []).unwrap(); + let mut cfg = base_cfg(&dir).with_p2p_listen("127.0.0.1:0".parse().unwrap()); + cfg.asmap = Some(PathBuf::from("empty.dat")); + cfg.max_run_secs = Some(0); + boot_refused(cfg, "Sanity check of asmap file").await; + + let td = TestDatadir::new().unwrap(); + let dir = td.path(); + std::fs::write(dir.join("one.dat"), [0u8]).unwrap(); + let mut cfg = base_cfg(&dir).with_p2p_listen("127.0.0.1:0".parse().unwrap()); + cfg.asmap = Some(PathBuf::from("one.dat")); + cfg.max_run_secs = Some(0); + boot_refused(cfg, "Sanity check of asmap file").await; + + let td = TestDatadir::new().unwrap(); + let dir = td.path(); + let mut cfg = base_cfg(&dir).with_p2p_listen("127.0.0.1:0".parse().unwrap()); + cfg.asmap = Some(PathBuf::from("no-such-asmap")); + cfg.max_run_secs = Some(0); + boot_refused(cfg, "Failed to open asmap file").await; + }; + tokio::time::timeout(wall, fut).await.unwrap_or_else(|_| { + panic!("asmap_refused_file_uses_prefix_netgroups wall timeout ({wall:?})") + }); +} diff --git a/crates/rbitcoin-test/tests/scenarios.rs b/crates/rbitcoin-test/tests/scenarios.rs index 5a4960f850..d2f5bc3205 100644 --- a/crates/rbitcoin-test/tests/scenarios.rs +++ b/crates/rbitcoin-test/tests/scenarios.rs @@ -2841,7 +2841,7 @@ fn block_cache_and_mempool_hub_surface() { // BlockCache: push chain, locator, headers, truncate, depth eviction. let cache = BlockCache::with_body_depth(4); assert!(cache.is_empty()); - assert_eq!(cache.len(), 0); + assert!(cache.tip_height().is_none()); for b in &blocks { cache.push_best(b.clone()).unwrap(); } @@ -2858,11 +2858,11 @@ fn block_cache_and_mempool_hub_surface() { "body depth eviction" ); assert!(cache.hash_at_height(0).is_some(), "hash chain retained"); - let loc = cache.locator(); - assert!(!loc.is_empty()); + let genesis_hash = cache.hash_at_height(0).unwrap(); let stop = BlockHash::from_byte_array([0u8; 32]); - let hdrs = cache.headers_after_locator(&loc[loc.len().saturating_sub(1)..], stop); + let hdrs = cache.headers_after_locator(&[genesis_hash], stop); assert!(!hdrs.is_empty()); + assert_ne!(hdrs[0].block_hash(), genesis_hash); // Bad extension rejected. let mut bad = blocks.last().unwrap().clone(); bad.header.prev_blockhash = BlockHash::from_byte_array([0xee; 32]); @@ -2872,7 +2872,6 @@ fn block_cache_and_mempool_hub_surface() { cache.clear(); assert!(cache.is_empty()); let empty = BlockCache::new(); - assert!(!empty.locator().is_empty()); assert!(empty .headers_after_locator(&[], BlockHash::from_byte_array([0u8; 32])) .is_empty());