diff --git a/crates/rbitcoin-consensus/src/index_writebehind.rs b/crates/rbitcoin-consensus/src/index_writebehind.rs index f0fa8034b..f3c92121f 100644 --- a/crates/rbitcoin-consensus/src/index_writebehind.rs +++ b/crates/rbitcoin-consensus/src/index_writebehind.rs @@ -546,7 +546,15 @@ fn run_writebehind( .name("rbtc-idx-cpu".into()) .spawn_scoped(scope, || cpu_worker(query, rx, &resync, &stages)) .expect("spawn index cpu worker"); - let io = io_loop(query, stop, &resync, &stages, &tx, on_filters_caught_up); + let io = io_loop( + query, + stop, + &resync, + &stages, + &tx, + PROGRESS_EVERY, + on_filters_caught_up, + ); drop(tx); let cpu = worker.join().expect("index cpu worker"); io.and(cpu) @@ -565,6 +573,7 @@ fn io_loop( resync: &AtomicBool, stages: &IndexStageMs, tx: &std::sync::mpsc::SyncSender, + progress_every: Duration, on_filters_caught_up: impl FnOnce(), ) -> Result<(), ConsensusError> { let mut on_filters_caught_up = Some(on_filters_caught_up); @@ -621,7 +630,7 @@ fn io_loop( }); } if let Some(p) = pass.as_mut() { - if p.last_log.elapsed() >= PROGRESS_EVERY { + if p.last_log.elapsed() >= progress_every { rbitcoin_log::info!( "{}", format_index_build_progress(&IndexBuildProgress { @@ -687,7 +696,7 @@ mod tests { }); rbitcoin_log::capture_logs(false); rbitcoin_log::capture_logs(true); - io_loop(query, stop, resync, stages, tx, on_caught).unwrap(); + io_loop(query, stop, resync, stages, tx, PROGRESS_EVERY, on_caught).unwrap(); let logs = rbitcoin_log::take_logs(); rbitcoin_log::capture_logs(false); stop.store(true, Ordering::Relaxed); @@ -702,8 +711,12 @@ mod tests { /// Seal from the current watermarks through the released tip. /// `hold_first` blocks the first window so a later window in the same - /// pass can cross [`PROGRESS_EVERY`]. - fn drive_index_pass(query: &Query, hold_first: Duration) -> Vec { + /// pass can cross `progress_every`. + fn drive_index_pass( + query: &Query, + hold_first: Duration, + progress_every: Duration, + ) -> Vec { let stop = AtomicBool::new(false); let resync = AtomicBool::new(false); let stages = IndexStageMs::default(); @@ -740,7 +753,7 @@ mod tests { }); rbitcoin_log::capture_logs(false); rbitcoin_log::capture_logs(true); - io_loop(query, &stop, &resync, &stages, &tx, || { + io_loop(query, &stop, &resync, &stages, &tx, progress_every, || { stop.store(true, Ordering::Relaxed); }) .unwrap(); @@ -1025,7 +1038,13 @@ mod tests { let pass_tip = resume + WINDOW_HEIGHTS; connect_coinbases(&q, resume, pass_tip, &mut prev_fk, &mut prev_hash); q.release_index_writebehind(Height(pass_tip)); - let lines = drive_index_pass(&q, PROGRESS_EVERY + Duration::from_secs(2)); + assert_eq!(PROGRESS_EVERY.as_secs(), 10); + let progress_every = Duration::from_secs(1); + let lines = drive_index_pass( + &q, + progress_every + Duration::from_millis(50), + progress_every, + ); let heights = pass_tip + 1 - resume; assert!( lines @@ -1049,7 +1068,7 @@ mod tests { let short_tip = short_from + 10; connect_coinbases(&q, short_from, short_tip, &mut prev_fk, &mut prev_hash); q.release_index_writebehind(Height(short_tip)); - let lines = drive_index_pass(&q, Duration::ZERO); + let lines = drive_index_pass(&q, Duration::ZERO, PROGRESS_EVERY); assert!( lines .iter() @@ -1291,7 +1310,7 @@ mod tests { let stages = IndexStageMs::default(); let (tx, rx) = sync_channel(1); drop(rx); - io_loop(&q, &stop, &resync, &stages, &tx, || {}).unwrap(); + io_loop(&q, &stop, &resync, &stages, &tx, PROGRESS_EVERY, || {}).unwrap(); let fin = finished_index_stages(); progress::capture_finished(false); assert_eq!(fin, [("blockfilter build", 0, 0, u64::from(tip) + 1)]); diff --git a/crates/rbitcoin-net/src/tx_relay.rs b/crates/rbitcoin-net/src/tx_relay.rs index ca9fca141..f88f52dd2 100644 --- a/crates/rbitcoin-net/src/tx_relay.rs +++ b/crates/rbitcoin-net/src/tx_relay.rs @@ -3772,11 +3772,9 @@ mod tests { let _ = std::fs::remove_dir_all(&mdir); } - #[test] - fn expire_stale_drops_old_tx_without_a_new_accept() { - let (store_dir, q, cbs) = pad_one_cb(); + fn expire_stale_drops_old_tx_without_a_new_accept(q: &Arc, cbs: &[Txid]) { let dir = tmp(); - let hub = MempoolHub::open(&dir, q).unwrap(); + let hub = MempoolHub::open(&dir, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); let tx = spend_true(cbs[0], 1_000, ScriptBuf::from_bytes(vec![0x51])); let tid = tx.compute_txid(); @@ -3788,7 +3786,6 @@ mod tests { assert_eq!(hub.live_count(), 0); assert!(!hub.contains(&tid)); let _ = std::fs::remove_dir_all(&dir); - let _ = std::fs::remove_dir_all(&store_dir); } /// One scan visits 256 live txs. With more than that expired, the survivors @@ -3828,11 +3825,9 @@ mod tests { let _ = std::fs::remove_dir_all(&store_dir); } - #[test] - fn try_expire_stale_skips_when_the_order_lock_is_held() { - let (store_dir, q, cbs) = pad_one_cb(); + fn try_expire_stale_skips_when_the_order_lock_is_held(q: &Arc, cbs: &[Txid]) { let dir = tmp(); - let hub = MempoolHub::open(&dir, q).unwrap(); + let hub = MempoolHub::open(&dir, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); hub.note_mock_now(1); let tx = spend_true(cbs[0], 1_000, ScriptBuf::from_bytes(vec![0x51])); @@ -3847,7 +3842,6 @@ mod tests { assert_eq!(hub.expire_stale(), 1); assert!(!hub.contains(&tid)); let _ = std::fs::remove_dir_all(&dir); - let _ = std::fs::remove_dir_all(&store_dir); } fn record_fee_sample(hub: &MempoolHub, height: u32, rate_sat_kvb: u64) { @@ -4491,40 +4485,30 @@ mod tests { /// Non-coinbase, no BIP68 time-lock: no `block_tx_fks` and no create MTP. /// A satisfied time-lock spend must survive `evict_after_reorg`. - #[test] - fn get_coin_skips_block_tx_fks_and_mtp_without_time_lock() { + fn get_coin_skips_block_tx_fks_and_mtp_without_time_lock(q: &Arc, cbs: &[Txid]) { use rbitcoin_consensus::{accept_and_connect_block, ChainParams, Milestone}; use rbitcoin_primitives::Height; use rbitcoin_store::script_hash; - let store_dir = tmp(); - let q = Query::open_or_create_tiny(&store_dir).unwrap(); let params = ChainParams::regtest(); - let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); - accept_and_connect_block(&q, ¶ms, Height::GENESIS, &genesis, Milestone::NONE).unwrap(); - let (tip, tip_time, cbs) = rbitcoin_consensus::pad_empty_from( - &q, - ¶ms, - genesis.block_hash(), - genesis.header.time, - 1, - 102, - 3, - ); + let tip_h = q.tip_height().expect("tip").0; + let fk = q.tip_header_fk().unwrap().expect("tip header"); + let rec = q.get_header(fk).unwrap(); + let tip = bitcoin::BlockHash::from_byte_array(rec.hash); + let next = tip_h + 1; let spk = ScriptBuf::from_bytes(vec![0x51]); let confirmed = spend_true(cbs[0], 1_000, spk.clone()); let b = rbitcoin_consensus::mine_regtest_paying( tip, - tip_time + 600, - 103, + rec.timestamp + 600, + next, spk.clone(), vec![confirmed.clone()], ); - accept_and_connect_block(&q, ¶ms, Height(103), &b, Milestone::NONE).unwrap(); + accept_and_connect_block(q, ¶ms, Height(next), &b, Milestone::NONE).unwrap(); q.apply_sh_pending().unwrap(); - let q = Arc::new(q); let mp = tmp(); - let hub = MempoolHub::open(&mp, Arc::clone(&q)).unwrap(); + let hub = MempoolHub::open(&mp, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); let _ = hub.sample_reset_perf(); let child = Transaction { @@ -4670,34 +4654,14 @@ mod tests { "evict_after_reorg must not drop a still-valid BIP68 time lock" ); let _ = std::fs::remove_dir_all(&mp); - let _ = std::fs::remove_dir_all(&store_dir); } /// Confirm/RBF unindex must drop `relay_seq` / `accept_at` for the gone /// wtxid and leave a still-live sibling indexed. - #[test] - fn unindex_drops_relay_seq_and_accept_at() { - use rbitcoin_consensus::{accept_and_connect_block, ChainParams, Milestone}; - use rbitcoin_primitives::Height; - - let store_dir = tmp(); - let q = Query::open_or_create_tiny(&store_dir).unwrap(); - let params = ChainParams::regtest(); - let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); - accept_and_connect_block(&q, ¶ms, Height::GENESIS, &genesis, Milestone::NONE).unwrap(); - let (_tip, _tip_time, cbs) = rbitcoin_consensus::pad_empty_from( - &q, - ¶ms, - genesis.block_hash(), - genesis.header.time, - 1, - 102, - 2, - ); - let q = Arc::new(q); + fn unindex_drops_relay_seq_and_accept_at(q: &Arc, cbs: &[Txid]) { let spk = ScriptBuf::from_bytes(vec![0x51]); let mp = tmp(); - let hub = MempoolHub::open(&mp, Arc::clone(&q)).unwrap(); + let hub = MempoolHub::open(&mp, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); hub.note_mock_now(10); let gone = spend_true(cbs[0], 1_000, spk.clone()); @@ -4716,34 +4680,15 @@ mod tests { assert!(!hub.tx_inv_due(&gone_w)); assert!(hub.tx_inv_due(&stay_w)); let _ = std::fs::remove_dir_all(&mp); - let _ = std::fs::remove_dir_all(&store_dir); } /// Without `setmocktime`, INV age must still elapse on wall clock /// (`mempool_accept_wtxid` wait_for_broadcast; mock_now==0 must not freeze). - #[test] - fn tx_inv_due_uses_wall_clock_when_mocktime_unset() { + fn tx_inv_due_uses_wall_clock_when_mocktime_unset(q: &Arc, cbs: &[Txid]) { use bitcoin::script::ScriptBuf; - use rbitcoin_consensus::{accept_and_connect_block, ChainParams, Milestone}; - use rbitcoin_primitives::Height; - let store_dir = tmp(); - let q = Query::open_or_create_tiny(&store_dir).unwrap(); - let params = ChainParams::regtest(); - let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); - accept_and_connect_block(&q, ¶ms, Height::GENESIS, &genesis, Milestone::NONE).unwrap(); - let (_tip, _tip_time, cbs) = rbitcoin_consensus::pad_empty_from( - &q, - ¶ms, - genesis.block_hash(), - genesis.header.time, - 1, - 102, - 1, - ); - let q = Arc::new(q); let mp = tmp(); - let hub = MempoolHub::open(&mp, Arc::clone(&q)).unwrap(); + let hub = MempoolHub::open(&mp, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); assert_eq!(hub.mock_now.load(Ordering::Relaxed), 0); let tx = spend_true(cbs[0], 1_000, ScriptBuf::from_bytes(vec![0x51])); @@ -4769,7 +4714,6 @@ mod tests { ); assert!(hub.any_tx_inv_due()); let _ = std::fs::remove_dir_all(&mp); - let _ = std::fs::remove_dir_all(&store_dir); } /// While relay is off, per-block remove is deferred; enabling relay runs purge. @@ -4995,25 +4939,7 @@ mod tests { let _ = std::fs::remove_dir_all(&store_dir); } - #[test] - fn startup_recomputes_unknown_sigops_with_configured_reserve() { - use rbitcoin_consensus::{accept_and_connect_block, ChainParams, Milestone}; - use rbitcoin_primitives::Height; - - let store_dir = tmp(); - let q = Query::open_or_create_tiny(&store_dir).unwrap(); - let params = ChainParams::regtest(); - let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); - accept_and_connect_block(&q, ¶ms, Height::GENESIS, &genesis, Milestone::NONE).unwrap(); - let (_tip, _tip_time, cbs) = rbitcoin_consensus::pad_empty_from( - &q, - ¶ms, - genesis.block_hash(), - genesis.header.time, - 1, - 102, - 1, - ); + fn startup_recomputes_unknown_sigops_with_configured_reserve(q: &Arc, cbs: &[Txid]) { let tx = Transaction { version: Version::TWO, lock_time: LockTime::ZERO, @@ -5053,7 +4979,7 @@ mod tests { } let hub = MempoolHub::open_with_weight_persist_and_sigop_reserve( &mempool_dir, - Arc::new(q), + Arc::clone(q), rbitcoin_mempool::DEFAULT_MAX_MEMPOOL_WEIGHT, true, Some(0), @@ -5076,7 +5002,6 @@ mod tests { assert!(hub.select_block_template(core_reserve).is_empty()); assert!(hub.contains(&txid)); let _ = std::fs::remove_dir_all(&mempool_dir); - let _ = std::fs::remove_dir_all(&store_dir); } #[test] @@ -5404,8 +5329,25 @@ mod tests { let _ = std::fs::remove_dir_all(&store_dir); } - fn pad_one_cb() -> (std::path::PathBuf, Arc, Vec) { - pad_cbs(1) + /// One 3-coinbase maturity pad. Each chapter opens its own mempool. + /// The last chapter connects a block on this chain. + #[test] + fn shared_maturity_pad_journey() { + let (store_dir, q, cbs) = pad_cbs(3); + expire_stale_drops_old_tx_without_a_new_accept(&q, &cbs); + try_expire_stale_skips_when_the_order_lock_is_held(&q, &cbs); + unindex_drops_relay_seq_and_accept_at(&q, &cbs); + tx_inv_due_uses_wall_clock_when_mocktime_unset(&q, &cbs); + startup_recomputes_unknown_sigops_with_configured_reserve(&q, &cbs); + submit_package_rpc_admits_cpfp_below_minrelay(&q, &cbs); + mempool_under_pressure(&q, &cbs); + accept_package_child_fail_restores_rbf_victims(&q, &cbs); + fee_snapshot_live_totals_match_list_live_meta(&q, &cbs); + accept_commit_does_not_query_under_write(&q, &cbs); + mempool_tx_snapshot_two_live_and_accept_while_held(&q, &cbs); + mempool_tx_snapshot_refresh_reuses_tx_arc(&q, &cbs); + get_coin_skips_block_tx_fks_and_mtp_without_time_lock(&q, &cbs); + let _ = std::fs::remove_dir_all(&store_dir); } /// Coinbases at heights `1..=n`, each with 100 confirmations. @@ -5431,11 +5373,9 @@ mod tests { (store_dir, Arc::new(q), cbs) } - #[test] - fn submit_package_rpc_admits_cpfp_below_minrelay() { - let (store_dir, q, cbs) = pad_one_cb(); + fn submit_package_rpc_admits_cpfp_below_minrelay(q: &Arc, cbs: &[Txid]) { let dir = tmp(); - let hub = MempoolHub::open(&dir, q).unwrap(); + let hub = MempoolHub::open(&dir, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); let parent = spend_true(cbs[0], 1, ScriptBuf::from_bytes(vec![0x51])); assert!( @@ -5470,14 +5410,11 @@ mod tests { assert!(hub.try_contains(&parent.compute_txid())); assert!(hub.try_contains(&child.compute_txid())); let _ = std::fs::remove_dir_all(&dir); - let _ = std::fs::remove_dir_all(&store_dir); } - #[test] - fn mempool_under_pressure() { - let (store_dir, q, cbs) = pad_one_cb(); + fn mempool_under_pressure(q: &Arc, cbs: &[Txid]) { let dir = tmp(); - let hub = MempoolHub::open(&dir, q).unwrap(); + let hub = MempoolHub::open(&dir, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); let parent = spend_true(cbs[0], 1, ScriptBuf::from_bytes(vec![0x51])); assert!(matches!( @@ -5583,32 +5520,12 @@ mod tests { "child of rejected parent must poison descendants" ); let _ = std::fs::remove_dir_all(&dir); - let _ = std::fs::remove_dir_all(&store_dir); } - #[test] - fn accept_package_child_fail_restores_rbf_victims() { - use rbitcoin_consensus::{accept_and_connect_block, ChainParams, Milestone}; - use rbitcoin_primitives::Height; - - let store_dir = tmp(); - let q = Query::open_or_create_tiny(&store_dir).unwrap(); - let params = ChainParams::regtest(); - let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); - accept_and_connect_block(&q, ¶ms, Height::GENESIS, &genesis, Milestone::NONE).unwrap(); - let (_tip, _tip_time, cbs) = rbitcoin_consensus::pad_empty_from( - &q, - ¶ms, - genesis.block_hash(), - genesis.header.time, - 1, - 102, - 1, - ); - let q = Arc::new(q); + fn accept_package_child_fail_restores_rbf_victims(q: &Arc, cbs: &[Txid]) { let spk = ScriptBuf::from_bytes(vec![0x51]); let mp = tmp(); - let hub = MempoolHub::open(&mp, Arc::clone(&q)).unwrap(); + let hub = MempoolHub::open(&mp, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); let low = Transaction { version: Version::TWO, @@ -5678,7 +5595,6 @@ mod tests { "hub package rollback must restore the RBF victim" ); let _ = std::fs::remove_dir_all(&mp); - let _ = std::fs::remove_dir_all(&store_dir); } #[test] @@ -6002,28 +5918,9 @@ mod tests { } /// Fee-snapshot refresh publishes live count/vsize/total_fee (GET /mempool). - #[test] - fn fee_snapshot_live_totals_match_list_live_meta() { - use rbitcoin_consensus::{accept_and_connect_block, ChainParams, Milestone}; - use rbitcoin_primitives::Height; - - let store_dir = tmp(); + fn fee_snapshot_live_totals_match_list_live_meta(q: &Arc, cbs: &[Txid]) { let mp_dir = tmp(); - let q = Query::open_or_create_tiny(&store_dir).unwrap(); - let params = ChainParams::regtest(); - let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); - accept_and_connect_block(&q, ¶ms, Height::GENESIS, &genesis, Milestone::NONE).unwrap(); - let (_tip, _time, cbs) = rbitcoin_consensus::pad_empty_from( - &q, - ¶ms, - genesis.block_hash(), - genesis.header.time, - 1, - 102, - 1, - ); - let q = Arc::new(q); - let hub = MempoolHub::open(&mp_dir, Arc::clone(&q)).unwrap(); + let hub = MempoolHub::open(&mp_dir, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); let a = spend_true(cbs[0], 1_000, ScriptBuf::from_bytes(vec![0x51])); hub.accept_tx(&a).expect("admit"); @@ -6049,7 +5946,6 @@ mod tests { (expect_count, expect_vsize, expect_fee) ); let _ = std::fs::remove_dir_all(&mp_dir); - let _ = std::fs::remove_dir_all(&store_dir); } /// Production accept must not run on a tokio worker (reactor starvation). @@ -6311,11 +6207,8 @@ mod tests { let _ = std::fs::remove_dir_all(&store_dir); } - #[test] - fn accept_commit_does_not_query_under_write() { - use rbitcoin_consensus::{accept_and_connect_block, ChainParams, Milestone}; + fn accept_commit_does_not_query_under_write(q: &Arc, cbs: &[Txid]) { use rbitcoin_mempool::UtxoProvider; - use rbitcoin_primitives::Height; use std::sync::atomic::{AtomicU64, Ordering}; use std::thread; @@ -6340,23 +6233,8 @@ mod tests { } } - let store_dir = tmp(); let mp_dir = tmp(); - let q = Query::open_or_create_tiny(&store_dir).unwrap(); - let params = ChainParams::regtest(); - let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); - accept_and_connect_block(&q, ¶ms, Height::GENESIS, &genesis, Milestone::NONE).unwrap(); - let (_tip, _tip_time, cbs) = rbitcoin_consensus::pad_empty_from( - &q, - ¶ms, - genesis.block_hash(), - genesis.header.time, - 1, - 102, - 1, - ); - let q = Arc::new(q); - let hub = MempoolHub::open(&mp_dir, Arc::clone(&q)).unwrap(); + let hub = MempoolHub::open(&mp_dir, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); let hits = Arc::new(AtomicU64::new(0)); let probe = ProbeUtxo { @@ -6372,7 +6250,6 @@ mod tests { "QueryUtxoProvider must not run while inner write is held" ); let _ = std::fs::remove_dir_all(&mp_dir); - let _ = std::fs::remove_dir_all(&store_dir); } #[test] @@ -6455,29 +6332,11 @@ mod tests { let _ = std::fs::remove_dir_all(&store_dir); } - #[test] - fn mempool_tx_snapshot_two_live_and_accept_while_held() { - use rbitcoin_consensus::{accept_and_connect_block, ChainParams, Milestone}; - use rbitcoin_primitives::Height; + fn mempool_tx_snapshot_two_live_and_accept_while_held(q: &Arc, cbs: &[Txid]) { use std::thread; - let store_dir = tmp(); let mp_dir = tmp(); - let q = Query::open_or_create_tiny(&store_dir).unwrap(); - let params = ChainParams::regtest(); - let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); - accept_and_connect_block(&q, ¶ms, Height::GENESIS, &genesis, Milestone::NONE).unwrap(); - let (_tip, _time, cbs) = rbitcoin_consensus::pad_empty_from( - &q, - ¶ms, - genesis.block_hash(), - genesis.header.time, - 1, - 102, - 3, - ); - let q = Arc::new(q); - let hub = MempoolHub::open(&mp_dir, Arc::clone(&q)).unwrap(); + let hub = MempoolHub::open(&mp_dir, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); let spk = ScriptBuf::from_bytes(vec![0x51]); let a = spend_true(cbs[0], 1_000, spk.clone()); @@ -6514,31 +6373,11 @@ mod tests { ); assert!(snap2.get(&cid).unwrap().json.get().is_none()); let _ = std::fs::remove_dir_all(&mp_dir); - let _ = std::fs::remove_dir_all(&store_dir); } - #[test] - fn mempool_tx_snapshot_refresh_reuses_tx_arc() { - use rbitcoin_consensus::{accept_and_connect_block, ChainParams, Milestone}; - use rbitcoin_primitives::Height; - - let store_dir = tmp(); + fn mempool_tx_snapshot_refresh_reuses_tx_arc(q: &Arc, cbs: &[Txid]) { let mp_dir = tmp(); - let q = Query::open_or_create_tiny(&store_dir).unwrap(); - let params = ChainParams::regtest(); - let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); - accept_and_connect_block(&q, ¶ms, Height::GENESIS, &genesis, Milestone::NONE).unwrap(); - let (_tip, _time, cbs) = rbitcoin_consensus::pad_empty_from( - &q, - ¶ms, - genesis.block_hash(), - genesis.header.time, - 1, - 102, - 2, - ); - let q = Arc::new(q); - let hub = MempoolHub::open(&mp_dir, Arc::clone(&q)).unwrap(); + let hub = MempoolHub::open(&mp_dir, Arc::clone(q)).unwrap(); hub.set_relay_enabled(true); let a = spend_true(cbs[0], 1_000, ScriptBuf::from_bytes(vec![0x51])); hub.accept_tx(&a).expect("a"); @@ -6564,6 +6403,5 @@ mod tests { assert!(!Arc::ptr_eq(&e3a.tx, &e3b.tx), "new admit gets its own Arc"); assert!(e3b.json.get().is_none()); let _ = std::fs::remove_dir_all(&mp_dir); - let _ = std::fs::remove_dir_all(&store_dir); } } diff --git a/crates/rbitcoin-query/src/query_resume_journey.rs b/crates/rbitcoin-query/src/query_resume_journey.rs index fe1a5a282..77446fc2d 100644 --- a/crates/rbitcoin-query/src/query_resume_journey.rs +++ b/crates/rbitcoin-query/src/query_resume_journey.rs @@ -79,8 +79,18 @@ fn resume_most_work_header_path() { let x_tip = x.last().unwrap(); let x_tip_fk = q.get_header_by_hash(&x_tip.hash).unwrap().unwrap().0; - put_header_fork(&q, (x_tip_fk, x_tip.hash), 1_000, 12_000); - let path = without_w(32); + // Longer than the 32-cap. Scored on a stack a recursive walk of this + // band would overflow; the production walk is a heap stack. + const DEEP: u32 = 256; + put_header_fork(&q, (x_tip_fk, x_tip.hash), 1_000, DEEP); + let path = std::thread::scope(|scope| { + std::thread::Builder::new() + .stack_size(32 * 1024) + .spawn_scoped(scope, || without_w(32)) + .expect("spawn") + .join() + .expect("deep resume walk") + }); assert_eq!(path.len(), 32, "capped walk length"); assert_eq!((path[0].height, path[31].height), (1, 32)); diff --git a/crates/rbitcoin-sv2/src/listener_tests.rs b/crates/rbitcoin-sv2/src/listener_tests.rs index 1920cc04e..cd4a41725 100644 --- a/crates/rbitcoin-sv2/src/listener_tests.rs +++ b/crates/rbitcoin-sv2/src/listener_tests.rs @@ -1,12 +1,14 @@ -use crate::test_chain::padded_chain; +use crate::test_chain::shared_regtest; use crate::testutil::TpClient; use crate::{run_sv2_tp, Sv2TpConfig, MAX_SESSIONS, MAX_STALE_GRACE, SETUP_TIMEOUT, WRITE_TIMEOUT}; use common_messages_sv2::{ SetupConnectionError, SetupConnectionSuccess, MESSAGE_TYPE_SETUP_CONNECTION_ERROR, MESSAGE_TYPE_SETUP_CONNECTION_SUCCESS, }; +use rbitcoin_net::ChainHub; use std::io; use std::net::SocketAddr; +use std::sync::Arc; use std::time::{Duration, Instant}; use template_distribution_sv2::MESSAGE_TYPE_REQUEST_TRANSACTION_DATA_ERROR; @@ -23,6 +25,24 @@ async fn connect_when_free(addr: SocketAddr, pk: [u8; 32]) -> TpClient { } } +/// One stale regtest chain. Each beat is its own listener. The two setup +/// deadlines overlap. The write-deadline pin is process-wide, so that beat +/// runs after the others have accepted. +#[tokio::test] +async fn listener_rejects_bad_sessions_on_one_chain() { + let tc = shared_regtest(0); + let chain = Arc::clone(&tc.chain); + tokio::join!( + setup_connection_success_errors_and_session_cap(Arc::clone(&chain)), + authority_key_prints_in_key_utils_base58check(Arc::clone(&chain)), + silent_sockets_are_dropped_at_the_setup_deadline(Arc::clone(&chain)), + session_without_constraints_is_dropped_at_the_setup_deadline(Arc::clone(&chain)), + oversized_client_frame_closes_the_session(Arc::clone(&chain)), + out_of_range_cert_validity_or_stale_grace_refuses_to_start(Arc::clone(&chain)), + ); + client_that_stops_reading_is_dropped_at_the_write_deadline(chain).await; +} + async fn expect_error(c: &mut TpClient, flags: u32, code: &str) { let mut f = c.recv().await.expect("setup reply"); assert_eq!(f.msg_type, MESSAGE_TYPE_SETUP_CONNECTION_ERROR); @@ -36,12 +56,10 @@ async fn expect_error(c: &mut TpClient, flags: u32, code: &str) { ); } -#[tokio::test] -async fn setup_connection_success_errors_and_session_cap() { - let tc = padded_chain("sv2-listener", 0); +async fn setup_connection_success_errors_and_session_cap(chain: Arc) { let tp = run_sv2_tp(Sv2TpConfig { listen: "127.0.0.1:0".parse().unwrap(), - chain: tc.chain.clone(), + chain, authority_secret: [7; 32], cert_validity: Duration::from_secs(3600), stale_grace: Duration::from_secs(10), @@ -94,14 +112,12 @@ async fn setup_connection_success_errors_and_session_cap() { /// key-utils 1.2.0 vector: SRI clients configure the TP authority key in /// this form, so the handle must print it, and a client must connect with it. -#[tokio::test] -async fn authority_key_prints_in_key_utils_base58check() { +async fn authority_key_prints_in_key_utils_base58check(chain: Arc) { let secret = bitcoin::base58::decode_check("zmBEmPhqo3A92FkiLVvyCz6htc3e53ph3ZbD4ASqGaLjwnFLi") .expect("vector secret"); - let tc = padded_chain("sv2-authority-key", 0); let tp = run_sv2_tp(Sv2TpConfig { listen: "127.0.0.1:0".parse().unwrap(), - chain: std::sync::Arc::clone(&tc.chain), + chain, authority_secret: secret.try_into().expect("32-byte secret"), cert_validity: Duration::from_secs(3600), stale_grace: Duration::from_secs(10), @@ -127,15 +143,13 @@ async fn authority_key_prints_in_key_utils_base58check() { /// Silent sockets take every slot at accept; the setup deadline must close /// them so a real client gets in. -#[tokio::test] -async fn silent_sockets_are_dropped_at_the_setup_deadline() { +async fn silent_sockets_are_dropped_at_the_setup_deadline(chain: Arc) { use tokio::io::AsyncReadExt; - let tc = padded_chain("sv2-setup-deadline", 0); let setup_timeout = Duration::from_millis(300); let tp = run_sv2_tp(Sv2TpConfig { listen: "127.0.0.1:0".parse().unwrap(), - chain: std::sync::Arc::clone(&tc.chain), + chain, authority_secret: [7; 32], cert_validity: Duration::from_secs(3600), stale_grace: Duration::from_secs(10), @@ -169,13 +183,11 @@ async fn silent_sockets_are_dropped_at_the_setup_deadline() { /// A TDP session without `CoinbaseOutputConstraints` never gets a template /// and never writes, so the setup deadline also covers the first constraints. /// Other frames before them do not reset it. -#[tokio::test] -async fn session_without_constraints_is_dropped_at_the_setup_deadline() { - let tc = padded_chain("sv2-constraints-deadline", 0); +async fn session_without_constraints_is_dropped_at_the_setup_deadline(chain: Arc) { let setup_timeout = Duration::from_millis(300); let tp = run_sv2_tp(Sv2TpConfig { listen: "127.0.0.1:0".parse().unwrap(), - chain: std::sync::Arc::clone(&tc.chain), + chain, authority_secret: [7; 32], cert_validity: Duration::from_secs(3600), stale_grace: Duration::from_secs(10), @@ -223,16 +235,14 @@ async fn session_without_constraints_is_dropped_at_the_setup_deadline() { /// A client that floods requests and never reads jams the TP's writes; the /// write deadline must close the session instead of stalling it forever. -#[tokio::test] -async fn client_that_stops_reading_is_dropped_at_the_write_deadline() { - let tc = padded_chain("sv2-write-deadline", 0); +async fn client_that_stops_reading_is_dropped_at_the_write_deadline(chain: Arc) { let write_timeout = Duration::from_millis(200); // Pin before accept. Drop clears it so other tests keep autotune. #[cfg(target_os = "linux")] let _send_buf = crate::test_send_buffer::pin(8 * 1024); let tp = run_sv2_tp(Sv2TpConfig { listen: "127.0.0.1:0".parse().unwrap(), - chain: std::sync::Arc::clone(&tc.chain), + chain, authority_secret: [7; 32], cert_validity: Duration::from_secs(3600), stale_grace: Duration::from_secs(10), @@ -259,6 +269,7 @@ async fn client_that_stops_reading_is_dropped_at_the_write_deadline() { // leaves the session idle so the deadline never starts. Yield so the // session task runs; a ready send does not, and the flood only fills // the kernel buffer. + let started = Instant::now(); let outcome = tokio::time::timeout(Duration::from_secs(5), async { for id in 1..=1_000_000u64 { c.request_transaction_data(id).await?; @@ -273,17 +284,20 @@ async fn client_that_stops_reading_is_dropped_at_the_write_deadline() { Ok(Ok(())) => panic!("socket buffers never filled"), Err(_) => panic!("session must close once its write stalls past the deadline"), } + assert!( + started.elapsed() < Duration::from_secs(2), + "stalled write took {:?}", + started.elapsed() + ); tp.shutdown().await; } /// The largest legitimate client frame (a `SubmitSolution` with a full /// `B064K` coinbase) keeps the session; a larger frame closes it. -#[tokio::test] -async fn oversized_client_frame_closes_the_session() { - let tc = padded_chain("sv2-frame-cap", 0); +async fn oversized_client_frame_closes_the_session(chain: Arc) { let tp = run_sv2_tp(Sv2TpConfig { listen: "127.0.0.1:0".parse().unwrap(), - chain: std::sync::Arc::clone(&tc.chain), + chain, authority_secret: [7; 32], cert_validity: Duration::from_secs(3600), stale_grace: Duration::from_secs(10), @@ -318,9 +332,7 @@ async fn oversized_client_frame_closes_the_session() { tp.shutdown().await; } -#[tokio::test] -async fn out_of_range_cert_validity_or_stale_grace_refuses_to_start() { - let tc = padded_chain("sv2-listener-range", 0); +async fn out_of_range_cert_validity_or_stale_grace_refuses_to_start(chain: Arc) { for (cert_validity, stale_grace) in [ (Duration::from_secs(u64::from(u32::MAX) + 1), Duration::ZERO), ( @@ -330,7 +342,7 @@ async fn out_of_range_cert_validity_or_stale_grace_refuses_to_start() { ] { let e = run_sv2_tp(Sv2TpConfig { listen: "127.0.0.1:0".parse().unwrap(), - chain: tc.chain.clone(), + chain: Arc::clone(&chain), authority_secret: [7; 32], cert_validity, stale_grace, diff --git a/crates/rbitcoin-sv2/src/template_tests.rs b/crates/rbitcoin-sv2/src/template_tests.rs index 51f0394b5..37bc7e4d4 100644 --- a/crates/rbitcoin-sv2/src/template_tests.rs +++ b/crates/rbitcoin-sv2/src/template_tests.rs @@ -1,4 +1,4 @@ -use crate::test_chain::{padded_chain, padded_chain_with, TestChain}; +use crate::test_chain::{copy_chain, padded_chain_with, shared_regtest, TestChain}; use crate::testutil::TpClient; use crate::{run_sv2_tp, Sv2TpConfig, SETUP_TIMEOUT, WRITE_TIMEOUT}; use bitcoin::consensus::encode::serialize; @@ -25,6 +25,23 @@ const MAX_BLOCK_WEIGHT: u64 = 4_000_000; const OP_TRUE: u8 = 0x51; const OP_CHECKSIG: u8 = 0xac; +fn mock_live_tip(tc: &TestChain) { + let tip_time = tc.chain.tip_header().expect("tip").time; + tc.chain.clock.set_mock(i64::from(tip_time)); +} + +/// Stale tip, two hubs from one pad: an empty pool and a heavy pool. Each +/// leaves IBD on its own, so the latch cannot be shared. +#[tokio::test(flavor = "multi_thread")] +async fn ibd_holds_the_last_budget_and_an_empty_template() { + let heavy = shared_regtest(1); + let empty = copy_chain(&heavy, "sv2-ibd-empty"); + tokio::join!( + sync_gate_holds_constraints_until_a_fresh_tip(&empty), + constraints_while_ibd_keep_the_last_budget(&heavy), + ); +} + fn spend(coinbase: Txid, fee: u64, script_pubkey: ScriptBuf) -> Transaction { Transaction { version: Version::TWO, @@ -156,9 +173,8 @@ async fn expect_prev_hash(c: &mut TpClient, tc: &TestChain, template_id: u64, ne #[tokio::test(flavor = "multi_thread")] async fn template_budget_fees_coinbase_and_merkle_path() { - let tc = padded_chain("sv2-template", 3); - // Tip time as "now": the padded chain is out of IBD for this test. - tc.chain.clock.set_mock(i64::from(tc.tip_time)); + let tc = shared_regtest(3); + mock_live_tip(&tc); let cheap = ScriptBuf::from_bytes(vec![OP_TRUE]); let a = spend(tc.coinbases[0], 3_000, cheap.clone()); let b = spend(tc.coinbases[1], 2_000, cheap); @@ -214,8 +230,8 @@ async fn template_budget_fees_coinbase_and_merkle_path() { #[tokio::test(flavor = "multi_thread")] async fn template_resent_constraints_do_not_rebuild() { - let tc = padded_chain("sv2-template-resent", 1); - tc.chain.clock.set_mock(i64::from(tc.tip_time)); + let tc = shared_regtest(1); + mock_live_tip(&tc); // 16_000 sigop cost: excluded under a u16::MAX client reserve. let heavy = spend( tc.coinbases[0], @@ -254,8 +270,8 @@ async fn template_resent_constraints_do_not_rebuild() { #[tokio::test(flavor = "multi_thread")] async fn template_constraint_rebuilds_are_rate_limited() { - let tc = padded_chain("sv2-template-rate", 1); - tc.chain.clock.set_mock(i64::from(tc.tip_time)); + let tc = shared_regtest(1); + mock_live_tip(&tc); // 16_000 sigop cost: excluded under a u16::MAX client reserve. let heavy = spend( tc.coinbases[0], @@ -295,9 +311,7 @@ async fn template_constraint_rebuilds_are_rate_limited() { tp.shutdown().await; } -#[tokio::test(flavor = "multi_thread")] -async fn sync_gate_holds_constraints_until_a_fresh_tip() { - let tc = padded_chain("sv2-gate", 0); +async fn sync_gate_holds_constraints_until_a_fresh_tip(tc: &TestChain) { let tp = run_sv2_tp(Sv2TpConfig { listen: "127.0.0.1:0".parse().unwrap(), chain: Arc::clone(&tc.chain), @@ -332,7 +346,7 @@ async fn sync_gate_holds_constraints_until_a_fresh_tip() { .expect("template after the fresh tip") .expect("message") }; - check_template(&mut c, &tc, first, &[], true).await; + check_template(&mut c, tc, first, &[], true).await; tp.shutdown().await; } @@ -340,9 +354,7 @@ async fn sync_gate_holds_constraints_until_a_fresh_tip() { /// A constraints flood while the node is in IBD must not close the session. /// Nothing is queued to build, so the post-template 8-replacement close does /// not apply. The tip that leaves IBD builds the last budget. -#[tokio::test(flavor = "multi_thread")] -async fn constraints_while_ibd_keep_the_last_budget() { - let tc = padded_chain("sv2-ibd-hold", 1); +async fn constraints_while_ibd_keep_the_last_budget(tc: &TestChain) { // 16_000 sigop cost: excluded under a u16::MAX client reserve. let heavy = spend( tc.coinbases[0], @@ -392,7 +404,7 @@ async fn constraints_while_ibd_keep_the_last_budget() { .expect("template after the fresh tip") .expect("message") }; - check_template(&mut c, &tc, first, &[], true).await; + check_template(&mut c, tc, first, &[], true).await; let extra = tokio::time::timeout(Duration::from_millis(500), c.recv()).await; assert!(extra.is_err(), "one template for the last budget"); @@ -411,7 +423,7 @@ struct FirstTemplate { } async fn first_template(tc: &TestChain) -> FirstTemplate { - tc.chain.clock.set_mock(i64::from(tc.tip_time)); + mock_live_tip(tc); let tp = run_sv2_tp(Sv2TpConfig { listen: "127.0.0.1:0".parse().unwrap(), chain: Arc::clone(&tc.chain), @@ -484,9 +496,9 @@ async fn first_template(tc: &TestChain) -> FirstTemplate { /// solution that meets it still becomes the tip. #[tokio::test(flavor = "multi_thread")] async fn submit_solution_checks_pow_before_accept() { + let tc = shared_regtest(0); use template_distribution_sv2::MESSAGE_TYPE_REQUEST_TRANSACTION_DATA_SUCCESS; - let tc = padded_chain("sv2-pow-precheck", 0); tc.chain.set_prefill_compact(true); let FirstTemplate { tp, @@ -527,6 +539,7 @@ async fn submit_solution_checks_pow_before_accept() { let solved = header.block_hash(); assert_eq!(tc.chain.tip_header().unwrap().block_hash(), solved); assert_eq!(tc.chain.cmpct_prefill_indexes(&solved), Some(vec![0])); + tc.chain.set_prefill_compact(false); tp.shutdown().await; } @@ -536,7 +549,7 @@ async fn submit_solution_checks_pow_before_accept() { /// with. The txid, and so the merkle root, does not cover the witness. #[tokio::test(flavor = "multi_thread")] async fn submit_solution_without_coinbase_witness_is_accepted() { - let tc = padded_chain("sv2-no-cb-witness", 0); + let tc = shared_regtest(0); let FirstTemplate { tp, mut c, @@ -573,7 +586,7 @@ async fn submit_solution_without_coinbase_witness_is_accepted() { /// stays bare: a filled witness without a commitment is not valid (BIP141). #[tokio::test(flavor = "multi_thread")] async fn submit_solution_without_witness_commitment_is_accepted() { - let tc = padded_chain("sv2-no-commitment", 0); + let tc = shared_regtest(0); let FirstTemplate { tp, mut c, @@ -615,7 +628,7 @@ async fn submit_solution_without_witness_commitment_is_accepted() { /// still consensus-valid (above MTP, under now + 2h), so it becomes the tip. #[tokio::test(flavor = "multi_thread")] async fn submit_solution_ahead_of_the_rolled_time_is_accepted() { - let tc = padded_chain("sv2-fast-clock", 0); + let tc = shared_regtest(0); let FirstTemplate { tp, mut c, @@ -643,8 +656,8 @@ async fn submit_solution_ahead_of_the_rolled_time_is_accepted() { #[tokio::test(flavor = "multi_thread")] async fn constraints_flood_closes_the_session() { - let tc = padded_chain("sv2-template-flood", 0); - tc.chain.clock.set_mock(i64::from(tc.tip_time)); + let tc = shared_regtest(0); + mock_live_tip(&tc); let tp = run_sv2_tp(Sv2TpConfig { listen: "127.0.0.1:0".parse().unwrap(), chain: Arc::clone(&tc.chain), @@ -687,8 +700,8 @@ async fn constraints_flood_closes_the_session() { #[tokio::test(flavor = "multi_thread")] async fn tip_event_rebuilds_a_template_built_on_its_prev_hash() { - let tc = padded_chain("sv2-template-tip-race", 1); - tc.chain.clock.set_mock(i64::from(tc.tip_time)); + let tc = shared_regtest(1); + mock_live_tip(&tc); let paid = spend( tc.coinbases[0], 10_000, @@ -719,8 +732,9 @@ async fn tip_event_rebuilds_a_template_built_on_its_prev_hash() { let tip = tc.chain.tip_header().expect("tip").block_hash(); let h = tc.chain.query.tip_height().expect("tip height").0; let script = ScriptBuf::from_bytes(vec![OP_TRUE]); - let x = mine_regtest_paying(tip, tc.tip_time + 1, h + 1, script, vec![paid.clone()]); - let y = mine_empty_regtest(x.block_hash(), tc.tip_time + 2, h + 2); + let now = tc.chain.tip_header().expect("tip").time; + let x = mine_regtest_paying(tip, now + 1, h + 1, script, vec![paid.clone()]); + let y = mine_empty_regtest(x.block_hash(), now + 2, h + 2); let loaded = tc .chain .confirm_wire_load_phase(&[(Height(h + 1), x), (Height(h + 2), y)]) diff --git a/crates/rbitcoin-sv2/src/test_chain.rs b/crates/rbitcoin-sv2/src/test_chain.rs index 8e43ab2de..c151c112a 100644 --- a/crates/rbitcoin-sv2/src/test_chain.rs +++ b/crates/rbitcoin-sv2/src/test_chain.rs @@ -3,6 +3,7 @@ use rbitcoin_consensus::{accept_and_connect_block, pad_empty_from, ChainParams, use rbitcoin_net::{ChainHub, MempoolHub}; use rbitcoin_primitives::Height; use rbitcoin_query::testutil::{tiny_query_labeled, TempDir}; +use std::path::Path; use std::sync::Arc; pub(crate) struct TestChain { @@ -49,3 +50,67 @@ pub(crate) fn padded_chain_with(label: &str, spendable: u32, params: ChainParams tip_time, } } + +/// A private hub copied from one process-wide regtest pad (3 spendable +/// coinbases). `spendable` is how many of those coinbases the chapter uses. +/// The pad is connected once; each caller gets a copy so parallel tests do +/// not share a tip or a mempool. +pub(crate) fn shared_regtest(spendable: u32) -> TestChain { + use std::sync::OnceLock; + assert!( + spendable <= 3, + "shared regtest pad has 3 spendable coinbases" + ); + static PAD: OnceLock = OnceLock::new(); + let src = PAD.get_or_init(|| { + let tc = padded_chain("sv2-shared", 3); + tc.chain.query.flush().expect("flush shared pad"); + tc + }); + // The pad is quiescent after init. Copies only read it. + copy_quiescent(src, "sv2-shared") +} + +/// Second hub on a copy of `src`'s store. The mempool is empty either way. +/// `ChainHub::in_ibd` latches per hub, so a chapter that leaves IBD cannot +/// share the hub with a chapter that must start stale. +pub(crate) fn copy_chain(src: &TestChain, label: &str) -> TestChain { + src.chain.query.flush().expect("flush before copy"); + copy_quiescent(src, label) +} + +fn copy_quiescent(src: &TestChain, label: &str) -> TestChain { + let dir = TempDir::labeled(label).expect("temp"); + copy_tree_except(src._dir.path(), dir.path(), "mempool"); + let q = rbitcoin_query::Query::open_or_create_tiny(dir.path()).expect("open copy"); + let chain = Arc::new(ChainHub::new(q, src.chain.params.clone(), Milestone::NONE)); + let mp = dir.path().join("mempool"); + std::fs::create_dir_all(&mp).unwrap(); + let mempool = MempoolHub::open(&mp, Arc::clone(&chain.query)).unwrap(); + mempool.set_relay_enabled(true); + assert!(chain.attach_mempool(Arc::clone(&mempool)).is_ok()); + TestChain { + _dir: dir, + chain, + mempool, + coinbases: src.coinbases.clone(), + tip_time: src.tip_time, + } +} + +fn copy_tree_except(src: &Path, dst: &Path, skip: &str) { + for ent in std::fs::read_dir(src).expect("read store") { + let ent = ent.expect("store entry"); + if ent.file_name() == skip { + continue; + } + let to = dst.join(ent.file_name()); + let ty = ent.file_type().expect("file type"); + if ty.is_dir() { + std::fs::create_dir_all(&to).unwrap(); + copy_tree_except(&ent.path(), &to, skip); + } else { + std::fs::copy(ent.path(), &to).unwrap(); + } + } +}