diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bb26b385a..3028cfba5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -99,6 +99,8 @@ jobs: run: ./scripts/tmpfs-test-runner.test.sh - name: overlay-functional runner self-test run: ./scripts/overlay-functional/run.sh.test.sh + - name: journeys manifest + run: ./scripts/journeys-manifest.test.sh - uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 with: shared-key: ci-clippy diff --git a/Cargo.toml b/Cargo.toml index a35d19023..59372fca4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -100,4 +100,8 @@ debug = "line-tables-only" debug = false [profile.test.package."*"] +# `*` is dependencies only (workspace crates keep the profile above). +# Optimized crypto and codec crates make block hashing in tests cheap. +# rust-cache holds the dep build across CI runs. +opt-level = 2 debug = false diff --git a/TESTING.md b/TESTING.md index 3b4844bd2..fa1706d62 100644 --- a/TESTING.md +++ b/TESTING.md @@ -181,7 +181,7 @@ absolute path. Keep default-tier fixtures small: on tmpfs, test bytes are RAM. | `rbitcoin-store --lib` | **<45 s** | Catalog-run fixtures stay tens of tiny files, not thousands | | `rbitcoin-consensus --lib` | **<30 s** | Pure-result rule: [True journeys](#true-journeys). Mainnet 866342 (~1.6 s) is the historical prevout pin — one zstd decode, overweight on a clone. | | `rbitcoin-query --lib` | **<20 s** | | -| `rbitcoin-test --test scenarios` | **<15 s** | Prefer `pad_empty_from` / shared mature helpers | +| `rbitcoin-test --test journeys` | **<15 s** for the old scenarios module | Prefer `pad_empty_from` / shared mature helpers. Default journeys are one binary (`scenarios::` and the rest). | | **Full** `cargo test --workspace` | **≤3 min** warm | Stretch **<2 min** | **New default-suite test rule:** if a new or expanded default test routinely takes **>2 s wall** on a warm tree, the PR must **justify** it (what contract needs that cost, why a smaller N / unit cannot hit the branch). Prefer `#[ignore]` + reason string for true microbenches / host-only forensics. diff --git a/changelog.d/suite-under-150s.md b/changelog.d/suite-under-150s.md new file mode 100644 index 000000000..0129f501e --- /dev/null +++ b/changelog.d/suite-under-150s.md @@ -0,0 +1,15 @@ +Changed + +- **P2P shutdown wakes dial, accept, and connect-retry.** Those tasks were blocked on a channel or a tick, so every shutdown sat out the full 750 ms grace and then aborted them. The grace remains for a session that is still unwinding. +- **`--max-run-secs` exits when the deadline is reached.** The tip loop used to notice it only on the 5 s perf tick. +- **IBD densify assign-stop is a field on the IBD config.** `RBITCOIN_BLOCK_QUEUE_BYTES` / `_GB` are still read once when a production config is built. Tests pass the cap in, so they no longer serialize on a process-wide env lock. +- **Test dependencies compile at opt-level 2.** Workspace crates stay unoptimized. Dev and test builds still keep line tables. +- **The hub `OP_TRUE` signet generate test uses regtest difficulty.** Accept still requires the BIP325 solution. The real signet target stays covered by the consensus miner test. +- **`NODE_COMPACT_FILTERS` lives on the peer hub.** Version, feeler, addr, and `getnetworkinfo` read that node's bit. There is no process-wide flag. +- **`getblocktemplate`, `getrawmempool`, and `generate` stay off while the pool is not maintained.** Relay off (still catching up) returns `-10`. A node with no mempool still serves `getblocktemplate` on the chain tip and an empty `getrawmempool`. `-blocksonly` returns a permanent error so a miner does not retry. `generate` during catch-up mines the coinbase only. Once relay is on, `getblocktemplate` and Stratum v2 build on the tip recorded after the pool strip. A Stratum v2 session that is still holding tries again on the template interval, so the tip that ends catch-up is not missed when relay turns on after that event. A block connect and an initial-download confirm batch hold that mark from before the chain tip moves until the drop is stored. A disconnect, including a header rewind, moves it to the fork point the pool was put back on. There is no chain scan under the mempool lock. +- **A node built with the journey `test-hooks` feature warns at startup.** That build uses the short reconnect and SV2 cooldown. Release, nix, and `cargo build -p rbitcoin-node` do not. +- **`cargo llvm-cov` no longer builds the workspace twice.** The coverage script was compiling plain node and cli bins before the instrumented test build. The suite already builds the bins it execs. + +Fixed + +- **Default journeys link and run as one test binary.** Doctests are off on library crates that have none. The node, cli, and journey-node bins do not build an empty test harness. Journeys compile a 50ms SAM retry, a 250ms connect redial, and a 100ms SV2 budget cooldown. The node binary keeps 500ms, 2s, and 1s. diff --git a/crates/rbitcoin-bench/Cargo.toml b/crates/rbitcoin-bench/Cargo.toml index f0ea9fc97..114d474be 100644 --- a/crates/rbitcoin-bench/Cargo.toml +++ b/crates/rbitcoin-bench/Cargo.toml @@ -18,6 +18,9 @@ mutants = [] default = [] cli = [] +[lib] +doctest = false + [dependencies] rbitcoin-primitives = { workspace = true } bitcoin = { workspace = true } diff --git a/crates/rbitcoin-cli/Cargo.toml b/crates/rbitcoin-cli/Cargo.toml index 07dc04876..1b5649529 100644 --- a/crates/rbitcoin-cli/Cargo.toml +++ b/crates/rbitcoin-cli/Cargo.toml @@ -10,6 +10,10 @@ description = "rbitcoin RPC CLI client" [[bin]] name = "rbitcoin-cli" path = "src/main.rs" +test = false + +[lib] +doctest = false [features] mutants = [] diff --git a/crates/rbitcoin-consensus/Cargo.toml b/crates/rbitcoin-consensus/Cargo.toml index cd85a764a..04ccd1116 100644 --- a/crates/rbitcoin-consensus/Cargo.toml +++ b/crates/rbitcoin-consensus/Cargo.toml @@ -13,6 +13,9 @@ corpus = ["dep:serde_json"] # Nightly cargo-mutants only. Off for the node build. mutants = ["dep:mutants"] +[lib] +doctest = false + [dependencies] rbitcoin-primitives = { workspace = true } arc-swap = { workspace = true } diff --git a/crates/rbitcoin-electrum/Cargo.toml b/crates/rbitcoin-electrum/Cargo.toml index a7260e6cc..cc30818de 100644 --- a/crates/rbitcoin-electrum/Cargo.toml +++ b/crates/rbitcoin-electrum/Cargo.toml @@ -10,6 +10,9 @@ description = "Electrum protocol server (scripthash JSON-RPC over TCP)" [features] mutants = [] +[lib] +doctest = false + [dependencies] rbitcoin-primitives = { workspace = true } rbitcoin-log = { workspace = true } diff --git a/crates/rbitcoin-esplora/Cargo.toml b/crates/rbitcoin-esplora/Cargo.toml index 22341a077..fcd308b33 100644 --- a/crates/rbitcoin-esplora/Cargo.toml +++ b/crates/rbitcoin-esplora/Cargo.toml @@ -10,6 +10,9 @@ description = "Esplora-compatible REST (electrs HTTP drop-in except address-pref [features] mutants = [] +[lib] +doctest = false + [dependencies] rbitcoin-primitives = { workspace = true } rbitcoin-log = { workspace = true } diff --git a/crates/rbitcoin-log/Cargo.toml b/crates/rbitcoin-log/Cargo.toml index 583792e1c..e462b8e7c 100644 --- a/crates/rbitcoin-log/Cargo.toml +++ b/crates/rbitcoin-log/Cargo.toml @@ -7,6 +7,9 @@ rust-version.workspace = true repository.workspace = true description = "Minimal leveled stderr logger and long-stage progress registry (no external deps)" +[lib] +doctest = false + [dependencies] [lints] diff --git a/crates/rbitcoin-mempool/Cargo.toml b/crates/rbitcoin-mempool/Cargo.toml index a930d05ab..9fa26c3a0 100644 --- a/crates/rbitcoin-mempool/Cargo.toml +++ b/crates/rbitcoin-mempool/Cargo.toml @@ -7,6 +7,9 @@ rust-version.workspace = true repository.workspace = true description = "Cluster mempool with InRam buffers + private sidecar files under datadir/mempool/" +[lib] +doctest = false + [dependencies] rbitcoin-log = { workspace = true } rbitcoin-consensus = { workspace = true } diff --git a/crates/rbitcoin-net/Cargo.toml b/crates/rbitcoin-net/Cargo.toml index 7b8d356e7..6be5bbf5f 100644 --- a/crates/rbitcoin-net/Cargo.toml +++ b/crates/rbitcoin-net/Cargo.toml @@ -9,6 +9,12 @@ description = "Bitcoin P2P (BIP324, IBD, tip follow, compact v2, tx relay)" [features] mutants = [] +# Journeys turn this on. The node binary does not, so its redial pass stays +# 2s and a SAM retry stays 500ms. +test-hooks = [] + +[lib] +doctest = false [dependencies] rbitcoin-primitives = { workspace = true } diff --git a/crates/rbitcoin-net/src/chain.rs b/crates/rbitcoin-net/src/chain.rs index abb7e5f83..d420f9483 100644 --- a/crates/rbitcoin-net/src/chain.rs +++ b/crates/rbitcoin-net/src/chain.rs @@ -1817,19 +1817,33 @@ impl ChainHub { .into_iter() .map(|(h, raw)| (h, BlockHash::from_byte_array(raw))) .collect(); + // Held across the write so the chain tip cannot move while the + // relay-on purge is deciding which tip it stripped. + let mp_hold = self.mempool().cloned(); + let mut template_hold = mp_hold.as_ref().map(|mp| mp.template_hold()); confirm_write_phase(&self.query, &self.params, self.milestone, batch) .map_err(NetError::from_confirm)?; - self.note_confirmed_tip(&meta, &headers)?; + self.note_confirmed_tip_while(&meta, &headers, template_hold.as_mut())?; Ok(meta .iter() .map(|&(height, _)| AcceptOutcome::Accepted { height }) .collect()) } + #[cfg(test)] pub(crate) fn note_confirmed_tip( &self, need_meta: &[(u32, BlockHash)], headers: &[Header], + ) -> Result<(), NetError> { + self.note_confirmed_tip_while(need_meta, headers, None) + } + + pub(crate) fn note_confirmed_tip_while( + &self, + need_meta: &[(u32, BlockHash)], + headers: &[Header], + mut hold: Option<&mut crate::tx_relay::TemplateHold<'_>>, ) -> Result<(), NetError> { if headers.len() != need_meta.len() { return Err(NetError::store(rbitcoin_store::StoreError::Corrupt( @@ -1838,18 +1852,28 @@ impl ChainHub { } if let Some(mp) = self.mempool() { mp.clear_recent_rejects(); + let mut stripped: Option<[u8; 32]> = None; if mp.relay_enabled() { for &(height, hash) in need_meta { mp.note_block_fee_history(Height(height)); - if let Ok(Some(block)) = + let Ok(Some(block)) = self.query.reconstruct_block_by_hash(&hash.to_byte_array()) - { - mp.note_recent_confirmed(&block.txdata); - let ids: Vec<_> = block.txdata.iter().map(|t| t.compute_txid()).collect(); - let spent = spent_outpoints(&block.txdata); - let n = mp.remove_for_block_spent(&ids, &spent); - rbitcoin_log::debug!("mempool: removed {n} confirmed tx(s) @ {hash}"); - } + else { + break; + }; + mp.note_recent_confirmed(&block.txdata); + let ids: Vec<_> = block.txdata.iter().map(|t| t.compute_txid()).collect(); + let spent = spent_outpoints(&block.txdata); + let n = mp.remove_for_block_spent(&ids, &spent); + rbitcoin_log::debug!("mempool: removed {n} confirmed tx(s) @ {hash}"); + stripped = Some(hash.to_byte_array()); + } + } + if let Some(hash) = stripped { + if let Some(hold) = hold.as_mut() { + hold.commit(hash); + } else { + mp.note_template_prev(hash); } } } @@ -2718,6 +2742,11 @@ impl ChainHub { if keep_height > tip { return Err(NetError::Protocol("rewind above tip")); } + // Before the commit lock, same order as `disconnect_to`. This rewind + // already holds that lock, so it cannot call `disconnect_to`, and the + // mark would stay on the block that was just disconnected. + let mp_hold = self.mempool().cloned(); + let mut template_hold = mp_hold.as_ref().map(|mp| mp.template_hold()); let _commit = if wait { self.query .lock_tip_commit() @@ -2728,6 +2757,12 @@ impl ChainHub { return Ok(false); }; self.disconnect_to_in_commit(keep_height)?; + if let Some(hold) = template_hold.as_mut() { + if let Ok(Some((_, rec))) = self.query.header_at_height(Height(keep_height)) { + hold.commit(rec.hash); + } + } + drop(template_hold); Ok(true) } @@ -3073,6 +3108,10 @@ impl ChainHub { crate::tip_accept::on_tip_accept_thread(), "connect_at must run on tip-accept" ); + // Held until this connect returns, so a template cannot observe the + // new query tip before the pool strip is recorded. + let mp_hold = self.mempool().cloned(); + let mut template_hold = mp_hold.as_ref().map(|mp| mp.template_hold()); let hash = block.block_hash(); let header = block.header; tip_accept_stats_reset(&self.query); @@ -3142,6 +3181,9 @@ impl ChainHub { rbitcoin_log::debug!("mempool: removed {n} confirmed tx(s) @ height {height}"); if mp.relay_enabled() { mp.note_block_fee_history(Height(height)); + if let Some(hold) = template_hold.as_mut() { + hold.commit(hash.to_byte_array()); + } } } let mp_strip_ns = t_mp.elapsed().as_nanos() as u64; @@ -3154,6 +3196,9 @@ impl ChainHub { // uses note_confirmed_tip without this line; IBD retains periodic status. info!("{}", log_update_tip_line(height, &hash, &header, tx_count)); log_tip_accept_sh(&self.query, height, tx_count, wall_ns, mp_strip_ns, pres_ns); + // Record the stripped tip before the announcement, so a builder that + // wakes on this event already sees the new mark. + drop(template_hold); let event = TipEvent { height, hash, @@ -3168,11 +3213,23 @@ impl ChainHub { } fn disconnect_to(&self, keep_height: u32) -> Result<(), NetError> { + // Before the commit lock: the relay-on purge takes this same mutex + // and then reads the query. The pool matches `keep_height` once the + // disconnected transactions are back in it. + let mp_hold = self.mempool().cloned(); + let mut template_hold = mp_hold.as_ref().map(|mp| mp.template_hold()); let _commit = self .query .lock_tip_commit() .map_err(|e| NetError::Store(e.to_string()))?; - self.disconnect_to_in_commit(keep_height) + self.disconnect_to_in_commit(keep_height)?; + if let Some(hold) = template_hold.as_mut() { + if let Ok(Some((_, rec))) = self.query.header_at_height(Height(keep_height)) { + hold.commit(rec.hash); + } + } + drop(template_hold); + Ok(()) } fn disconnect_to_in_commit(&self, keep_height: u32) -> Result<(), NetError> { @@ -6246,7 +6303,7 @@ mod tests { { let hub = hub.clone(); thread::spawn(move || { - let until = Instant::now() + Duration::from_secs(2); + let until = Instant::now() + Duration::from_millis(200); while Instant::now() < until { assert_eq!(hub.best_header_height(), 2); assert!(hub.chaintips().iter().any(|t| t.hash == tip)); diff --git a/crates/rbitcoin-net/src/i2p_sam.rs b/crates/rbitcoin-net/src/i2p_sam.rs index 2eee039bb..1a3fb5e1e 100644 --- a/crates/rbitcoin-net/src/i2p_sam.rs +++ b/crates/rbitcoin-net/src/i2p_sam.rs @@ -9,6 +9,14 @@ use std::sync::Mutex; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; +/// Pause before a SAM retry. Thirty-two session attempts are about 16s, so a +/// router that is not accepting yet still comes up. +#[cfg(not(any(test, feature = "test-hooks")))] +const SAM_RETRY: std::time::Duration = std::time::Duration::from_millis(500); +/// Crate tests and journeys. The node binary keeps 500ms. +#[cfg(any(test, feature = "test-hooks"))] +const SAM_RETRY: std::time::Duration = std::time::Duration::from_millis(50); + static INSTALLED: Mutex> = Mutex::new(None); struct Installed { @@ -193,7 +201,7 @@ impl I2pSam { Ok(v) => return Ok(v), Err(e) if sam_retry(&e) => { last = Some(e); - tokio::time::sleep(std::time::Duration::from_millis(500)).await; + tokio::time::sleep(SAM_RETRY).await; } Err(e) => return Err(e), } @@ -274,7 +282,7 @@ impl I2pSam { } Err(e) if stream_retry(&e) => { last = Some(e); - tokio::time::sleep(std::time::Duration::from_millis(500)).await; + tokio::time::sleep(SAM_RETRY).await; } Err(e) => return Err(e), } @@ -312,7 +320,7 @@ impl I2pDialer { Err(e) if stream_socket_dead(&e) => return Err(e), Err(e) if stream_retry(&e) => { last = Some(e); - tokio::time::sleep(std::time::Duration::from_millis(500)).await; + tokio::time::sleep(SAM_RETRY).await; } Err(e) => return Err(e), } diff --git a/crates/rbitcoin-net/src/ibd/assign.rs b/crates/rbitcoin-net/src/ibd/assign.rs index d19e2e2ec..df7431833 100644 --- a/crates/rbitcoin-net/src/ibd/assign.rs +++ b/crates/rbitcoin-net/src/ibd/assign.rs @@ -241,7 +241,8 @@ pub(crate) fn assign_work_ordered( return; } - let (bq_stop, bq_bytes, bq_count) = hub.query.block_queue_stats(); + let (bq_bytes, bq_count) = hub.query.block_queue_stats(); + let bq_stop = cfg.assign_stop_bytes; st.intake_stop = bq_stop; st.intake_queued = bq_bytes; @@ -443,7 +444,7 @@ fn assign_densify( } let densify_hi = path_lo.saturating_add(CONTIG_DENSIFY_AHEAD); - let depth_bytes = hub.query.block_queue_stats().1; + let depth_bytes = hub.query.block_queue_stats().0; let fetched_hi = hub .query .block_queue_max_height() @@ -455,7 +456,7 @@ fn assign_densify( densify_hi, depth_bytes, tip_rate_blocks_per_s, - rbitcoin_query::bq_assign_stop_bytes(), + cfg.assign_stop_bytes, fetched_hi, ); @@ -1642,7 +1643,6 @@ pub(in crate::ibd) mod tests { #[test] fn download_gate_stops_getdata_when_tip_plus_one_unconfirmable() { - let _env = lock_default_assign_stop(); let (dir, hub) = tmp_hub(); hub.ensure_genesis().unwrap(); let mut st = IbdWorkState::new(vec![dummy_slot(0)], hub.tip_hash(), hub.tip_height()); @@ -1662,7 +1662,6 @@ pub(in crate::ibd) mod tests { #[test] fn clear_inflight_add_peer_pop_need_and_tip_holes() { - let _env = lock_default_assign_stop(); let (dir, hub) = tmp_hub(); let mut st = IbdWorkState::new(vec![dummy_slot(0), dummy_slot(1)], None, Some(0)); let hash = h(10); @@ -1992,8 +1991,8 @@ pub(in crate::ibd) mod tests { // Quarter of the configured assign-stop. One new hash reserves 4 MiB, // so the stop must still fit that reserve on top of the queued quarter. let stop: u64 = 8 * 1024 * 1024; - std::env::set_var("RBITCOIN_BLOCK_QUEUE_BYTES", stop.to_string()); - let (mut st, cfg, stats) = gap_work(hub); + let (mut st, mut cfg, stats) = gap_work(hub); + cfg.assign_stop_bytes = stop; plant_work_path(&mut st, path_lo, path_lo.saturating_add(80)); let n = (stop / 4) as usize; let payload = vec![0u8; n]; @@ -2016,7 +2015,6 @@ pub(in crate::ibd) mod tests { .contains_key(&h(path_lo.saturating_add(TIP_HOLE_MAX as u32))), "densify stays off at the byte quarter" ); - std::env::remove_var("RBITCOIN_BLOCK_QUEUE_BYTES"); drain_block_queue(hub); let (mut st, cfg, stats) = gap_work(hub); @@ -2050,7 +2048,6 @@ pub(in crate::ibd) mod tests { fn ibd_cover_tip_holes() { use super::super::peer_io::ibd_mono_ms; use super::super::progress::claim_ready; - let _env = lock_default_assign_stop(); let (dir, hub) = tmp_hub(); hub.ensure_genesis().unwrap(); let (slots, mut wire) = wired_slots(6); @@ -2538,42 +2535,6 @@ pub(in crate::ibd) mod tests { let _ = std::fs::remove_dir_all(dir); } - /// Serialize env mutators — parallel suite races `bq_assign_stop_bytes`. - static BQ_ASSIGN_STOP_ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); - - pub(in crate::ibd) struct AssignStopEnvRestore( - Option, - Option, - ); - impl Drop for AssignStopEnvRestore { - fn drop(&mut self) { - match self.0.take() { - Some(v) => std::env::set_var("RBITCOIN_BLOCK_QUEUE_BYTES", v), - None => std::env::remove_var("RBITCOIN_BLOCK_QUEUE_BYTES"), - } - match self.1.take() { - Some(v) => std::env::set_var("RBITCOIN_BLOCK_QUEUE_GB", v), - None => std::env::remove_var("RBITCOIN_BLOCK_QUEUE_GB"), - } - } - } - - /// Tuple fields drop in order: the env is restored before the lock is - /// released, so the next holder's own setting survives. - pub(in crate::ibd) fn lock_default_assign_stop( - ) -> (AssignStopEnvRestore, std::sync::MutexGuard<'static, ()>) { - let g = BQ_ASSIGN_STOP_ENV_LOCK - .lock() - .unwrap_or_else(|e| e.into_inner()); - let restore = AssignStopEnvRestore( - std::env::var_os("RBITCOIN_BLOCK_QUEUE_BYTES"), - std::env::var_os("RBITCOIN_BLOCK_QUEUE_GB"), - ); - std::env::remove_var("RBITCOIN_BLOCK_QUEUE_BYTES"); - std::env::remove_var("RBITCOIN_BLOCK_QUEUE_GB"); - (restore, g) - } - /// Densify on one hub at genesis across IBD restarts: assign depth, the /// tip batch, band limits from the body queue, per-peer caps, hung /// owners, and reorg need. @@ -2583,7 +2544,6 @@ pub(in crate::ibd) mod tests { use super::super::dial::release_peer_block_work; use super::super::peer_io::ibd_mono_ms; use rbitcoin_query::{soft_confirm_window_n, BQ_SOFT_FREE_BYTES}; - let _env = lock_default_assign_stop(); let (dir, hub) = tmp_hub(); hub.ensure_genesis().unwrap(); let (slots, mut wire) = wired_slots(3); @@ -2725,7 +2685,7 @@ pub(in crate::ibd) mod tests { for ht in 1u32..=10 { enqueue(&mut st, ht, b"x"); } - assert!(hub.query.block_queue_stats().1 < BQ_SOFT_FREE_BYTES); + assert!(hub.query.block_queue_stats().0 < BQ_SOFT_FREE_BYTES); hub.query.set_lookup_taken_hi(Some(20)); assign_work_ordered(&mut st, &hub, &cfg, &stats, AssignDepth::Full, Some(0.1)); hub.query.set_lookup_taken_hi(None); @@ -2782,7 +2742,7 @@ pub(in crate::ibd) mod tests { enqueue(&mut st, 1, &chunk); enqueue(&mut st, 2, &chunk); drop(chunk); - assert!(hub.query.block_queue_stats().1 > BQ_SOFT_FREE_BYTES); + assert!(hub.query.block_queue_stats().0 > BQ_SOFT_FREE_BYTES); assert_eq!(soft_confirm_window_n(Some(0.1)), 6); assign_work_ordered(&mut st, &hub, &cfg, &stats, AssignDepth::Full, Some(0.1)); let issued = issued_heights(&st); @@ -2795,21 +2755,21 @@ pub(in crate::ibd) mod tests { restart_on(&mut st, &hub, &mut wire, &[0, 1]); cfg.window = 128; cfg.per_peer = 64; - std::env::set_var("RBITCOIN_BLOCK_QUEUE_BYTES", "2048"); + cfg.assign_stop_bytes = 2048; plant_work_path(&mut st, 1, 500); for ht in 1u32..=TIP_HOLE_MAX as u32 { enqueue(&mut st, ht, b"x"); } enqueue(&mut st, 500, &[0u8; 4096]); - assert!(hub.query.block_queue_stats().1 >= 2048); + assert!(hub.query.block_queue_stats().0 >= 2048); assert_eq!(hub.query.block_queue_max_height(), Some(500)); assign_work_ordered(&mut st, &hub, &cfg, &stats, AssignDepth::Full, Some(5.0)); - std::env::remove_var("RBITCOIN_BLOCK_QUEUE_BYTES"); assert!( st.inflight.is_empty(), "queue already at the assign-stop: no new getdata; issued={:?}", issued_heights(&st) ); + cfg.assign_stop_bytes = rbitcoin_query::BQ_ASSIGN_STOP_BYTES; // Tip+1 in hand and tip+2 the pre-hole. Past it, densify re-gets a // zombie pending (no wire) and a wrong first-wins body by path hash. @@ -3178,16 +3138,6 @@ pub(in crate::ibd) mod tests { #[test] fn hostile_peer_session() { - let _g = BQ_ASSIGN_STOP_ENV_LOCK - .lock() - .unwrap_or_else(|e| e.into_inner()); - let _restore = AssignStopEnvRestore( - std::env::var_os("RBITCOIN_BLOCK_QUEUE_BYTES"), - std::env::var_os("RBITCOIN_BLOCK_QUEUE_GB"), - ); - std::env::remove_var("RBITCOIN_BLOCK_QUEUE_GB"); - std::env::remove_var("RBITCOIN_BLOCK_QUEUE_BYTES"); - let (dir, hub) = tmp_hub(); hub.ensure_genesis().unwrap(); let four_mib = GETDATA_RESERVE_BYTES; @@ -3216,15 +3166,15 @@ pub(in crate::ibd) mod tests { assert_eq!(st.inflight.len(), before); assert!(!st.inflight.contains_key(&h(33))); - std::env::set_var("RBITCOIN_BLOCK_QUEUE_BYTES", "1000"); plant_work_path(&mut st, 1, 2); let queued = vec![0u8; 1000]; hub.query .block_queue_enqueue(50, h(50).to_byte_array(), 50, &queued) .unwrap(); - assert!(hub.query.block_queue_stats().1 >= 1000); + assert!(hub.query.block_queue_stats().0 >= 1000); let stats = LoopStats::default(); - let cfg = IbdConfig::for_test(); + let mut cfg = IbdConfig::for_test(); + cfg.assign_stop_bytes = 1000; assign_work_ordered(&mut st, &hub, &cfg, &stats, AssignDepth::Full, None); assert!( !st.inflight.contains_key(&h(1)) && !st.inflight.contains_key(&h(2)), @@ -3239,7 +3189,6 @@ pub(in crate::ibd) mod tests { /// cursor has moved past it. A holder whose peer is gone does not cover it. #[test] fn tip_hole_in_front_of_queued_bodies_is_requested() { - let _env = lock_default_assign_stop(); let (dir, hub) = tmp_hub(); hub.ensure_genesis().unwrap(); let (slots, mut wire) = wired_slots(2); diff --git a/crates/rbitcoin-net/src/ibd/confirm/mod.rs b/crates/rbitcoin-net/src/ibd/confirm/mod.rs index 2dd297544..29e0ed2cd 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/mod.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/mod.rs @@ -966,7 +966,7 @@ impl AccountedHeap { fn from_hub(hub: &ChainHub, inflight_bytes: u64, wire_bytes: u64) -> Self { let owned = hub.query.process_owned_size_snapshot(); - let (_stop, bq_bytes, _count) = hub.query.block_queue_stats(); + let (bq_bytes, _count) = hub.query.block_queue_stats(); let head = owned.head; Self { bq_bytes, @@ -1961,6 +1961,10 @@ pub(crate) fn spawn_confirm_engine( .iter() .map(|&(h, raw)| (h, BlockHash::from_byte_array(raw))) .collect(); + // Same lock as a block connect. The chain tip must not move + // while the relay-on purge is choosing the tip it stripped. + let mp_hold = hub_wb.mempool().cloned(); + let mut template_hold = mp_hold.as_ref().map(|mp| mp.template_hold()); match rbitcoin_query::confirm_write_phase_or_return( &hub_wb.query, &hub_wb.params, @@ -1968,7 +1972,9 @@ pub(crate) fn spawn_confirm_engine( batch, ) { Ok(_fks) => { - if let Err(e) = hub_wb.note_confirmed_tip(&meta, &headers) { + if let Err(e) = + hub_wb.note_confirmed_tip_while(&meta, &headers, template_hold.as_mut()) + { warn!("ibd: confirm write note tip: {e}"); } let t_deq = Instant::now(); @@ -2041,7 +2047,11 @@ pub(crate) fn spawn_confirm_engine( &heights_hashes, ) { Ok(()) => { - if let Err(e) = hub_wb.note_confirmed_tip(&meta, &headers) { + if let Err(e) = hub_wb.note_confirmed_tip_while( + &meta, + &headers, + template_hold.as_mut(), + ) { warn!("ibd: confirm write note tip: {e}"); } for (h, raw) in &heights_hashes { diff --git a/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs b/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs index d34657e74..b787f8d53 100644 --- a/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs +++ b/crates/rbitcoin-net/src/ibd/events/confirm_reject_tests.rs @@ -1090,7 +1090,7 @@ fn apply_peer_event_block_framed_bq_horizon_and_headers_done() { local, None, ); - assert_eq!(hub.query.block_queue_stats().2, 1); + assert_eq!(hub.query.block_queue_stats().1, 1); let b2 = shell(h1, 2, 2); let h2 = b2.block_hash(); @@ -1304,7 +1304,7 @@ fn mine( #[allow(clippy::cognitive_complexity)] // one chain, many fork and reject arms #[test] fn ibd_bad_prev_fork() { - use super::super::assign::tests::{dummy_slot, lock_default_assign_stop}; + use super::super::assign::tests::dummy_slot; use super::super::assign::{assign_work_ordered, AssignDepth}; use super::super::confirm::{ConfirmEvent, ConfirmFeed}; use super::super::exit::header_lag_behind_peers; @@ -1324,7 +1324,6 @@ fn ibd_bad_prev_fork() { use std::sync::atomic::AtomicU32; use std::time::{Duration, Instant}; - let _env = lock_default_assign_stop(); let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("bad-prev-fork"); hub.ensure_genesis().unwrap(); let gen = hub.tip_hash().unwrap(); @@ -1433,7 +1432,7 @@ fn ibd_bad_prev_fork() { local, Some(&feed), ); - assert_eq!(hub.query.block_queue_stats().2, 0, "undecodable wire"); + assert_eq!(hub.query.block_queue_stats().1, 0, "undecodable wire"); assert_eq!(feed.size_snap().0, 0); assert!(st.inflight.is_empty() && !st.body.is_pending(&a1)); assert!(!st.body.is_rejected(&a1), "the hash stays fetchable"); @@ -1460,7 +1459,7 @@ fn ibd_bad_prev_fork() { Some(&feed), ); assert!(st.body.is_pending(&a1) && st.inflight.is_empty()); - assert_eq!(hub.query.block_queue_stats().2, 1); + assert_eq!(hub.query.block_queue_stats().1, 1); assert_eq!(feed.size_snap().0, 1); } @@ -2429,11 +2428,9 @@ fn wire_pad_copy_keeps_the_other_tip() { /// honest body from another peer connects. #[test] fn ibd_mutated_body_is_refetched_not_blacklisted() { - use super::super::assign::tests::lock_default_assign_stop; use bitcoin::{ScriptBuf, Transaction, Witness}; use std::sync::Arc; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("mutated-body", 5); let (t, cbs) = (rig.t, rig.cbs.clone()); let honest = rbitcoin_consensus::mine_regtest_paying( @@ -2529,13 +2526,11 @@ fn ibd_mutated_body_is_refetched_not_blacklisted() { /// block is claimed while the one before it is still in the pipeline. #[test] fn isolation_claims_one_block_at_a_time() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::state::InflightReq; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; use std::time::{Duration, Instant}; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("isolate-serial", 1); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -2582,12 +2577,10 @@ fn isolation_claims_one_block_at_a_time() { /// honest bodies connect. #[test] fn ibd_batched_mutated_body_is_isolated_to_its_block() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::state::InflightReq; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("batched-mutated", 3); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -2652,14 +2645,12 @@ fn ibd_batched_mutated_body_is_isolated_to_its_block() { /// tip+2 alone is rejected as invalid. #[test] fn batched_write_reject_offers_the_wave_back() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::confirm::ConfirmEvent; use super::super::state::InflightReq; use bitcoin::{ScriptBuf, Sequence}; use rbitcoin_consensus::mine_regtest_paying; use std::time::{Duration, Instant}; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("write-reject", 2); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -2756,12 +2747,10 @@ fn batched_write_reject_offers_the_wave_back() { /// is archived, not connected): tip+1 is invalid, and tip+2 is not judged. #[test] fn batched_spend_of_a_later_block_is_rejected_alone() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::state::InflightReq; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("forward-spend", 2); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -2810,12 +2799,10 @@ fn batched_spend_of_a_later_block_is_rejected_alone() { /// the archive, and that retry must still pin the same-block parent. #[test] fn archived_retry_pins_a_same_block_parent() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::state::InflightReq; use bitcoin::{Amount, ScriptBuf}; use rbitcoin_consensus::{grind_regtest_pow, mine_regtest_paying}; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("archived-same-block", 2); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -2867,12 +2854,10 @@ fn archived_retry_pins_a_same_block_parent() { /// first, noted in flight with no outs. #[test] fn archived_wave_pins_a_parent_from_the_previous_chunk() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::state::InflightReq; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("archived-cross-chunk", 1); let (t, cbs) = (rig.t, rig.cbs.clone()); let op_true = ScriptBuf::from_bytes(vec![0x51]); @@ -2950,12 +2935,10 @@ fn archived_wave_pins_a_parent_from_the_previous_chunk() { /// connects with no later wave, disconnect, or reorg to re-arm lookup. #[test] fn load_engine_fault_near_the_tip_retakes_the_wave() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::state::InflightReq; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("load-engine-fault", 1); let (t, cbs) = (rig.t, rig.cbs.clone()); let b1 = mine_regtest_paying( @@ -2988,13 +2971,11 @@ fn load_engine_fault_near_the_tip_retakes_the_wave() { /// outputs read again, both blocks connect. #[test] fn batched_pin_engine_fault_retries_one_block_at_a_time() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::state::InflightReq; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; use std::time::{Duration, Instant}; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("pin-engine-fault", 1); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -3085,14 +3066,12 @@ fn isolated_soft_reject_does_not_count_toward_cascade_halt() { /// and tip+2 alone is rejected as invalid. #[test] fn batched_scripts_reject_offers_the_wave_back() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::confirm::ConfirmEvent; use super::super::state::InflightReq; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; use std::time::{Duration, Instant}; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("scripts-reject", 2); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -3177,14 +3156,12 @@ fn batched_scripts_reject_offers_the_wave_back() { /// is still the peer dropped, and both honest bodies then connect. #[test] fn isolated_mutated_head_punishes_its_sender_when_lookup_runs_ahead() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::confirm::ConfirmEvent; use super::super::state::InflightReq; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; use std::time::{Duration, Instant}; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("mutated-head-ahead", 3); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -3270,7 +3247,6 @@ fn isolated_mutated_head_punishes_its_sender_when_lookup_runs_ahead() { /// the live sender of the replacement. #[test] fn mutated_verdict_punishes_its_copy_not_a_replacement() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::assign::{assign_work_ordered, AssignDepth}; use super::super::confirm::ConfirmEvent; use super::super::state::InflightReq; @@ -3278,7 +3254,6 @@ fn mutated_verdict_punishes_its_copy_not_a_replacement() { use rbitcoin_consensus::mine_regtest_paying; use std::time::{Duration, Instant}; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("mutated-replaced", 3); let (t, cbs) = (rig.t, rig.cbs.clone()); let b1 = mine_regtest_paying( @@ -3353,13 +3328,12 @@ fn mutated_verdict_punishes_its_copy_not_a_replacement() { #[test] fn mutated_body_from_an_earlier_run_punishes_no_current_peer() { use super::super::archive::rehydrate_block_queue_into_confirm; - use super::super::assign::tests::{dummy_slot, lock_default_assign_stop}; + use super::super::assign::tests::dummy_slot; use super::super::dial::next_peer_id; use super::super::state::InflightReq; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("sender-across-runs", 1); let (t, cbs) = (rig.t, rig.cbs.clone()); let b1 = mine_regtest_paying( @@ -3462,7 +3436,7 @@ fn ibd_flag_zero_tx_block_is_consensus_invalid() { ); assert!(st.slots[0].alive, "Core's encoding is not undecodable wire"); assert!(st.body.is_pending(&hash)); - assert_eq!(hub.query.block_queue_stats().2, 1); + assert_eq!(hub.query.block_queue_stats().1, 1); let (ev_tx, ev_rx) = channel(); let (engine, _queues) = spawn_confirm_engine( @@ -3534,11 +3508,9 @@ fn ibd_flag_zero_tx_block_is_consensus_invalid() { /// not halt, and it follows the other valid tip+1. #[test] fn ibd_spend_of_an_unknown_txid_is_invalid() { - use super::super::assign::tests::lock_default_assign_stop; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("unknown-parent", 2); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -3582,11 +3554,9 @@ fn ibd_spend_of_an_unknown_txid_is_invalid() { /// marked invalid, as it would be without the restart. #[test] fn spend_of_an_unknown_txid_after_restart_is_invalid() { - use super::super::assign::tests::lock_default_assign_stop; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("unknown-parent-restart", 1).restart(|_| {}); let t = rig.t; let ghost = WireRig::spend(bitcoin::Txid::from_byte_array([0xcd; 32])); @@ -3616,12 +3586,10 @@ fn spend_of_an_unknown_txid_after_restart_is_invalid() { /// still be draining. The parent is on the chain, so tip+2 connects. #[test] fn lone_spend_of_the_block_just_written_connects() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::state::InflightReq; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("spend-just-written", 1); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -3665,12 +3633,10 @@ fn lone_spend_of_the_block_just_written_connects() { /// view), so the new chain's block that spends it is invalid. #[test] fn spend_of_a_reorged_out_tx_is_invalid() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::reorg::maybe_rewind_to_best_work; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("reorged-parent", 1); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -3725,12 +3691,10 @@ fn spend_of_a_reorged_out_tx_is_invalid() { /// invalid without waiting for its pending mark to go stale. #[test] fn verdict_applied_after_the_tip_moved_asks_for_the_body_again() { - use super::super::assign::tests::lock_default_assign_stop; use super::super::reorg::apply_header_rewind; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("verdict-tip-moved", 1); let t = rig.t; let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -3770,11 +3734,9 @@ fn verdict_applied_after_the_tip_moved_asks_for_the_body_again() { /// connected chain: a store fault, so IBD halts and the hash stays valid. #[test] fn spend_of_a_create_the_head_lost_is_not_invalid() { - use super::super::assign::tests::lock_default_assign_stop; use bitcoin::ScriptBuf; use rbitcoin_consensus::mine_regtest_paying; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new("lost-head-row", 1); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -3819,11 +3781,9 @@ fn spend_of_a_create_the_head_lost_is_not_invalid() { /// connected chain: a store fault, so the block is not marked invalid. #[test] fn spend_of_a_create_the_head_lost_before_a_roll_is_not_invalid() { - use super::super::assign::tests::lock_default_assign_stop; use bitcoin::ScriptBuf; use rbitcoin_consensus::{mine_empty_regtest, mine_regtest_paying}; - let _env = lock_default_assign_stop(); let mut rig = WireRig::new_small_head("lost-head-row-roll", 1); let (t, cbs) = (rig.t, rig.cbs.clone()); let spk = ScriptBuf::from_bytes(vec![0x51]); @@ -3894,7 +3854,7 @@ fn spend_of_a_create_the_head_lost_before_a_roll_is_not_invalid() { /// for it again, or confirm waits at that height forever. #[test] fn ibd_refused_body_is_asked_again() { - use super::super::assign::tests::{dummy_slot, lock_default_assign_stop}; + use super::super::assign::tests::dummy_slot; use super::super::assign::{assign_work_ordered, AssignDepth}; use super::super::confirm::{ConfirmEvent, ConfirmFeed}; use super::super::path::seed_work_path_from_store; @@ -3908,7 +3868,6 @@ fn ibd_refused_body_is_asked_again() { use std::sync::atomic::AtomicU32; use std::time::Instant; - let _env = lock_default_assign_stop(); let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("refused-body-reask"); hub.ensure_genesis().unwrap(); let gen = hub.tip_hash().unwrap(); diff --git a/crates/rbitcoin-net/src/ibd/events/ibd_memory_tests.rs b/crates/rbitcoin-net/src/ibd/events/ibd_memory_tests.rs index fbe4ec3b1..c74375cc0 100644 --- a/crates/rbitcoin-net/src/ibd/events/ibd_memory_tests.rs +++ b/crates/rbitcoin-net/src/ibd/events/ibd_memory_tests.rs @@ -121,7 +121,7 @@ fn unsolicited_body_is_not_copied_into_the_queue() { let mut st = IbdWorkState::new(Vec::new(), hub.tip_hash(), hub.tip_height()); st.record_height(hash, 1); st.header_fks.insert(hash, rbitcoin_primitives::Fk(1)); - let before = hub.query.block_queue_stats().1; + let before = hub.query.block_queue_stats().0; let write_next = AtomicU32::new(0); apply_block_framed( &mut st, @@ -136,7 +136,7 @@ fn unsolicited_body_is_not_copied_into_the_queue() { }, ); assert_eq!( - hub.query.block_queue_stats().1, + hub.query.block_queue_stats().0, before, "unsolicited body must not be copied into the queue" ); @@ -156,10 +156,10 @@ fn unsolicited_body_is_not_copied_into_the_queue() { }, ); assert!( - hub.query.block_queue_stats().1 > before, + hub.query.block_queue_stats().0 > before, "a requested body is still queued" ); - let queued = hub.query.block_queue_stats().1; + let queued = hub.query.block_queue_stats().0; st.inflight .insert(hash, super::super::state::InflightReq::new(0)); apply_block_framed( @@ -175,7 +175,7 @@ fn unsolicited_body_is_not_copied_into_the_queue() { }, ); assert_eq!( - hub.query.block_queue_stats().1, + hub.query.block_queue_stats().0, queued, "a second copy of a queued hash is dropped" ); diff --git a/crates/rbitcoin-net/src/ibd/mod.rs b/crates/rbitcoin-net/src/ibd/mod.rs index fa3e9f8a2..2a192c519 100644 --- a/crates/rbitcoin-net/src/ibd/mod.rs +++ b/crates/rbitcoin-net/src/ibd/mod.rs @@ -190,6 +190,10 @@ pub struct IbdConfig { pub peers: Option>>, /// Outbound TCP: direct or SOCKS5. pub dialer: crate::socks::Dialer, + /// Densify assign-stop in bytes. `u64::MAX` is unlimited. + /// Production copies [`rbitcoin_query::bq_assign_stop_bytes`] once at + /// startup. Tests pass the cap here instead of mutating process env. + pub assign_stop_bytes: u64, } impl Default for IbdConfig { @@ -203,6 +207,7 @@ impl Default for IbdConfig { connect_timeout: Duration::from_secs(8), peers: None, dialer: crate::socks::Dialer::Direct, + assign_stop_bytes: rbitcoin_query::bq_assign_stop_bytes(), } } } @@ -219,6 +224,7 @@ impl IbdConfig { connect_timeout: Duration::from_millis(400), peers: None, dialer: crate::socks::Dialer::Direct, + assign_stop_bytes: rbitcoin_query::BQ_ASSIGN_STOP_BYTES, } } } @@ -576,7 +582,7 @@ pub async fn ibd_cancellable( if cadence.assign_due(now_cadence, st.inflight.is_empty()) { let tip_rate_opt = tip_rate_tracker.eta_rate(now_cadence); let _ = hub.query.block_queue_update_soft_pressure(tip_rate_opt); - let (_bq_budget, bq_bytes, bq_count) = hub.query.block_queue_stats(); + let (bq_bytes, bq_count) = hub.query.block_queue_stats(); let bq_window_covered = rbitcoin_query::soft_confirm_window_covered( bq_count as u32, bq_bytes, @@ -886,7 +892,7 @@ pub async fn ibd_cancellable( g.inflight.clone() }; let ready_n = confirm::confirm_ready_count(&hub.query, path_lo, &inflight_h); - let (_bq_budget, bq_bytes, bq_count) = hub.query.block_queue_stats(); + let (bq_bytes, bq_count) = hub.query.block_queue_stats(); tip_rate_tracker.push(now, prog.tip); let eta_rate = tip_rate_tracker.eta_rate(now); diff --git a/crates/rbitcoin-net/src/ibd/progress.rs b/crates/rbitcoin-net/src/ibd/progress.rs index 875a49d05..71b876a04 100644 --- a/crates/rbitcoin-net/src/ibd/progress.rs +++ b/crates/rbitcoin-net/src/ibd/progress.rs @@ -473,7 +473,7 @@ pub(crate) fn spawn_status_thread( let mut snap = board.snap.lock().unwrap_or_else(|e| e.into_inner()).clone(); snap.hole = board.hole.load(std::sync::atomic::Ordering::Relaxed); let published = read_progress_published(&query); - let (_budget, bq_bytes, bq_count) = query.block_queue_stats(); + let (bq_bytes, bq_count) = query.block_queue_stats(); let now = Instant::now(); log_progress_tick( &mut clock, diff --git a/crates/rbitcoin-net/src/lib.rs b/crates/rbitcoin-net/src/lib.rs index 24fa6739b..edf1950d9 100644 --- a/crates/rbitcoin-net/src/lib.rs +++ b/crates/rbitcoin-net/src/lib.rs @@ -58,9 +58,8 @@ pub use net_permissions::{ pub use netaddr::{is_cjdns_ip, NetAddr, OnlyNet}; pub use netgroup::netgroup; pub use peer::{ - drain_pending_now, flush_tx_invs, force_announce_txid, local_service_flags, - local_service_flags_pruned, run_feeler_timed, set_compact_filters_service, PendingBlocks, - V2PlainSession, MAX_SERVE_BLOCKS, + drain_pending_now, flush_tx_invs, force_announce_txid, local_service_flags, run_feeler_timed, + PendingBlocks, V2PlainSession, MAX_SERVE_BLOCKS, }; pub use peer_dos::DEFAULT_MAX_INBOUND; pub use peers::{ @@ -80,7 +79,7 @@ pub use seeds::{ pub use serve_perf::{ format_serve_perf, sample_reset_serve_perf, serve_perf_totals, ServePerfSample, }; -pub use service::P2PNode; +pub use service::{compiled_with_test_hooks, P2PNode}; pub use socks::{install_i2p_dialer, Dialer}; pub use tx_relay::{ ElectrumMempoolItem, FeeHistoryBackfillStats, MempoolAnnounce, MempoolHub, MempoolPerfSample, diff --git a/crates/rbitcoin-net/src/peer.rs b/crates/rbitcoin-net/src/peer.rs index f04a2b7cb..2f7c9cab4 100644 --- a/crates/rbitcoin-net/src/peer.rs +++ b/crates/rbitcoin-net/src/peer.rs @@ -30,7 +30,7 @@ use rbitcoin_primitives::Height; use rbitcoin_query::Query; use std::collections::{BTreeSet, HashMap, HashSet, VecDeque}; use std::net::SocketAddr; -use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::net::TcpStream; @@ -328,34 +328,21 @@ fn stash_pending_block(pending: &mut PendingBlocks, hash: BlockHash, block: bitc pending.map.insert(hash, block); } -/// Services we advertise once store-backed reconstruct serve is available. -pub fn local_service_flags() -> ServiceFlags { - local_service_flags_pruned(false) -} - -static COMPACT_FILTERS_ADVERTISED: AtomicBool = AtomicBool::new(false); - -/// Version messages include `NODE_COMPACT_FILTERS` while this is set. +/// Service bits for one node. /// -/// The node sets it once basic filters first reach the tip, so a peer never -/// hears the bit from a node that would answer its requests with silence. -/// A peer that handshook earlier does not learn of a later flip. -pub fn set_compact_filters_service(on: bool) { - COMPACT_FILTERS_ADVERTISED.store(on, Ordering::Release); -} - /// BIP159: a pruned node offers `NETWORK_LIMITED`, not `NETWORK`. -pub fn local_service_flags_pruned(pruned: bool) -> ServiceFlags { +/// `NODE_COMPACT_FILTERS` is that hub's bit. No hub means this node has +/// not caught filters up. +pub fn local_service_flags(pruned: bool, peers: Option<&crate::peers::PeerHub>) -> ServiceFlags { let mut base = if pruned { ServiceFlags::NETWORK_LIMITED | ServiceFlags::WITNESS | ServiceFlags::P2P_V2 } else { crate::seeds::required_seed_services() }; - if COMPACT_FILTERS_ADVERTISED.load(Ordering::Acquire) { - base.add(ServiceFlags::COMPACT_FILTERS) - } else { - base + if peers.is_some_and(|p| p.compact_filters_advertised()) { + base.add(ServiceFlags::COMPACT_FILTERS); } + base } /// NETWORK_LIMITED is enough when the tip is shallower than this (~24h at 10m). @@ -514,7 +501,10 @@ fn queue_addr_list( addrs: Vec<(u32, crate::NetAddr)>, v2: bool, ) -> Result<(), NetError> { - let services = local_service_flags(); + let services = session + .and_then(|s| s.peer_hub()) + .map(|h| local_service_flags(h.is_pruned(), Some(&h))) + .unwrap_or_else(|| local_service_flags(false, None)); if v2 { let list: Vec = addrs .into_iter() @@ -894,6 +884,7 @@ pub async fn run_feeler( their_addr: SocketAddr, start_height: i32, user_agent: &str, + peers: &crate::peers::PeerHub, ) -> Result<(), NetError> { run_feeler_timed( HANDSHAKE_TIMEOUT, @@ -903,11 +894,13 @@ pub async fn run_feeler( their_addr, start_height, user_agent, + peers, ) .await } /// Feeler handshake with an explicit timeout (production uses [`HANDSHAKE_TIMEOUT`]). +#[allow(clippy::too_many_arguments)] // call-site args stay unbundled pub async fn run_feeler_timed( limit: Duration, stream: TcpStream, @@ -916,6 +909,7 @@ pub async fn run_feeler_timed( their_addr: SocketAddr, start_height: i32, user_agent: &str, + peers: &crate::peers::PeerHub, ) -> Result<(), NetError> { tokio::time::timeout( limit, @@ -926,6 +920,7 @@ pub async fn run_feeler_timed( their_addr, start_height, user_agent, + peers, ), ) .await @@ -939,9 +934,10 @@ async fn run_feeler_inner( their_addr: SocketAddr, start_height: i32, user_agent: &str, + peers: &crate::peers::PeerHub, ) -> Result<(), NetError> { let (mut reader, mut writer, _wire, _tcp_shutdown) = open_v2(stream, magic, false).await?; - let services = local_service_flags(); + let services = local_service_flags(peers.is_pruned(), Some(peers)); let now = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_secs() as i64) @@ -1031,7 +1027,7 @@ async fn application_handshake( policy: HandshakePolicy<'_>, ) -> Result { let pruned = policy.peers.map(|p| p.is_pruned()).unwrap_or(false); - let services = local_service_flags_pruned(pruned); + let services = local_service_flags(pruned, policy.peers); let now = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_secs() as i64) diff --git a/crates/rbitcoin-net/src/peer_tests.rs b/crates/rbitcoin-net/src/peer_tests.rs index c15bbab9a..ebd59caa9 100644 --- a/crates/rbitcoin-net/src/peer_tests.rs +++ b/crates/rbitcoin-net/src/peer_tests.rs @@ -383,7 +383,7 @@ fn live_follow_dec_on_drop() { #[test] fn local_service_flags_include_network_witness_v2() { - let f = local_service_flags(); + let f = local_service_flags(false, None); assert!(f.has(ServiceFlags::NETWORK)); assert!(f.has(ServiceFlags::WITNESS)); assert!(f.has(ServiceFlags::P2P_V2)); @@ -391,12 +391,28 @@ fn local_service_flags_include_network_witness_v2() { #[test] fn local_service_flags_pruned_are_limited_not_network() { - let f = local_service_flags_pruned(true); + let f = local_service_flags(true, None); assert!(f.has(ServiceFlags::NETWORK_LIMITED)); assert!(!f.has(ServiceFlags::NETWORK)); assert!(f.has(ServiceFlags::WITNESS)); assert!(f.has(ServiceFlags::P2P_V2)); - assert_eq!(local_service_flags_pruned(false), local_service_flags()); +} + +#[test] +fn compact_filter_bit_stays_on_the_node_that_caught_up() { + use bitcoin::p2p::ServiceFlags; + let quiet = crate::peers::PeerHub::new(); + let caught_up = crate::peers::PeerHub::new(); + caught_up.set_compact_filters_service(true); + assert!( + !local_service_flags(false, Some(&quiet)).has(ServiceFlags::COMPACT_FILTERS), + "a neighbor catching up must not advertise filters for this node" + ); + assert!(local_service_flags(false, Some(&caught_up)).has(ServiceFlags::COMPACT_FILTERS)); + assert!( + !local_service_flags(false, None).has(ServiceFlags::COMPACT_FILTERS), + "no session table means this node has not caught filters up" + ); } #[test] @@ -3361,6 +3377,7 @@ async fn handshake_timeout_after_silence() { addr, 0, "/rbitcoin:test/", + crate::peers::PeerHub::new().as_ref(), ) .await }), diff --git a/crates/rbitcoin-net/src/peers.rs b/crates/rbitcoin-net/src/peers.rs index 6518d0193..d1b52f1e1 100644 --- a/crates/rbitcoin-net/src/peers.rs +++ b/crates/rbitcoin-net/src/peers.rs @@ -551,7 +551,7 @@ impl LivePeer { self.next_local_addr_send .compare_exchange(prev, next, Ordering::Relaxed, Ordering::Relaxed) .ok()?; - let services = crate::peer::local_service_flags_pruned(hub.is_pruned()); + let services = crate::peer::local_service_flags(hub.is_pruned(), Some(&hub)); let t = now as u32; if self.wants_addrv2() { let mut v = Vec::new(); @@ -1474,6 +1474,8 @@ pub struct PeerHub { /// Core `-i2psam`: I2P rows may enter addrman. Off, they are still relayed. i2p_reachable: AtomicBool, pruned: AtomicBool, + /// `NODE_COMPACT_FILTERS` once this node's basic filters reach the tip. + compact_filters: AtomicBool, asmap: Mutex>>, /// Tip-mode mempool for Core `EraseForPeer` on disconnect. mempool: Mutex>>, @@ -1569,6 +1571,7 @@ impl PeerHub { cjdns_reachable: AtomicBool::new(false), i2p_reachable: AtomicBool::new(false), pruned: AtomicBool::new(false), + compact_filters: AtomicBool::new(false), asmap: Mutex::new(None), mempool: Mutex::new(None), net_perms: Mutex::new(crate::net_permissions::NetPermTable::default()), @@ -1688,6 +1691,16 @@ impl PeerHub { self.pruned.load(Ordering::Relaxed) } + /// Version, feeler, addr, and `getnetworkinfo` include `NODE_COMPACT_FILTERS` + /// after this node's filters reach the tip. + pub fn set_compact_filters_service(&self, on: bool) { + self.compact_filters.store(on, Ordering::Release); + } + + pub fn compact_filters_advertised(&self) -> bool { + self.compact_filters.load(Ordering::Acquire) + } + pub fn set_listen_port(&self, port: u16) { self.listen_port.store(port, Ordering::Relaxed); } @@ -1853,7 +1866,7 @@ impl PeerHub { &self, bind: SocketAddr, ) -> Vec<(u32, bitcoin::p2p::address::Address)> { - let services = crate::peer::local_service_flags(); + let services = crate::peer::local_service_flags(self.is_pruned(), Some(self)); self.addr_response_net(bind, false) .into_iter() .filter_map(|(t, a)| match a { diff --git a/crates/rbitcoin-net/src/service.rs b/crates/rbitcoin-net/src/service.rs index 0a2af1f64..9a74683fa 100644 --- a/crates/rbitcoin-net/src/service.rs +++ b/crates/rbitcoin-net/src/service.rs @@ -47,6 +47,9 @@ pub struct P2PNode { /// Shared inbound slots across all listen sockets. inbound_sem: Arc, dialer: crate::socks::Dialer, + /// Wakes accept, dial, and connect-retry so `shutdown` does not sit out + /// its grace waiting on a recv or a 2s tick. + stop_tx: tokio::sync::watch::Sender, } impl P2PNode { @@ -149,6 +152,7 @@ impl P2PNode { let cache = hub.cache.clone(); let query = hub.query.clone(); let shutdown = Arc::new(AtomicBool::new(false)); + let (stop_tx, _stop_rx) = tokio::sync::watch::channel(false); let (listener, local_addr) = if let Some(addr) = listen { let listener = TcpListener::bind(addr).await?; @@ -182,6 +186,7 @@ impl P2PNode { inbound_sem.clone(), shutdown.clone(), session_tasks.clone(), + stop_tx.subscribe(), )); } @@ -193,8 +198,17 @@ impl P2PNode { let dial_shutdown = shutdown.clone(); let sessions_dial = session_tasks.clone(); let dialer_task = dialer.clone(); + let mut dial_stop = stop_tx.subscribe(); let dial_task = tokio::spawn(async move { - while let Some(req) = dial_rx.recv().await { + loop { + let req = tokio::select! { + biased; + _ = dial_stop.changed() => break, + r = dial_rx.recv() => match r { + Some(req) => req, + None => break, + }, + }; if dial_shutdown.load(Ordering::SeqCst) { break; } @@ -230,12 +244,17 @@ impl P2PNode { // pool (`PeerHub::redial_remembered_off_runtime`). let retry_peers = peers.clone(); let retry_shutdown = shutdown.clone(); + let mut retry_stop = stop_tx.subscribe(); let retry_task = tokio::spawn(async move { - let mut interval = tokio::time::interval(Duration::from_secs(2)); + let mut interval = tokio::time::interval(REDIAL_INTERVAL); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); interval.tick().await; loop { - interval.tick().await; + tokio::select! { + biased; + _ = retry_stop.changed() => break, + _ = interval.tick() => {} + } if retry_shutdown.load(Ordering::SeqCst) { break; } @@ -259,6 +278,7 @@ impl P2PNode { max_inbound, inbound_sem, dialer, + stop_tx, }) } @@ -290,6 +310,7 @@ impl P2PNode { self.inbound_sem.clone(), self.shutdown.clone(), self.session_tasks.clone(), + self.stop_tx.subscribe(), ); self.tasks.push(accept_task); Ok(local_addr) @@ -434,6 +455,7 @@ impl P2PNode { pub async fn shutdown(mut self) { self.shutdown.store(true, Ordering::SeqCst); + let _ = self.stop_tx.send(true); for p in self.peers.live_peers() { p.request_disconnect(); } @@ -463,6 +485,7 @@ impl Drop for P2PNode { // The connect-retry interval runs until this flag or an abort. // Dropping the handle without `shutdown` must not pin the test runtime. self.shutdown.store(true, Ordering::SeqCst); + let _ = self.stop_tx.send(true); for t in &self.tasks { t.abort(); } @@ -493,15 +516,20 @@ fn spawn_inbound_accept( inbound_sem: Arc, shutdown: Arc, session_tasks: Arc>>>, + mut stop: tokio::sync::watch::Receiver, ) -> JoinHandle<()> { tokio::spawn(async move { loop { if shutdown.load(Ordering::SeqCst) { break; } - let accept = tokio::time::timeout(Duration::from_millis(200), listener.accept()).await; + let accept = tokio::select! { + biased; + _ = stop.changed() => break, + a = listener.accept() => a, + }; match accept { - Ok(Ok((stream, peer_addr))) => { + Ok((stream, peer_addr)) => { if peers.inbound_discouraged(peer_addr) { rbitcoin_log::info!("p2p: reject discouraged inbound {peer_addr}"); drop(stream); @@ -579,8 +607,14 @@ fn spawn_inbound_accept( let _ = ah_tx.send(h.abort_handle()); push_session_task(&sessions, h); } - Ok(Err(_)) => break, - Err(_) => continue, + Err(e) => { + if accept_error_ends_listen(&e) { + rbitcoin_log::warn!("p2p: accept ended: {e}"); + break; + } + rbitcoin_log::warn!("p2p: accept retry: {e}"); + tokio::time::sleep(Duration::from_millis(50)).await; + } } } }) @@ -603,6 +637,36 @@ struct PreparedOutbound { id: u64, } +/// True when this crate was compiled with the `test-hooks` feature. +/// +/// Journeys turn that on. A workspace build unifies it into the node binary, +/// which then uses the short redial and SAM retry. The node warns at startup. +pub fn compiled_with_test_hooks() -> bool { + cfg!(feature = "test-hooks") +} + +/// A transient accept error should not stop the listen loop. +/// +/// `ConnectionAborted` is a peer that vanished during accept. +/// `Interrupted` is a signal. `EMFILE` / `ENFILE` (24 / 23) clear once a +/// socket is closed. Anything else, including a closed listener, ends accept. +pub(crate) fn accept_error_ends_listen(err: &std::io::Error) -> bool { + if matches!( + err.kind(), + std::io::ErrorKind::ConnectionAborted | std::io::ErrorKind::Interrupted + ) { + return false; + } + !matches!(err.raw_os_error(), Some(23 | 24)) +} + +/// How often a down `--connect` / `addnode` target is tried again. +#[cfg(not(any(test, feature = "test-hooks")))] +const REDIAL_INTERVAL: Duration = Duration::from_secs(2); +/// Crate tests and journeys. The node binary keeps the 2s pass. +#[cfg(any(test, feature = "test-hooks"))] +const REDIAL_INTERVAL: Duration = Duration::from_millis(250); + /// Bound on the TCP / SOCKS / SAM connect of an outbound dial, the same 8 s /// as the IBD and tip-follow dials ([`crate::connect_timeout_for`] gives I2P /// longer). Without it a proxy that accepts and never answers holds the @@ -762,7 +826,16 @@ async fn run_outbound_session_with_abort( } let stream = dialer.connect_net(peer_net).await?; let height = hub.tip_height().map(|h| h as i32).unwrap_or(0); - return crate::peer::run_feeler(stream, magic, local, peer_addr, height, &user_agent).await; + return crate::peer::run_feeler( + stream, + magic, + local, + peer_addr, + height, + &user_agent, + peers.as_ref(), + ) + .await; } let prepared = prepare_outbound_session( peer, @@ -795,6 +868,20 @@ pub fn magic_for_params(params: &ChainParams) -> Magic { mod tests { use super::*; + #[test] + fn accept_retries_a_full_fd_table_and_stops_on_a_dead_listener() { + let aborted = std::io::Error::new(std::io::ErrorKind::ConnectionAborted, "reset"); + let interrupted = std::io::Error::new(std::io::ErrorKind::Interrupted, "signal"); + let emfile = std::io::Error::from_raw_os_error(24); + let enfile = std::io::Error::from_raw_os_error(23); + let dead = std::io::Error::new(std::io::ErrorKind::InvalidInput, "closed"); + assert!(!accept_error_ends_listen(&aborted)); + assert!(!accept_error_ends_listen(&interrupted)); + assert!(!accept_error_ends_listen(&emfile)); + assert!(!accept_error_ends_listen(&enfile)); + assert!(accept_error_ends_listen(&dead)); + } + #[test] fn magic_for_all_networks() { assert_eq!( @@ -861,6 +948,36 @@ mod tests { ); } + /// Dial and connect-retry block on a channel and a tick. Shutdown must + /// wake them; the grace is only for a session that is still unwinding. + #[tokio::test] + async fn shutdown_finishes_without_the_dial_grace() { + let n = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let dir = std::env::temp_dir().join(format!("rbitcoin-shutdown-grace-{n}")); + let _ = std::fs::remove_dir_all(&dir); + let node = P2PNode::start_outbound_only( + Query::open_or_create_tiny(&dir).unwrap(), + ChainParams::regtest(), + Milestone::NONE, + "/rbitcoin:test/".into(), + 0, + crate::socks::Dialer::Direct, + ) + .await + .unwrap(); + let t0 = tokio::time::Instant::now(); + node.shutdown().await; + let elapsed = t0.elapsed(); + assert!( + elapsed < Duration::from_millis(200), + "dial and retry must observe shutdown, took {elapsed:?}" + ); + let _ = std::fs::remove_dir_all(&dir); + } + /// One datadir, two node lifetimes. The first is dropped without /// shutdown. The second binds loopback as with onion listen, adds a /// second listener, and shuts down with one session task that finishes diff --git a/crates/rbitcoin-net/src/tx_relay.rs b/crates/rbitcoin-net/src/tx_relay.rs index 0b9df3442..2dad7aaa5 100644 --- a/crates/rbitcoin-net/src/tx_relay.rs +++ b/crates/rbitcoin-net/src/tx_relay.rs @@ -744,6 +744,9 @@ pub struct MempoolHub { fee_deltas: Mutex>, /// Monotonic template generation (admit / remove / prioritise). GBT longpoll. template_updates: AtomicU64, + /// Tip whose mempool strip has finished. Block templates build on this. + /// The query tip is visible earlier, before `remove_for_block_spent`. + template_prev: Mutex>, /// INV immediately instead of waiting on mocktime. immediate_relay: AtomicBool, /// Last `setmocktime` (0 = wall). Used to age mempool txs for delayed INV. @@ -820,6 +823,30 @@ struct CommitOutcome { prevouts: Vec, } +/// Lock held for the whole of a block connect. `commit` records the hash +/// whose pool strip has finished; the drop writes it. +pub(crate) struct TemplateHold<'a> { + updates: &'a AtomicU64, + slot: std::sync::MutexGuard<'a, Option<[u8; 32]>>, + commit: Option<[u8; 32]>, +} + +impl TemplateHold<'_> { + pub(crate) fn commit(&mut self, hash: [u8; 32]) { + self.commit = Some(hash); + } +} + +impl Drop for TemplateHold<'_> { + fn drop(&mut self) { + let Some(hash) = self.commit else { + return; + }; + *self.slot = Some(hash); + self.updates.fetch_add(1, Ordering::Relaxed); + } +} + impl MempoolHub { /// Short-id walks that entered the mempool, including a `try_read` miss. #[cfg(test)] @@ -956,6 +983,7 @@ impl MempoolHub { ), fee_deltas: Mutex::new(HashMap::new()), template_updates: AtomicU64::new(0), + template_prev: Mutex::new(None), age_inv: Mutex::new(BTreeMap::new()), min_live_accept_at: AtomicU64::new(u64::MAX), tip_ctx: Mutex::new(None), @@ -1309,15 +1337,34 @@ impl MempoolHub { /// (mainnet: 40k+ live after offline). One purge at tip-mode entry is enough /// before relay. pub fn set_relay_enabled(&self, on: bool) { - let was = self.relay_enabled.swap(on, Ordering::SeqCst); - if on && !was { + if on && !self.relay_enabled() { + // Same lock `connect_at` holds from before the query tip moves + // until the strip is recorded. Take it before the purge so the + // two never invert lock order with the mempool write lock. + let mut slot = self.template_prev.lock().unwrap(); let n = self.purge_confirmed_on_chain(); if n > 0 { rbitcoin_log::info!( "mempool: purged {n} leftover tx(s) at tip-mode entry (deferred during IBD)" ); } + let marked = if let Some(h) = self.query.tip_height() { + if let Ok(Some((_, rec))) = self.query.header_at_height(h) { + let changed = *slot != Some(rec.hash); + *slot = Some(rec.hash); + changed + } else { + false + } + } else { + false + }; + drop(slot); + if marked { + self.note_template_update(); + } } + self.relay_enabled.store(on, Ordering::SeqCst); } pub fn relay_enabled(&self) -> bool { @@ -1535,7 +1582,7 @@ impl MempoolHub { if live.is_empty() { return 0; } - let to_drop = self.live_confirmed_strong(&live); + let to_drop = self.live_confirmed_strong(&live).unwrap_or_default(); let mut g = self.lock_write(); let drop_set: HashSet = to_drop .iter() @@ -1548,7 +1595,9 @@ impl MempoolHub { .map(|(t, _)| *t) .filter(|t| !drop_set.contains(t)) .collect(); - let spent_ops = self.spent_chain_prevouts(&g, &remain); + let spent_ops = self + .spent_chain_prevouts(&g, &remain, &drop_set) + .unwrap_or_default(); let confirmed: Vec = drop_set.into_iter().collect(); let (_, conflicts) = g .remove_block_txids(&confirmed, &spent_ops) @@ -1572,27 +1621,32 @@ impl MempoolHub { gone.len() } - fn live_confirmed_strong(&self, live: &[Txid]) -> Vec { + fn live_confirmed_strong(&self, live: &[Txid]) -> Result, ()> { let tid_bytes: Vec<[u8; 32]> = live.iter().map(|t| t.to_byte_array()).collect(); let hits = self .query .store() .get_fk_by_txid_batch(&tid_bytes) - .unwrap_or_default(); + .map_err(|_| ())?; let mut to_drop = Vec::new(); for (tid_b, row) in hits { let Some((fk, _)) = row else { continue }; - if self.query.store().is_confirmed_strong(fk).unwrap_or(false) { + if self.query.store().is_confirmed_strong(fk).map_err(|_| ())? { to_drop.push(Txid::from_byte_array(tid_b)); } } - to_drop + Ok(to_drop) } - fn spent_chain_prevouts(&self, g: &ActiveMempool, remain: &[Txid]) -> Vec { - let op_fk = self.chain_create_fks(g, remain); + fn spent_chain_prevouts( + &self, + g: &ActiveMempool, + remain: &[Txid], + confirmed_resident: &HashSet, + ) -> Result, ()> { + let op_fk = self.chain_create_fks(g, remain, confirmed_resident)?; if op_fk.is_empty() { - return Vec::new(); + return Ok(Vec::new()); } let mut fk_vouts: HashMap> = HashMap::new(); for (op, fk) in &op_fk { @@ -1606,9 +1660,10 @@ impl MempoolHub { (fk, v) }) .collect(); - let Ok(unspent) = self.query.unspent_create_vouts_batch(&items) else { - return Vec::new(); - }; + let unspent = self + .query + .unspent_create_vouts_batch(&items) + .map_err(|_| ())?; let mut spent_pair: HashSet<(u64, u32)> = HashSet::new(); for ((fk, vouts), keep) in items.iter().zip(unspent.iter()) { let keep_set: HashSet = keep.iter().copied().collect(); @@ -1618,14 +1673,19 @@ impl MempoolHub { } } } - op_fk + Ok(op_fk .into_iter() .filter(|(op, fk)| spent_pair.contains(&(fk.0, op.vout))) .map(|(op, _)| op) - .collect() + .collect()) } - fn chain_create_fks(&self, g: &ActiveMempool, remain: &[Txid]) -> HashMap { + fn chain_create_fks( + &self, + g: &ActiveMempool, + remain: &[Txid], + confirmed_resident: &HashSet, + ) -> Result, ()> { let mut op_fk: HashMap = HashMap::new(); let mut need_id: Vec<[u8; 32]> = Vec::new(); for tid in remain { @@ -1633,7 +1693,10 @@ impl MempoolHub { let aux = g.vin_aux(tid); for (i, inp) in tx.input.iter().enumerate() { let op = inp.previous_output; - if g.graph.contains(&op.txid) { + // Still-unconfirmed parents are mempool dependencies, not chain + // outputs. A parent this call already knows is confirmed is on + // chain even while it sits in the pool, so its output may be spent. + if g.graph.contains(&op.txid) && !confirmed_resident.contains(&op.txid) { continue; } if let Some(fk) = aux.get(i).and_then(|a| a.create_fk) { @@ -1644,13 +1707,15 @@ impl MempoolHub { } } if need_id.is_empty() { - return op_fk; + return Ok(op_fk); } need_id.sort_unstable(); need_id.dedup(); - let Ok(rows) = self.query.store().get_fk_by_txid_batch(&need_id) else { - return op_fk; - }; + let rows = self + .query + .store() + .get_fk_by_txid_batch(&need_id) + .map_err(|_| ())?; let mut fk_of: HashMap<[u8; 32], Fk> = HashMap::new(); for (tid, row) in rows { if let Some((fk, _)) = row { @@ -1661,7 +1726,9 @@ impl MempoolHub { let Some(tx) = g.get_tx(tid) else { continue }; for inp in &tx.input { let op = inp.previous_output; - if op_fk.contains_key(&op) || g.graph.contains(&op.txid) { + if op_fk.contains_key(&op) + || (g.graph.contains(&op.txid) && !confirmed_resident.contains(&op.txid)) + { continue; } if let Some(&fk) = fk_of.get(&op.txid.to_byte_array()) { @@ -1669,7 +1736,7 @@ impl MempoolHub { } } } - op_fk + Ok(op_fk) } pub fn subscribe_announces(&self) -> broadcast::Receiver { @@ -3452,6 +3519,27 @@ impl MempoolHub { self.template_updates.fetch_add(1, Ordering::Relaxed); } + /// The pool has been stripped for `hash`. Templates build on this tip. + pub fn note_template_prev(&self, hash: [u8; 32]) { + *self.template_prev.lock().unwrap() = Some(hash); + self.note_template_update(); + } + + /// Held across a block connect: from before the query tip moves until the + /// strip has been recorded. [`Self::note_template_prev`] takes the same + /// mutex, so callers must record the hash through [`TemplateHold::commit`]. + pub(crate) fn template_hold(&self) -> TemplateHold<'_> { + TemplateHold { + updates: &self.template_updates, + slot: self.template_prev.lock().unwrap(), + commit: None, + } + } + + pub fn template_prev(&self) -> Option<[u8; 32]> { + *self.template_prev.lock().unwrap() + } + /// Generation for `getblocktemplate.longpollid`. pub fn template_updates(&self) -> u64 { self.template_updates.load(Ordering::Relaxed) diff --git a/crates/rbitcoin-node/Cargo.toml b/crates/rbitcoin-node/Cargo.toml index 35749196d..99aade9d2 100644 --- a/crates/rbitcoin-node/Cargo.toml +++ b/crates/rbitcoin-node/Cargo.toml @@ -10,6 +10,10 @@ description = "rbitcoin full node binary and lifecycle" [[bin]] name = "rbitcoin-node" path = "src/main.rs" +test = false + +[lib] +doctest = false [features] mutants = [] diff --git a/crates/rbitcoin-node/src/config.rs b/crates/rbitcoin-node/src/config.rs index a71d3c3f6..0a49dd3db 100644 --- a/crates/rbitcoin-node/src/config.rs +++ b/crates/rbitcoin-node/src/config.rs @@ -10,6 +10,7 @@ use rbitcoin_primitives::{ use rbitcoin_store::HeadScale; use std::net::SocketAddr; use std::path::{Path, PathBuf}; +use std::time::Duration; /// Default max concurrent inbound P2P sessions (same as net `DEFAULT_MAX_INBOUND`). pub const DEFAULT_MAX_INBOUND: u32 = rbitcoin_net::DEFAULT_MAX_INBOUND as u32; @@ -346,12 +347,14 @@ pub struct NodeConfig { /// Validity of each per-connection Noise certificate. Default 3600 s. pub sv2_tp_cert_validity_secs: u64, /// How long a template on a replaced tip still answers. Default 10 s. - pub sv2_tp_stale_grace_secs: u64, + /// Conf and CLI set whole seconds; tests may set a shorter duration. + pub sv2_tp_stale_grace: Duration, /// Fee gain over a session's last template that pushes a rebuild on the /// same tip. Default 1000 sat. pub sv2_tp_fee_delta: u64, - /// Minimum seconds from a session's last push to a fee push. Default 5. - pub sv2_tp_template_interval_secs: u64, + /// Minimum time from a session's last push to a fee push. Default 5 s. + /// Conf and CLI set whole seconds (at least 1); tests may set the floor. + pub sv2_tp_template_interval: Duration, /// Skip script/prevout checks for blocks at or below this height (0 = off). pub milestone_height: u32, /// Set when conf or CLI applied `milestone` (including 0). @@ -434,9 +437,9 @@ impl Default for NodeConfig { sv2_tp_listen: None, sv2_tp_authority_sec: None, sv2_tp_cert_validity_secs: 3600, - sv2_tp_stale_grace_secs: 10, + sv2_tp_stale_grace: Duration::from_secs(10), sv2_tp_fee_delta: rbitcoin_sv2::FEE_DELTA, - sv2_tp_template_interval_secs: rbitcoin_sv2::TEMPLATE_INTERVAL.as_secs(), + sv2_tp_template_interval: rbitcoin_sv2::TEMPLATE_INTERVAL, milestone_height: 0, milestone_explicit: false, inhibit_suspend: false, @@ -1170,10 +1173,14 @@ impl NodeConfig { } "sv2_tp_stale_grace" => { let max = rbitcoin_sv2::MAX_STALE_GRACE.as_secs(); - self.sv2_tp_stale_grace_secs = - val.parse().ok().filter(|&s| s <= max).ok_or_else(|| { + let secs = val + .parse() + .ok() + .filter(|&s: &u64| s <= max) + .ok_or_else(|| { NodeError::Config(format!("conf sv2_tp_stale_grace: want seconds <= {max}")) })?; + self.sv2_tp_stale_grace = Duration::from_secs(secs); } "sv2_tp_fee_delta" => { self.sv2_tp_fee_delta = val @@ -1182,15 +1189,16 @@ impl NodeConfig { } "sv2_tp_template_interval" => { let max = rbitcoin_sv2::MAX_TEMPLATE_INTERVAL.as_secs(); - self.sv2_tp_template_interval_secs = val + let secs = val .parse() .ok() - .filter(|&s| (1..=max).contains(&s)) + .filter(|&s: &u64| (1..=max).contains(&s)) .ok_or_else(|| { NodeError::Config(format!( "conf sv2_tp_template_interval: want seconds in 1..={max}" )) })?; + self.sv2_tp_template_interval = Duration::from_secs(secs); } "sv2_tp_cert_validity" => { self.sv2_tp_cert_validity_secs = val @@ -1801,9 +1809,9 @@ mod tests { let mut c = NodeConfig::default().with_datadir(tmp()); assert_eq!(c.sv2_tp_listen, None); assert_eq!(c.sv2_tp_cert_validity_secs, 3600); - assert_eq!(c.sv2_tp_stale_grace_secs, 10); + assert_eq!(c.sv2_tp_stale_grace, Duration::from_secs(10)); assert_eq!(c.sv2_tp_fee_delta, 1000); - assert_eq!(c.sv2_tp_template_interval_secs, 5); + assert_eq!(c.sv2_tp_template_interval, Duration::from_secs(5)); for (k, v) in [ ("sv2_tp_listen", "127.0.0.1:8442"), ("sv2_tp_cert_validity", "600"), @@ -1815,9 +1823,9 @@ mod tests { } assert_eq!(c.sv2_tp_listen, Some("127.0.0.1:8442".parse().unwrap())); assert_eq!(c.sv2_tp_cert_validity_secs, 600); - assert_eq!(c.sv2_tp_stale_grace_secs, 0); + assert_eq!(c.sv2_tp_stale_grace, Duration::ZERO); assert_eq!(c.sv2_tp_fee_delta, 0); - assert_eq!(c.sv2_tp_template_interval_secs, 86400); + assert_eq!(c.sv2_tp_template_interval, Duration::from_secs(86400)); let no_auth = c.validate().unwrap_err(); assert!( format!("{no_auth}").contains("sv2-tp-authority-sec"), diff --git a/crates/rbitcoin-node/src/regtest_rpc.rs b/crates/rbitcoin-node/src/regtest_rpc.rs index ddd7a03d3..87e980aab 100644 --- a/crates/rbitcoin-node/src/regtest_rpc.rs +++ b/crates/rbitcoin-node/src/regtest_rpc.rs @@ -36,6 +36,9 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// `OP_TRUE` signet generation on regtest difficulty. Parent bits are + /// `0x207fffff`, so the grind is a handful of nonces. Accept still checks + /// the BIP325 solution, which the regtest miner does not attach. #[test] fn hub_op_true_signet_generate_one() { let n = SystemTime::now() @@ -44,8 +47,8 @@ mod tests { .as_nanos(); let dir = std::env::temp_dir().join(format!("rbitcoin-node-signet-gen-{n}")); std::fs::create_dir_all(&dir).unwrap(); - let params = - ChainParams::custom_signet(ScriptBuf::from_bytes(vec![0x51]), 10 * 60).unwrap(); + let mut params = ChainParams::regtest(); + params.signet_challenge = Some(ScriptBuf::from_bytes(vec![0x51])); let hub = Arc::new(ChainHub::new( Query::open_or_create_tiny(dir.join("store")).unwrap(), params, @@ -57,6 +60,12 @@ mod tests { .unwrap(); assert_eq!(hashes.len(), 1); assert_eq!(miner.0.tip_height(), Some(1)); + let bits = miner.0.tip_header().expect("tip").bits; + assert_eq!( + bits.to_consensus(), + rbitcoin_consensus::REGTEST_POW_BITS, + "the signet miner must grind the parent target, which is regtest here" + ); let _ = std::fs::remove_dir_all(&dir); } } diff --git a/crates/rbitcoin-node/src/run.rs b/crates/rbitcoin-node/src/run.rs index e42fa11b1..ec128303f 100644 --- a/crates/rbitcoin-node/src/run.rs +++ b/crates/rbitcoin-node/src/run.rs @@ -178,6 +178,7 @@ async fn start_rpc_listener( .unwrap_or_else(|_| format!("/rbitcoin:{}/", env!("CARGO_PKG_VERSION"))), ), alert_notify: config.alert_notify.clone(), + blocksonly: config.mempool.blocksonly, }; let op_true_signet = config.network == Network::Signet && config @@ -295,6 +296,12 @@ pub fn run_node(config: NodeConfig) -> Result { /// hold the process). #[allow(clippy::cognitive_complexity)] // node bring-up / P2P follow loop pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { + if rbitcoin_net::compiled_with_test_hooks() || rbitcoin_sv2::compiled_with_test_hooks() { + warn!( + "node: built with test-hooks — connect redial, SAM retry, and the SV2 \ + budget cooldown are the short test values, not the operator ones" + ); + } let status = NodeStatus::new(config.network, config.shindex); let _health = match config.listen.health { Some(addr) => { @@ -728,10 +735,11 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { let index_entry = rbitcoin_query::index_tip_entry(&node.hub.query); if index_entry.advertise_filters { info!("blockfilter: already at tip; advertising NODE_COMPACT_FILTERS"); - rbitcoin_net::set_compact_filters_service(true); + node.peers.set_compact_filters_service(true); } if index_entry.spawn { let advertise_later = !index_entry.advertise_filters; + let filter_peers = Arc::clone(&node.peers); index_writebehind = Some(rbitcoin_query::spawn_index_writebehind( Arc::clone(&node.hub.query), Arc::clone(&shutdown.flag), @@ -743,7 +751,7 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { // when served filters reach the tip, not during materialize. move || { if advertise_later { - rbitcoin_net::set_compact_filters_service(true); + filter_peers.set_compact_filters_service(true); } }, )); @@ -882,7 +890,9 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { sh_tip_ready, config.listen.esplora.clone(), config.network, - config.esplora_block_template, + config + .esplora_block_template + .then_some(config.mempool.blocksonly), &shutdown, Arc::clone(&node.hub), &mempool, @@ -1132,7 +1142,9 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { true, config.listen.esplora.clone(), config.network, - config.esplora_block_template, + config + .esplora_block_template + .then_some(config.mempool.blocksonly), &shutdown, Arc::clone(&node.hub), &mempool, @@ -1192,6 +1204,7 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { &mut perf_tick, &mut tip_rx, &mut stale_poll, + deadline, ) .await; if let Some(ref h) = rpc_handle { @@ -2042,7 +2055,8 @@ async fn start_esplora_if_ready( sh_tip_ready: bool, listen: Option, network: Network, - enable_block_template: bool, + // `Some(blocksonly)` turns on `/block-template`. `None` leaves it off. + block_template: Option, shutdown: &Shutdown, hub: Arc, mempool: &std::sync::Arc, @@ -2061,7 +2075,7 @@ async fn start_esplora_if_ready( rbitcoin_primitives::Network::Regtest => bitcoin::Network::Regtest, }; let mut ecfg = EsploraConfig::with_listen(listen, btc_net); - if enable_block_template { + if let Some(blocksonly) = block_template { let q = Arc::clone(&hub.query); let mp = Arc::clone(mempool); let chain = Arc::clone(&hub); @@ -2083,6 +2097,7 @@ async fn start_esplora_if_ready( active: Arc::new(std::sync::Mutex::new(RpcActive::default())), alert_notify: None, alert_fired: Arc::new(AtomicBool::new(false)), + blocksonly, }; gbt_template(&ctx).map_err(|v| { v.get("message") @@ -2122,11 +2137,11 @@ async fn start_sv2_tp(config: &NodeConfig, hub: &Arc) -> Option { @@ -2468,10 +2483,17 @@ pub(crate) async fn tip_follow_next_wake( perf: &mut tokio::time::Interval, tip_rx: &mut broadcast::Receiver, stale: &mut tokio::time::Interval, + run_deadline: Option, ) -> TipFollowWake { tokio::select! { biased; _ = shutdown => TipFollowWake::Stop, + _ = async { + match run_deadline { + Some(d) => tokio::time::sleep(d.saturating_duration_since(Instant::now())).await, + None => std::future::pending::<()>().await, + } + } => TipFollowWake::Stop, _ = async { match rpc_stop { Some(tick) => { @@ -2724,6 +2746,7 @@ mod tests { &mut perf, &mut tip_rx, &mut stale, + None, ), ) .await @@ -2739,6 +2762,34 @@ mod tests { ); } + /// `--max-run-secs` is a deadline, not "exit on the next 5s perf tick". + #[tokio::test] + async fn max_run_deadline_wakes_without_the_perf_tick() { + let mut perf = tokio::time::interval(Duration::from_secs(30)); + perf.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + perf.tick().await; + let mut stale = tokio::time::interval(Duration::from_secs(30)); + stale.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + stale.tick().await; + let (_tx, mut tip_rx) = broadcast::channel::(8); + let start = Instant::now(); + let wake = tokio::time::timeout( + Duration::from_millis(500), + tip_follow_next_wake( + std::future::pending(), + None, + &mut perf, + &mut tip_rx, + &mut stale, + Some(Instant::now() + Duration::from_millis(40)), + ), + ) + .await + .expect("deadline must wake before the perf tick"); + assert!(matches!(wake, TipFollowWake::Stop)); + assert!(start.elapsed() < Duration::from_millis(200)); + } + #[test] fn catch_up_ok_genesis_zero_accepts_is_incomplete() { assert_eq!(catch_up_after_ok(0, 0, false), CatchUp::Incomplete); diff --git a/crates/rbitcoin-primitives/Cargo.toml b/crates/rbitcoin-primitives/Cargo.toml index 9f4e433dd..526c13d65 100644 --- a/crates/rbitcoin-primitives/Cargo.toml +++ b/crates/rbitcoin-primitives/Cargo.toml @@ -10,6 +10,9 @@ description = "Shared types for the rbitcoin node" [features] mutants = [] +[lib] +doctest = false + [dependencies] bitcoin_hashes = { workspace = true } # Same crate `bitcoin::hex` re-exports. Not the `bitcoin` crate: store depends diff --git a/crates/rbitcoin-query/Cargo.toml b/crates/rbitcoin-query/Cargo.toml index 30c906644..1421e0d5a 100644 --- a/crates/rbitcoin-query/Cargo.toml +++ b/crates/rbitcoin-query/Cargo.toml @@ -7,6 +7,9 @@ rust-version.workspace = true repository.workspace = true description = "Query API over the rbitcoin store" +[lib] +doctest = false + [dependencies] rbitcoin-consensus = { workspace = true } rbitcoin-primitives = { workspace = true } diff --git a/crates/rbitcoin-query/src/bq_lock_pin.rs b/crates/rbitcoin-query/src/bq_lock_pin.rs index 313cd909e..bbec74376 100644 --- a/crates/rbitcoin-query/src/bq_lock_pin.rs +++ b/crates/rbitcoin-query/src/bq_lock_pin.rs @@ -106,12 +106,12 @@ fn stats_follow_enqueue_and_dequeue() { let q = open(); q.block_queue_enqueue(5, [5u8; 32], 1, b"aaaaa").unwrap(); q.block_queue_enqueue(9, [9u8; 32], 1, b"bbb").unwrap(); - let (_stop, bytes, count) = q.block_queue_stats(); + let (bytes, count) = q.block_queue_stats(); assert_eq!(count, 2); assert_eq!(bytes, 8); assert_eq!(q.block_queue_max_height(), Some(9)); q.block_queue_dequeue_height(9).unwrap(); - let (_stop, bytes, count) = q.block_queue_stats(); + let (bytes, count) = q.block_queue_stats(); assert_eq!(count, 1); assert_eq!(bytes, 5); assert_eq!(q.block_queue_max_height(), Some(5)); diff --git a/crates/rbitcoin-query/src/combined_stage.rs b/crates/rbitcoin-query/src/combined_stage.rs index 094837722..8cee5cb7f 100644 --- a/crates/rbitcoin-query/src/combined_stage.rs +++ b/crates/rbitcoin-query/src/combined_stage.rs @@ -216,7 +216,7 @@ mod tests { let id = q .block_queue_enqueue(42, [0xCDu8; 32], 7, &payload) .unwrap(); - assert_eq!(q.block_queue_stats().2, 1); + assert_eq!(q.block_queue_stats().1, 1); let _ = id; assert!(q.block_queue_has_height(42)); assert_eq!( @@ -225,12 +225,12 @@ mod tests { ); // Confirm-write hook: dequeue by height. assert_eq!(q.block_queue_dequeue_height(42).unwrap(), 1); - assert_eq!(q.block_queue_stats().2, 0); + assert_eq!(q.block_queue_stats().1, 0); // Restart: RAM queue is empty (by design — redownload, no double disk write). drop(q); let q2 = Query::open_or_create_tiny(dir.path()).unwrap(); assert!(!q2.block_queue_has_height(42)); - assert_eq!(q2.block_queue_stats().2, 0); + assert_eq!(q2.block_queue_stats().1, 0); let _ = std::fs::remove_dir_all(dir); } @@ -242,13 +242,13 @@ mod tests { let p2 = vec![2u8; 64 * 1024]; let o1 = q.block_queue_offer(1, [1u8; 32], 1, &p1).unwrap(); assert!(o1.queue_id > 0); - assert_eq!(q.block_queue_stats().2, 1); + assert_eq!(q.block_queue_stats().1, 1); let o2 = q.block_queue_offer(2, [2u8; 32], 2, &p2).unwrap(); assert!(o2.queue_id > 0); - assert_eq!(q.block_queue_stats().2, 2); + assert_eq!(q.block_queue_stats().1, 2); let n = q.block_queue_dequeue_height(1).unwrap(); assert_eq!(n, 1); - assert_eq!(q.block_queue_stats().2, 1); + assert_eq!(q.block_queue_stats().1, 1); assert!(q.block_queue_has_height(2)); assert!(!q.block_queue_has_height(1)); assert_eq!( @@ -269,7 +269,7 @@ mod tests { Some(wire.as_slice()) ); assert!(q.block_queue_has_height(10)); - assert_eq!(q.block_queue_stats().2, 1, "peek does not dequeue"); + assert_eq!(q.block_queue_stats().1, 1, "peek does not dequeue"); assert!(q.block_queue_payload(999).unwrap().is_none()); let _ = std::fs::remove_dir_all(dir); } @@ -357,14 +357,14 @@ mod tests { let chunk = vec![0u8; 80 * 1024 * 1024]; q.block_queue_enqueue(1, [1u8; 32], 1, &chunk).unwrap(); q.block_queue_enqueue(2, [2u8; 32], 2, &chunk).unwrap(); - assert!(q.block_queue_stats().1 > BQ_SOFT_FREE_BYTES); + assert!(q.block_queue_stats().0 > BQ_SOFT_FREE_BYTES); assert!( q.block_queue_update_soft_pressure(None), "bytes over free floor → restricted" ); // Drop one chunk → under free floor → unrestricted. q.block_queue_dequeue_height(1).unwrap(); - assert!(q.block_queue_stats().1 < BQ_SOFT_FREE_BYTES); + assert!(q.block_queue_stats().0 < BQ_SOFT_FREE_BYTES); assert!( !q.block_queue_update_soft_pressure(None), "bytes under free floor → unrestricted" @@ -557,7 +557,7 @@ mod tests { assert!(q.block_queue_raw_payload(7).unwrap().is_none()); assert!(q.block_queue_payload(7).unwrap().unwrap().is_empty()); assert_eq!( - q.block_queue_stats().1, + q.block_queue_stats().0, 64 + 6, "charge 64 + leftover raw wire-8" ); diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 4c354e080..bb976bbfb 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/crates/rbitcoin-query/src/lib.rs @@ -71,8 +71,8 @@ pub use regtest_pad::pad_empty_from; pub use resolved_wire::{BlockQueueWaveIntake, ResolvedWire}; pub use soft_densify::{ bq_assign_stop_bytes, soft_ahead_quarter_full, soft_assign_restricted, - soft_confirm_window_covered, soft_confirm_window_n, soft_densify_band_hi, BQ_SOFT_FREE_BYTES, - TIP_HOLE_MIN_AHEAD_BLOCKS, + soft_confirm_window_covered, soft_confirm_window_n, soft_densify_band_hi, BQ_ASSIGN_STOP_BYTES, + BQ_SOFT_FREE_BYTES, TIP_HOLE_MIN_AHEAD_BLOCKS, }; pub use sp_read::{tweaks_at_height, tweaks_for_height, tweaks_from_thin_and_body}; pub use sp_tweaks::{ThinTweakRangeLimits, ThinTweakRow}; @@ -1497,13 +1497,14 @@ impl Query { self.tx_index.load(std::sync::atomic::Ordering::SeqCst) } - /// In-RAM block queue stats: `(assign_stop_bytes, bytes, count)`. + /// In-RAM block queue stats: `(bytes, count)`. /// - /// Bytes are process heap (wire payloads). First field is the densify - /// assign-stop ([`bq_assign_stop_bytes`]; `u64::MAX` when unlimited). - pub fn block_queue_stats(&self) -> (u64, u64, usize) { + /// Bytes are process heap (wire payloads). The densify cap is + /// [`bq_assign_stop_bytes`], copied onto the IBD config when that config + /// is built. + pub fn block_queue_stats(&self) -> (u64, usize) { let g = self.block_queue.lock().unwrap(); - (bq_assign_stop_bytes(), g.bytes(), g.count()) + (g.bytes(), g.count()) } /// In-RAM entry count (soft time-depth meter). diff --git a/crates/rbitcoin-rpc/Cargo.toml b/crates/rbitcoin-rpc/Cargo.toml index 811ceba3d..9fcd79837 100644 --- a/crates/rbitcoin-rpc/Cargo.toml +++ b/crates/rbitcoin-rpc/Cargo.toml @@ -10,6 +10,9 @@ description = "Core-class JSON-RPC HTTP server (subset; not full Core parity)" [features] mutants = [] +[lib] +doctest = false + [dependencies] rbitcoin-primitives = { workspace = true } rbitcoin-log = { workspace = true } diff --git a/crates/rbitcoin-rpc/src/methods/mempool.rs b/crates/rbitcoin-rpc/src/methods/mempool.rs index 112912035..0063f4aa0 100644 --- a/crates/rbitcoin-rpc/src/methods/mempool.rs +++ b/crates/rbitcoin-rpc/src/methods/mempool.rs @@ -206,6 +206,10 @@ pub(crate) fn mempool_graph_json(mp: &MempoolHub, txid: &Txid, fee: u64, weight: } pub(crate) fn getrawmempool(ctx: &RpcContext, params: &RpcParams) -> Result { + let listing = ctx.pool_listing(); + if !matches!(listing, PoolListing::Ready) { + return Err(pool_listing_unavailable(listing)); + } params.reject_unknown(&["verbose", "mempool_sequence"])?; let verbose = params.opt_bool(0, "verbose")?.unwrap_or(false); let want_seq = params.opt_bool(1, "mempool_sequence")?.unwrap_or(false); diff --git a/crates/rbitcoin-rpc/src/methods/mine.rs b/crates/rbitcoin-rpc/src/methods/mine.rs index 139c0a119..b803e728a 100644 --- a/crates/rbitcoin-rpc/src/methods/mine.rs +++ b/crates/rbitcoin-rpc/src/methods/mine.rs @@ -149,10 +149,19 @@ pub(crate) fn generate_with_mempool( script: ScriptBuf, ) -> Result { let miner = require_regtest_miner(ctx, "generate")?; - let extras: Vec = mempool_block_txs(ctx) - .into_iter() - .map(|(tx, _)| Transaction::clone(&tx)) - .collect(); + let listing = ctx.pool_listing(); + if matches!(listing, PoolListing::Blocksonly) { + return Err(pool_listing_unavailable(listing)); + } + // Catch-up leaves the pool unstripped. Mine the coinbase only. + let extras: Vec = if matches!(listing, PoolListing::Ready) { + mempool_block_txs(ctx) + .into_iter() + .map(|(tx, _)| Transaction::clone(&tx)) + .collect() + } else { + Vec::new() + }; let hashes = miner .generate_to_script(nblocks, script, extras.clone()) .map_err(|e| rpc_error(ERR_MISC, e))?; @@ -540,6 +549,10 @@ pub(crate) fn gbt_rules(req: Option<&Value>) -> Result, Value> { } pub(crate) fn getblocktemplate(ctx: &RpcContext, params: &RpcParams) -> Result { + let listing = ctx.pool_listing(); + if !matches!(listing, PoolListing::Ready) { + return Err(pool_listing_unavailable(listing)); + } params.reject_unknown(&["template_request"])?; let req = params.get(0, "template_request"); let _rules = gbt_rules(req)?; @@ -567,9 +580,6 @@ pub(crate) fn getblocktemplate(ctx: &RpcContext, params: &RpcParams) -> Result String { - let tip = if let Some(h) = ctx.query.tip_height() { - ctx.query - .header_at_height(h) - .ok() - .flatten() - .map(|(_, rec)| hash_hex_display(&rec.hash)) - .unwrap_or_default() - } else { - String::new() +/// Where a template may build. The stripped tip wins. The query tip is only +/// a stand-in before any strip has been recorded. +enum GbtTip { + /// `(height, display hash, time, bits)`. + Ready(u32, String, u32, u32), + /// `template_prev` names a block that is not the header at its height. + Stale, +} + +fn anchor_of(height: u32, hash: [u8; 32], time: u32, bits: u32) -> GbtTip { + GbtTip::Ready(height, hash_hex_display(&hash), time, bits) +} + +fn gbt_tip(ctx: &RpcContext) -> Result { + if let Some(hash) = ctx.mempool.as_ref().and_then(|m| m.template_prev()) { + let Some(height) = ctx + .query + .height_of_hash(&hash) + .map_err(|e| rpc_error(ERR_MISC, e.to_string()))? + else { + return Ok(GbtTip::Stale); + }; + let Some((_, rec)) = ctx + .query + .header_at_height(height) + .map_err(|e| rpc_error(ERR_MISC, e.to_string()))? + else { + return Ok(GbtTip::Stale); + }; + if rec.hash != hash { + return Ok(GbtTip::Stale); + } + return Ok(anchor_of(height.0, rec.hash, rec.timestamp, rec.bits)); + } + let Some(h) = ctx.query.tip_height() else { + return Err(rpc_error(ERR_MISC, "no tip")); }; + let (_, rec) = ctx + .query + .header_at_height(h) + .map_err(|e| rpc_error(ERR_MISC, e.to_string()))? + .ok_or_else(|| rpc_error(ERR_MISC, "tip header missing"))?; + Ok(anchor_of(h.0, rec.hash, rec.timestamp, rec.bits)) +} + +pub(crate) fn gbt_longpoll_id(ctx: &RpcContext) -> String { let updates = ctx .mempool .as_ref() .map(|m| m.template_updates()) .unwrap_or(0); - format!("{tip}{updates}") + match gbt_tip(ctx) { + Ok(GbtTip::Ready(_, hex, _, _)) => format!("{hex}{updates}"), + _ => format!("stale{updates}"), + } } pub fn gbt_template(ctx: &RpcContext) -> Result { - let tip_h = ctx.query.tip_height().map(|h| h.0).unwrap_or(0); - let next_h = tip_h.saturating_add(1); - let (prev_hex, tip_time, tip_bits) = if let Some(h) = ctx.query.tip_height() { - let (_, rec) = ctx - .query - .header_at_height(h) - .map_err(|e| rpc_error(ERR_MISC, e.to_string()))? - .ok_or_else(|| rpc_error(ERR_MISC, "tip header missing"))?; - (hash_hex_display(&rec.hash), rec.timestamp, rec.bits) - } else { - return Err(rpc_error(ERR_MISC, "no tip")); + let listing = ctx.pool_listing(); + if !matches!(listing, PoolListing::Ready) { + return Err(pool_listing_unavailable(listing)); + } + let GbtTip::Ready(tip_h, prev_hex, tip_time, tip_bits) = gbt_tip(ctx)? else { + return Err(rpc_error(ERR_MISC, "template tip is not on the best chain")); }; + let updates = ctx + .mempool + .as_ref() + .map(|m| m.template_updates()) + .unwrap_or(0); + let next_h = tip_h.saturating_add(1); let params = ctx .chain .as_ref() @@ -677,7 +726,7 @@ pub fn gbt_template(ctx: &RpcContext) -> Result { })); } let subsidy = rbitcoin_consensus::block_subsidy(next_h, ¶ms) as u64; - let longpollid = gbt_longpoll_id(ctx); + let longpollid = format!("{prev_hex}{updates}"); let target = bitcoin::Target::from_compact(bitcoin::CompactTarget::from_consensus(bits)); if let Some(c) = ctx.chain.as_ref() { c.note_gbt_assembled(); diff --git a/crates/rbitcoin-rpc/src/methods/mod.rs b/crates/rbitcoin-rpc/src/methods/mod.rs index 18cca5f41..2cf9a0f7a 100644 --- a/crates/rbitcoin-rpc/src/methods/mod.rs +++ b/crates/rbitcoin-rpc/src/methods/mod.rs @@ -99,6 +99,9 @@ pub struct RpcContext { pub alert_notify: Option, /// Latches after the first alertnotify invocation. pub alert_fired: Arc, + /// `-blocksonly`: the pool accepts local transactions and does not relay + /// them. Mining and raw-mempool listing stay off for the life of the node. + pub blocksonly: bool, /// In-flight RPC methods for `getrpcinfo.active_commands`. pub active: Arc>, } @@ -205,10 +208,48 @@ impl RpcRegtest for HubRegtest { } } +/// Whether `getblocktemplate`, `getrawmempool`, and pool-reading `generate` +/// may use the live pool. +pub(crate) enum PoolListing { + /// Relay is on. The pool is stripped before each tip announcement. + Ready, + /// Relay is still off because catch-up has not finished the purge. + Downloading, + /// `-blocksonly`. Relay never turns on. + Blocksonly, +} + impl RpcContext { pub fn uptime_secs(&self) -> u64 { self.start.elapsed().as_secs() } + + /// Blocksonly wins over catch-up so a miner does not retry forever. + /// No pool is not catch-up: there is nothing to strip. + pub(crate) fn pool_listing(&self) -> PoolListing { + if self.blocksonly { + return PoolListing::Blocksonly; + } + if self.mempool.as_ref().is_none_or(|m| m.relay_enabled()) { + PoolListing::Ready + } else { + PoolListing::Downloading + } + } +} + +pub(crate) fn pool_listing_unavailable(listing: PoolListing) -> Value { + match listing { + PoolListing::Blocksonly => rpc_error( + ERR_MISC, + "blocksonly: this node does not serve a block template or the raw mempool", + ), + PoolListing::Downloading => rpc_error( + ERR_CLIENT_IN_INITIAL_DOWNLOAD, + "Bitcoin is downloading blocks...", + ), + PoolListing::Ready => rpc_error(ERR_MISC, "pool listing is available"), + } } /// JSON-RPC error object (Core-ish codes). @@ -217,6 +258,8 @@ pub fn rpc_error(code: i64, message: impl Into) -> Value { } pub const ERR_MISC: i64 = -1; +/// Core `RPC_CLIENT_IN_INITIAL_DOWNLOAD`. +pub const ERR_CLIENT_IN_INITIAL_DOWNLOAD: i64 = -10; /// Core `RPC_TYPE_ERROR`. pub const ERR_TYPE_ERROR: i64 = -3; /// Core `RPC_INVALID_ADDRESS_OR_KEY`. diff --git a/crates/rbitcoin-rpc/src/methods/net.rs b/crates/rbitcoin-rpc/src/methods/net.rs index 86a784099..37e736f92 100644 --- a/crates/rbitcoin-rpc/src/methods/net.rs +++ b/crates/rbitcoin-rpc/src/methods/net.rs @@ -293,7 +293,11 @@ pub(crate) fn getnodeaddresses(ctx: &RpcContext, params: &RpcParams) -> Result Value { } else { (0, ctx.connections.load(Ordering::Relaxed), 0) }; - let flags = rbitcoin_net::local_service_flags_pruned(ctx.query.prune_seqsigwit()); + let flags = + rbitcoin_net::local_service_flags(ctx.query.prune_seqsigwit(), ctx.peers.as_deref()); let svc_bits = flags.to_u64(); json!({ "version": RPC_COMPAT_VERSION, diff --git a/crates/rbitcoin-rpc/src/methods_tests.rs b/crates/rbitcoin-rpc/src/methods_tests.rs index 0074cd6f8..6e9d23a7d 100644 --- a/crates/rbitcoin-rpc/src/methods_tests.rs +++ b/crates/rbitcoin-rpc/src/methods_tests.rs @@ -43,6 +43,7 @@ fn open_ctx(dir: &TempDir) -> RpcContext { alert_notify: None, alert_fired: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + blocksonly: false, } } @@ -621,6 +622,123 @@ fn blocksonly_sendraw_admits_valid_tx() { let _ = std::fs::remove_dir_all(&dir); } +#[test] +fn relay_off_refuses_pool_listing_and_mines_coinbase_only() { + let (ctx, dir, _hub) = ctx_from_mature_pad(300_000_000); + let (hex, spend) = mature_coinbase_spend_hex(&ctx, 50_0000_0000 - 1_000); + dispatch(&ctx, "sendrawtransaction", vec![json!(hex)]).unwrap(); + ctx.mempool.as_ref().unwrap().set_relay_enabled(false); + let hashes = dispatch(&ctx, "generate", vec![json!(1)]).unwrap(); + let hash = hashes[0].as_str().unwrap(); + let block = dispatch(&ctx, "getblock", vec![json!(hash)]).unwrap(); + assert_eq!(block["nTx"].as_u64(), Some(1), "{block}"); + assert!(ctx + .mempool + .as_ref() + .unwrap() + .contains(&spend.compute_txid())); + let gbt = dispatch(&ctx, "getblocktemplate", vec![json!({"rules": ["segwit"]})]).unwrap_err(); + assert_eq!(gbt["code"], -10, "{gbt}"); + let raw = dispatch(&ctx, "getrawmempool", vec![]).unwrap_err(); + assert_eq!(raw["code"], -10, "{raw}"); + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] +fn template_follows_the_stripped_tip() { + let (ctx, dir, _hub) = ctx_from_mature_pad(300_000_000); + let tip = dispatch(&ctx, "getbestblockhash", vec![]).unwrap(); + let (_, genesis) = ctx + .query + .header_at_height(Height(0)) + .unwrap() + .expect("genesis"); + ctx.mempool + .as_ref() + .unwrap() + .note_template_prev(genesis.hash); + let gbt = dispatch(&ctx, "getblocktemplate", vec![json!({"rules": ["segwit"]})]).unwrap(); + let prev = gbt["previousblockhash"].as_str().unwrap(); + assert_ne!(prev, tip.as_str().unwrap(), "{gbt}"); + assert_eq!(gbt["height"].as_u64(), Some(1), "{gbt}"); + assert!( + gbt["longpollid"].as_str().unwrap().starts_with(prev), + "{gbt}" + ); + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] +fn template_follows_the_fork_after_invalidate() { + let (ctx, dir, _hub) = ctx_from_mature_pad(300_000_000); + let tip = dispatch(&ctx, "getbestblockhash", vec![]).unwrap(); + let blk = dispatch(&ctx, "getblock", vec![tip.clone(), json!(1)]).unwrap(); + let parent = blk["previousblockhash"].clone(); + let height = blk["height"].as_u64().unwrap(); + dispatch(&ctx, "invalidateblock", vec![tip]).unwrap(); + let gbt = dispatch(&ctx, "getblocktemplate", vec![json!({"rules": ["segwit"]})]).unwrap(); + assert_eq!(gbt["previousblockhash"], parent, "{gbt}"); + assert_eq!(gbt["height"].as_u64(), Some(height), "{gbt}"); + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] +fn template_follows_the_fork_after_header_rewind() { + let (ctx, dir, hub) = ctx_from_mature_pad(300_000_000); + let tip = dispatch(&ctx, "getbestblockhash", vec![]).unwrap(); + let blk = dispatch(&ctx, "getblock", vec![tip, json!(1)]).unwrap(); + let parent = blk["previousblockhash"].clone(); + let height = blk["height"].as_u64().unwrap(); + hub.rewind_to_height(u32::try_from(height - 1).unwrap()) + .unwrap(); + let gbt = dispatch(&ctx, "getblocktemplate", vec![json!({"rules": ["segwit"]})]).unwrap(); + assert_eq!(gbt["previousblockhash"], parent, "{gbt}"); + assert_eq!(gbt["height"].as_u64(), Some(height), "{gbt}"); + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] +fn template_follows_blocks_the_hub_connects() { + let (ctx, dir, _hub) = ctx_regtest_hub(); + dispatch(&ctx, "generate", vec![json!(2)]).unwrap(); + let gbt = dispatch(&ctx, "getblocktemplate", vec![json!({"rules": ["segwit"]})]).unwrap(); + assert_eq!(gbt["height"].as_u64(), Some(3), "{gbt}"); + let mut bare = ctx_without_chain(&ctx); + bare.mempool = None; + let bare_gbt = dispatch( + &bare, + "getblocktemplate", + vec![json!({"rules": ["segwit"]})], + ) + .unwrap(); + assert_eq!(bare_gbt["height"].as_u64(), Some(3), "{bare_gbt}"); + assert_eq!(dispatch(&bare, "getrawmempool", vec![]).unwrap(), json!([])); + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] +fn blocksonly_refuses_template_mempool_and_generate() { + let (mut ctx, dir, _hub) = ctx_from_mature_pad(300_000_000); + ctx.blocksonly = true; + ctx.mempool.as_ref().unwrap().set_relay_enabled(true); + let (hex, spend) = mature_coinbase_spend_hex(&ctx, 50_0000_0000 - 1_000); + let ok = dispatch(&ctx, "sendrawtransaction", vec![json!(hex)]).unwrap(); + assert_eq!(ok.as_str().unwrap(), &spend.compute_txid().to_string()); + for (method, params) in [ + ("getblocktemplate", vec![json!({"rules": ["segwit"]})]), + ("getrawmempool", vec![]), + ("generate", vec![json!(1)]), + ] { + let err = dispatch(&ctx, method, params).unwrap_err(); + assert_ne!(err["code"], -10, "{method}: {err}"); + assert!( + err["message"].as_str().unwrap_or("").contains("blocksonly"), + "{method}: {err}" + ); + } + let _ = std::fs::remove_dir_all(&dir); +} + #[test] fn parse_wrapped_multi_sh_wsh_and_wsh() { let pk = "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798"; @@ -777,10 +895,23 @@ fn all_methods_callable_empty_or_error() { alert_notify: None, alert_fired: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + blocksonly: false, }; let mem2 = dispatch(&ctx2, "getmempoolinfo", vec![]).unwrap(); assert_eq!(mem2["loaded"], true); - let _ = dispatch(&ctx2, "getrawmempool", vec![json!(true)]).unwrap(); + let raw = dispatch(&ctx2, "getrawmempool", vec![json!(true)]).unwrap(); + assert_eq!(raw, json!({})); + let gbt = dispatch( + &ctx2, + "getblocktemplate", + vec![json!({"rules": ["segwit"]})], + ) + .unwrap_err(); + assert!( + gbt["message"].as_str().unwrap_or("").contains("no tip"), + "{gbt}" + ); + assert_ne!(gbt["code"], -10, "{gbt}"); let _ = dispatch(&ctx2, "estimatesmartfee", vec![json!(1)]).unwrap(); let _ = dispatch(&ctx2, "sendrawtransaction", vec![json!("00")]); let info = dispatch(&ctx2, "getblockchaininfo", vec![]).unwrap(); @@ -1167,6 +1298,9 @@ fn regtest_ctx( max_wu: u64, ) -> (RpcContext, TempDir, Arc) { let mp = MempoolHub::open_with_weight(dir.join("mempool"), hub.query.clone(), max_wu).unwrap(); + // The node attaches this pool before it serves RPC. Connects and + // disconnects move the template mark on that same pool. + assert!(hub.attach_mempool(Arc::clone(&mp)).is_ok()); mp.set_relay_enabled(true); let ctx = RpcContext { query: hub.query.clone(), @@ -1191,6 +1325,7 @@ fn regtest_ctx( alert_notify: None, alert_fired: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + blocksonly: false, }; (ctx, dir, hub) } @@ -1540,14 +1675,23 @@ fn gettxout_leftover_is_connected_not_unconfirmed() { .unwrap(); assert_eq!(shown["coinbase"], true, "{shown}"); assert!(shown["confirmations"].as_u64().unwrap() >= 1, "{shown}"); - ctx.mempool.as_ref().unwrap().set_relay_enabled(false); dispatch(&ctx, "generate", vec![json!(1)]).unwrap(); let tid = spend.compute_txid(); + let txid = hash_hex_display(&tid.to_byte_array()); assert!( - ctx.mempool.as_ref().unwrap().contains(&tid), - "relay off must leave the confirmed tx in the hub" + !ctx.mempool.as_ref().unwrap().contains(&tid), + "generate must drop a tx it just confirmed" + ); + let gbt = dispatch(&ctx, "getblocktemplate", vec![json!({"rules": ["segwit"]})]).unwrap(); + let republished = gbt["transactions"] + .as_array() + .unwrap() + .iter() + .any(|t| t["txid"].as_str() == Some(txid.as_str())); + assert!( + !republished, + "template must not contain a tx the new tip already confirmed: {gbt}" ); - let txid = hash_hex_display(&tid.to_byte_array()); let utxo = dispatch(&ctx, "gettxout", vec![json!(txid), json!(0)]).unwrap(); assert_ne!( utxo["confirmations"], 0, @@ -4293,6 +4437,7 @@ fn ctx_without_chain(src: &RpcContext) -> RpcContext { active: Arc::clone(&src.active), alert_notify: src.alert_notify.clone(), alert_fired: Arc::clone(&src.alert_fired), + blocksonly: src.blocksonly, } } diff --git a/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs b/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs index 99a727789..0436feb7f 100644 --- a/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs +++ b/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs @@ -439,7 +439,7 @@ fn chain_ops_network_identity(ctx: &RpcContext) { net["subversion"].as_str().unwrap(), rbitcoin_primitives::rbitcoin_subversion(env!("CARGO_PKG_VERSION"), &[] as &[&str]).unwrap() ); - let flags = rbitcoin_net::local_service_flags(); + let flags = rbitcoin_net::local_service_flags(false, None); let bits = flags.to_u64(); let hex = format!("{bits:016x}"); assert_eq!(net["localservices"].as_str(), Some(hex.as_str())); @@ -2543,11 +2543,12 @@ fn chain_ops_invalidate_and_precious( dispatch(ctx, "invalidateblock", vec![tip.clone()]).unwrap(); assert_eq!(tip_count(ctx), tip_h - 1); assert!(gettxout().is_null(), "a disconnected Class A row is not a UTXO"); - let archived = tma(&hex); - assert_eq!(archived["allowed"], true, "{archived}"); + let back = tma(&hex); + assert_eq!(back["reject-reason"], "txn-already-in-mempool", "{back}"); dispatch(ctx, "reconsiderblock", vec![tip.clone()]).unwrap(); assert_eq!(tip_count(ctx), tip_h); assert_eq!(best_hash(ctx), tip); + assert_eq!(tma(&hex)["reject-reason"], "txn-already-known"); let (_, parent) = hub.query.header_at_height(Height(tip_h as u32 - 1)).unwrap().unwrap(); let sibling = rbitcoin_consensus::mine_regtest_paying( diff --git a/crates/rbitcoin-rpc/src/server.rs b/crates/rbitcoin-rpc/src/server.rs index a3f9232d6..504530125 100644 --- a/crates/rbitcoin-rpc/src/server.rs +++ b/crates/rbitcoin-rpc/src/server.rs @@ -145,6 +145,8 @@ pub struct RpcConfig { pub rest: bool, /// `--alert-notify` (`%s` = warning text). pub alert_notify: Option, + /// `-blocksonly`. Mining and raw-mempool listing stay off. + pub blocksonly: bool, } /// Live RPC server handle. @@ -243,6 +245,7 @@ pub async fn run_rpc( active: Arc::clone(&active), alert_notify: config.alert_notify.clone(), alert_fired: Arc::new(AtomicBool::new(false)), + blocksonly: config.blocksonly, }); let n = work_queue_permits(config.work_queue); @@ -954,6 +957,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, q, None, None, None, None, None) .await @@ -994,6 +998,7 @@ mod tests { rest: true, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, q, Some(mp), None, None, None, None) .await @@ -1094,6 +1099,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, q, None, None, None, None, None).await.unwrap(); tokio::time::sleep(std::time::Duration::from_millis(40)).await; @@ -1164,6 +1170,7 @@ mod tests { rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, q, Some(mp), None, None, None, None) .await @@ -1246,6 +1253,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, q, None, None, None, None, None).await.unwrap(); tokio::time::sleep(Duration::from_millis(40)).await; @@ -1288,6 +1296,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, q, None, None, None, None, None).await.unwrap(); tokio::time::sleep(Duration::from_millis(40)).await; @@ -1356,6 +1365,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, query, None, None, None, Some(Arc::new(hub)), None) .await @@ -1490,6 +1500,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, query, None, None, None, Some(Arc::new(hub)), None) .await @@ -1555,6 +1566,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, query, None, None, None, Some(Arc::new(hub)), None) .await @@ -1619,6 +1631,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, query, None, None, None, Some(Arc::new(hub)), None) .await @@ -1689,6 +1702,7 @@ mod tests { work_queue: Some(1), rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, query, None, None, None, Some(Arc::new(hub)), None) .await @@ -1748,6 +1762,7 @@ mod tests { rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, q, Some(mp), None, None, None, None) .await @@ -1832,6 +1847,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, q, Some(mp), None, None, None, None) .await @@ -2013,6 +2029,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc( without_cookie, @@ -2055,6 +2072,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; // The cookie is TCP-only: a socket-only listener must not silently ignore it. std::fs::write(&cookie_path, "mempool:secret").unwrap(); @@ -2216,6 +2234,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let err = run_rpc(cfg, q, None, None, None, None, None) .await @@ -2245,6 +2264,7 @@ mod tests { work_queue: None, rest: false, alert_notify: None, + blocksonly: false, }; let handle = run_rpc(cfg, q, None, None, None, None, None).await.unwrap(); tokio::time::sleep(std::time::Duration::from_millis(80)).await; @@ -2300,6 +2320,7 @@ mod tests { logpath: String::new(), active: Arc::new(std::sync::Mutex::new(crate::methods::RpcActive::default())), alert_notify: None, + blocksonly: false, alert_fired: Arc::new(AtomicBool::new(false)), }); (ctx, dir) diff --git a/crates/rbitcoin-store/Cargo.toml b/crates/rbitcoin-store/Cargo.toml index bebfdb226..be934917f 100644 --- a/crates/rbitcoin-store/Cargo.toml +++ b/crates/rbitcoin-store/Cargo.toml @@ -7,6 +7,9 @@ rust-version.workspace = true repository.workspace = true description = "Relational archive store (Class A/B/C; fd-only bulk IO)" +[lib] +doctest = false + [dependencies] rbitcoin-primitives = { workspace = true } rbitcoin-log = { workspace = true } diff --git a/crates/rbitcoin-sv2/Cargo.toml b/crates/rbitcoin-sv2/Cargo.toml index 1327dd4e0..47e8d1d79 100644 --- a/crates/rbitcoin-sv2/Cargo.toml +++ b/crates/rbitcoin-sv2/Cargo.toml @@ -9,6 +9,12 @@ description = "Stratum v2 Template Distribution Protocol server (Noise over TCP) [features] mutants = [] +# Journeys turn this on. The node binary does not, so a changed SV2 budget +# still waits 1s. +test-hooks = [] + +[lib] +doctest = false [dependencies] rbitcoin-log = { workspace = true } diff --git a/crates/rbitcoin-sv2/src/lib.rs b/crates/rbitcoin-sv2/src/lib.rs index e62da566f..3dbe8c88a 100644 --- a/crates/rbitcoin-sv2/src/lib.rs +++ b/crates/rbitcoin-sv2/src/lib.rs @@ -47,7 +47,19 @@ pub const TEMPLATE_INTERVAL: Duration = Duration::from_secs(5); /// A client cycling budgets costs at most one mempool-locked build per this /// interval. A changed budget can lag by up to this long. Tip events are /// not delayed. +#[cfg(not(any(test, feature = "test-hooks")))] pub const CONSTRAINTS_COOLDOWN: Duration = Duration::from_secs(1); +/// Crate tests and journeys. The node binary keeps 1s. +#[cfg(any(test, feature = "test-hooks"))] +pub const CONSTRAINTS_COOLDOWN: Duration = Duration::from_millis(100); + +/// True when this crate was compiled with the `test-hooks` feature. +/// +/// A workspace build unifies that into the node, which then uses the short +/// budget cooldown. The node warns at startup. +pub fn compiled_with_test_hooks() -> bool { + cfg!(feature = "test-hooks") +} /// Lower bound on [`Sv2TpConfig::template_interval`]. CPU trade: each check /// after a mempool change is one build under the mempool read lock, so a diff --git a/crates/rbitcoin-sv2/src/session.rs b/crates/rbitcoin-sv2/src/session.rs index 3332eb875..e3ab65d61 100644 --- a/crates/rbitcoin-sv2/src/session.rs +++ b/crates/rbitcoin-sv2/src/session.rs @@ -178,6 +178,7 @@ pub(crate) async fn serve( superseded: 0, holding: false, held_logged: false, + hold_check_at: None, fee_check_at: None, seen_updates: 0, }; @@ -207,10 +208,13 @@ struct Session { rebuild_at: Option, /// Budgets that replaced a queued rebuild's since the last build. superseded: u32, - /// Last `publish` returned without a template because the node was in IBD. - /// Cleared when a template is sent. + /// Last `publish` returned without a template because the node was in IBD + /// or relay was off. Cleared when a template is sent. holding: bool, held_logged: bool, + /// While `holding`, the next time to try again. Relay turns on after the + /// catch-up tip, so that tip's event can run before the pool is ready. + hold_check_at: Option, /// Next fee check: one interval after the last push or check. fee_check_at: Option, /// `MempoolHub::template_updates` read before the last build. @@ -264,6 +268,14 @@ impl Session { { self.check_fees().await?; } + // Same interval, only while no template has been sent. The + // counter the fee check watches is not armed until then, and + // relay can turn on without another tip. + _ = tokio::time::sleep_until(self.hold_check_at.unwrap_or_else(Instant::now)), + if self.hold_check_at.is_some() => + { + self.publish().await?; + } _ = tokio::time::sleep_until(setup_deadline), if self.constraints.is_none() => { return Err(missed_setup_deadline()); } @@ -288,9 +300,8 @@ impl Session { c.coinbase_output_max_additional_sigops, )); // Same budget: the last publish already applied it. A new - // budget while holding for IBD waits for the tip that leaves - // IBD; that tip calls `publish` and must not find a queued - // rebuild or a flood close in its place. + // budget while holding is applied on the next hold check. + // Publishing here would queue a rebuild that the check drops. let same = c == self.constraints; self.constraints = c; if same || self.holding { @@ -328,8 +339,16 @@ impl Session { self.rebuild_at = None; self.superseded = 0; match self.build().await? { - Some(t) => self.send(t).await, - None => Ok(()), + Some(t) => { + self.hold_check_at = None; + self.send(t).await + } + None => { + if self.holding { + self.hold_check_at = Some(Instant::now() + self.fee_push.interval); + } + Ok(()) + } } } @@ -370,8 +389,10 @@ impl Session { Ok(()) } - /// Template for the session's constraints. `None` before constraints or - /// while in IBD: leaving IBD always comes with a new tip, which publishes. + /// Template for the session's constraints. `None` before constraints, while + /// in IBD, or while relay is off (including `-blocksonly`). A holding + /// session tries again on the template interval, so relay turning on + /// after the catch-up tip still publishes. async fn build(&mut self) -> io::Result> { let Some((size, sigops)) = self.constraints else { return Ok(None); @@ -384,13 +405,16 @@ impl Session { let stats = Arc::clone(&self.stats); let t = tokio::task::spawn_blocking(move || { let _g = BlockingRegion::enter(); - if c.in_ibd() { + let relay_off = c.mempool().is_none_or(|m| !m.relay_enabled()); + if c.in_ibd() || relay_off { return Ok(None); } let started = std::time::Instant::now(); let t = template::build(&c, size, sigops)?; - stats.builds.note(micros(started.elapsed())); - Ok::<_, io::Error>(Some(t)) + if t.is_some() { + stats.builds.note(micros(started.elapsed())); + } + Ok::<_, io::Error>(t) }) .await .map_err(io::Error::other)??; @@ -398,7 +422,7 @@ impl Session { if t.is_none() { self.holding = true; if !self.held_logged { - rbitcoin_log::info!("sv2: holding templates until the node leaves IBD"); + rbitcoin_log::info!("sv2: holding templates until the mempool is maintained"); self.held_logged = true; } } @@ -632,14 +656,14 @@ mod tests { use tokio::time::Instant; #[test] - fn constraints_rebuild_at_uses_the_one_second_cooldown() { + fn constraints_rebuild_at_uses_the_cooldown() { let built_at = Instant::now(); - let soon = built_at + Duration::from_millis(999); + let soon = built_at + crate::CONSTRAINTS_COOLDOWN - Duration::from_millis(1); assert_eq!( constraints_rebuild_at(built_at, soon), - Some(built_at + Duration::from_secs(1)) + Some(built_at + crate::CONSTRAINTS_COOLDOWN) ); - let due = built_at + Duration::from_secs(1); + let due = built_at + crate::CONSTRAINTS_COOLDOWN; assert_eq!(constraints_rebuild_at(built_at, due), None); } diff --git a/crates/rbitcoin-sv2/src/template.rs b/crates/rbitcoin-sv2/src/template.rs index e2a1ecf3f..77a6322d1 100644 --- a/crates/rbitcoin-sv2/src/template.rs +++ b/crates/rbitcoin-sv2/src/template.rs @@ -75,22 +75,33 @@ impl Template { } } -/// Template on the current tip for one client's coinbase constraints. The +/// Template on the stripped tip for one client's coinbase constraints. The /// client's sigops replace the default reserve (Core `BlockAssembler`). -/// Takes the mempool lock and reads the store: blocking region only. +/// `Ok(None)` when that tip has not been recorded, or is no longer the +/// header at its height. Takes the mempool lock and reads the store: +/// blocking region only. pub(crate) fn build( chain: &ChainHub, max_additional_size: u32, max_additional_sigops: u16, -) -> io::Result