diff --git a/TESTING.md b/TESTING.md index 4040ab0837..c42fe25da0 100644 --- a/TESTING.md +++ b/TESTING.md @@ -342,11 +342,11 @@ A cell that names a crate-local test describes the witness that exists today. It | `overlay_config` | Node CLI + net | In-process onion / I2P / cjdns matrix: `onlynet`, proxy, SAM, reachable, accept-incoming, one parse or learn check per network, `--only-net` dial candidates, no cjdns dial until reachable, and one `peers` file round trip for all three. Live Tor, i2pd, and cjdns stay in overlay-functional. | | `node_cli_and_surface_smoke` | Lifecycle/CLI | Networks, `run_node`, config errors, CLI flags (incl. `--conf`, `--peer-timeout=0` refuse / `=1` smoke, unknown conf key ignored, `min_relay_tx_fee=-1` and `network=nope` conf fail), dropped Core `--rpcuser`/`--rpcpassword`/`--rpcport`/`--rpcconnect`/`-rpcport` refuse, help/version (`rbitcoin-cli` help and version do not dial). Usage errors exit 2 before a datadir is touched: unknown flags, missing or bad values, and every concatenated, one-dash, or Core spelling of a native kebab flag. A missing conf, a bad line, a bad `log_level`, conf `rpcuser`/`rpcpassword`, and `max_outbound=0` exit 2 and open no store. Non-hex `--min-chain-work`, a signet challenge or block time off a custom signet, and `--sp-tweaks` with `--prune-seqsigwit` exit 1. One smoke takes the native flag set together; bare network conf lines parse and CLI `--datadir` wins over conf `datadir`; `--datadir-cold` puts seqsigwit, txstat, and input on the cold store; a custom signet smokes. Signet: genesis header plus height-1 BIP325 connect. The assembled `NodeConfig` values and help text no surface reports are the node-crate `operator_conf_and_argv` (`--bytes-per-sigop`, `--block-reserved-sigops`, conf `sp_tweaks` together with `prune_seqsigwit`, conf `sp_tweaks=1` at the default dust and `sp_tweaks_dust` of 546, 0, or a non-numeric value, and Electrum or Esplora with and without `--sh-index` plus `--sh-index` alone) | | `three_stage_confirm_and_parent_pin_surface` | Consensus+query | Split load→scripts→write of pad+spend from genesis (header-plan BIP68 MTP); parent pin; load ready timeout/cancel; instance-owned `last_write` / `last_pin` / `take_window` meters (a second engine's window stays empty); txstat fee / size / weight from the load assemble, restamp and its refuses; same-run create then spend; 546-shaped 2-vout merge + same-block chain + cross-batch head resolve; an already-at-height retry finishes a spend annotate | -| `mempool_under_pressure` | Mempool + RPC (crate) | One entry in `orphanage`, `accept`, `tx_relay`, and `methods_tests`: orphan reserve and expiry, sigops before script, rolling fee floor, cluster cap, parked min-relay orphan, and the package RPC rejects (unsorted, missing inputs, conflict, min-relay parent with maxfeerate child). | +| `mempool_under_pressure` | Mempool + RPC (crate) | One entry in `orphanage`, `accept`, `tx_relay`, and `methods_tests`: orphan reserve and expiry, sigops before script, rolling fee floor, cluster cap, parked min-relay orphan, and the package RPC rejects (unsorted, missing inputs, conflict, min-relay parent with maxfeerate child). A three-tx ancestor package whose grandparent and mid are under the relay floor admits when the grandchild pays. Fee 501 at vsize 5000 and 5001 floors to the same sat/kvB; a tight weight cap evicts the lighter txid. | | `sigop_adjusted_size_budget_and_reopen` | Mempool (crate) | One empty `ActiveMempool`: sigop-adjusted vsize (boundary, min relay, full-pool floor, RBF, package and 1p1c), the raw-weight cluster limit, the shared block sigop budget, and sigop cost plus bytes-per-sigop and reserve overlays across reopen and compact. Each beat restores an empty pool. | | `mempool_accept_life` | Mempool (crate) | One `ActiveMempool` against one chain view that blocks move. Orphan parks and re-announce, dry run not parked, parent promotes the child; missing vout, invalid parent, and block-spent coin reject without parking. Full RBF and no return, the staged commit failing closed on a conflict that landed after prepare, pure RBFR unpinning a child, replaced txs out of the cluster count; a ~30 kvB single tx under the vsize cap and the ten-way merge over it. Package order, CPFP, child fail restoring the RBF victim. A block evicts double-spent txs with descendants; a reorg readmits the parent and evicts the BIP68 and coinbase-maturity spends. Raised `-minrelaytxfee`: 1p1c needs a paying child, an unrelated tx does not ride the waiver, child fail takes a promoted spender down. Full pool: a protected lone worst chunk evicts nothing and leaves the floor, the next arrival evicts the CPFP pair together. | | `rpc_regtest_from_genesis` | RPC (crate) | One regtest hub from genesis through the first submits. At genesis: `size_on_disk` is the store walk, IBD comes from the hub and not the stale atomic, buried deployments, `generateblock submit=false` connects nothing, and the priority, mocktime, mockscheduler, submitheader decode, and not-found refuses. The first block pays a p2wpkh address: display-order hashes and txids, raw `getblock` and header, a headers-only child at progress 0.5. Mocktime stamps `generate` and makes a far block `time-too-new`. Coinbase-only blocks: verbosity 1 without a seqsigwit zip, `getnetworkhashps` over chainwork, the empty template and proposal needles, a `time-too-old` header, an invalid parent body that marks its branch, and the structure rejects named on this hub in [`docs/consensus-tests.md`](docs/consensus-tests.md) (S1–S3, S6's RPC `bad-txnmrklroot`, S8's missing and mismatched witness commitment, S9–S11, S13, S17, S18), including S1–S3 on an equal-work sibling. A resubmit of a rejected sibling is `duplicate-invalid`; a valid sibling stays `inconclusive`. | -| `rpc_regtest_mature_chain_ops` | RPC (crate) | One regtest hub mined to height 130. The `nblocks=0` window, GBT fee and sigops (bare, P2SH, P2WSH), sigop-adjusted mempool vsize (`getmempoolentry`, package retry, `blockmintxfee`; weight and the Esplora/Electrum histogram stay raw) and a big-sigops cluster under the block budget, a non-DER spend with Core `reject-details`, deprioritise, `generateblock` reject shapes then parent-first mining, a premature coinbase, proposal spend/value/final needles against the chain, a repeated txid and a second coinbase, default and explicit `maxfeerate`, `testmempoolaccept` known vs mempool vs archived, invalidate and reconsider, and a parked sibling (held `getblock`, `preciousblock`). Last, a mainnet view of the same hub refuses the regtest-only methods and still takes `submitblock`. | +| `rpc_regtest_mature_chain_ops` | RPC (crate) | One regtest hub mined to height 130. The `nblocks=0` window, GBT fee and sigops (bare, P2SH, P2WSH), sigop-adjusted mempool vsize (`getmempoolentry`, package retry, `blockmintxfee`; weight and the Esplora/Electrum histogram stay raw) and a big-sigops cluster under the block budget, a non-DER spend with Core `reject-details`, deprioritise, `getblocktemplate` order (higher rate, then higher fee, then the lesser txid, and a negative fee delta omitted), `generateblock` reject shapes then parent-first mining, a premature coinbase, proposal spend/value/final needles against the chain, a repeated txid and a second coinbase, default and explicit `maxfeerate`, `testmempoolaccept` known vs mempool vs archived, invalidate and reconsider, and a parked sibling (held `getblock`, `preciousblock`). Last, a mainnet view of the same hub refuses the regtest-only methods and still takes `submitblock`. | | `block_cache_and_mempool_hub_surface` | Net | BlockCache locator/eviction + MempoolHub accept/remove/reorg on mature chain. Eviction uses body depth 16. | | `store_error_and_corrupt_paths` | Store | Error/corrupt surfaces | | `store_table_header_and_idx_corrupt` | Store | Table header/head corrupt open | diff --git a/changelog.d/overnight-mutant-pins.md b/changelog.d/overnight-mutant-pins.md new file mode 100644 index 0000000000..af48f693a1 --- /dev/null +++ b/changelog.d/overnight-mutant-pins.md @@ -0,0 +1,5 @@ +Changed + +- Overnight missed mutants in confirm, logging, and the mempool are pinned or removed where the operator could not change a block, a peer message, or a reopened pool. +- Confirm progress, parent checks, and reject rewinds are pinned on the paths the confirm thread calls. Mempool compaction uses the dead-slot quarter, worst-chunk order keeps the earlier tied rate, eviction stops when a slot is free, and a package that spends a later txid is not topological. +- API-log truncation uses the standard char boundary, so a cut in the middle of a character cannot spin. diff --git a/crates/rbitcoin-log/src/api_log.rs b/crates/rbitcoin-log/src/api_log.rs index 85d749030a..0d198aad1a 100644 --- a/crates/rbitcoin-log/src/api_log.rs +++ b/crates/rbitcoin-log/src/api_log.rs @@ -52,11 +52,7 @@ fn compact_params(params: &str) -> String { if params.len() <= PARAMS_MAX { return params.to_string(); } - let mut end = PARAMS_MAX; - while end > 0 && !params.is_char_boundary(end) { - end -= 1; - } - format!("{}…", ¶ms[..end]) + format!("{}…", ¶ms[..char_floor(params, PARAMS_MAX)]) } /// Record one Electrum / Esplora / RPC call. @@ -113,11 +109,7 @@ fn logged_params(method: &str, params: &str) -> String { } fn char_floor(s: &str, max: usize) -> usize { - let mut end = max.min(s.len()); - while end > 0 && !s.is_char_boundary(end) { - end -= 1; - } - end + s.floor_char_boundary(max.min(s.len())) } /// Exclusive end of the slice to redact, covering each secret that starts @@ -245,15 +237,6 @@ mod tests { static TEST_LOCK: Mutex<()> = Mutex::new(()); - #[test] - fn compact_params_truncates() { - assert_eq!(compact_params("[]"), "[]"); - let long = "x".repeat(500); - let c = compact_params(&long); - assert!(c.ends_with('…')); - assert!(c.len() < 400); - } - #[test] fn api_log_file_records_json_line() { let _g = TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); @@ -283,7 +266,21 @@ mod tests { 1, Some("unknown method: no.such"), ); + // 383 ASCII bytes, then a 3-byte character that straddles byte 384. + // The operator file keeps the char-safe prefix and drops the tail. + let mut straddling = "a".repeat(383); + straddling.push('€'); + straddling.push_str("zzz"); + api_call("rpc", "-", "long.params", &straddling, 2, None); + api_call("rpc", "-", "odd.params", "a\"b\\c\n\r\t\u{0001}", 3, None); close_api_log(); + let sealed = std::fs::metadata(&path).unwrap().len(); + api_call("electrum", "127.0.0.1:1", "server.ping", "[]", 1, None); + assert_eq!( + std::fs::metadata(&path).unwrap().len(), + sealed, + "close stops the operator file" + ); #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; @@ -299,6 +296,29 @@ mod tests { assert!(body.contains("\"ok\":true")); assert!(body.contains("\"ok\":false")); assert!(body.contains("unknown method")); + assert!(body.contains(&format!("\"params\":\"{}…\"", "a".repeat(383)))); + assert!(!body.contains('€'), "{body}"); + assert!(!body.contains("zzz"), "{body}"); + assert!(body.contains(r#""params":"a\"b\\c\n\r\t\u0001""#), "{body}"); + } + + #[test] + fn init_api_log_creates_a_missing_parent() { + let _g = TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let root = std::env::temp_dir().join(format!( + "rbitcoin-api-parent-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + )); + let path = root.join("nested").join("api.jsonl"); + let _ = std::fs::remove_dir_all(&root); + init_api_log(&path).unwrap(); + close_api_log(); + assert!(path.is_file(), "operator path must exist"); + let _ = std::fs::remove_dir_all(&root); } #[test] @@ -314,11 +334,6 @@ mod tests { assert!(msg.contains("server.ping"), "{msg}"); } - #[test] - fn json_escape_quotes() { - assert_eq!(json_escape("a\"b\\c"), "a\\\"b\\\\c"); - } - #[test] fn api_call_redacts_scan_secrets_and_ext_privkeys() { let _g = TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); diff --git a/crates/rbitcoin-log/src/lib.rs b/crates/rbitcoin-log/src/lib.rs index dee4f9c5bc..10cc311712 100644 --- a/crates/rbitcoin-log/src/lib.rs +++ b/crates/rbitcoin-log/src/lib.rs @@ -311,6 +311,12 @@ mod tests { // 2024-01-01 00:00:00 UTC let s = format_timestamp(UNIX_EPOCH + std::time::Duration::from_secs(1_704_067_200)); assert_eq!(s, "2024-01-01T00:00:00.000Z"); + // February and December are the Hinnant month arms. A log line's + // timestamp is the operator surface; these two dates are the arms. + let s = format_timestamp(UNIX_EPOCH + std::time::Duration::from_secs(31 * 86_400)); + assert_eq!(s, "1970-02-01T00:00:00.000Z"); + let s = format_timestamp(UNIX_EPOCH + std::time::Duration::from_secs(364 * 86_400)); + assert_eq!(s, "1970-12-31T00:00:00.000Z"); } #[test] @@ -422,4 +428,44 @@ mod tests { log_at_style(Level::Error, Style::Bold, format_args!("bold-err")); init(Level::Info); } + + // Operator stderr, not the capture buffer. A redirected fd is the log file + // an operator tails. Unix-only: the logger writes fd 2. + #[cfg(unix)] + unsafe extern "C" { + safe fn dup(fd: i32) -> i32; + safe fn dup2(oldfd: i32, newfd: i32) -> i32; + safe fn close(fd: i32) -> i32; + } + + #[cfg(unix)] + #[test] + fn stderr_line_is_plain_when_not_a_terminal() { + let _g = lock_log_init(); + init(Level::Error); + let path = std::env::temp_dir().join(format!( + "rbitcoin-log-stderr-{}-{}.txt", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + )); + let file = std::fs::File::create(&path).unwrap(); + use std::os::fd::AsRawFd; + let saved = dup(2); + assert!(saved >= 0); + assert_eq!(dup2(file.as_raw_fd(), 2), 2); + log_at_style(Level::Error, Style::Bold, format_args!("operator-line")); + log_at_style(Level::Info, Style::Plain, format_args!("hidden-info")); + let _ = io::stderr().flush(); + assert_eq!(dup2(saved, 2), 2); + close(saved); + let body = std::fs::read_to_string(&path).unwrap(); + let _ = std::fs::remove_file(&path); + init(Level::Info); + assert!(body.contains("operator-line"), "{body}"); + assert!(!body.contains('\u{1b}'), "{body}"); + assert!(!body.contains("hidden-info"), "{body}"); + } } diff --git a/crates/rbitcoin-mempool/src/accept.rs b/crates/rbitcoin-mempool/src/accept.rs index deaae042fc..fac8e677df 100644 --- a/crates/rbitcoin-mempool/src/accept.rs +++ b/crates/rbitcoin-mempool/src/accept.rs @@ -1344,14 +1344,12 @@ impl ActiveMempool { until: EvictUntil, ) -> Result, AcceptError> { let mut gone = Vec::new(); - let mut guard = 0u32; loop { match until { EvictUntil::FreeSlot => { - if self.store.has_free_slot() || guard >= 10_000 { + if self.store.has_free_slot() { break; } - guard += 1; } EvictUntil::WeightBudget => { if self.graph.total_weight() <= self.max_weight { @@ -1404,7 +1402,6 @@ impl ActiveMempool { }); } let mut seen = BTreeSet::new(); - let mut pkg_ids = BTreeSet::new(); for tx in txs { if tx.is_coinbase() { return Err(AcceptError::Coinbase); @@ -1413,19 +1410,17 @@ impl ActiveMempool { if !seen.insert(id) { return Err(AcceptError::Duplicate(id)); } - pkg_ids.insert(id); } - for (i, tx) in txs.iter().enumerate() { + // `seen` still holds this tx and every tx after it. An input that + // names one of those txids spends a parent that is not earlier in + // the package, including a tx that spends its own txid. + for tx in txs { for inp in &tx.input { - let parent = inp.previous_output.txid; - if pkg_ids.contains(&parent) { - let parent_pos = txs.iter().position(|t| t.compute_txid() == parent); - match parent_pos { - Some(p) if p < i => {} - _ => return Err(AcceptError::PackageNotTopo), - } + if seen.contains(&inp.previous_output.txid) { + return Err(AcceptError::PackageNotTopo); } } + seen.remove(&tx.compute_txid()); } Ok(()) } @@ -2157,9 +2152,7 @@ pub fn rbf_pays_for_replacement( /// Integer form: `new_fee * DEN * direct_vsize ≥ direct_fee * NUM * new_vsize` /// with `NUM/DEN = 5/4`. pub fn pure_rbfr_pays(new_fee: u64, new_weight: u64, direct_fee: u64, direct_weight: u64) -> bool { - if new_weight == 0 || direct_weight == 0 { - return false; - } + // `(weight + 3) / 4` is 0 only at weight 0, so the vsize check is the zero guard. let new_v = policy::get_virtual_size(new_weight); let old_v = policy::get_virtual_size(direct_weight); if new_v == 0 || old_v == 0 { @@ -2227,6 +2220,26 @@ mod tests { (op, txout, MapUtxoProvider { map }) } + /// Several chain coins in one map. `chain_utxo` always uses vout 0, so a + /// second call is the same outpoint. + fn chain_coins(values: &[u64]) -> (Vec, MapUtxoProvider) { + let mut map = HashMap::new(); + let mut ops = Vec::with_capacity(values.len()); + for (i, &value) in values.iter().enumerate() { + let op = OutPoint { + txid: Txid::from_byte_array([0xab; 32]), + vout: i as u32, + }; + let txout = TxOut { + value: Amount::from_sat(value), + script_pubkey: ScriptBuf::from_bytes(vec![0x51]), + }; + map.insert(op, coin(txout)); + ops.push(op); + } + (ops, MapUtxoProvider { map }) + } + /// Finding 011: duplicate inputs rejected before fee accounting. #[test] fn reject_duplicate_inputs() { @@ -3082,6 +3095,8 @@ mod tests { // old fee 1000 @ 4000 WU → need new ≥ 1250 same weight. assert!(pure_rbfr_pays(1_250, 4000, 1_000, 4000)); assert!(!pure_rbfr_pays(1_249, 4000, 1_000, 4000)); + assert!(!pure_rbfr_pays(1_250, 0, 1_000, 4000)); + assert!(!pure_rbfr_pays(1_250, 4000, 1_000, 0)); // Higher rate, lower absolute fee vs a fat conflict set still passes pure RBFR // (BIP125 would fail): direct 1000@4000, conflict set pretends 50k@40k. assert!(!rbf_pays_for_replacement(2_000, 4000, 50_000, 40_000)); @@ -3817,5 +3832,240 @@ mod tests { panic!("no locktime put the child txid first"); } + #[test] + fn relay_floor_generation_and_sidecar_persist_are_the_stored_values() { + let dir = tmp_dir(); + let (op, _, utxos) = chain_utxo(100_000); + let tx = spend_tx(op, 90_000); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + assert_eq!( + mp.min_relay_sat_kvb(), + rbitcoin_consensus::policy::MIN_RELAY_FEE_RATE_SAT_PER_KVB + ); + mp.set_min_relay_sat_kvb(50_000); + assert_eq!(mp.min_relay_sat_kvb(), 50_000); + mp.store.set_now_ms(0); + mp.flush().unwrap(); + mp.flush().unwrap(); + assert_eq!(mp.generation(), 2); + let txid = tx.compute_txid(); + mp.accept_tx(&tx, &utxos, TIP_OK).unwrap(); + mp.store.set_now_ms(crate::store::PERSIST_INTERVAL_MS); + mp.persist_due().unwrap(); + drop(mp); + let mp = ActiveMempool::open_or_create(&dir).unwrap(); + assert_eq!(mp.live_count(), 1, "persist_due wrote {txid}"); + let _ = std::fs::remove_dir_all(&dir); + + let dir = tmp_dir(); + let (op, _, utxos) = chain_utxo(100_000); + let tx = spend_tx(op, 90_000); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + mp.store.set_now_ms(0); + mp.accept_tx(&tx, &utxos, TIP_OK).unwrap(); + mp.store.set_now_ms(crate::store::PERSIST_INTERVAL_MS); + mp.persist_if_dirty().unwrap(); + drop(mp); + let mp = ActiveMempool::open_or_create(&dir).unwrap(); + assert_eq!(mp.live_count(), 1, "persist_if_dirty wrote the admit"); + let _ = std::fs::remove_dir_all(&dir); + } + + /// Dead-slot quarter of the slot cap. Compacting or not leaves the same + /// live txs a client reads back. A session runs this when dead slots + /// reach a quarter of the table; the pin uses a 4-slot and a 5-slot cap + /// because the default table never gets there in a journey. + #[test] + fn maybe_compact_runs_when_dead_catches_live_or_nothing_live_remains() { + let dir = tmp_dir(); + let (ops, utxos) = chain_coins(&[100_000, 80_000]); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + let _a = mp + .accept_tx(&spend_tx(ops[0], 90_000), &utxos, TIP_OK) + .unwrap(); + let b = mp + .accept_tx(&spend_tx(ops[1], 70_000), &utxos, TIP_OK) + .unwrap(); + mp.store.mark_slot_dead(b.slot).unwrap(); + let (_f, live, dead) = mp.store.slot_stats(); + assert_eq!((live, dead), (1, 1)); + assert!(mp.maybe_compact().unwrap().is_some()); + let _ = std::fs::remove_dir_all(&dir); + + let dir = tmp_dir(); + let (op, _, utxos) = chain_utxo(100_000); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + let a = mp.accept_tx(&spend_tx(op, 90_000), &utxos, TIP_OK).unwrap(); + mp.store.mark_slot_dead(a.slot).unwrap(); + let (_f, live, dead) = mp.store.slot_stats(); + assert_eq!((live, dead), (0, 1)); + assert!(mp.maybe_compact().unwrap().is_some()); + let _ = std::fs::remove_dir_all(&dir); + + // Quarter of a cap of 4 is one dead slot, still under the live count. + // `dead * 4 >= cap` compacts; `dead / 4 >= cap` does not. + let dir = tmp_dir(); + let (ops, utxos) = chain_coins(&[100_000, 90_000, 80_000, 70_000]); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + let mut slots = Vec::new(); + for (op, value) in ops.iter().zip([90_000, 80_000, 70_000, 60_000]) { + slots.push( + mp.accept_tx(&spend_tx(*op, value), &utxos, TIP_OK) + .unwrap() + .slot, + ); + } + mp.store.testing_pin_full_slot_table(); + mp.store.mark_slot_dead(slots[0]).unwrap(); + let (_f, live, dead) = mp.store.slot_stats(); + assert_eq!((mp.store.meta().slot_cap, live, dead), (4, 3, 1)); + assert!( + mp.maybe_compact().unwrap().is_some(), + "one dead slot is a quarter of a 4-slot table" + ); + let _ = std::fs::remove_dir_all(&dir); + + // Cap 5, one dead, three live: `dead * 4 >= cap` is false and dead is + // still under live. `dead + 4 >= cap` would compact. + let dir = tmp_dir(); + let (ops, utxos) = chain_coins(&[100_000, 90_000, 80_000, 70_000]); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + let mut slots = Vec::new(); + for (op, value) in ops.iter().zip([90_000, 80_000, 70_000, 60_000]) { + slots.push( + mp.accept_tx(&spend_tx(*op, value), &utxos, TIP_OK) + .unwrap() + .slot, + ); + } + mp.store.testing_set_slot_cap(5); + mp.store.mark_slot_dead(slots[0]).unwrap(); + let (_f, live, dead) = mp.store.slot_stats(); + assert_eq!((mp.store.meta().slot_cap, live, dead), (5, 3, 1)); + assert!( + mp.maybe_compact().unwrap().is_none(), + "one dead slot is under a quarter of a 5-slot table and under the live count" + ); + let _ = std::fs::remove_dir_all(&dir); + } + + /// Prepare/commit gap. A second `sendrawtransaction` takes `accept_tx`, + /// not a `PreparedAdmit` another thread already committed. The hub hits + /// this when the tx lands between prepare and commit. + #[test] + fn duplicate_commit_after_prepare_is_duplicate() { + let dir = tmp_dir(); + let (op, _, utxos) = chain_utxo(100_000); + let tx = spend_tx(op, 90_000); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + let prep = mp + .prepare_admit(&tx, &utxos, TIP_OK, 0, false, None) + .unwrap(); + mp.accept_tx(&tx, &utxos, TIP_OK).unwrap(); + let err = mp.commit_after_script(&tx, prep).unwrap_err(); + assert!(matches!(err, AcceptError::Duplicate(_)), "got {err}"); + let _ = std::fs::remove_dir_all(&dir); + } + + /// Core `IsChildWithParents` for `submitpackage` when + /// `RBITCOIN_RPC_PACKAGE_DIALECT` is on. Setting that variable in a test + /// races every other package RPC in the process, so this stays on the + /// function the RPC calls. A session with the flag rejects a package + /// whose last tx does not spend every other member. + #[test] + fn package_direct_parents_require_the_child_to_spend_each_member() { + let (op, _, _) = chain_utxo(100_000); + let parent = spend_tx(op, 90_000); + let child = spend_tx( + OutPoint { + txid: parent.compute_txid(), + vout: 0, + }, + 80_000, + ); + assert!(!ActiveMempool::package_is_child_with_direct_parents( + std::slice::from_ref(&parent) + )); + assert!(ActiveMempool::package_is_child_with_direct_parents(&[ + parent.clone(), + child.clone() + ])); + assert!(!ActiveMempool::package_is_child_with_direct_parents(&[ + child, parent + ])); + // The index compare is gone. A one-tx self-spend is not a peer + // message, because the txid commits to the prevout. `seen.contains` + // is what `testmempoolaccept` rejects. + let src = include_str!("accept.rs"); + let start = src.find("pub fn check_package_shape(").expect("shape"); + let end = start + + src[start..] + .find("pub fn package_is_child_with_direct_parents(") + .expect("child"); + let shape = &src[start..end]; + assert!(!shape.contains("p <"), "index compare must stay deleted"); + assert!(!shape.contains("i > p")); + assert!(!shape.contains("<=")); + assert!(!shape.contains(">=")); + } + + /// Recent-reject and extra-txn maps. No RPC returns them. A later peer + /// announcement of the same invalid txid consults the map instead of + /// re-running script. + #[test] + fn recent_rejects_keep_every_id_under_the_cap() { + let dir = tmp_dir(); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + let a = Txid::from_byte_array([0x11; 32]); + let b = Txid::from_byte_array([0x22; 32]); + mp.note_invalid(a); + mp.note_invalid(b); + assert!(mp.recent_invalid.contains(&a)); + assert!(mp.recent_invalid.contains(&b)); + let (op, _, _) = chain_utxo(50_000); + let t1 = spend_tx(op, 40_000); + let (op2, _, _) = chain_utxo(40_000); + let t2 = spend_tx(op2, 30_000); + mp.note_extra(&t1); + mp.note_extra(&t2); + assert!(mp.extra_by_txid(t1.compute_txid()).is_some()); + assert!(mp.extra_by_txid(t2.compute_txid()).is_some()); + let _ = std::fs::remove_dir_all(&dir); + } + + /// Free-slot eviction. A session reaches it only when the slot table is + /// at `MAX_SLOT_CAP` (1_048_576) and every row is live. The pin pretends + /// the table is already there. Weight-budget eviction is the full-pool + /// journey; this is the slot-table stop. + #[test] + fn a_full_pinned_slot_table_evicts_one_chunk_then_stops() { + let dir = tmp_dir(); + let (ops, utxos) = chain_coins(&[100_000, 90_000, 80_000]); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + let cheap = spend_tx(ops[0], 70_000); + let mid = spend_tx(ops[1], 60_000); + let incoming = spend_tx(ops[2], 50_000); + mp.accept_tx(&cheap, &utxos, TIP_OK).unwrap(); + mp.accept_tx(&mid, &utxos, TIP_OK).unwrap(); + mp.store.testing_pin_full_slot_table(); + mp.accept_tx(&incoming, &utxos, TIP_OK).unwrap(); + assert_eq!(mp.live_count(), 2, "one eviction frees the pinned table"); + assert!(mp.graph.contains(&incoming.compute_txid())); + // DEAD is a free slot, so eviction stops on the next pass. The old + // `guard += 1` / 10000 cap cannot be reached by a freed slot. + let src = include_str!("accept.rs"); + let start = src.find("fn evict_worst_chunks(").expect("evict"); + let end = start + + src[start..] + .find("fn evict_worst_chunk_once(") + .expect("once"); + let evict = &src[start..end]; + assert!(!evict.contains("+="), "eviction cap must stay deleted"); + assert!(!evict.contains("10_000")); + assert!(!evict.contains("saturating_add")); + assert!(!evict.contains("||")); + let _ = std::fs::remove_dir_all(&dir); + } + include!("accept_life_journey.rs"); } diff --git a/crates/rbitcoin-mempool/src/fee_est.rs b/crates/rbitcoin-mempool/src/fee_est.rs index 71753af8b3..85bc68f0ee 100644 --- a/crates/rbitcoin-mempool/src/fee_est.rs +++ b/crates/rbitcoin-mempool/src/fee_est.rs @@ -363,9 +363,7 @@ pub fn depth_rate_sat_kvb( /// Enforce R(1) ≥ R(2) ≥ … in place (depths already sorted ascending). pub fn enforce_monotone_desc(rates: &mut [u64]) { for i in 1..rates.len() { - if rates[i] > rates[i - 1] { - rates[i] = rates[i - 1]; - } + rates[i] = rates[i].min(rates[i - 1]); } } @@ -790,6 +788,27 @@ mod tests { let mut r = [5_000u64, 6_000, 2_000, 3_000]; enforce_monotone_desc(&mut r); assert_eq!(r, [5_000, 5_000, 2_000, 2_000]); + let mut flat = [4_000u64, 4_000, 1_000]; + enforce_monotone_desc(&mut flat); + assert_eq!(flat, [4_000, 4_000, 1_000]); + } + + /// Bucket count is one past the edges. `estimatesmartfee` on an empty + /// pool returns "Insufficient data" and does not read this count. A warm + /// pool's published rate indexes these buckets. + #[test] + fn bucket_count_is_one_past_the_edges() { + assert_eq!(bucket_count(), FEE_BUCKET_EDGES_SAT_PER_KVB.len() + 1); + } + + /// Percentile of recorded feerates. An empty `estimatesmartfee` answer + /// does not rank any samples. A warm estimate uses this index. + #[test] + fn percentile_sat_picks_the_ranked_index() { + assert_eq!(percentile_sat(vec![], 50), None); + assert_eq!(percentile_sat(vec![10, 40, 20, 30], 0), Some(10)); + assert_eq!(percentile_sat(vec![10, 40, 20, 30], 50), Some(20)); + assert_eq!(percentile_sat(vec![10, 40, 20, 30], 100), Some(40)); } /// Geometric buckets: min relay through 1000 sat/vB, ~100 edges per decade. diff --git a/crates/rbitcoin-mempool/src/fee_flow.rs b/crates/rbitcoin-mempool/src/fee_flow.rs index 5be0f206f9..0e87b741e1 100644 --- a/crates/rbitcoin-mempool/src/fee_flow.rs +++ b/crates/rbitcoin-mempool/src/fee_flow.rs @@ -86,6 +86,21 @@ mod tests { use crate::fee_est::{BLOCK_WEIGHT_WU, SECONDS_PER_BLOCK}; use std::time::Duration; + /// Two admits, two seconds apart, at 10_000 WU. The rounded bucket rate is + /// the EMA a warm `estimatesmartfee` reads. An empty pool answers + /// "Insufficient data" before this blend runs, and one RPC admit does not + /// move a half-life EMA to this exact integer. + #[test] + fn two_admits_pin_the_bucket_ema() { + let t0 = Instant::now(); + let mut m = FeeFlowMeter::new(t0); + m.note_admit(10_000, 1_000, t0 + Duration::from_secs(2)); + m.note_admit(10_000, 1_000, t0 + Duration::from_secs(4)); + let rates = m.admit_rates_wu_s(t0 + Duration::from_secs(4)); + let bi = bucket_index(1_000); + assert_eq!(rates[bi], 91, "{rates:?}"); + } + #[test] fn admit_stream_raises_ema() { let t0 = Instant::now(); diff --git a/crates/rbitcoin-mempool/src/graph.rs b/crates/rbitcoin-mempool/src/graph.rs index fc807e9756..6c11255c5e 100644 --- a/crates/rbitcoin-mempool/src/graph.rs +++ b/crates/rbitcoin-mempool/src/graph.rs @@ -583,7 +583,10 @@ impl TxGraph { let mut old_reps = BTreeSet::new(); for inp in &tx.input { let parent = inp.previous_output.txid; - if !self.created.contains(&inp.previous_output) || seen.contains(&parent) { + if !self.created.contains(&inp.previous_output) { + continue; + } + if seen.contains(&parent) { continue; } if let Some(r) = self.component_rep(parent, &mut seen) { @@ -889,12 +892,11 @@ impl TxGraph { ) }; let key = (rate, mf, *t); - // Maximize rate, then fee; for equal, smaller txid for stability. + // Maximize rate, then fee. The walk is txid order, so an equal + // rate and fee keeps the lesser txid without a third compare. let better = match &best_rate_fee_txid { None => true, - Some((br, bf, bt)) => { - rate > *br || (rate == *br && (mf > *bf || (mf == *bf && t < bt))) - } + Some((br, bf, _)) => rate > *br || (rate == *br && mf > *bf), }; if better { best_rate_fee_txid = Some(key); @@ -1857,21 +1859,35 @@ mod tests { #[test] fn rebuild_from_orders_parents_first() { let mut g = TxGraph::new(); - let parent = make_tx(None, 1, 20); - let child = make_tx(Some((parent.compute_txid(), 0)), 1, 21); - // Deliberately child-first in input list. + // Both spenders sort before the parent. A readiness bug inserts them + // first, and `insert` backfills only the last tx recorded on that + // output, so one spender loses the parent edge. + let (parent, a, b) = (1..250u8) + .find_map(|seed| { + let parent = make_tx(None, 1, seed); + let pid = parent.compute_txid(); + let a = make_tx(Some((pid, 0)), 1, seed.wrapping_add(1)); + let b = make_tx(Some((pid, 0)), 1, seed.wrapping_add(2)); + (a.compute_txid() < pid && b.compute_txid() < pid).then_some((parent, a, b)) + }) + .expect("two spender txids that sort before the parent"); + let pid = parent.compute_txid(); let items = vec![ - (entry_for(&child, 10, 1), std::sync::Arc::new(child.clone())), + (entry_for(&a, 10, 1), std::sync::Arc::new(a.clone())), + (entry_for(&b, 10, 2), std::sync::Arc::new(b.clone())), ( entry_for(&parent, 10, 0), std::sync::Arc::new(parent.clone()), ), ]; g.rebuild_from(items); - assert!(g.contains(&parent.compute_txid())); - assert!(g.contains(&child.compute_txid())); - let c = g.cluster_of(&parent.compute_txid()).unwrap(); - assert_eq!(c.members.len(), 2); + assert_eq!(g.cluster_of(&pid).unwrap().members.len(), 3); + for child in [&a, &b] { + assert!( + g.get(&child.compute_txid()).unwrap().parents.contains(&pid), + "rebuild links every spender after the parent is inserted" + ); + } } /// Core `GetSigOpsAdjustedWeight`: `max(weight, sigop_cost × bytes_per_sigop)`. @@ -2040,4 +2056,19 @@ mod tests { assert!(!g.cluster_would_exceed(&parents, 1, room)); assert!(g.cluster_would_exceed(&parents, 1, room + 4)); } + + /// Empty-pool cluster cap. `getmempoolinfo` on an empty pool does not + /// report this constant. The first admit compares the new tx against it. + #[test] + fn empty_graph_reports_the_cluster_weight_cap() { + let g = TxGraph::new(); + assert!(g.is_empty()); + assert_eq!(g.total_weight(), 0); + assert_eq!(g.cluster_weight_limit(), MAX_CLUSTER_WEIGHT); + let tx = make_tx(None, 1, 9); + let mut g = g; + g.insert(entry_for(&tx, 1, 0), &tx); + assert!(!g.is_empty()); + assert!(g.total_weight() > 1); + } } diff --git a/crates/rbitcoin-mempool/src/orphanage.rs b/crates/rbitcoin-mempool/src/orphanage.rs index 4ce5984ddb..ad76c8f209 100644 --- a/crates/rbitcoin-mempool/src/orphanage.rs +++ b/crates/rbitcoin-mempool/src/orphanage.rs @@ -164,9 +164,12 @@ impl Orphanage { break; } } - if self.by_txid.len() >= self.max_count - || self.total_weight.saturating_add(weight) > self.max_weight - { + if self.by_txid.len() >= self.max_count { + return false; + } + // Room left is the cap minus what is already parked. A tx that lands + // on the cap still fits; one past it does not. + if self.max_weight.saturating_sub(self.total_weight) < weight { return false; } for p in &missing { @@ -733,4 +736,155 @@ mod tests { "the next expiry bound drops the orphan that survived the first" ); } + + /// Orphan count, the per-peer reserve, and taking children of one parent. + /// A peer tx parks an orphan. The reserve choice (drop the unannounced + /// tx, keep the peer's) is not a field `getmempoolinfo` returns. + #[test] + fn orphan_counters_limits_and_child_take() { + let parent = txid_n(21); + let other = txid_n(22); + let tx = make_orphan(parent, 1); + let w = tx.weight().to_wu(); + assert!(w > 1); + let mut miss = BTreeSet::new(); + miss.insert(parent); + miss.insert(other); + + let mut empty = Orphanage::with_limits(w, 4); + assert!(empty.is_empty()); + assert_eq!(empty.total_weight(), 0); + assert!(empty.insert_from_at(tx.clone(), miss.clone(), None, 0)); + assert!(!empty.is_empty()); + assert_eq!(empty.total_weight(), w); + + let mut counted = Orphanage::with_limits(DEFAULT_ORPHAN_MAX_WEIGHT, 1); + assert!(counted.insert_from_at(tx.clone(), miss.clone(), None, 0)); + let extra = make_orphan(parent, 3); + assert!( + counted.insert_from_at(extra, miss_one_parent(parent), None, 0), + "count cap evicts the older orphan" + ); + assert_eq!(counted.len(), 1); + + let mut linked = Orphanage::with_limits(DEFAULT_ORPHAN_MAX_WEIGHT, 4); + let body = make_orphan(parent, 4); + let id = body.compute_txid(); + assert!(linked.insert_from_at(body, miss.clone(), None, 0)); + let taken = linked.take_children_of(&parent); + assert_eq!(taken.len(), 1); + assert!( + !linked + .by_parent + .get(&other) + .is_some_and(|set| set.contains(&id)), + "the other missing parent drops the taken child" + ); + + let mut reserved = Orphanage::with_limits(DEFAULT_ORPHAN_MAX_WEIGHT, 2); + let protected = make_orphan(parent, 5); + let protected_id = protected.compute_txid(); + let loose = make_orphan(parent, 6); + let loose_id = loose.compute_txid(); + assert!(reserved.insert_from_at(protected, miss_one_parent(parent), Some(1), 0)); + assert!(reserved.insert_from_at(loose, miss_one_parent(parent), None, 0)); + let newer = make_orphan(parent, 7); + let newer_id = newer.compute_txid(); + assert!(reserved.insert_from_at(newer, miss_one_parent(parent), Some(2), 0)); + assert!( + reserved.contains(&protected_id), + "peer reserve keeps the older announcer" + ); + assert!(reserved.contains(&newer_id)); + assert!( + !reserved.contains(&loose_id), + "the unannounced orphan is the one evicted" + ); + } + + /// `with_limits` floors the weight cap at one max-size orphan. A tx that + /// lands on that cap is admitted. One past it evicts a single orphan. + /// The default orphan pool is far above one max-size tx. A journey cannot + /// fill it. Every orphan a peer sends still runs this compare. + #[test] + fn weight_cap_admits_exact_and_evicts_one_past() { + let parent = txid_n(21); + let cap = MAX_ORPHAN_TX_WEIGHT; + let mut exact = Orphanage::with_limits(cap, 100_000); + let seed = make_orphan(parent, 1); + let seed_w = seed.weight().to_wu(); + assert!(exact.insert_from_at(seed, miss_one_parent(parent), None, 0)); + let filler = orphan_of_weight(parent, cap - seed_w); + assert_eq!(filler.weight().to_wu(), cap - seed_w); + assert!( + exact.insert_from_at(filler, miss_one_parent(parent), None, 0), + "a tx that lands on the weight cap still fits" + ); + assert_eq!(exact.len(), 2); + assert_eq!(exact.total_weight(), cap); + let spill = make_orphan(parent, 2); + assert!(exact.total_weight().saturating_add(spill.weight().to_wu()) > cap); + assert_ne!( + exact.total_weight().saturating_add(spill.weight().to_wu()), + cap + ); + assert!( + exact.insert_from_at(spill, miss_one_parent(parent), None, 0), + "one eviction makes room past the cap" + ); + assert_eq!(exact.len(), 2); + assert!(exact.total_weight() <= cap); + + // Dropping only the oldest still leaves the new tx over the cap. + let mut twice = Orphanage::with_limits(cap, 100_000); + let mut oldest = orphan_of_weight(parent, 50_000); + oldest.lock_time = LockTime::from_height(1).unwrap(); + let oldest_id = oldest.compute_txid(); + let mut second = orphan_of_weight(parent, 300_000); + second.lock_time = LockTime::from_height(2).unwrap(); + let second_id = second.compute_txid(); + let mut incoming = orphan_of_weight(parent, 200_000); + incoming.lock_time = LockTime::from_height(3).unwrap(); + assert!(twice.insert_from_at(oldest, miss_one_parent(parent), None, 0)); + assert!(twice.insert_from_at(second, miss_one_parent(parent), None, 0)); + assert_eq!(twice.total_weight(), 350_000); + assert!(twice.total_weight().saturating_add(200_000) > cap); + assert!(300_000u64.saturating_add(200_000) > cap); + assert!( + twice.insert_from_at(incoming.clone(), miss_one_parent(parent), None, 0), + "the weight cap evicts until the incoming orphan fits" + ); + assert_eq!(twice.len(), 1); + assert_eq!(twice.total_weight(), incoming.weight().to_wu()); + assert!(twice.contains(&incoming.compute_txid())); + assert!(!twice.contains(&oldest_id)); + assert!(!twice.contains(&second_id)); + } + + fn miss_one_parent(parent: Txid) -> BTreeSet { + let mut miss = BTreeSet::new(); + miss.insert(parent); + miss + } + + /// Pad a witness until the tx weighs `target` wu. Witness bytes move the + /// weight by one, so every target above the bare orphan is reachable. + fn orphan_of_weight(parent: Txid, target: u64) -> Transaction { + let mut n = 1usize; + for _ in 0..8 { + let mut tx = make_orphan(parent, 9); + tx.lock_time = LockTime::from_height(50_000).unwrap(); + tx.input[0].witness = Witness::from_slice(&[vec![0x11; n]]); + let got = tx.weight().to_wu(); + if got == target { + return tx; + } + if got > target { + n -= (got - target) as usize; + } else { + n += (target - got) as usize; + } + } + panic!("no witness length weighs {target}"); + } } diff --git a/crates/rbitcoin-mempool/src/packed.rs b/crates/rbitcoin-mempool/src/packed.rs index 35c3a6e0e0..ffa1ac1406 100644 --- a/crates/rbitcoin-mempool/src/packed.rs +++ b/crates/rbitcoin-mempool/src/packed.rs @@ -14,7 +14,7 @@ use bitcoin::{Amount, ScriptBuf, Sequence, Transaction, TxIn, TxOut, Txid, Witne use rbitcoin_primitives::{read_compact_size, write_compact_size, Fk}; use rbitcoin_store::OutputRecord; -const VIN_SEQ_FINAL: u8 = 1 << 0; +const VIN_SEQ_FINAL: u8 = 1; const VIN_EMPTY_SCRIPT: u8 = 1 << 1; const VIN_EMPTY_WITNESS: u8 = 1 << 2; const VIN_HAS_CREATE_FK: u8 = 1 << 3; @@ -352,6 +352,101 @@ mod tests { assert_eq!(serialize(&got.tx), raw, "consensus serialize pin"); assert_eq!(got.vins[0].script_hash, Some(sh)); assert_eq!(got.vins[0].create_fk, Some(Fk(42))); + // SEQ_FINAL | EMPTY_SCRIPT | CREATE_FK | SCRIPT_HASH. Witness is present. + assert_eq!(packed[98], 0b0001_1011, "vin flags {:08b}", packed[98]); + } + + /// Vin flag bits and a script or witness that ends the record. Encode and + /// decode share the constants, so a reopen of a tx this process wrote + /// still round-trips if a bit number changes. The byte is the schema an + /// older mempool file uses. + #[test] + fn packed_vin_flags_and_exact_tail_roundtrip() { + let mut bare = sample_tx(false); + bare.input[0].witness = Witness::new(); + let txid = bare.compute_txid(); + let wtxid = bare.compute_wtxid(); + let packed = encode_packed_live(&bare, &txid, &wtxid, 1, 400, Some(1), &[]).unwrap(); + // SEQ_FINAL | EMPTY_SCRIPT | EMPTY_WITNESS, no aux. The witness/script + // tails are absent, so this pins the three flag bits an operator reload reads. + assert_eq!(packed[98], 0b0000_0111, "vin flags {:08b}", packed[98]); + let got = decode_packed_live(&packed).unwrap(); + assert_eq!(serialize(&got.tx), serialize(&bare)); + + bare.input[0].script_sig = ScriptBuf::from_bytes(vec![0x51]); + bare.input[0].sequence = Sequence::ZERO; + let txid = bare.compute_txid(); + let wtxid = bare.compute_wtxid(); + let packed = encode_packed_live(&bare, &txid, &wtxid, 1, 400, Some(1), &[]).unwrap(); + // Non-empty script, empty witness, non-final sequence: only EMPTY_WITNESS. + assert_eq!(packed[98], 0b0000_0100, "vin flags {:08b}", packed[98]); + let got = decode_packed_live(&packed).unwrap(); + assert_eq!(got.tx.input[0].script_sig.as_bytes(), &[0x51]); + assert_eq!(serialize(&got.tx), serialize(&bare)); + + // No outputs: the script or the witness is the last byte of the vin + // slice, so `end == buf.len()` is a real record and one byte short is not. + let script_tail = encode_tail(true, false); + assert!(decode_packed_live(&script_tail).is_ok()); + let mut torn = script_tail.clone(); + torn.pop(); + assert!(matches!( + decode_packed_live(&torn), + Err(MempoolError::Corrupt("packed script")) + )); + let witness_tail = encode_tail(false, true); + assert!(decode_packed_live(&witness_tail).is_ok()); + let mut torn = witness_tail; + torn.pop(); + assert!(matches!( + decode_packed_live(&torn), + Err(MempoolError::Corrupt("packed witness")) + )); + } + + /// One input, no outputs. `script` puts a one-byte scriptSig last; otherwise + /// a one-byte witness item is last. + fn encode_tail(script: bool, witness: bool) -> Vec { + let mut tx = sample_tx(false); + tx.output.clear(); + tx.input[0].witness = Witness::new(); + tx.input[0].script_sig = ScriptBuf::new(); + tx.input[0].sequence = Sequence::ZERO; + if script { + tx.input[0].script_sig = ScriptBuf::from_bytes(vec![0x51]); + } + if witness { + let mut w = Witness::new(); + w.push([0x51]); + tx.input[0].witness = w; + } + let (txid, wtxid) = (tx.compute_txid(), tx.compute_wtxid()); + encode_packed_live(&tx, &txid, &wtxid, 1, 400, Some(1), &[]).unwrap() + } + + /// Input and output counts past 100_000. A live mempool tx cannot be that + /// large. Opening a torn `tx.body` hits this decoder and must not allocate. + #[test] + fn packed_live_rejects_only_past_the_count_cap() { + let tx = sample_tx(false); + let (txid, wtxid) = (tx.compute_txid(), tx.compute_wtxid()); + let seed = encode_packed_live(&tx, &txid, &wtxid, 1, 400, Some(1), &[]).unwrap(); + let with_counts = |n_in: u64, n_out: u64| { + let mut buf = seed[..96].to_vec(); + write_compact_size(&mut buf, n_in); + write_compact_size(&mut buf, n_out); + buf + }; + let counts = |buf: &[u8]| { + matches!( + decode_packed_live(buf), + Err(MempoolError::Corrupt("packed live counts")) + ) + }; + assert!(counts(&with_counts(100_001, 0))); + assert!(counts(&with_counts(0, 100_001))); + assert!(!counts(&with_counts(100_000, 0))); + assert!(!counts(&with_counts(0, 100_000))); } #[test] diff --git a/crates/rbitcoin-mempool/src/store.rs b/crates/rbitcoin-mempool/src/store.rs index 487e4439d6..7d19718001 100644 --- a/crates/rbitcoin-mempool/src/store.rs +++ b/crates/rbitcoin-mempool/src/store.rs @@ -269,6 +269,11 @@ impl Mempool { Ok(()) } + /// Two slot records share one `pwrite` when the next record starts where the last ended. + fn adjacent_slot_records(prev_end: usize, off: usize) -> bool { + prev_end == off + } + /// `pwrite` LIVE rows whose payload starts at or past `old_persisted`. /// /// Adjacent new slots are one write. Does not rewrite the rest of the table. @@ -286,7 +291,7 @@ impl Mempool { } let rec_end = off + SLOT_REC; if let Some((_, end)) = ranges.last_mut() { - if *end == off { + if Self::adjacent_slot_records(*end, off) { *end = rec_end; continue; } @@ -323,7 +328,7 @@ impl Mempool { vins: &[VinAux], ) -> Result { let payload = encode_packed_live(tx, txid, wtxid, fee_sat, weight, Some(sigop_cost), vins)?; - if payload.len() > u32::MAX as usize { + if u32::try_from(payload.len()).is_err() { return Err(MempoolError::Corrupt("tx body too large")); } let payload_len = payload.len(); @@ -659,6 +664,18 @@ impl Mempool { self.grow_pinned = true; } + /// Tests set the logical slot cap. The slots image must already cover it. + #[cfg(test)] + pub(crate) fn testing_set_slot_cap(&mut self, cap: u32) { + let need = SLOTS_HEADER + (cap as usize) * SLOT_REC; + assert!( + cap >= 1 && self.slots.len() >= need, + "slot image is shorter than cap {cap}" + ); + self.slots[8..12].copy_from_slice(&cap.to_le_bytes()); + self.slot_cap = cap; + } + /// Double slot capacity (up to [`MAX_SLOT_CAP`]) and extend the slots image with FREE records. pub fn grow_slots(&mut self) -> Result<(), MempoolError> { if self.grow_pinned || self.slot_cap >= MAX_SLOT_CAP { @@ -1567,11 +1584,35 @@ pub(crate) mod tests { let t2 = Txid::from_byte_array([0x02; 32]); mp.append_live_tx(&raw, &t2, &raw.compute_wtxid(), 2, 400, 0, &[]) .unwrap(); + let t3 = Txid::from_byte_array([0x03; 32]); + mp.append_live_tx(&raw, &t3, &raw.compute_wtxid(), 3, 400, 0, &[]) + .unwrap(); + // A byte just past the new slot records. A widened record range rewrites it. + let past = SLOTS_HEADER + 3 * SLOT_REC; + { + let mut slots = OpenOptions::new() + .write(true) + .open(dir.join("slots")) + .unwrap(); + slots.seek(SeekFrom::Start(past as u64)).unwrap(); + slots.write_all(&[0xab]).unwrap(); + slots.sync_all().unwrap(); + } + // Cap the walk at the live rows so a divided slot offset cannot alias + // slot 1 from an index thousands of rows later. + mp.testing_pin_full_slot_table(); mp.set_now_ms(PERSIST_INTERVAL_MS * 2); mp.persist_due().unwrap(); + let on_disk = fs::read(dir.join("slots")).unwrap(); + assert_eq!( + on_disk[past], 0xab, + "persist must not rewrite the slot past the new records" + ); + assert_eq!(on_disk[SLOTS_HEADER + SLOT_REC], SLOT_LIVE); + assert_eq!(on_disk[SLOTS_HEADER + 2 * SLOT_REC], SLOT_LIVE); drop(mp); let mp = Mempool::open_or_create(&dir).unwrap(); - assert_eq!(mp.load_live_txs().unwrap().len(), 2); + assert_eq!(mp.load_live_txs().unwrap().len(), 3); let _ = fs::remove_dir_all(&dir); } @@ -1882,4 +1923,95 @@ pub(crate) mod tests { } let _ = fs::remove_dir_all(&dir); } + + /// A clean pool skips the next meta fsync. The txs are already on disk + /// either way. The client reads the same set. The skip is what a quiet + /// node does on the persist interval. + #[test] + fn clear_dirty_keeps_a_second_interval_from_rewriting_meta() { + let dir = tmp_dir(); + let mut mp = Mempool::open_or_create(&dir).unwrap(); + mp.set_now_ms(0); + let tid = Txid::from_byte_array([0x71; 32]); + put_live(&mut mp, &tid, 1, 400); + mp.set_now_ms(PERSIST_INTERVAL_MS); + mp.persist_due().unwrap(); + let syncs = mp.meta_syncs(); + mp.set_now_ms(PERSIST_INTERVAL_MS * 2); + mp.persist_due().unwrap(); + assert_eq!( + mp.meta_syncs(), + syncs, + "a clean pool does not fsync meta again" + ); + let _ = fs::remove_dir_all(&dir); + } + + /// Two new slot rows share one `pwrite` when they are adjacent. Separate + /// writes leave the same bytes a reopen reads. The merge is the persist + /// path for a burst of admits. + #[test] + fn adjacent_slot_records_touch_when_the_next_starts_at_the_end() { + assert!(Mempool::adjacent_slot_records(64, 64)); + assert!(!Mempool::adjacent_slot_records(64, 112)); + } + + #[test] + fn abandon_live_clears_every_slot_on_reopen() { + let dir = tmp_dir(); + let mut mp = Mempool::open_or_create(&dir).unwrap(); + put_live(&mut mp, &Txid::from_byte_array([0x81; 32]), 1, 400); + put_live(&mut mp, &Txid::from_byte_array([0x82; 32]), 1, 400); + mp.flush().unwrap(); + // Two live rows and a cap of 2: slot 1 is only reached by `slot * SLOT_REC`. + mp.testing_pin_full_slot_table(); + assert_eq!(mp.abandon_live().unwrap(), 2); + drop(mp); + let mp = Mempool::open_or_create(&dir).unwrap(); + assert_eq!(mp.live_count(), 0); + assert!(mp.load_live_txs().unwrap().is_empty()); + let _ = fs::remove_dir_all(&dir); + } + + /// Free, live, and dead slot counts. No RPC returns the split. Eviction + /// and compact both read it to decide whether a slot is reusable. + #[test] + fn slot_stats_counts_free_live_and_dead() { + let dir = tmp_dir(); + let mut mp = Mempool::open_or_create(&dir).unwrap(); + let cap = mp.meta().slot_cap; + let slot = put_live(&mut mp, &Txid::from_byte_array([0x91; 32]), 1, 400); + let (free, live, dead) = mp.slot_stats(); + assert_eq!((free, live, dead), (cap - 1, 1, 0)); + mp.mark_slot_dead(slot).unwrap(); + let (free, live, dead) = mp.slot_stats(); + assert_eq!((free, live, dead), (cap - 1, 0, 1)); + let _ = fs::remove_dir_all(&dir); + } + + /// A live slot whose body length runs past the file. Compact refuses it. + /// A normal reopen never builds this row. Startup compact of a torn + /// slots file is the path that returns the error. + #[test] + fn compact_rejects_a_live_slot_whose_body_range_is_not_in_the_file() { + let dir = tmp_dir(); + let mut mp = Mempool::open_or_create(&dir).unwrap(); + put_live(&mut mp, &Txid::from_byte_array([0xa1; 32]), 1, 400); + let off = SLOTS_HEADER; + // A prefix-sized span inside the file is a copy, not a range error. + mp.slots[off + 12..off + 16].copy_from_slice(&(BODY_TX_PREFIX as u32).to_le_bytes()); + let (n, _) = mp.compact().unwrap(); + assert_eq!(n, 1); + + put_live(&mut mp, &Txid::from_byte_array([0xa2; 32]), 1, 400); + let off = SLOTS_HEADER; + let logical = mp.body_logical_len().unwrap(); + let past = (logical as u32).saturating_add(8); + mp.slots[off + 12..off + 16].copy_from_slice(&past.to_le_bytes()); + match mp.compact() { + Err(MempoolError::Corrupt("live slot body range")) => {} + other => panic!("expected live slot body range, got {other:?}"), + } + let _ = fs::remove_dir_all(&dir); + } } diff --git a/crates/rbitcoin-net/src/ibd/confirm/mod.rs b/crates/rbitcoin-net/src/ibd/confirm/mod.rs index 4697546232..7a15ed865c 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/mod.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/mod.rs @@ -7,6 +7,7 @@ use bitcoin::hashes::Hash; use bitcoin::BlockHash; use rbitcoin_consensus::WireLoadPipeline; use rbitcoin_log::{debug, info, warn}; +use std::cell::Cell; use std::collections::HashMap; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; @@ -543,6 +544,49 @@ fn reoffer_blocks_to_body_queue<'a>( offer_back(hub, prepare_offer_back(hub, items)); } +/// A stale load epoch offers the wave back. Already-connected hashes are +/// dropped by [`prepare_offer_back`]; a second filter here cannot change +/// the queue. +fn reoffer_unconnected_bodies<'a>(hub: &ChainHub, items: impl IntoIterator>) { + reoffer_blocks_to_body_queue(hub, items); +} + +/// The script wave that goes back on the body queue. A one-block verdict +/// drops its block. A retry, an engine fault, or a session fault keeps it. +fn script_reject_wave( + retried: bool, + session_fault: bool, + class: ConfirmRejectClass, + wave: &[T], +) -> &[T] { + if retried || session_fault || class == ConfirmRejectClass::EngineFault { + wave + } else { + &[] + } +} + +fn rearm_script_reject( + hub: &ChainHub, + feed: &ConfirmFeed, + retried: bool, + session_fault: bool, + class: ConfirmRejectClass, + wave: &[(u32, BlockHash, &bitcoin::Block)], +) { + let rearm = if retried { + ConfirmRejectClass::Cascade + } else { + class + }; + rearm_after_reject( + hub, + feed, + rearm, + script_reject_wave(retried, session_fault, class, wave), + ); +} + /// A write or scripts reject: lookup took the wave's bodies off the body /// queue. Re-arm lookup at the tip, which bumps the generation so load /// drops its look-ahead plan before the next wave. When the wave is retried @@ -589,6 +633,20 @@ fn rearm_after_reject( true } +/// A cancelled write does not re-arm. Load then drops the plan it stamped +/// for that wave. Every other class re-arms, which already bumps the generation. +fn apply_write_reject( + hub: &ChainHub, + feed: &ConfirmFeed, + class: ConfirmRejectClass, + wave: &[(u32, BlockHash, &bitcoin::Block)], + reset_lookahead: &AtomicBool, +) { + if !rearm_after_reject(hub, feed, class, wave) { + reset_lookahead.store(true, Ordering::Release); + } +} + /// The class a load reject of `wave` is rewound and sent with. A lone /// block's verdict holds only while its parent is the tip; otherwise it is /// a cascade. This is load's one tip sample: the rewind and the event both @@ -641,6 +699,25 @@ fn load_fail_rewind_wave( feed.notify(); } +/// A pin-stage engine fault that is not a uring session fault is a stale +/// plan and rewinds as a cascade. A session fault keeps the engine class. +fn rewind_after_load_reject( + feed: &ConfirmFeed, + hub: &ChainHub, + lookup_ahead: &mut LoadAheadState, + first_h: u32, + class: ConfirmRejectClass, + session_fault: bool, + wave: &[OfferBack<'_>], +) { + let rewind = if class == ConfirmRejectClass::EngineFault && !session_fault { + ConfirmRejectClass::Cascade + } else { + class + }; + load_fail_rewind_wave(feed, hub, lookup_ahead, first_h, rewind, wave); +} + /// Peer id the wire was enqueued with: `sender` on [`ConfirmEvent::Reject`]. fn wire_sender(wire: &rbitcoin_query::ResolvedWire) -> Option { wire.sender.and_then(|s| usize::try_from(s).ok()) @@ -741,15 +818,21 @@ pub(crate) enum ConfirmEvent { }, } -fn publish_confirm_progress(query: &rbitcoin_query::Query) { - use std::cell::Cell; - thread_local! { - static LAST: Cell> = const { Cell::new(None) }; - } - let due = LAST.with(|cell| match cell.get() { - Some(t) if t.elapsed() < Duration::from_secs(5) => false, +thread_local! { + static CONFIRM_PROGRESS_LAST: Cell> = const { Cell::new(None) }; +} + +#[cfg(test)] +fn clear_confirm_progress_clock() { + CONFIRM_PROGRESS_LAST.with(|cell| cell.set(None)); +} + +/// Publish operator snapshots when `now` is at least 5s after the last publish. +fn publish_confirm_progress_at(query: &rbitcoin_query::Query, now: Instant) { + let due = CONFIRM_PROGRESS_LAST.with(|cell| match cell.get() { + Some(t) if now.saturating_duration_since(t) < Duration::from_secs(5) => false, _ => { - cell.set(Some(Instant::now())); + cell.set(Some(now)); true } }); @@ -758,6 +841,10 @@ fn publish_confirm_progress(query: &rbitcoin_query::Query) { } } +fn publish_confirm_progress(query: &rbitcoin_query::Query) { + publish_confirm_progress_at(query, Instant::now()); +} + #[allow(clippy::too_many_arguments)] // call-site args stay unbundled fn emit_confirm_reject( tx: &std::sync::mpsc::Sender, @@ -827,8 +914,8 @@ pub(crate) const WRITE_QUEUE_CAP_DEFAULT: usize = 14; /// Blocks added to or removed from the extra in-flight window per wave. pub(crate) const INFLIGHT_RETAIN_STEP: u32 = 64; -/// Accounted-heap line. Under it the window can grow; over it the window shrinks. -pub(crate) const INFLIGHT_RETAIN_OWNED_LIMIT: u64 = 1024 * 1024 * 1024; +/// Accounted-heap line (1 GiB). Under it the window can grow; over it the window shrinks. +pub(crate) const INFLIGHT_RETAIN_OWNED_LIMIT: u64 = 1 << 30; /// `height_by_hash` entry estimate from the old `ibd: sizes` accounted total. const H2H_ENTRY_BYTES: u64 = 48; /// Height-fence run estimate from that same total. @@ -1483,6 +1570,35 @@ pub(crate) fn format_stamp_reject_missing_prevout( /// before, and scripts and write never wait on load. const PARENT_CHECK_WAIT: Duration = Duration::from_secs(10); +thread_local! { + /// `Some` advances the parent-check deadline without sleeping. Unset uses + /// [`Instant::now`] and a 2ms sleep, which is the confirm thread. + static PARENT_CHECK_CLOCK: Cell> = const { Cell::new(None) }; +} + +fn parent_now() -> Instant { + PARENT_CHECK_CLOCK.with(|cell| cell.get().unwrap_or_else(Instant::now)) +} + +#[cfg(test)] +fn set_parent_check_clock(now: Option) { + PARENT_CHECK_CLOCK.with(|cell| cell.set(now)); +} + +fn parent_check_wait_slice() { + let advanced = PARENT_CHECK_CLOCK.with(|cell| { + if let Some(t) = cell.get() { + cell.set(Some(t + Duration::from_millis(500))); + true + } else { + false + } + }); + if !advanced { + std::thread::sleep(Duration::from_millis(2)); + } +} + /// A fresh read of a lone block's parents after its load stamp missed one. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum ParentCheck { @@ -1508,17 +1624,19 @@ fn check_missing_parent(hub: &ChainHub, feed: &ConfirmFeed, block: &bitcoin::Blo let on_tip = || hub.tip_hash() == Some(prev); let ready = || on_tip() && hub.query.head_covers_fence(); let stop = || feed.stopped() || hub.query.confirm_cancelled(); - let deadline = Instant::now() + PARENT_CHECK_WAIT; + let deadline = parent_now() + PARENT_CHECK_WAIT; while !ready() { - if stop() || Instant::now() >= deadline { + if stop() || parent_now() >= deadline { return if on_tip() && !feed.stopped() { ParentCheck::Engine } else { ParentCheck::OffTip }; } - std::thread::sleep(Duration::from_millis(2)); + parent_check_wait_slice(); } + // Same HDD gate as lookup and load: this read scans `tx.head`. + hub.query.store().spindle_wait(); let read = rbitcoin_consensus::parent_missing_from_chain(&hub.query, block, stop); if stop() || !ready() || hub.query.take_disconnect(&mut disconnects).is_some() { return ParentCheck::OffTip; @@ -2013,9 +2131,7 @@ pub(crate) fn spawn_confirm_engine( .zip(failed.wire_blocks()) .map(|(&(h, ha), b)| (h, ha, b.as_ref())) .collect(); - if !rearm_after_reject(&hub_wb, &feed_wb, class, &wave) { - load_ahead_reset_wb.store(true, Ordering::Release); - } + apply_write_reject(&hub_wb, &feed_wb, class, &wave, &load_ahead_reset_wb); confirm_thr_stats::add_write_work(&stats, t_rearm.elapsed()); let _ = emit_confirm_reject( &event_tx_wb, @@ -2106,20 +2222,18 @@ pub(crate) fn spawn_confirm_engine( let retried = class.isolate_if_batched(meta.heights_hashes.len()) == ConfirmRejectClass::Cascade; let t_rearm = Instant::now(); - let wave: Vec<(u32, BlockHash, &bitcoin::Block)> = if retried - || session_fault - || class == ConfirmRejectClass::EngineFault - { - std::iter::once(&meta) - .chain(dropped) - .flat_map(|m| m.heights_hashes.iter().zip(&m.wire_blocks)) - .map(|(&(h, raw), b)| (h, BlockHash::from_byte_array(raw), b.as_ref())) - .collect() - } else { - Vec::new() - }; + let full: Vec<(u32, BlockHash, &bitcoin::Block)> = std::iter::once(&meta) + .chain(dropped) + .flat_map(|m| m.heights_hashes.iter().zip(&m.wire_blocks)) + .map(|(&(h, raw), b)| (h, BlockHash::from_byte_array(raw), b.as_ref())) + .collect(); if session_fault { - requeue_on_uring_recover(&hub_sc, &feed_sc, "ibd-confirm-scripts", &wave); + requeue_on_uring_recover( + &hub_sc, + &feed_sc, + "ibd-confirm-scripts", + script_reject_wave(retried, session_fault, class, &full), + ); confirm_thr_stats::add_script_work(&stats, t_rearm.elapsed()); warn!("ibd: confirm scripts uring recover @ {}: {e}", meta.first_h); return true; @@ -2137,12 +2251,7 @@ pub(crate) fn spawn_confirm_engine( .confirm_reject_stops .fetch_add(1, Ordering::Relaxed); warn!("ibd: confirm scripts reject @ {height} (batch first {hash}): {e}"); - let rearm = if retried { - ConfirmRejectClass::Cascade - } else { - class - }; - rearm_after_reject(&hub_sc, &feed_sc, rearm, &wave); + rearm_script_reject(&hub_sc, &feed_sc, retried, session_fault, class, &full); confirm_thr_stats::add_script_work(&stats, t_rearm.elapsed()); let _ = emit_confirm_reject( &event_tx_sc, @@ -2226,12 +2335,11 @@ pub(crate) fn spawn_confirm_engine( "ibd: confirm load drop stale plan epoch={claim_epoch} live={}", feed_load.epoch() ); - reoffer_blocks_to_body_queue( + reoffer_unconnected_bodies( &hub_load, - lb.items.iter().filter_map(|(h, raw, w)| { + lb.items.iter().map(|(h, raw, w)| { let hash = BlockHash::from_byte_array(*raw); - (!hub_load.has_block(&hash)) - .then_some((*h, hash, w.block.as_ref(), w.sender)) + (*h, hash, w.block.as_ref(), w.sender) }), ); feed_load.finish(lb.items.iter().map(|(h, _, _)| *h)); @@ -2556,17 +2664,13 @@ pub(crate) fn spawn_confirm_engine( // A pin-stage invariant miss can be a stale-fk plan // race, not a store fault: retry it one block at a // time, as a cascade. A session fault is not a race. - let rewind = if class == ConfirmRejectClass::EngineFault && !session_fault { - ConfirmRejectClass::Cascade - } else { - class - }; - load_fail_rewind_wave( + rewind_after_load_reject( &feed_load, &hub_load, &mut lookup_ahead, expect_h, - rewind, + class, + session_fault, &wave, ); confirm_thr_stats::add_load_prune(&stats, t_rewind.elapsed()); @@ -2697,7 +2801,10 @@ pub(crate) fn spawn_confirm_engine( max_blocks, max_inputs, ) { - Ok(wave) if !wave.items.is_empty() || wave.undecodable.is_some() => { + Ok(wave) if wave.items.is_empty() && wave.undecodable.is_none() => { + lookup_faults.on_success(); + } + Ok(wave) => { lookup_faults.on_success(); did = true; if let Some((h, hash)) = wave.undecodable { @@ -2792,9 +2899,6 @@ pub(crate) fn spawn_confirm_engine( } feed.notify(); } - Ok(_) => { - lookup_faults.on_success(); - } Err(e) => { if e.is_uring_session_fault() { hub.query.uring_recover_or_abort("ibd-confirm-lookup"); diff --git a/crates/rbitcoin-net/src/ibd/confirm/tests.rs b/crates/rbitcoin-net/src/ibd/confirm/tests.rs index 95762b6fe9..3dc24e036b 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/tests.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/tests.rs @@ -111,8 +111,11 @@ fn owned_heap_steps_inflight_around_one_gib_only_when_a_prune_would_drop() { 0, "nothing to save does not extend the window" ); + assert_eq!(INFLIGHT_RETAIN_OWNED_LIMIT, 1_073_741_824); assert_eq!(retain.observe(under, true), INFLIGHT_RETAIN_STEP); + assert_eq!(retain.extra(), INFLIGHT_RETAIN_STEP); assert_eq!(retain.observe(under, true), INFLIGHT_RETAIN_STEP * 2); + assert_eq!(retain.extra(), INFLIGHT_RETAIN_STEP * 2); assert_eq!( retain.observe(under, false), INFLIGHT_RETAIN_STEP * 2, @@ -2168,6 +2171,22 @@ fn reject_rearms_lookup_and_requeues_a_retried_wave() { ); assert!(!feed.single_block()); + let feed = ConfirmFeed::new(); + hub.query.set_lookup_taken_hi(Some(2)); + assert!(rearm_after_reject( + &hub, + &feed, + ConfirmRejectClass::Cascade, + &wave[..1], + )); + assert!( + !feed.single_block(), + "a one-block retry is already one block" + ); + if hub.query.block_queue_has_height(1) { + hub.query.block_queue_dequeue_height(1).unwrap(); + } + for class in [ ConfirmRejectClass::Cascade, ConfirmRejectClass::ConsensusInvalid, @@ -2210,3 +2229,351 @@ fn reject_rearms_lookup_and_requeues_a_retried_wave() { ); let _ = std::fs::remove_dir_all(dir); } + +/// A cancelled write drops the load plan. A retried or engine-fault script +/// wave goes back on the body queue. A peer's invalid block is a one-block +/// consensus reject and does not take these arms. The confirm thread does. +#[test] +fn write_cancel_and_script_reject_requeue() { + use super::{apply_write_reject, rearm_script_reject}; + use rbitcoin_consensus::mine_empty_regtest; + use std::sync::atomic::{AtomicBool, Ordering}; + + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("script-reject-requeue"); + hub.ensure_genesis().unwrap(); + let tip = hub.tip_hash().unwrap(); + let time = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest) + .header + .time; + let b1 = mine_empty_regtest(tip, time + 600, 1); + let b2 = mine_empty_regtest(b1.block_hash(), time + 1200, 2); + let wave = [(1, b1.block_hash(), &b1), (2, b2.block_hash(), &b2)]; + + let reset = AtomicBool::new(false); + let feed = ConfirmFeed::new(); + apply_write_reject(&hub, &feed, ConfirmRejectClass::Cancelled, &[], &reset); + assert!( + reset.load(Ordering::Acquire), + "a cancelled write drops the load plan" + ); + let reset = AtomicBool::new(false); + apply_write_reject( + &hub, + &feed, + ConfirmRejectClass::ConsensusInvalid, + &wave[..1], + &reset, + ); + assert!( + !reset.load(Ordering::Acquire), + "a consensus write re-arms and keeps the load plan" + ); + + let feed = ConfirmFeed::new(); + rearm_script_reject( + &hub, + &feed, + true, + false, + ConfirmRejectClass::ConsensusInvalid, + &wave, + ); + assert!( + hub.query.block_queue_has_height(1) && hub.query.block_queue_has_height(2), + "a retried script wave is back on the queue" + ); + for ht in [1, 2] { + hub.query.block_queue_dequeue_height(ht).unwrap(); + } + let feed = ConfirmFeed::new(); + rearm_script_reject( + &hub, + &feed, + false, + false, + ConfirmRejectClass::EngineFault, + &wave, + ); + assert!( + hub.query.block_queue_has_height(1) && hub.query.block_queue_has_height(2), + "an engine-fault script wave is back on the queue" + ); + assert!(!feed.single_block()); + let _ = std::fs::remove_dir_all(dir); +} + +/// The 5s cut is the gap between two publishes on the confirm thread. +/// A wall-clock journey would sleep to tell `<` from `<=`. The operator +/// sees the progress line; this passes the timestamp the thread would. +#[test] +fn confirm_progress_publishes_on_a_five_second_boundary() { + use super::{clear_confirm_progress_clock, publish_confirm_progress_at}; + use std::time::{Duration, Instant}; + + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("confirm-progress"); + hub.ensure_genesis().unwrap(); + clear_confirm_progress_clock(); + rbitcoin_query::reset_progress_snapshot_counts(); + let t0 = Instant::now() + Duration::from_secs(10_000); + publish_confirm_progress_at(&hub.query, t0); + assert_eq!(rbitcoin_query::progress_snapshot_counts().0, 1); + publish_confirm_progress_at(&hub.query, t0 + Duration::from_secs(4)); + assert_eq!( + rbitcoin_query::progress_snapshot_counts().0, + 1, + "four seconds does not publish again" + ); + publish_confirm_progress_at(&hub.query, t0 + Duration::from_secs(5)); + assert_eq!( + rbitcoin_query::progress_snapshot_counts().0, + 2, + "five seconds publishes again" + ); + clear_confirm_progress_clock(); + let _ = std::fs::remove_dir_all(dir); +} + +fn cover_parent_fence(hub: &crate::chain::ChainHub) { + let max = hub.query.store().fence_max_connected_fk(); + if max > 0 { + hub.query.note_head_drain_fk(max); + } + assert!( + hub.query.head_covers_fence(), + "drain {} fence {}", + hub.query.head_drain_fk(), + hub.query.store().fence_max_connected_fk() + ); +} + +fn uncover_parent_fence(hub: &crate::chain::ChainHub) { + if !hub.query.head_covers_fence() { + return; + } + let store = hub.query.store(); + let first = store.fence_max_connected_fk().saturating_add(10).max(1); + let header_fk = rbitcoin_primitives::Fk(first); + store + .header_txs + .put_range(header_fk, rbitcoin_primitives::Fk(first), 1) + .unwrap(); + store + .height_fence_extend(rbitcoin_primitives::Height(50), header_fk) + .unwrap(); + assert!( + !hub.query.head_covers_fence(), + "a fence run past the drain is not covered" + ); +} + +struct ParentClockGuard; +impl Drop for ParentClockGuard { + fn drop(&mut self) { + super::set_parent_check_clock(None); + } +} + +/// Parent stamp: off-tip waits out the deadline, a stop does not, an uncovered +/// fence on the tip is an engine miss, and a disconnect during the read retries. +/// The clock is injected so the test does not sleep 10s. A peer block on a +/// stopped node, or a reorg during the parent read, is what a session runs. +#[test] +fn parent_check_reads_the_tip() { + use super::{check_missing_parent, parent_now, set_parent_check_clock, ParentCheck}; + use rbitcoin_consensus::mine_empty_regtest; + use std::sync::mpsc; + use std::time::{Duration, Instant}; + + let _guard = ParentClockGuard; + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("parent-check-read"); + hub.ensure_genesis().unwrap(); + cover_parent_fence(&hub); + let tip = hub.tip_hash().unwrap(); + let time = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest) + .header + .time; + let on_tip = mine_empty_regtest(tip, time + 600, 1); + let off = mine_empty_regtest( + bitcoin::BlockHash::from_byte_array([0x44; 32]), + time + 600, + 1, + ); + + let feed = ConfirmFeed::new(); + feed.request_stop(); + let start = Instant::now() + Duration::from_secs(3_600); + set_parent_check_clock(Some(start)); + assert_eq!(check_missing_parent(&hub, &feed, &off), ParentCheck::OffTip); + assert!( + parent_now().saturating_duration_since(start) < Duration::from_secs(1), + "a stopped feed does not wait out the parent deadline" + ); + set_parent_check_clock(Some(start)); + assert_eq!( + check_missing_parent(&hub, &feed, &on_tip), + ParentCheck::OffTip, + "confirm is stopping, so a tip parent is a retry" + ); + assert!(parent_now().saturating_duration_since(start) < Duration::from_secs(1)); + + let feed = ConfirmFeed::new(); + set_parent_check_clock(Some(start)); + assert_eq!( + check_missing_parent(&hub, &feed, &off), + ParentCheck::OffTip, + "an off-tip parent retries after the deadline" + ); + assert!( + parent_now().saturating_duration_since(start) >= Duration::from_secs(9), + "the off-tip wait runs to the deadline" + ); + + uncover_parent_fence(&hub); + set_parent_check_clock(Some(start)); + assert_eq!( + check_missing_parent(&hub, &feed, &on_tip), + ParentCheck::Engine, + "the tip's parent is missing from tx.head after the deadline" + ); + assert!( + parent_now().saturating_duration_since(start) >= Duration::from_secs(9), + "an uncovered fence waits out the deadline" + ); + + cover_parent_fence(&hub); + set_parent_check_clock(Some(start)); + // The parent read waits on the HDD spindle, after it has sampled the + // disconnect generation. A reorg noted while that read is parked is a retry. + hub.query.store().set_datadir_hdd(true); + let (tx, rx) = mpsc::channel(); + std::thread::scope(|scope| { + let hold = hub.query.store().spindle_hold(); + scope.spawn(|| { + let _ = tx.send(check_missing_parent(&hub, &feed, &on_tip)); + }); + std::thread::sleep(Duration::from_millis(200)); + assert!( + rx.try_recv().is_err(), + "the parent read did not wait for the spindle" + ); + hub.query.note_disconnect_height(1); + drop(hold); + let saw = rx + .recv_timeout(Duration::from_secs(2)) + .expect("parent read did not finish after the spindle"); + assert_eq!( + saw, + ParentCheck::OffTip, + "a disconnect during the parent read retries the block" + ); + }); + let _ = std::fs::remove_dir_all(dir); +} + +/// A stale load offers a body the node has not connected and skips one it has. +/// The confirm thread does this when its load epoch is no longer current. +/// A peer does not send that epoch. +#[test] +fn unconnected_stale_body_returns_to_the_queue() { + use super::reoffer_unconnected_bodies; + use rbitcoin_consensus::mine_empty_regtest; + + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("stale-body"); + hub.ensure_genesis().unwrap(); + let tip = hub.tip_hash().unwrap(); + let time = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest) + .header + .time; + let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest); + let fresh = mine_empty_regtest(tip, time + 600, 1); + assert!(hub.has_block(&genesis.block_hash())); + assert!(!hub.has_block(&fresh.block_hash())); + reoffer_unconnected_bodies( + &hub, + [ + (1, genesis.block_hash(), &genesis, None), + (2, fresh.block_hash(), &fresh, None), + ], + ); + assert!( + !hub.query.block_queue_has_height(1), + "a connected body stays off the queue" + ); + assert!( + hub.query.block_queue_has_height(2), + "an unconnected body is offered back" + ); + let _ = std::fs::remove_dir_all(dir); +} + +/// A lone consensus load reject drops the block. `&&` becoming `||` treats +/// that reject as a cascade and puts the body back. The confirm thread +/// classifies the load error. A peer sees the block stay off the tip, which +/// a cascade that immediately retries can also produce. +#[test] +fn load_consensus_rewind_drops_the_block() { + use super::{rewind_after_load_reject, ConfirmRejectClass, LoadAheadState}; + use rbitcoin_consensus::mine_empty_regtest; + + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("load-rewind"); + hub.ensure_genesis().unwrap(); + let tip = hub.tip_hash().unwrap(); + let time = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest) + .header + .time; + let head = mine_empty_regtest(tip, time + 600, 1); + let feed = ConfirmFeed::new(); + let mut st = LoadAheadState::new(&hub); + rewind_after_load_reject( + &feed, + &hub, + &mut st, + 1, + ConfirmRejectClass::ConsensusInvalid, + false, + &[(1, head.block_hash(), &head, None)], + ); + assert!( + !hub.query.block_queue_has_height(1), + "a one-block consensus reject drops the body" + ); + assert!(!feed.single_block()); + let _ = std::fs::remove_dir_all(dir); +} + +/// A stopped confirm does not wait out the parent deadline. Off-tip and +/// on-tip both retry. Same stop arm as `parent_check_reads_the_tip`, on the +/// real clock, bounded so a passing run stays under two seconds. +#[test] +fn stopped_parent_check_is_off_tip() { + use super::{check_missing_parent, ParentCheck}; + use rbitcoin_consensus::mine_empty_regtest; + use std::time::{Duration, Instant}; + + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("parent-check-stop"); + hub.ensure_genesis().unwrap(); + let tip = hub.tip_hash().unwrap(); + let time = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest) + .header + .time; + let on_tip = mine_empty_regtest(tip, time + 600, 1); + let off = mine_empty_regtest( + bitcoin::BlockHash::from_byte_array([0x44; 32]), + time + 600, + 1, + ); + let feed = ConfirmFeed::new(); + feed.request_stop(); + let t0 = Instant::now(); + assert_eq!(check_missing_parent(&hub, &feed, &off), ParentCheck::OffTip); + assert!( + t0.elapsed() < Duration::from_secs(2), + "a stopped feed does not wait out the parent deadline" + ); + assert_eq!( + check_missing_parent(&hub, &feed, &on_tip), + ParentCheck::OffTip, + "confirm is stopping, so a tip parent is a retry" + ); + let _ = std::fs::remove_dir_all(dir); +} diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 6660310749..80260f6fb4 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/crates/rbitcoin-query/src/lib.rs @@ -1189,7 +1189,7 @@ impl Query { } /// Record a tip shrink so load can drop in-flight layers for that height. - pub(crate) fn note_disconnect_height(&self, height: u32) { + pub fn note_disconnect_height(&self, height: u32) { self.disconnect_height .store(height, AtomicOrdering::Release); self.disconnect_gen.fetch_add(1, AtomicOrdering::Release); diff --git a/crates/rbitcoin-rpc/src/methods_tests.rs b/crates/rbitcoin-rpc/src/methods_tests.rs index 39bc6449c2..1c5eacc722 100644 --- a/crates/rbitcoin-rpc/src/methods_tests.rs +++ b/crates/rbitcoin-rpc/src/methods_tests.rs @@ -2477,6 +2477,7 @@ fn mempool_under_pressure() { use bitcoin::{OutPoint, Sequence, Transaction, TxIn, TxOut, Witness}; let (ctx, dir, _hub) = ctx_from_mature_pad(300_000_000); + pressure_same_rate_evicts_the_lighter_chunk(&ctx); let spk = ScriptBuf::from_bytes(vec![0x51]); let (parent_hex, parent) = spend_generated_coinbase(&ctx, 1, 50_0000_0000 - 1_000, spk.clone()); let child = child_of(&parent, 1_000); @@ -2599,6 +2600,182 @@ fn mempool_under_pressure() { let _ = std::fs::remove_dir_all(&dir); } +/// Fee 501 at vsize 5000 and 5001 both floor to 100 sat/kvB and both clear +/// min relay. The lighter txid is linearized first, so the two children stay +/// two chunks. A weight cap that fits the parent and the heavier child drops +/// the lighter one. `<=` in the worst-chunk walk would drop the heavier one. +fn pressure_same_rate_evicts_the_lighter_chunk(ctx: &RpcContext) { + chain_ops_mine_pool_empty(ctx); + let tip = dispatch(ctx, "getblockcount", vec![]) + .unwrap() + .as_u64() + .unwrap(); + if tip < 105 { + dispatch(ctx, "generate", vec![json!(105 - tip)]).unwrap(); + } + let mp = ctx.mempool.as_ref().unwrap(); + mp.set_max_weight(300_000_000); + let cb = generated_coinbase_value(ctx, 4); + let each = (cb - 10_000) / 2; + let light_lens = p2wsh_item_lens(5_000); + let heavy_lens = p2wsh_item_lens(5_001); + let light_proto = p2wsh_with_lens(bitcoin::OutPoint::null(), each, 501, 1, &light_lens); + let heavy_proto = p2wsh_with_lens(bitcoin::OutPoint::null(), each, 501, 1, &heavy_lens); + let parent = parent_paying( + ctx, + 4, + each, + p2wsh_spk(&light_proto), + p2wsh_spk(&heavy_proto), + ); + let pid = parent.compute_txid(); + dispatch(ctx, "sendrawtransaction", vec![json!(tx_hex(&parent))]).unwrap(); + let (light, heavy) = (1..400u32) + .find_map(|n| { + let light = p2wsh_with_lens( + bitcoin::OutPoint { txid: pid, vout: 0 }, + each, + 501, + n, + &light_lens, + ); + let heavy = p2wsh_with_lens( + bitcoin::OutPoint { txid: pid, vout: 1 }, + each, + 501, + n + 400, + &heavy_lens, + ); + (light.compute_txid() < heavy.compute_txid()).then_some((light, heavy)) + }) + .expect("a lighter txid before the heavier sibling"); + assert_eq!( + rbitcoin_consensus::policy::fee_rate_sat_per_kvb(501, light.weight().to_wu()), + rbitcoin_consensus::policy::fee_rate_sat_per_kvb(501, heavy.weight().to_wu()), + ); + dispatch(ctx, "sendrawtransaction", vec![json!(tx_hex(&light))]).unwrap(); + dispatch(ctx, "sendrawtransaction", vec![json!(tx_hex(&heavy))]).unwrap(); + let probe_cb = generated_coinbase_value(ctx, 5); + let (probe_hex, probe) = spend_generated_coinbase(ctx, 5, probe_cb - 20_000, true_spk()); + mp.set_max_weight(parent.weight().to_wu() + heavy.weight().to_wu() + probe.weight().to_wu()); + dispatch(ctx, "sendrawtransaction", vec![json!(probe_hex)]).unwrap(); + let raw = dispatch(ctx, "getrawmempool", vec![]).unwrap(); + let listed = |tx: &Transaction| { + let want = tx.compute_txid().to_string(); + raw.as_array() + .unwrap() + .iter() + .any(|id| id.as_str() == Some(want.as_str())) + }; + assert!(listed(&parent), "parent stays"); + assert!(listed(&heavy), "heavier sibling stays"); + assert!( + !listed(&light), + "the earlier equal-rate chunk is the one evicted" + ); + assert!(listed(&probe), "the arriving tx stays"); + chain_ops_mine_pool_empty(ctx); + mp.set_max_weight(300_000_000); +} + +fn p2wsh_script(n_drops: usize) -> ScriptBuf { + let mut bytes = vec![0x75u8; n_drops]; + bytes.push(0x51); + ScriptBuf::from_bytes(bytes) +} + +fn witness_script(tx: &Transaction) -> ScriptBuf { + let last = tx.input[0].witness.iter().last().expect("witness script"); + ScriptBuf::from_bytes(last.to_vec()) +} + +fn p2wsh_spk(tx: &Transaction) -> ScriptBuf { + ScriptBuf::new_p2wsh(&bitcoin::WScriptHash::hash(witness_script(tx).as_bytes())) +} + +/// Witness-item lengths for a P2WSH `OP_DROP`…`OP_TRUE` spend of `target_vsize`. +/// Stack items stay at or under 520 bytes. The shape does not depend on the prevout. +fn p2wsh_item_lens(target_vsize: u64) -> Vec { + let mut lens = Vec::new(); + loop { + let tx = p2wsh_with_lens(bitcoin::OutPoint::null(), 1_000_000, 1, 1, &lens); + let vsize = rbitcoin_consensus::policy::get_virtual_size(tx.weight().to_wu()); + if vsize == target_vsize { + return lens; + } + assert!( + vsize < target_vsize, + "vsize {vsize} overshot {target_vsize}" + ); + if lens.last().copied().unwrap_or(520) >= 520 { + lens.push(1); + } else { + *lens.last_mut().unwrap() += 1; + } + assert!( + lens.len() < 180, + "witness pad did not reach vsize {target_vsize}" + ); + } +} + +/// P2WSH `OP_DROP`…`OP_TRUE` spend. `lens` is from [`p2wsh_item_lens`]. +fn p2wsh_with_lens( + prev: bitcoin::OutPoint, + value_in: u64, + fee: u64, + nonce: u32, + lens: &[usize], +) -> Transaction { + use bitcoin::absolute::LockTime; + use bitcoin::transaction::Version as TxVersion; + use bitcoin::{Sequence, TxIn, TxOut, Witness}; + let script = p2wsh_script(lens.len()); + let mut stack: Vec> = lens.iter().copied().map(|n| vec![0u8; n]).collect(); + stack.push(script.to_bytes()); + let refs: Vec<&[u8]> = stack.iter().map(Vec::as_slice).collect(); + Transaction { + version: TxVersion::TWO, + lock_time: LockTime::ZERO, + input: vec![TxIn { + previous_output: prev, + script_sig: ScriptBuf::new(), + sequence: Sequence::from_consensus(0x8000_0000 | nonce), + witness: Witness::from_slice(&refs), + }], + output: vec![TxOut { + value: Amount::from_sat(value_in - fee), + script_pubkey: ScriptBuf::from_bytes(vec![0x51]), + }], + } +} + +fn parent_paying( + ctx: &RpcContext, + height: u32, + each: u64, + script0: ScriptBuf, + script1: ScriptBuf, +) -> Transaction { + use bitcoin::TxOut; + let (_, proto) = spend_generated_coinbase(ctx, height, 1, ScriptBuf::from_bytes(vec![0x51])); + Transaction { + version: proto.version, + lock_time: proto.lock_time, + input: proto.input, + output: vec![ + TxOut { + value: Amount::from_sat(each), + script_pubkey: script0, + }, + TxOut { + value: Amount::from_sat(each), + script_pubkey: script1, + }, + ], + } +} + #[test] fn pressure_tiny_weight() { use bitcoin::absolute::LockTime; @@ -2673,10 +2850,12 @@ fn pressure_tiny_weight() { fn pressure_admit_then_cluster(ctx: &RpcContext, spk: &ScriptBuf) { use bitcoin::consensus::encode::serialize; - let (chain_hex, chain_parent) = - spend_generated_coinbase(ctx, 2, 50_0000_0000 - 1_000, spk.clone()); - let chain_child = child_of(&chain_parent, 1_000); - let chain_grand = child_of(&chain_child, 1_000); + // Parent and child are under the relay floor. The grandchild pays for + // the chain. A package walk that stops at the direct parent does not + // admit them. + let (chain_hex, chain_parent) = spend_generated_coinbase(ctx, 2, 50_0000_0000 - 1, spk.clone()); + let chain_child = child_of(&chain_parent, 1); + let chain_grand = child_of(&chain_child, 80_000); let chain_child_hex = hex_encode(serialize(&chain_child)); let chain = dispatch( ctx, @@ -2689,11 +2868,10 @@ fn pressure_admit_then_cluster(ctx: &RpcContext, spk: &ScriptBuf) { ) .unwrap(); assert_eq!(chain["package_msg"], "success", "{chain}"); - assert!(ctx - .mempool - .as_ref() - .unwrap() - .contains(&chain_parent.compute_txid())); + let admitted = ctx.mempool.as_ref().unwrap(); + assert!(admitted.contains(&chain_parent.compute_txid()), "{chain}"); + assert!(admitted.contains(&chain_child.compute_txid()), "{chain}"); + assert!(admitted.contains(&chain_grand.compute_txid()), "{chain}"); let pair = dispatch( ctx, "submitpackage", diff --git a/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs b/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs index 1a5c250695..88adb772ea 100644 --- a/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs +++ b/crates/rbitcoin-rpc/src/regtest_chain_ops_journey.rs @@ -126,6 +126,7 @@ fn rpc_regtest_mature_chain_ops() { chain_ops_submitpackage_sigop_vsize(&ctx, &mut cbs); chain_ops_big_sigops_cluster(&ctx, &mut cbs); chain_ops_prioritise(&ctx, &mut cbs); + chain_ops_template_orders_rate_then_fee(&ctx, &mut cbs); let fee_block = chain_ops_generateblock_and_parent_first(&ctx, &mut cbs); chain_ops_proposal_spends(&ctx, &mut cbs); chain_ops_submit_repeated_txids(&ctx, &hub, &mut cbs); @@ -1390,6 +1391,159 @@ fn true_spk() -> ScriptBuf { ScriptBuf::from_bytes(vec![0x51]) } +fn gbt_txids(ctx: &RpcContext) -> Vec { + let tmpl = dispatch(ctx, "getblocktemplate", vec![json!({"rules": ["segwit"]})]).unwrap(); + tmpl["transactions"] + .as_array() + .unwrap() + .iter() + .map(|t| t["txid"].as_str().unwrap().to_string()) + .collect() +} + +fn assert_template_before(ids: &[String], earlier: &str, later: &str) { + let a = ids.iter().position(|t| t == earlier); + let b = ids.iter().position(|t| t == later); + assert!( + a.is_some() && b.is_some() && a.unwrap() < b.unwrap(), + "{earlier} before {later} in {ids:?}" + ); +} + +fn op_true_child(parent: &Transaction, vout: u32, fee: u64, nonce: u32) -> Transaction { + use bitcoin::absolute::LockTime; + use bitcoin::transaction::Version as TxVersion; + use bitcoin::{OutPoint, Sequence, TxIn, TxOut, Witness}; + Transaction { + version: TxVersion::TWO, + lock_time: LockTime::ZERO, + input: vec![TxIn { + previous_output: OutPoint { + txid: parent.compute_txid(), + vout, + }, + script_sig: ScriptBuf::new(), + sequence: Sequence::from_consensus(0x8000_0000 | nonce), + witness: Witness::new(), + }], + output: vec![TxOut { + value: parent.output[vout as usize].value - Amount::from_sat(fee), + script_pubkey: true_spk(), + }], + } +} + +fn send_tx(ctx: &RpcContext, tx: &Transaction) -> String { + dispatch(ctx, "sendrawtransaction", vec![json!(tx_hex(tx))]) + .unwrap() + .as_str() + .unwrap() + .to_string() +} + +/// `getblocktemplate` lists a cluster in linearization order: higher rate, +/// then higher fee, then the lesser txid. A negative `prioritisetransaction` +/// drops that child from the template. +fn chain_ops_template_orders_rate_then_fee(ctx: &RpcContext, cbs: &mut TrueCoinbases) { + let h = cbs.take(); + let cb = generated_coinbase_value(ctx, h); + let each = (cb - 10_000) / 2; + let parent = parent_paying(ctx, h, each, true_spk(), true_spk()); + send_tx(ctx, &parent); + let (hi_rate, lo_rate) = (1..400u32) + .find_map(|n| { + let hi = op_true_child(&parent, 0, 8_000, n); + let lo = op_true_child(&parent, 1, 1_000, n + 400); + (hi.compute_txid() > lo.compute_txid()).then_some((hi, lo)) + }) + .expect("higher-fee child txid after the lower-fee child"); + assert_eq!(hi_rate.weight(), lo_rate.weight()); + let hi_rate_id = send_tx(ctx, &hi_rate); + let lo_rate_id = send_tx(ctx, &lo_rate); + let ids = gbt_txids(ctx); + assert_template_before(&ids, &hi_rate_id, &lo_rate_id); + chain_ops_mine_pool_empty(ctx); + + let h = cbs.take(); + let cb = generated_coinbase_value(ctx, h); + let each = (cb - 10_000) / 2; + let parent = parent_paying(ctx, h, each, true_spk(), true_spk()); + send_tx(ctx, &parent); + let a = op_true_child(&parent, 0, 2_000, 1); + let b = op_true_child(&parent, 1, 2_000, 2); + assert_eq!(a.weight(), b.weight()); + let a_id = send_tx(ctx, &a); + let b_id = send_tx(ctx, &b); + let (lesser, greater) = if a.compute_txid() < b.compute_txid() { + (a_id, b_id) + } else { + (b_id, a_id) + }; + let ids = gbt_txids(ctx); + assert_template_before(&ids, &lesser, &greater); + dispatch( + ctx, + "prioritisetransaction", + vec![json!(lesser.clone()), json!(0), json!(-2_001)], + ) + .unwrap(); + let ids = gbt_txids(ctx); + assert!( + !ids.iter().any(|t| t == &lesser), + "a negative fee delta leaves the template: {ids:?}" + ); + assert!(ids.iter().any(|t| t == &greater), "{ids:?}"); + dispatch( + ctx, + "prioritisetransaction", + vec![json!(lesser), json!(0), json!(2_001)], + ) + .unwrap(); + chain_ops_mine_pool_empty(ctx); + + // 400_000 sat at vsize 2000 and 200_000 sat at vsize 1000 floor to the + // same sat/kvB. The higher fee is the later txid, so the fee compare is + // what puts it first in the template. + let h = cbs.take(); + let cb = generated_coinbase_value(ctx, h); + let each = (cb - 10_000) / 2; + let hi_lens = p2wsh_item_lens(2_000); + let lo_lens = p2wsh_item_lens(1_000); + let hi_proto = p2wsh_with_lens(bitcoin::OutPoint::null(), each, 400_000, 1, &hi_lens); + let lo_proto = p2wsh_with_lens(bitcoin::OutPoint::null(), each, 200_000, 1, &lo_lens); + assert_eq!( + rbitcoin_consensus::policy::fee_rate_sat_per_kvb(400_000, hi_proto.weight().to_wu()), + rbitcoin_consensus::policy::fee_rate_sat_per_kvb(200_000, lo_proto.weight().to_wu()), + ); + let parent = parent_paying(ctx, h, each, p2wsh_spk(&hi_proto), p2wsh_spk(&lo_proto)); + let pid = parent.compute_txid(); + send_tx(ctx, &parent); + let (hi, lo) = (1..400u32) + .find_map(|n| { + let hi = p2wsh_with_lens( + bitcoin::OutPoint { txid: pid, vout: 0 }, + each, + 400_000, + n, + &hi_lens, + ); + let lo = p2wsh_with_lens( + bitcoin::OutPoint { txid: pid, vout: 1 }, + each, + 200_000, + n + 400, + &lo_lens, + ); + (hi.compute_txid() > lo.compute_txid()).then_some((hi, lo)) + }) + .expect("higher-fee txid after the lower-fee sibling"); + let hi_id = send_tx(ctx, &hi); + let lo_id = send_tx(ctx, &lo); + let ids = gbt_txids(ctx); + assert_template_before(&ids, &hi_id, &lo_id); + chain_ops_mine_pool_empty(ctx); +} + fn chain_ops_generateblock_and_parent_first(ctx: &RpcContext, cbs: &mut TrueCoinbases) -> u32 { // Core `rpc_generate.py` generateblock reject strings and codes. let op_true = "raw(51)"; diff --git a/crates/rbitcoin-test/tests/integration_multinode.rs b/crates/rbitcoin-test/tests/integration_multinode.rs index da09df120f..a2e8575ce3 100644 --- a/crates/rbitcoin-test/tests/integration_multinode.rs +++ b/crates/rbitcoin-test/tests/integration_multinode.rs @@ -190,11 +190,31 @@ fn llvm_cov_wall(default_secs: u64, llvm_secs: u64) -> Duration { } } -fn ephemeral_addr() -> SocketAddr { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let addr = listener.local_addr().unwrap(); - drop(listener); - addr +fn any_port() -> SocketAddr { + "127.0.0.1:0".parse().unwrap() +} + +/// The address `run_p2p` bound for `name`, from `{datadir}/run/{name}.addr`. +/// A bind-then-drop reserve races another test onto the same port. +async fn published_listener(datadir: &std::path::Path, name: &str) -> SocketAddr { + let path = datadir.join("run").join(format!("{name}.addr")); + let deadline = std::time::Instant::now() + Duration::from_secs(20); + loop { + if let Ok(text) = std::fs::read_to_string(&path) { + if let Ok(addr) = text.trim().parse::() { + if addr.port() != 0 { + return addr; + } + } + } + if std::time::Instant::now() >= deadline { + panic!( + "bound {name} address was not published at {}", + path.display() + ); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } } async fn wait_listeners(addrs: &[SocketAddr]) { @@ -3004,21 +3024,19 @@ async fn node_run_p2p_short() { let seed = start_node(&seed_dir).await; seed_chain(&seed, 3).await; let seed_addr = seed.local_addr; - let rpc_addr = ephemeral_addr(); - let electrum_addr = ephemeral_addr(); - let esplora_addr = ephemeral_addr(); + let node_path = node_dir.path().to_path_buf(); let mut cfg = NodeConfig::default() .with_datadir(node_dir.path()) .with_network(Network::Regtest) - .with_p2p_listen("127.0.0.1:0".parse().unwrap()) + .with_p2p_listen(any_port()) .with_tiny_heads(); cfg.listen.connect = vec![NetAddr::Ip(seed_addr)]; cfg.listen.use_seeds = false; - cfg.listen.electrum = Some(electrum_addr); - cfg.listen.esplora = Some(rbitcoin_esplora::EsploraListen::Tcp(esplora_addr)); + cfg.listen.electrum = Some(any_port()); + cfg.listen.esplora = Some(rbitcoin_esplora::EsploraListen::Tcp(any_port())); cfg.shindex = true; - cfg.rpc.listen = Some(rpc_addr); + cfg.rpc.listen = Some(any_port()); std::fs::write(node_dir.path().join("rpc.token"), "pass").unwrap(); cfg.max_run_secs = Some(60); cfg.mempool.blocksonly = true; @@ -3026,6 +3044,9 @@ async fn node_run_p2p_short() { let seed_peers = seed.peers.clone(); let pin = tokio::spawn(async move { + let rpc_addr = published_listener(&node_path, "rpc").await; + let electrum_addr = published_listener(&node_path, "electrum").await; + let esplora_addr = published_listener(&node_path, "esplora").await; wait_listeners(&[rpc_addr, electrum_addr, esplora_addr]).await; let deadline = Instant::now() + Duration::from_secs(20);