From 01742248c7e7f0956333babd2cf65014e81ac3c3 Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Thu, 15 Oct 2026 22:37:24 -0600 Subject: [PATCH 1/5] mempool: keep each entry's accept time across a restart Entry time lived only in the hub's RAM maps, filled on admit. Hub open loaded the live set but never indexed it, so after a restart getmempoolentry.time was 0 and -mempoolexpiry never removed a loaded entry. The mempool sidecar is schema 4: each packed record stores its accept time after sigops. The mempool stamps it from the hub's mock clock (now shared, so setmocktime still reaches it), AcceptResult carries it, and the hub indexes admits with that same value. At open the hub indexes loaded entries with their stored time, without queueing an INV. Schema 1-3 records migrate on open with the migrate's wall-clock second. Refs #1029 Co-Authored-By: Claude Opus 5.5 --- crates/rbitcoin-mempool/src/accept.rs | 47 ++- crates/rbitcoin-mempool/src/packed.rs | 109 +++++-- crates/rbitcoin-mempool/src/store.rs | 339 +++++++++++++++++---- crates/rbitcoin-net/src/tx_relay.rs | 112 +++++-- crates/rbitcoin-test/tests/core_analogs.rs | 4 +- docs/ibd-memory.md | 2 +- docs/io-modality.md | 2 +- docs/operator/operations.md | 6 +- 8 files changed, 514 insertions(+), 107 deletions(-) diff --git a/crates/rbitcoin-mempool/src/accept.rs b/crates/rbitcoin-mempool/src/accept.rs index 7209d95c3..36e165ebe 100644 --- a/crates/rbitcoin-mempool/src/accept.rs +++ b/crates/rbitcoin-mempool/src/accept.rs @@ -11,6 +11,7 @@ use bitcoin::{OutPoint, Transaction, TxOut, Txid, Wtxid}; use rbitcoin_consensus::policy::{self, PolicyResult}; use rbitcoin_primitives::Fk; use std::collections::{BTreeMap, BTreeSet, HashSet, VecDeque}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::Instant; @@ -169,6 +170,8 @@ pub struct AcceptResult { /// Sigop-adjusted weight; `vsize = ceil(weight / 4)` is Core's `m_vsize`. pub weight: u64, pub slot: u32, + /// Node-time seconds stamped on the durable record (Core `nTime`). + pub accept_secs: u64, /// Mempool txids removed by full-RBF / RBFR when admitting this tx (empty if no conflict). pub replaced: Vec, /// Electrum scripthashes (SHA256 of output scriptPubKeys) of **replaced** bodies, @@ -447,6 +450,8 @@ struct IngestedLive { graph: TxGraph, bodies: std::collections::HashMap>, vin_aux: std::collections::HashMap>, + /// `(txid, wtxid, accept_secs)` per loaded record. + accept_secs: Vec<(Txid, Wtxid, u64)>, } /// Mempool with RAM TxGraph layered on durable store. @@ -484,9 +489,31 @@ pub struct ActiveMempool { /// Bodies removed for replacement when `finish_commit` later returns /// `Err`. Success moves them onto [`AcceptResult::replaced_txs`]. undone_replacements: Vec, + /// `setmocktime` seconds, shared with the hub's clock (0 = wall). + node_clock: Arc, + /// Accept times of the records loaded at open, until the hub takes them. + loaded_accept: Vec<(Txid, Wtxid, u64)>, } impl ActiveMempool { + /// Share the hub's mock clock so stamped accept times follow `setmocktime`. + pub fn set_node_clock(&mut self, clock: Arc) { + self.node_clock = clock; + } + + fn node_now_secs(&self) -> u64 { + match self.node_clock.load(Ordering::Relaxed) { + 0 => unix_ms() / 1000, + mock => mock, + } + } + + /// `(txid, wtxid, accept_secs)` of every record loaded at open. Empty + /// after the first call; the hub indexes them for expiry and RPC. + pub fn take_loaded_accept_times(&mut self) -> Vec<(Txid, Wtxid, u64)> { + std::mem::take(&mut self.loaded_accept) + } + /// Evictions from a commit that returned `Err`. Empty after a success. pub fn take_failed_evictions(&mut self) -> Vec { std::mem::take(&mut self.last_evicted) @@ -606,6 +633,8 @@ impl ActiveMempool { extra_compact: VecDeque::new(), last_evicted: Vec::new(), undone_replacements: Vec::new(), + node_clock: Arc::new(AtomicU64::new(0)), + loaded_accept: ingested.accept_secs, }) } @@ -695,8 +724,10 @@ impl ActiveMempool { let mut bodies = std::collections::HashMap::new(); let mut vin_aux = std::collections::HashMap::new(); let mut items = Vec::with_capacity(loaded.len()); + let mut accept_secs = Vec::with_capacity(loaded.len()); for live in loaded { let p = live.packed; + accept_secs.push((p.txid, p.wtxid, p.accept_secs)); let tx = Arc::new(p.tx); let entry = TxEntry { txid: p.txid, @@ -719,6 +750,7 @@ impl ActiveMempool { graph, bodies, vin_aux, + accept_secs, } } @@ -1111,6 +1143,7 @@ impl ActiveMempool { let mut evicted = self.ensure_free_slot(Some(txid))?; let aux = Self::vin_aux_from_prep(tx, &prep); + let accept_secs = self.node_now_secs(); let t_dur = Instant::now(); let slot = match self.store.append_live_tx( tx, @@ -1119,6 +1152,7 @@ impl ActiveMempool { fee_sat, weight, prep.sigop_cost, + accept_secs, &aux, ) { Ok(slot) => slot, @@ -1174,6 +1208,7 @@ impl ActiveMempool { fee_sat, weight: adj_weight, slot, + accept_secs, replaced: conflict_set.into_iter().collect(), replaced_scripthashes, replaced_txs, @@ -1194,6 +1229,7 @@ impl ActiveMempool { fee_sat, weight, slot: 0, + accept_secs: self.node_now_secs(), replaced: conflict_set.into_iter().collect(), replaced_scripthashes: Vec::new(), replaced_txs: Vec::new(), @@ -2815,7 +2851,16 @@ mod tests { // Admitted by a pre-sigop-check build. let (pid, pw) = (probe.compute_txid(), probe.compute_wtxid()); mp.store - .append_live_tx(&probe, &pid, &pw, 1_000, probe.weight().to_wu(), 0, &[]) + .append_live_tx( + &probe, + &pid, + &pw, + 1_000, + probe.weight().to_wu(), + 0, + 1_700_000_000, + &[], + ) .unwrap(); mp.flush().unwrap(); } diff --git a/crates/rbitcoin-mempool/src/packed.rs b/crates/rbitcoin-mempool/src/packed.rs index ffa1ac140..442142b98 100644 --- a/crates/rbitcoin-mempool/src/packed.rs +++ b/crates/rbitcoin-mempool/src/packed.rs @@ -1,7 +1,10 @@ -//! MEM_SCHEMA 3 packed live records (`fee‖weight‖sigops‖txid‖wtxid‖tx‖vin aux`). +//! MEM_SCHEMA 4 packed live records +//! (`fee‖weight‖sigops‖accept_secs‖txid‖wtxid‖tx‖vin aux`). //! -//! Schema 2 had no `sigops`; [`decode_packed_live_v2`] reads it for the soft -//! migrate, which writes [`SIGOP_COST_UNKNOWN`] until the hub recomputes. +//! Schema 3 had no `accept_secs` and schema 2 had no `sigops` either; +//! [`decode_packed_live_v3`] / [`decode_packed_live_v2`] read them for the soft +//! migrate, which stamps the migrate time and writes [`SIGOP_COST_UNKNOWN`] +//! until the hub recomputes. //! //! Vin codec mirrors Class A packed flags/witness but writes `prev_txid` when //! there is no Class A `create_fk`. Outputs reuse [`OutputRecord`] unspent @@ -22,7 +25,7 @@ const VIN_HAS_SCRIPT_HASH: u8 = 1 << 4; /// `sigops` value for a record migrated from schema ≤ 2 (cost not yet known). const SIGOP_COST_UNKNOWN: u64 = u64::MAX; -/// Byte offset of `sigops` in a schema-3 record. +/// Byte offset of `sigops` in a schema-3 or schema-4 record. pub(crate) const SIGOP_COST_OFF: usize = 16; /// Per-vin aux persisted with the packed body (SH index / tip-entry purge). @@ -34,19 +37,23 @@ pub struct VinAux { pub create_fk: Option, } -/// One LIVE body payload (schema 3). +/// One LIVE body payload (schema 4). #[derive(Debug, Clone)] pub struct PackedLive { pub fee_sat: u64, pub weight: u64, /// Full sigop cost; `None` for a record migrated from schema ≤ 2. pub sigop_cost: Option, + /// Node-time seconds at accept (Core `nTime`). A record read from schema + /// ≤ 3 has none and reads 0; the migrate stamps it. + pub accept_secs: u64, pub txid: Txid, pub wtxid: Wtxid, pub tx: Transaction, pub vins: Vec, } +#[allow(clippy::too_many_arguments)] // one field per record column pub fn encode_packed_live( tx: &Transaction, txid: &Txid, @@ -54,6 +61,7 @@ pub fn encode_packed_live( fee_sat: u64, weight: u64, sigop_cost: Option, + accept_secs: u64, vins: &[VinAux], ) -> Result, MempoolError> { if vins.len() != tx.input.len() && !vins.is_empty() { @@ -64,6 +72,7 @@ pub fn encode_packed_live( out.extend_from_slice(&weight.to_le_bytes()); let sigops = sigop_cost.unwrap_or(SIGOP_COST_UNKNOWN); out.extend_from_slice(&sigops.to_le_bytes()); + out.extend_from_slice(&accept_secs.to_le_bytes()); out.extend_from_slice(txid.as_byte_array()); out.extend_from_slice(wtxid.as_byte_array()); out.extend_from_slice(&tx.version.0.to_le_bytes()); @@ -81,16 +90,25 @@ pub fn encode_packed_live( } pub fn decode_packed_live(buf: &[u8]) -> Result { - decode_packed(buf, true) + decode_packed(buf, true, true) } -/// Schema-2 record (no `sigops` field) for the 2 → 3 soft migrate. +/// Schema-3 record (no `accept_secs` field) for the 3 → 4 soft migrate. +pub(crate) fn decode_packed_live_v3(buf: &[u8]) -> Result { + decode_packed(buf, true, false) +} + +/// Schema-2 record (no `sigops` or `accept_secs`) for the 2 → 4 soft migrate. pub(crate) fn decode_packed_live_v2(buf: &[u8]) -> Result { - decode_packed(buf, false) + decode_packed(buf, false, false) } -fn decode_packed(buf: &[u8], has_sigops: bool) -> Result { - let ids = if has_sigops { 24 } else { 16 }; +fn decode_packed( + buf: &[u8], + has_sigops: bool, + has_accept: bool, +) -> Result { + let ids = 16 + if has_sigops { 8 } else { 0 } + if has_accept { 8 } else { 0 }; if buf.len() < ids + 32 + 32 + 8 { return Err(MempoolError::Corrupt("packed live short")); } @@ -105,6 +123,11 @@ fn decode_packed(buf: &[u8], has_sigops: bool) -> Result Result Transaction { let mut witness = Witness::new(); witness.push([0x01, 0x02]); @@ -342,18 +370,25 @@ mod tests { script_hash: Some(sh), create_fk: Some(Fk(42)), }]; - let packed = encode_packed_live(&tx, &txid, &wtxid, 123, 400, Some(80), &aux).unwrap(); + let packed = + encode_packed_live(&tx, &txid, &wtxid, 123, 400, Some(80), 1_700_000_123, &aux) + .unwrap(); let got = decode_packed_live(&packed).unwrap(); assert_eq!(got.fee_sat, 123); assert_eq!(got.weight, 400); assert_eq!(got.sigop_cost, Some(80)); + assert_eq!(got.accept_secs, 1_700_000_123); assert_eq!(got.txid, txid); assert_eq!(got.wtxid, wtxid); 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]); + assert_eq!( + packed[VIN0_FLAGS], 0b0001_1011, + "vin flags {:08b}", + packed[VIN0_FLAGS] + ); } /// Vin flag bits and a script or witness that ends the record. Encode and @@ -366,10 +401,15 @@ mod tests { 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(); + let packed = + encode_packed_live(&bare, &txid, &wtxid, 1, 400, Some(1), 1_700_000_000, &[]).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]); + assert_eq!( + packed[VIN0_FLAGS], 0b0000_0111, + "vin flags {:08b}", + packed[VIN0_FLAGS] + ); let got = decode_packed_live(&packed).unwrap(); assert_eq!(serialize(&got.tx), serialize(&bare)); @@ -377,9 +417,14 @@ mod tests { 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(); + let packed = + encode_packed_live(&bare, &txid, &wtxid, 1, 400, Some(1), 1_700_000_000, &[]).unwrap(); // Non-empty script, empty witness, non-final sequence: only EMPTY_WITNESS. - assert_eq!(packed[98], 0b0000_0100, "vin flags {:08b}", packed[98]); + assert_eq!( + packed[VIN0_FLAGS], 0b0000_0100, + "vin flags {:08b}", + packed[VIN0_FLAGS] + ); 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)); @@ -421,7 +466,7 @@ mod tests { 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() + encode_packed_live(&tx, &txid, &wtxid, 1, 400, Some(1), 1_700_000_000, &[]).unwrap() } /// Input and output counts past 100_000. A live mempool tx cannot be that @@ -430,9 +475,10 @@ mod tests { 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 seed = + encode_packed_live(&tx, &txid, &wtxid, 1, 400, Some(1), 1_700_000_000, &[]).unwrap(); let with_counts = |n_in: u64, n_out: u64| { - let mut buf = seed[..96].to_vec(); + let mut buf = seed[..104].to_vec(); write_compact_size(&mut buf, n_in); write_compact_size(&mut buf, n_out); buf @@ -450,27 +496,38 @@ mod tests { } #[test] - fn unknown_sigops_and_schema2_record_decode_as_none() { + fn unknown_sigops_and_older_schema_records_decode() { let tx = sample_tx(false); let (txid, wtxid) = (tx.compute_txid(), tx.compute_wtxid()); - let v3 = encode_packed_live(&tx, &txid, &wtxid, 7, 400, None, &[]).unwrap(); - assert_eq!(decode_packed_live(&v3).unwrap().sigop_cost, None); + let v4 = encode_packed_live(&tx, &txid, &wtxid, 7, 400, None, 1_700_000_000, &[]).unwrap(); + assert_eq!(decode_packed_live(&v4).unwrap().sigop_cost, None); // A torn 8-byte write-back over 0xFF.. overestimates; any cost no block // can hold reads as unknown so the open-time pass recomputes it. for (stored, want) in [(80_000, Some(80_000)), (80_001, None)] { - let rec = encode_packed_live(&tx, &txid, &wtxid, 7, 400, Some(stored), &[]).unwrap(); + let rec = + encode_packed_live(&tx, &txid, &wtxid, 7, 400, Some(stored), 1_700_000_000, &[]) + .unwrap(); assert_eq!(decode_packed_live(&rec).unwrap().sigop_cost, want); } + let mut v3 = v4.clone(); + v3.drain(SIGOP_COST_OFF + 8..SIGOP_COST_OFF + 16); + let got = decode_packed_live_v3(&v3).unwrap(); + assert_eq!((got.fee_sat, got.weight, got.sigop_cost), (7, 400, None)); + assert_eq!(got.accept_secs, 0, "schema 3 has no accept time"); + assert_eq!(got.txid, txid); + assert_eq!(serialize(&got.tx), serialize(&tx)); let mut v2 = v3.clone(); v2.drain(SIGOP_COST_OFF..SIGOP_COST_OFF + 8); let got = decode_packed_live_v2(&v2).unwrap(); assert_eq!((got.fee_sat, got.weight, got.sigop_cost), (7, 400, None)); assert_eq!(got.txid, txid); assert_eq!(serialize(&got.tx), serialize(&tx)); - // Fixed prefix is 96 B (schema 3) / 88 B (schema 2): one byte short - // fails the length guard; exactly the prefix reaches the input count. + // Fixed prefix is 104 B (schema 4) / 96 B (schema 3) / 88 B (schema 2): + // one byte short fails the length guard; exactly the prefix reaches the + // input count. for (buf, min, decode) in [ - (&v3, 96, decode_packed_live as fn(&[u8]) -> _), + (&v4, 104, decode_packed_live as fn(&[u8]) -> _), + (&v3, 96, decode_packed_live_v3), (&v2, 88, decode_packed_live_v2), ] { assert!(matches!( diff --git a/crates/rbitcoin-mempool/src/store.rs b/crates/rbitcoin-mempool/src/store.rs index 7d1971800..fdcf86c8f 100644 --- a/crates/rbitcoin-mempool/src/store.rs +++ b/crates/rbitcoin-mempool/src/store.rs @@ -23,8 +23,8 @@ use crate::error::MempoolError; use crate::packed::{ - decode_packed_live, decode_packed_live_v2, encode_packed_live, PackedLive, VinAux, - SIGOP_COST_OFF, + decode_packed_live, decode_packed_live_v2, decode_packed_live_v3, encode_packed_live, + PackedLive, VinAux, SIGOP_COST_OFF, }; use bitcoin::consensus::encode::deserialize; use bitcoin::hashes::Hash; @@ -37,7 +37,7 @@ use std::time::Instant; /// File magic: `rBMP` (rbitcoin mempool). pub const MEM_MAGIC: [u8; 4] = *b"rBMP"; /// Schema version for the meta header. -pub const MEM_SCHEMA: u16 = 3; +pub const MEM_SCHEMA: u16 = 4; const META_LEN: usize = 64; /// Initial slot table capacity (records). @@ -325,9 +325,19 @@ impl Mempool { fee_sat: u64, weight: u64, sigop_cost: u64, + accept_secs: u64, vins: &[VinAux], ) -> Result { - let payload = encode_packed_live(tx, txid, wtxid, fee_sat, weight, Some(sigop_cost), vins)?; + let payload = encode_packed_live( + tx, + txid, + wtxid, + fee_sat, + weight, + Some(sigop_cost), + accept_secs, + vins, + )?; if u32::try_from(payload.len()).is_err() { return Err(MempoolError::Corrupt("tx body too large")); } @@ -466,14 +476,20 @@ impl Mempool { Ok((self.live_count, packed_len)) } - /// Recode leftover schema-1 (`fee‖weight‖raw_tx`) or schema-2 packed LIVE - /// slots into schema 3. + /// Recode leftover schema-1 (`fee‖weight‖raw_tx`), schema-2 or schema-3 + /// packed LIVE slots into schema 4. /// - /// Same install as compact (tmp+rename). Sigop cost is written unknown; + /// Same install as compact (tmp+rename). Older records carry no accept + /// time, so each gets the migrate's wall-clock second: expiry counts from + /// this open. Schema ≤ 2 sigop cost is written unknown; /// [`crate::ActiveMempool::recompute_missing_sigops`] fills it after open. /// Schema-1 vin aux is empty (SH reindex batch-fills). Other schemas refuse /// at header read ([`accepted_schema`]). fn migrate_to_current(&mut self, from: u16) -> Result<(), MempoolError> { + let accept_secs = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); let logical = body_logical_len(&self.body)?; let mut new_body = vec![0u8; BODY_HEADER]; new_body[0..4].copy_from_slice(&MEM_MAGIC); @@ -503,13 +519,16 @@ impl Mempool { fee_sat: u64::from_le_bytes(rec[0..8].try_into().unwrap()), weight: u64::from_le_bytes(rec[8..16].try_into().unwrap()), sigop_cost: None, + accept_secs: 0, txid: tx.compute_txid(), wtxid: tx.compute_wtxid(), tx, vins: Vec::new(), } - } else { + } else if from == 2 { decode_packed_live_v2(rec)? + } else { + decode_packed_live_v3(rec)? }; let payload = encode_packed_live( &old.tx, @@ -517,7 +536,8 @@ impl Mempool { &old.wtxid, old.fee_sat, old.weight, - None, + old.sigop_cost, + accept_secs, &old.vins, )?; let new_off = new_body.len() as u64; @@ -994,7 +1014,7 @@ pub(crate) fn quarantine_sidecar(dir: &Path) -> Result { } fn accepted_schema(schema: u16) -> Result { - if matches!(schema, 1 | 2) || schema == MEM_SCHEMA { + if matches!(schema, 1..=3) || schema == MEM_SCHEMA { Ok(schema) } else { Err(MempoolError::BadSchema(schema)) @@ -1143,9 +1163,19 @@ fn open_or_init_body(path: &Path) -> Result<(File, Vec), MempoolError> { pub(crate) mod tests { use super::*; - /// Rewrite a closed schema-3 pool under `dir` as schema 2 (drop each - /// record's sigops field, stamp headers 2) — pre-upgrade fixture. + /// Rewrite a closed current pool under `dir` as schema 2 (drop each + /// record's sigops and accept time, stamp headers 2) — pre-upgrade fixture. pub(crate) fn downgrade_to_schema2(dir: &Path) { + downgrade_to(dir, 2); + } + + /// Rewrite a closed current pool under `dir` as schema 3 (drop each + /// record's accept time, stamp headers 3) — pre-upgrade fixture. + pub(crate) fn downgrade_to_schema3(dir: &Path) { + downgrade_to(dir, 3); + } + + fn downgrade_to(dir: &Path, schema: u16) { let mut slots = fs::read(dir.join("slots")).unwrap(); let body = fs::read(dir.join("tx.body")).unwrap(); let cap = u32::from_le_bytes(slots[8..12].try_into().unwrap()); @@ -1158,17 +1188,23 @@ pub(crate) mod tests { let b_off = u64::from_le_bytes(slots[off + 4..off + 12].try_into().unwrap()) as usize; let b_len = u32::from_le_bytes(slots[off + 12..off + 16].try_into().unwrap()) as usize; let mut rec = body[b_off..b_off + b_len].to_vec(); - rec.drain(SIGOP_COST_OFF..SIGOP_COST_OFF + 8); + // Accept time sits right after sigops; schema 2 drops both. + let from = if schema == 2 { + SIGOP_COST_OFF + } else { + SIGOP_COST_OFF + 8 + }; + rec.drain(from..SIGOP_COST_OFF + 16); slots[off + 4..off + 12].copy_from_slice(&(new_body.len() as u64).to_le_bytes()); slots[off + 12..off + 16].copy_from_slice(&(rec.len() as u32).to_le_bytes()); new_body.extend_from_slice(&rec); } let len = new_body.len() as u64; new_body[8..16].copy_from_slice(&len.to_le_bytes()); - new_body[4..6].copy_from_slice(&2u16.to_le_bytes()); - slots[4..6].copy_from_slice(&2u16.to_le_bytes()); + new_body[4..6].copy_from_slice(&schema.to_le_bytes()); + slots[4..6].copy_from_slice(&schema.to_le_bytes()); let mut meta = fs::read(dir.join("meta")).unwrap(); - meta[4..6].copy_from_slice(&2u16.to_le_bytes()); + meta[4..6].copy_from_slice(&schema.to_le_bytes()); fs::write(dir.join("tx.body"), new_body).unwrap(); fs::write(dir.join("slots"), slots).unwrap(); fs::write(dir.join("meta"), meta).unwrap(); @@ -1188,8 +1224,17 @@ pub(crate) mod tests { script_hash: Some(sh), create_fk: None, }]; - mp.append_live_tx(&tx, &tid, &tx.compute_wtxid(), 55, 400, 12, &aux) - .unwrap(); + mp.append_live_tx( + &tx, + &tid, + &tx.compute_wtxid(), + 55, + 400, + 12, + 1_700_000_000, + &aux, + ) + .unwrap(); mp.flush().unwrap(); } downgrade_to_schema2(&dir); @@ -1211,6 +1256,57 @@ pub(crate) mod tests { assert_eq!(mp.load_live_txs().unwrap()[0].packed.sigop_cost, Some(12)); } + /// Schema 3 records keep fee, weight, sigops and vin aux, and get the + /// migrate's wall-clock second as their accept time. + #[test] + fn schema3_converts_on_open_with_migrate_accept_time() { + let dir = tmp_dir(); + let tx = tiny_tx(); + let tid = tx.compute_txid(); + { + let mut mp = Mempool::open_or_create(&dir).unwrap(); + mp.append_live_tx( + &tx, + &tid, + &tx.compute_wtxid(), + 55, + 400, + 12, + 1_700_000_000, + &[], + ) + .unwrap(); + mp.flush().unwrap(); + } + downgrade_to_schema3(&dir); + let before = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let mp = Mempool::open_or_create(&dir).expect("schema 3 converts"); + for f in ["meta", "slots", "tx.body"] { + let b = fs::read(dir.join(f)).unwrap(); + assert_eq!(&b[4..6], &MEM_SCHEMA.to_le_bytes(), "{f} stamped"); + } + let live = mp.load_live_txs().unwrap(); + assert_eq!(live.len(), 1); + let p = &live[0].packed; + assert_eq!((p.fee_sat, p.weight, p.txid), (55, 400, tid)); + assert_eq!(p.sigop_cost, Some(12), "schema 3 cost kept"); + assert!( + p.accept_secs >= before && p.accept_secs < before + 60, + "migrate stamps its own time: {} vs {before}", + p.accept_secs + ); + drop(mp); + let mp = Mempool::open_or_create(&dir).unwrap(); + assert_eq!( + mp.load_live_txs().unwrap()[0].packed.accept_secs, + p.accept_secs, + "the stamp is durable, not redone each open" + ); + } + /// A legacy LIVE slot whose range runs past the body, or is shorter than /// the 16-byte fee‖weight prefix, refuses open as Corrupt (no panic). #[test] @@ -1226,8 +1322,17 @@ pub(crate) mod tests { let dir = tmp_dir(); { let mut mp = Mempool::open_or_create(&dir).unwrap(); - mp.append_live_tx(&tx, &tid, &tx.compute_wtxid(), 55, 400, 12, &[]) - .unwrap(); + mp.append_live_tx( + &tx, + &tid, + &tx.compute_wtxid(), + 55, + 400, + 12, + 1_700_000_000, + &[], + ) + .unwrap(); mp.flush().unwrap(); } downgrade_to_schema2(&dir); @@ -1290,8 +1395,17 @@ pub(crate) mod tests { fn put_live(mp: &mut Mempool, tid: &Txid, fee: u64, weight: u64) -> u32 { let tx = tiny_tx(); - mp.append_live_tx(&tx, tid, &tx.compute_wtxid(), fee, weight, 0, &[]) - .unwrap() + mp.append_live_tx( + &tx, + tid, + &tx.compute_wtxid(), + fee, + weight, + 0, + 1_700_000_000, + &[], + ) + .unwrap() } #[test] @@ -1457,8 +1571,17 @@ pub(crate) mod tests { mp.set_now_ms(0); let raw = tiny_tx(); let t1 = Txid::from_byte_array([0x01; 32]); - mp.append_live_tx(&raw, &t1, &raw.compute_wtxid(), 1, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t1, + &raw.compute_wtxid(), + 1, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); let persisted = mp.body_persisted_len(); mp.persist_slots_without_body().unwrap(); drop(mp); @@ -1483,8 +1606,17 @@ pub(crate) mod tests { mp.set_now_ms(0); let raw = tiny_tx(); let torn = Txid::from_byte_array([0x11; 32]); - mp.append_live_tx(&raw, &torn, &raw.compute_wtxid(), 1, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &torn, + &raw.compute_wtxid(), + 1, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); mp.flush().unwrap(); drop(mp); let mut body = fs::read(dir.join("tx.body")).unwrap(); @@ -1547,15 +1679,33 @@ pub(crate) mod tests { mp.set_now_ms(0); let raw = tiny_tx(); let t1 = Txid::from_byte_array([0x01; 32]); - mp.append_live_tx(&raw, &t1, &raw.compute_wtxid(), 1, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t1, + &raw.compute_wtxid(), + 1, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); mp.set_now_ms(PERSIST_INTERVAL_MS); mp.persist_due().unwrap(); let first_end = mp.body_persisted_len(); let body_after_first = fs::read(dir.join("tx.body")).unwrap(); let t2 = Txid::from_byte_array([0x02; 32]); - mp.append_live_tx(&raw, &t2, &raw.compute_wtxid(), 2, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t2, + &raw.compute_wtxid(), + 2, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); mp.set_now_ms(PERSIST_INTERVAL_MS * 2); mp.persist_due().unwrap(); let body_after_second = fs::read(dir.join("tx.body")).unwrap(); @@ -1577,16 +1727,43 @@ pub(crate) mod tests { mp.set_now_ms(0); let raw = tiny_tx(); let t1 = Txid::from_byte_array([0x01; 32]); - mp.append_live_tx(&raw, &t1, &raw.compute_wtxid(), 1, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t1, + &raw.compute_wtxid(), + 1, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); mp.set_now_ms(PERSIST_INTERVAL_MS); mp.persist_due().unwrap(); let t2 = Txid::from_byte_array([0x02; 32]); - mp.append_live_tx(&raw, &t2, &raw.compute_wtxid(), 2, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t2, + &raw.compute_wtxid(), + 2, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); let t3 = Txid::from_byte_array([0x03; 32]); - mp.append_live_tx(&raw, &t3, &raw.compute_wtxid(), 3, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t3, + &raw.compute_wtxid(), + 3, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); // A byte just past the new slot records. A widened record range rewrites it. let past = SLOTS_HEADER + 3 * SLOT_REC; { @@ -1659,6 +1836,7 @@ pub(crate) mod tests { 10, 400, 0, + 1_700_000_000, &[], ) .unwrap(); @@ -1732,7 +1910,16 @@ pub(crate) mod tests { // 5th append must grow, not Corrupt. let tid5 = Txid::from_byte_array([0x55; 32]); let tx = tiny_tx(); - let r = mp.append_live_tx(&tx, &tid5, &tx.compute_wtxid(), 1, 400, 0, &[]); + let r = mp.append_live_tx( + &tx, + &tid5, + &tx.compute_wtxid(), + 1, + 400, + 0, + 1_700_000_000, + &[], + ); assert!( r.is_ok(), "expected grow on full table, got {:?}", @@ -1751,13 +1938,31 @@ pub(crate) mod tests { let mut mp = Mempool::open_or_create(&dir).unwrap(); let raw = tiny_tx(); let t1 = Txid::from_byte_array([0x01; 32]); - mp.append_live_tx(&raw, &t1, &raw.compute_wtxid(), 1, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t1, + &raw.compute_wtxid(), + 1, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); mp.flush().unwrap(); let slots_first = fs::read(dir.join("slots")).unwrap(); let t2 = Txid::from_byte_array([0x02; 32]); - mp.append_live_tx(&raw, &t2, &raw.compute_wtxid(), 1, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t2, + &raw.compute_wtxid(), + 1, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); mp.flush().unwrap(); drop(mp); fs::write(dir.join("slots"), &slots_first).unwrap(); @@ -1768,8 +1973,17 @@ pub(crate) mod tests { let dir = tmp_dir(); let mut mp = Mempool::open_or_create(&dir).unwrap(); - mp.append_live_tx(&raw, &t1, &raw.compute_wtxid(), 1, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t1, + &raw.compute_wtxid(), + 1, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); let off = SLOTS_HEADER; mp.slots[off + 12..off + 16].copy_from_slice(&88u32.to_le_bytes()); let short = mp.load_live_txs().unwrap_err().to_string(); @@ -1804,11 +2018,29 @@ pub(crate) mod tests { let mut mp = Mempool::open_or_create(&dir).unwrap(); let raw = tiny_tx(); let t1 = Txid::from_byte_array([0x01; 32]); - mp.append_live_tx(&raw, &t1, &raw.compute_wtxid(), 1, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t1, + &raw.compute_wtxid(), + 1, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); let t2 = Txid::from_byte_array([0x02; 32]); - mp.append_live_tx(&raw, &t2, &raw.compute_wtxid(), 1, 400, 0, &[]) - .unwrap(); + mp.append_live_tx( + &raw, + &t2, + &raw.compute_wtxid(), + 1, + 400, + 0, + 1_700_000_000, + &[], + ) + .unwrap(); mp.flush().unwrap(); assert_eq!(mp.live_count(), 2); let meta_before_compact = fs::read(dir.join("meta")).unwrap(); @@ -1900,26 +2132,27 @@ pub(crate) mod tests { #[test] fn leftover_schema_unknown_is_refused() { let dir = tmp_dir(); + let next = MEM_SCHEMA + 1; fs::create_dir_all(&dir).unwrap(); { let mut meta = [0u8; META_LEN]; write_meta_bytes(&mut meta, 0, 4, 0); - meta[4..6].copy_from_slice(&4u16.to_le_bytes()); + meta[4..6].copy_from_slice(&next.to_le_bytes()); fs::write(dir.join("meta"), meta).unwrap(); let mut slots = vec![0u8; SLOTS_HEADER + 4 * SLOT_REC]; slots[0..4].copy_from_slice(&MEM_MAGIC); - slots[4..6].copy_from_slice(&4u16.to_le_bytes()); + slots[4..6].copy_from_slice(&next.to_le_bytes()); fs::write(dir.join("slots"), &slots).unwrap(); let mut body = vec![0u8; BODY_HEADER]; body[0..4].copy_from_slice(&MEM_MAGIC); - body[4..6].copy_from_slice(&4u16.to_le_bytes()); + body[4..6].copy_from_slice(&next.to_le_bytes()); body[8..16].copy_from_slice(&(BODY_HEADER as u64).to_le_bytes()); fs::write(dir.join("tx.body"), &body).unwrap(); } match Mempool::open_or_create(&dir) { - Err(MempoolError::BadSchema(4)) => {} - Ok(_) => panic!("expected BadSchema(4), got Ok"), - Err(e) => panic!("expected BadSchema(4), got {e}"), + Err(MempoolError::BadSchema(s)) if s == next => {} + Ok(_) => panic!("expected BadSchema({next}), got Ok"), + Err(e) => panic!("expected BadSchema({next}), got {e}"), } let _ = fs::remove_dir_all(&dir); } diff --git a/crates/rbitcoin-net/src/tx_relay.rs b/crates/rbitcoin-net/src/tx_relay.rs index c7d2c7fe3..29760580a 100644 --- a/crates/rbitcoin-net/src/tx_relay.rs +++ b/crates/rbitcoin-net/src/tx_relay.rs @@ -657,7 +657,8 @@ pub struct MempoolHub { /// INV immediately instead of waiting on mocktime. immediate_relay: AtomicBool, /// Last `setmocktime` (0 = wall). Used to age mempool txs for delayed INV. - mock_now: AtomicU64, + /// Shared with the mempool, which stamps it on each durable record. + mock_now: Arc, /// Mock/wall seconds when each live wtxid was accepted. accept_at: Mutex>, /// Monotonic accept generation for `p2p_tx_privacy` (skip pre-handshake txs). @@ -851,7 +852,7 @@ impl MempoolHub { wtxid_by_txid: Mutex::new(HashMap::new()), next_relay_seq: AtomicU64::new(1), immediate_relay: AtomicBool::new(false), - mock_now: AtomicU64::new(0), + mock_now: Arc::new(AtomicU64::new(0)), accept_at: Mutex::new(HashMap::new()), next_accept_gen: AtomicU64::new(1), accept_gen: Mutex::new(HashMap::new()), @@ -871,8 +872,19 @@ impl MempoolHub { parent_req: Mutex::new(parent_req::ParentTracker::new()), }; // Schema ≤ 2 records carry no sigop cost: fill (or evict) before serving. - hub.lock_write() - .recompute_missing_sigops(&QueryUtxoProvider::new(&hub.query)); + let loaded = { + let mut g = hub.lock_write(); + g.set_node_clock(Arc::clone(&hub.mock_now)); + g.recompute_missing_sigops(&QueryUtxoProvider::new(&hub.query)); + g.take_loaded_accept_times() + }; + // Stored accept times feed expiry and `getmempoolentry.time`. A loaded + // tx is not queued for INV, as before a restart dropped the maps. + for (txid, wtxid, at) in loaded { + if hub.contains(&txid) { + hub.insert_relay_maps(txid, wtxid, 0, at, false); + } + } { let mut u = hub.unbroadcast.lock().unwrap(); u.retain(|t| hub.contains(t)); @@ -1028,8 +1040,9 @@ impl MempoolHub { .unwrap_or(0) } - fn insert_relay_maps(&self, txid: Txid, wtxid: Wtxid, seq: u64) { - let at = self.relay_now_secs(); + /// Index a live tx accepted at `at` (node seconds). `age_inv` queues the + /// delayed INV for a fresh accept. + fn insert_relay_maps(&self, txid: Txid, wtxid: Wtxid, seq: u64, at: u64, age_inv: bool) { let gen = self.next_accept_gen.fetch_add(1, Ordering::Relaxed); let due = at.saturating_add(30); let mut by_tx = self.wtxid_by_txid.lock().unwrap(); @@ -1046,7 +1059,9 @@ impl MempoolHub { seqs.insert(wtxid, seq); ats.insert(wtxid, at); gens.insert(wtxid, gen); - age.insert((due, gen), (txid, wtxid)); + if age_inv { + age.insert((due, gen), (txid, wtxid)); + } order.insert((at, wtxid), txid); self.min_live_accept_at.fetch_min(at, Ordering::Relaxed); } @@ -2333,7 +2348,7 @@ impl MempoolHub { self.unindex_evicted(&r.evicted); let seq = self.next_relay_seq.fetch_add(1, Ordering::Relaxed); let w = tx.compute_wtxid(); - self.insert_relay_maps(r.txid, w, seq); + self.insert_relay_maps(r.txid, w, seq, r.accept_secs, true); self.reorg_servable.lock().unwrap().remove(&w); self.note_fee_flow_admit(r.weight, r.fee_sat); self.push_recent(tx, r); @@ -2759,7 +2774,7 @@ impl MempoolHub { self.unindex_evicted(&r.evicted); // Same relay-age clock as an individual admit (`mempool_limit.py` INV). let seq = self.next_relay_seq.fetch_add(1, Ordering::Relaxed); - self.insert_relay_maps(r.txid, tx.compute_wtxid(), seq); + self.insert_relay_maps(r.txid, tx.compute_wtxid(), seq, r.accept_secs, true); self.index_txid( r.txid, tx, @@ -3065,7 +3080,7 @@ impl MempoolHub { if let Ok(prevouts) = self.staged_reorg_admit(tx, &utxo) { let w = tx.compute_wtxid(); self.reorg_servable.lock().unwrap().insert(w); - self.insert_relay_maps(tx.compute_txid(), w, 0); + self.insert_relay_maps(tx.compute_txid(), w, 0, self.relay_now_secs(), true); admitted.push((tx, prevouts)); } } @@ -3724,13 +3739,24 @@ impl MempoolHub { } fn live_accept_result(&self, txid: &Txid) -> Option { - let g = self.lock_read(); - let e = g.graph.get(txid)?; + let (txid, fee_sat, weight, slot) = { + let g = self.lock_read(); + let e = g.graph.get(txid)?; + ( + e.txid, + e.fee_sat, + e.adjusted_weight(g.graph.bytes_per_sigop()), + e.slot, + ) + }; Some(AcceptResult { - txid: e.txid, - fee_sat: e.fee_sat, - weight: e.adjusted_weight(g.graph.bytes_per_sigop()), - slot: e.slot, + txid, + fee_sat, + weight, + slot, + accept_secs: self + .accept_time_txid(&txid) + .unwrap_or_else(|| self.relay_now_secs()), replaced: Vec::new(), replaced_scripthashes: Vec::new(), replaced_txs: Vec::new(), @@ -4451,6 +4477,37 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// A reopened pool keeps each entry's accept time: `getmempoolentry.time` + /// and `-mempoolexpiry` see the original accept, not 0 or the restart. + #[test] + fn reopen_keeps_accept_time_and_expiry() { + let (_store, q, cbs) = pad_cbs(3); + let mp = tmp(); + let hub = MempoolHub::open(&mp, Arc::clone(&q)).unwrap(); + let t0 = 1_700_000_000; + hub.note_mock_now(t0); + let tx = spend_true(cbs[0], 1_000, ScriptBuf::from_bytes(vec![0x51])); + let tid = tx.compute_txid(); + hub.accept_tx(&tx).expect("admit"); + assert_eq!(hub.accept_time_txid(&tid), Some(t0)); + hub.flush().expect("shutdown flush"); + drop(hub); + let hub2 = MempoolHub::open(&mp, Arc::clone(&q)).unwrap(); + assert!(hub2.contains(&tid)); + assert_eq!( + hub2.accept_time_txid(&tid), + Some(t0), + "accept time must survive a restart" + ); + hub2.set_expiry_hours(1); + hub2.note_mock_now(t0 + 3600 - 5); + assert_eq!(hub2.expire_stale(), 0, "not yet an hour old"); + hub2.note_mock_now(t0 + 3600 + 5); + assert_eq!(hub2.expire_stale(), 1, "a loaded entry must still expire"); + assert!(!hub2.contains(&tid)); + let _ = std::fs::remove_dir_all(&mp); + } + /// One scan visits 256 live txs. With more than that expired, the survivors /// are the newest accepts, and the next call continues from there. #[test] @@ -5093,7 +5150,7 @@ mod tests { { let mut store = rbitcoin_mempool::Mempool::open_or_create(&mp).unwrap(); store - .append_live_tx(&tx, &tid, &wtxid, 1_000, 400, 0, &[]) + .append_live_tx(&tx, &tid, &wtxid, 1_000, 400, 0, 1_700_000_000, &[]) .unwrap(); store.flush().unwrap(); } @@ -5125,7 +5182,7 @@ mod tests { for tx in [&ok, &gone] { let (tid, wtxid) = (tx.compute_txid(), tx.compute_wtxid()); store - .append_live_tx(tx, &tid, &wtxid, 1_000, 400, u64::MAX, &[]) + .append_live_tx(tx, &tid, &wtxid, 1_000, 400, u64::MAX, 1_700_000_000, &[]) .unwrap(); } store.flush().unwrap(); @@ -5155,7 +5212,7 @@ mod tests { { let mut store = rbitcoin_mempool::Mempool::open_or_create(&mp).unwrap(); store - .append_live_tx(&tx, &tid, &wtxid, 1_000, 400, 100, &[]) + .append_live_tx(&tx, &tid, &wtxid, 1_000, 400, 100, 1_700_000_000, &[]) .unwrap(); store.flush().unwrap(); } @@ -5879,7 +5936,13 @@ mod tests { Err(AcceptError::Orphaned { .. }) )); let orphan_w = orphan.compute_wtxid(); - hub.insert_relay_maps(orphan.compute_txid(), orphan_w, 1); + hub.insert_relay_maps( + orphan.compute_txid(), + orphan_w, + 1, + hub.relay_now_secs(), + true, + ); assert!(hub.relay_seq_of(&orphan_w).is_some()); hub.unindex_if_absent(&orphan); assert!( @@ -5888,7 +5951,13 @@ mod tests { ); let absent = spend_true(cbs[2], 3_000, ScriptBuf::from_bytes(vec![0x51])); let absent_w = absent.compute_wtxid(); - hub.insert_relay_maps(absent.compute_txid(), absent_w, 1); + hub.insert_relay_maps( + absent.compute_txid(), + absent_w, + 1, + hub.relay_now_secs(), + true, + ); assert!(hub.relay_seq_of(&absent_w).is_some()); hub.unindex_if_absent(&absent); assert!( @@ -6247,6 +6316,7 @@ mod tests { 4_999_999_001, tx.weight().to_wu(), u64::MAX, + 1_700_000_000, &[], ) .unwrap(); diff --git a/crates/rbitcoin-test/tests/core_analogs.rs b/crates/rbitcoin-test/tests/core_analogs.rs index dbfc4ae73..9bdf93f7f 100644 --- a/crates/rbitcoin-test/tests/core_analogs.rs +++ b/crates/rbitcoin-test/tests/core_analogs.rs @@ -427,9 +427,9 @@ fn pin_unreadable_sidecar_then_accept( std::fs::write(mp_dir.join("tx.body"), body_prefix).unwrap(); std::fs::write(mp_dir.join("slots"), slots_prefix).unwrap(); let mut body = std::fs::read(mp_dir.join("tx.body")).unwrap(); - // Schema-3 txid starts 24 bytes into the packed record. Flip it so the + // Schema-4 txid starts 32 bytes into the packed record. Flip it so the // slot's txid does not match the payload. The tx bytes still decode. - let txid_at = 16 + 24; + let txid_at = 16 + 32; assert!(body.len() > txid_at, "payload has a txid"); body[txid_at] ^= 0xff; std::fs::write(mp_dir.join("tx.body"), &body).unwrap(); diff --git a/docs/ibd-memory.md b/docs/ibd-memory.md index de8e5f792..e1fa7b972 100644 --- a/docs/ibd-memory.md +++ b/docs/ibd-memory.md @@ -167,7 +167,7 @@ intentional: | **Sealed BDZ `g`** | **0 heap** | Header only; 4 KiB `g` pages via uring stream (`KIND_MPHF_G`). Hot pages are kernel `RssFile`. FdOnly because a mapped miss serializes one fault per lookup thread; the ring keeps 128 pages in flight. Do not mmap packed `g`. | | **SH BDZ3 occupancy** | **supers heap** (`mphf_occ=`); ~150 MiB `file=` at ~1 B SH keys | Read-only map of the `NN.mphf` prefix through occ (not tags). Rank popcounts mapped bytes. | | **Class C L2 `strong_tx`** | **~177 MiB** | 1 bit/create, under the 256 MiB in-RAM cap. Stays process `Vec`: `MAP_SHARED` would write before the tip barrier; `MAP_PRIVATE` COWs `set_bit` back into anon. | -| **Mempool schema 3** | Slot table ~6 MiB at 128k + body weight | InRam `slots`/`body` Vecs + `pwrite`. Live set is also the graph. Not mapped. | +| **Mempool schema 4** | Slot table ~6 MiB at 128k + body weight | InRam `slots`/`body` Vecs + `pwrite`. Live set is also the graph. Not mapped. | | **`height_by_hash`** | **~60 MiB** | In-process confirmed hash→height map. Incremental on tip extend/shrink; full `0..=tip` walk on open / invalidate only. | | **mimalloc arenas** | **anon − accounted** after IBD | Product bins are global mimalloc. IBD allocates/frees GiB-class transients; `free` is not `munmap`. After tip catch-up, `anon=` can stay fat while `accounted=` collapses. That is allocator residue, not a store leak. Fuse mmap does not mean RSS equals fuse. Do not `malloc_trim` a mimalloc process. Optional operator: `MIMALLOC_PURGE_DELAY=0`. The `mimalloc` crate does not expose `mi_collect` on `MiMalloc` (no in-process purge hook). | | **Process baseline** | **~90 MiB** | Visible at genesis (`class_a=476`, `residual≈93`). Allocator arenas, rustc runtime, net. | diff --git a/docs/io-modality.md b/docs/io-modality.md index 07f5bae87..652c13e9a 100644 --- a/docs/io-modality.md +++ b/docs/io-modality.md @@ -241,7 +241,7 @@ Sealed `.fuse8` sidecars and SH BDZ3 occupancy prefixes are the mmap exceptions (read-only, not `MmapMut`). Packed `g` stays FdOnly. Class C is L2 write-behind (`flush_class_c_tip` before BQ dequeue) — **not** mapped (`strong_tx` write-behind is tip-last). -Mempool schema 3 is InRam Vecs + `pwrite`. +Mempool schema 4 is InRam Vecs + `pwrite`. Live head insert is page-coalesced pread → mutate → pwrite (not per-slot uring). Head resolve batches one pread per distinct probe page, then one loc batch diff --git a/docs/operator/operations.md b/docs/operator/operations.md index 53c583b96..72a6bae0d 100644 --- a/docs/operator/operations.md +++ b/docs/operator/operations.md @@ -583,8 +583,10 @@ failed window still logs `rbitcoin: tip revalidate …`. That is intentional mid-chain header graph poison still means a clean datadir. **Mempool recovery:** `{datadir}/mempool/` is a private sidecar (not Class A), -schema **3**. Leftover schema **1** (pre-packed `fee‖weight‖bitcoin-serialize`) -or **2** converts to schema 3 on the next open (Class A untouched). Wipe `{datadir}/mempool/` +schema **4** (each record keeps its accept time, so `getmempoolentry.time` and +`-mempoolexpiry` survive a restart). Leftover schema **1** (pre-packed +`fee‖weight‖bitcoin-serialize`), **2** or **3** converts to schema 4 on the next +open, and its records count their accept time from that open (Class A untouched). Wipe `{datadir}/mempool/` only if the sidecar is damaged or an unknown schema/old 4k-slot table was left wedged — the next start recreates it empty and redownloads unconfirmed txs. Do **not** wipe `store/` for mempool slot/full/schema errors. From dca5ac9507832a77f9e6fd7f084bd9bf279b23e1 Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Thu, 15 Oct 2026 22:37:24 -0600 Subject: [PATCH 2/5] net: rate limit addrv2 before it reaches the address book The addr token bucket gated only relay: on_addrv2 learned every row first, and the bucket started full at 1000. One peer could replace every never-connected address with nine 1000-entry messages. IBD peers' addresses were not limited at all. AddrTokenBucket follows Core v31.1: one token at connect, 0.1 per second up to 1000. Tokens are taken before learn_addrv2, and only the admitted prefix is learned and relayed. IBD peers spend a bucket of their own. A peer with the addr permission is not limited. getpeerinfo reports addr_processed and addr_rate_limited. Core credits 1000 tokens when it sends getaddr. Here only addr-fetch peers get getaddr, and their multi-address reply is dropped before learning, so there is no credit to add. Refs #1030 Co-Authored-By: Claude Opus 5.5 --- changelog.d/mempool-accept-time-addr-rate.md | 4 ++ crates/rbitcoin-net/src/ibd/events/mod.rs | 49 +++++++++++++- crates/rbitcoin-net/src/ibd/state.rs | 4 ++ crates/rbitcoin-net/src/peer.rs | 51 ++++++++++---- .../rbitcoin-net/src/peer_hostile_journey.rs | 66 ++++++++++++++++++- crates/rbitcoin-net/src/peers.rs | 51 ++++++++++---- crates/rbitcoin-rpc/src/methods/net.rs | 2 + docs/operator/operations.md | 6 +- 8 files changed, 202 insertions(+), 31 deletions(-) create mode 100644 changelog.d/mempool-accept-time-addr-rate.md diff --git a/changelog.d/mempool-accept-time-addr-rate.md b/changelog.d/mempool-accept-time-addr-rate.md new file mode 100644 index 000000000..40b22367a --- /dev/null +++ b/changelog.d/mempool-accept-time-addr-rate.md @@ -0,0 +1,4 @@ +Fixed + +- **Mempool entries keep their accept time across a restart.** The mempool sidecar is schema 4: each record stores its accept time, so `getmempoolentry.time` is no longer 0 after a restart and `-mempoolexpiry` still removes old entries. A schema 1–3 sidecar converts on the next open, and its entries count their accept time from that open (#1029). +- **Peer addresses are rate limited before they reach the address book.** Each peer starts with one address token and earns one per 10 s (up to 1000), as in Core. A peer can no longer replace the address book with a few 1000-entry `addrv2` messages. The IBD path uses the same bucket. `getpeerinfo` reports `addr_processed` and `addr_rate_limited`. Peers with the `addr` whitelist permission are not limited (#1030). diff --git a/crates/rbitcoin-net/src/ibd/events/mod.rs b/crates/rbitcoin-net/src/ibd/events/mod.rs index fb1593fe0..1936c0fdb 100644 --- a/crates/rbitcoin-net/src/ibd/events/mod.rs +++ b/crates/rbitcoin-net/src/ibd/events/mod.rs @@ -455,7 +455,8 @@ pub(crate) fn apply_peer_event( } } PeerEvent::Addrs { peer, addrs } => { - inject_learned_addrs(peer_book, &addrs, local_addr, peer); + let allow = take_addr_tokens(st, peer, addrs.len()); + inject_learned_addrs(peer_book, &addrs[..allow], local_addr, peer); } PeerEvent::Dead { peer, reason } => apply_peer_dead(st, peer_book, peer, reason), } @@ -871,8 +872,24 @@ fn apply_notfound(st: &mut IbdWorkState, peer: usize, hashes: Vec) { st.reopen_for_densify(&freed); } +/// How many of `want` addresses from `peer` may enter the dial book. The +/// bucket counts from the peer's connect on the IBD mono clock. +fn take_addr_tokens(st: &mut IbdWorkState, peer: usize, want: usize) -> usize { + let now = super::peer_io::ibd_mono_ms(); + let start = st + .slots + .iter() + .find(|s| s.id == peer) + .map_or(now, |s| s.connected_ms); + st.addr_buckets + .entry(peer) + .or_insert_with(|| crate::peer::AddrTokenBucket::new(start)) + .take(want, now) +} + fn apply_peer_dead(st: &mut IbdWorkState, peer_book: &mut AddrMan, peer: usize, reason: String) { warn!("ibd: peer[{peer}] dead: {reason}"); + st.addr_buckets.remove(&peer); super::header_walk::forget_walk_peer(st, peer); if let Some(s) = st.slots.iter().find(|s| s.id == peer) { if reason == "peer misbehavior threshold" { @@ -1368,6 +1385,36 @@ mod misbehavior_dead_tests { } } +#[cfg(test)] +mod addr_bucket_tests { + use super::super::assign::tests::dummy_slot; + use super::super::peer_io::ibd_mono_ms; + use super::super::state::IbdWorkState; + use super::{apply_peer_dead, take_addr_tokens}; + use crate::seeds::AddrMan; + + /// IBD peers spend the same addr token bucket as tip-follow sessions: one + /// token at connect, so a 1000-address flood teaches the book one address. + #[test] + fn ibd_peer_addr_flood_takes_one_token() { + let mut slot = dummy_slot(1); + slot.connected_ms = ibd_mono_ms(); + let mut st = IbdWorkState::new(vec![slot], None, Some(0)); + assert_eq!(take_addr_tokens(&mut st, 1, 1_000), 1); + assert_eq!( + take_addr_tokens(&mut st, 1, 1_000), + 0, + "the bucket is spent" + ); + let mut book = AddrMan::new(); + apply_peer_dead(&mut st, &mut book, 1, "bye".to_string()); + assert!( + st.addr_buckets.is_empty(), + "a dead peer's bucket is dropped" + ); + } +} + #[cfg(test)] mod header_batch_tests { use super::super::peer_io::empty_header_fk_bus; diff --git a/crates/rbitcoin-net/src/ibd/state.rs b/crates/rbitcoin-net/src/ibd/state.rs index 404a2d396..73b09e50e 100644 --- a/crates/rbitcoin-net/src/ibd/state.rs +++ b/crates/rbitcoin-net/src/ibd/state.rs @@ -157,6 +157,9 @@ pub(crate) struct IbdWorkState { pub assign_rot: usize, /// Stall-disconnect cooldowns (addr → until). pub addr_cooldown: HashMap, + /// Per-peer addr token buckets (peer id → bucket), as on tip-follow + /// sessions. Dropped when the peer dies. + pub addr_buckets: HashMap, /// Process-local stall/relative-slow kick counts (not persisted). pub addr_strikes: HashMap, /// Relative-slow hysteresis: peer id + first Gate B fail ms. @@ -266,6 +269,7 @@ impl IbdWorkState { header_req_seq: 0, assign_rot: 0, addr_cooldown: HashMap::new(), + addr_buckets: HashMap::new(), addr_strikes: HashMap::new(), relative_slow_suspect: None, relative_slow_last_kick_ms: 0, diff --git a/crates/rbitcoin-net/src/peer.rs b/crates/rbitcoin-net/src/peer.rs index 185727226..aeeb756cb 100644 --- a/crates/rbitcoin-net/src/peer.rs +++ b/crates/rbitcoin-net/src/peer.rs @@ -402,16 +402,41 @@ pub fn version_handshake_timeout_log(peer: u64) -> String { /// Max addresses in one ADDR / addrv2. Interop size — do not change without /// a named reason to diverge (see COMPAT.md). pub const MAX_ADDR_TO_SEND: usize = 1000; -/// Addresses one peer may relay before the bucket must refill (~0.1/s). -pub(crate) const ADDR_RELAY_BURST: f64 = 1000.0; -pub(crate) const ADDR_RELAY_PER_SEC: f64 = 0.1; +/// Core `MAX_ADDR_RATE_PER_SECOND`: a peer earns one address token per 10 s. +pub(crate) const MAX_ADDR_RATE_PER_SECOND: f64 = 0.1; +/// Core `MAX_ADDR_PROCESSING_TOKEN_BUCKET`: refill stops at this many tokens. +pub(crate) const MAX_ADDR_PROCESSING_TOKEN_BUCKET: f64 = MAX_ADDR_TO_SEND as f64; + +/// Core `Peer::m_addr_token_bucket`. It starts at one token. Each address a +/// peer sends costs one, and an address with no token left is dropped before +/// AddrMan and relay. +#[derive(Debug, Clone, Copy)] +pub(crate) struct AddrTokenBucket { + tokens: f64, + at_ms: u64, +} + +impl AddrTokenBucket { + pub(crate) fn new(now_ms: u64) -> Self { + Self { + tokens: 1.0, + at_ms: now_ms, + } + } -pub(crate) fn addr_relay_tokens(tokens: f64, at_ms: u64, now_ms: u64) -> f64 { - if now_ms <= at_ms { - return tokens.min(ADDR_RELAY_BURST); + /// Refill for the time since the last call (only below the cap), then + /// take up to `want` tokens. Returns how many addresses may be processed. + pub(crate) fn take(&mut self, want: usize, now_ms: u64) -> usize { + if self.tokens < MAX_ADDR_PROCESSING_TOKEN_BUCKET { + let dt = now_ms.saturating_sub(self.at_ms) as f64 / 1000.0; + self.tokens = + (self.tokens + dt * MAX_ADDR_RATE_PER_SECOND).min(MAX_ADDR_PROCESSING_TOKEN_BUCKET); + } + self.at_ms = self.at_ms.max(now_ms); + let take = (self.tokens.floor() as usize).min(want); + self.tokens -= take as f64; + take } - let dt = (now_ms - at_ms) as f64 / 1000.0; - (tokens + dt * ADDR_RELAY_PER_SEC).min(ADDR_RELAY_BURST) } fn addr_relay_key(msg: &bitcoin::p2p::address::AddrV2Message) -> u64 { @@ -2983,12 +3008,10 @@ fn on_addrv2( } if let Some(s) = session { if let Some(ph) = s.peer_hub() { - ph.learn_addrv2(list); - let now_ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0); - let allow = s.take_addr_relay(list.len(), now_ms); + // Node time, as the bucket's start (`connected_at`) is. + let now_ms = s.clock_now().saturating_mul(1000); + let allow = s.take_addr_tokens(list.len(), now_ms); + ph.learn_addrv2(&list[..allow]); let mut neighbors: Vec<_> = ph .live_peers() .into_iter() diff --git a/crates/rbitcoin-net/src/peer_hostile_journey.rs b/crates/rbitcoin-net/src/peer_hostile_journey.rs index 598e844a8..cea7cadb9 100644 --- a/crates/rbitcoin-net/src/peer_hostile_journey.rs +++ b/crates/rbitcoin-net/src/peer_hostile_journey.rs @@ -303,8 +303,20 @@ async fn hostile_addr_relay( let (quiet_tx, mut quiet_rx) = mpsc::unbounded_channel(); quiet.attach_out(quiet_tx); - assert_eq!(addr_relay_tokens(0.0, 1_000, 11_000), 1.0); - assert!(addr_relay_tokens(0.0, 1_000, 1_000) < 1.0); + let mut fresh = AddrTokenBucket::new(1_000); + assert_eq!(fresh.take(5, 1_000), 1, "a new bucket holds one token"); + assert_eq!(fresh.take(1, 10_999), 0, "under 10 s is less than a token"); + let mut ten = AddrTokenBucket::new(1_000); + assert_eq!(ten.take(1, 1_000), 1); + assert_eq!(ten.take(1, 11_000), 1, "10 s refills one token"); + let mut capped = AddrTokenBucket::new(0); + assert_eq!( + capped.take(5_000, 1_000_000_000), + MAX_ADDR_TO_SEND, + "refill stops at the cap" + ); + // A quiet peer refills its bucket: let 10_000 s of node time pass. + peers.set_mock_now(peer.connected_at_secs() + 10_000); let list: Vec = (1..=3u8) .map(|i| AddrV2Message { @@ -436,3 +448,53 @@ async fn hostile_addr_relay( assert_eq!(extra_relay, 0, "past the 1000-address burst nothing is relayed"); } + +fn addr_flood(first_octet: u8) -> Vec { + (0..MAX_ADDR_TO_SEND as u32) + .map(|i| AddrV2Message { + time: 1_700_000_000, + services: ServiceFlags::NETWORK, + addr: AddrV2::Ipv4(Ipv4Addr::new(first_octet, (i >> 8) as u8, i as u8, 1)), + port: 8333, + }) + .collect() +} + +/// Core v31.1 ADDR processing: a peer's token bucket starts at one address, +/// and an address with no token left is dropped before AddrMan, not only +/// before relay. A peer with the `addr` permission is not rate limited. +#[test] +fn addrv2_flood_is_rate_limited_before_addrman() { + for (perm, want_learned) in [(false, 1usize), (true, MAX_ADDR_TO_SEND)] { + let peers = crate::peers::PeerHub::new(); + if perm { + let mut t = crate::NetPermTable::default(); + t.whitelist + .push(crate::parse_whitelist("addr@127.0.0.1").unwrap()); + peers.set_net_perms(t); + } + let am = std::sync::Arc::new(std::sync::Mutex::new(crate::seeds::AddrMan::new())); + peers.set_addrman(am.clone()); + let bind = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 18444); + let ver = hostile_ver(bind); + let peer = peers.register(bind, bind, &ver, true, crate::peers::PeerConnType::Inbound); + let mut follow = PeerFollowState::new(); + on_addrv2(&mut follow, Some(peer.as_ref()), &addr_flood(11)).unwrap(); + assert_eq!( + am.lock().unwrap().len(), + want_learned, + "addr permission {perm}: addresses that reach AddrMan" + ); + let info = peers + .snapshot() + .into_iter() + .find(|p| p.id == peer.id) + .expect("peer info"); + let limited = (MAX_ADDR_TO_SEND - want_learned) as u64; + assert_eq!( + (info.addr_processed, info.addr_rate_limited), + (want_learned as u64, limited), + "getpeerinfo addr counters, addr permission {perm}" + ); + } +} diff --git a/crates/rbitcoin-net/src/peers.rs b/crates/rbitcoin-net/src/peers.rs index be7e3f2c4..db524ac93 100644 --- a/crates/rbitcoin-net/src/peers.rs +++ b/crates/rbitcoin-net/src/peers.rs @@ -359,9 +359,11 @@ pub struct LivePeer { inflight: Mutex>, /// Session writer. RPC/accept flushes tx INVs onto this (`p2p_blocksonly`). out_tx: Mutex>>, - /// Addr relay tokens still available, and the millisecond they were last filled. - addr_tokens: Mutex, - addr_token_ms: AtomicU64, + /// Addresses this peer may still send us (Core `m_addr_token_bucket`). + addr_bucket: Mutex, + /// `getpeerinfo.addr_processed` / `addr_rate_limited`. + addr_processed: AtomicU64, + addr_rate_limited: AtomicU64, /// Unix seconds when this session was registered. connected_at: AtomicU64, /// Netgroup fixed at accept. Inbound eviction compares this integer @@ -960,14 +962,28 @@ impl LivePeer { *self.out_tx.lock().unwrap_or_else(|e| e.into_inner()) = Some(tx); } - pub(crate) fn take_addr_relay(&self, want: usize, now_ms: u64) -> usize { - let mut tokens = self.addr_tokens.lock().unwrap_or_else(|e| e.into_inner()); - let at = self.addr_token_ms.load(Ordering::Relaxed); - let filled = crate::peer::addr_relay_tokens(*tokens, at, now_ms); - self.addr_token_ms.store(now_ms.max(at), Ordering::Relaxed); - let take = (filled.floor() as usize).min(want); - *tokens = filled - take as f64; - take + /// How many of the `want` addresses in one message this peer may have + /// processed (learned and relayed). The rest count as rate limited. A + /// peer with the `addr` permission spends tokens but is never limited. + pub(crate) fn take_addr_tokens(&self, want: usize, now_ms: u64) -> usize { + let took = self + .addr_bucket + .lock() + .unwrap_or_else(|e| e.into_inner()) + .take(want, now_ms); + let allow = if self + .net_perm_flags() + .has(crate::net_permissions::NetPermissionFlags::ADDR) + { + want + } else { + took + }; + self.addr_processed + .fetch_add(allow as u64, Ordering::Relaxed); + self.addr_rate_limited + .fetch_add((want - allow) as u64, Ordering::Relaxed); + allow } pub(crate) fn writer(&self) -> Option> { @@ -1230,6 +1246,8 @@ impl LivePeer { minfeefilter_sat_kvb: self.minfeefilter_sat_kvb.load(Ordering::Relaxed), last_inv_sequence: self.last_inv_sequence(), inv_to_send: self.inv_to_send(), + addr_processed: self.addr_processed.load(Ordering::Relaxed), + addr_rate_limited: self.addr_rate_limited.load(Ordering::Relaxed), bytesrecv, bytessent, inflight: self @@ -1342,6 +1360,10 @@ pub struct PeerInfo { pub last_inv_sequence: u64, /// `getpeerinfo.inv_to_send` (queued tx INV count). pub inv_to_send: u32, + /// `getpeerinfo.addr_processed`: addresses not dropped by the rate limit. + pub addr_processed: u64, + /// `getpeerinfo.addr_rate_limited`: addresses the rate limit dropped. + pub addr_rate_limited: u64, /// Raw TCP bytes (`getpeerinfo.bytesrecv`), including BIP324 handshake. pub bytesrecv: u64, /// Raw TCP bytes (`getpeerinfo.bytessent`), including BIP324 handshake. @@ -2331,8 +2353,11 @@ impl PeerHub { failed_cmpct: Mutex::new(HashSet::new()), inflight: Mutex::new(Vec::new()), out_tx: Mutex::new(None), - addr_tokens: Mutex::new(crate::peer::ADDR_RELAY_BURST), - addr_token_ms: AtomicU64::new(0), + addr_bucket: Mutex::new(crate::peer::AddrTokenBucket::new( + connected_at.saturating_mul(1000), + )), + addr_processed: AtomicU64::new(0), + addr_rate_limited: AtomicU64::new(0), connected_at: AtomicU64::new(connected_at), netgroup: crate::eviction::eviction_netgroup(endpoint.addr), inv_gen_floor: AtomicU64::new(0), diff --git a/crates/rbitcoin-rpc/src/methods/net.rs b/crates/rbitcoin-rpc/src/methods/net.rs index 78f5b037e..86a784099 100644 --- a/crates/rbitcoin-rpc/src/methods/net.rs +++ b/crates/rbitcoin-rpc/src/methods/net.rs @@ -115,6 +115,8 @@ pub(crate) fn peerinfo_json(ctx: &RpcContext, p: rbitcoin_net::PeerInfo) -> Valu "bytessent": p.bytessent, "last_inv_sequence": p.last_inv_sequence, "inv_to_send": p.inv_to_send, + "addr_processed": p.addr_processed, + "addr_rate_limited": p.addr_rate_limited, "inflight": p.inflight, "permissions": p.permissions, }); diff --git a/docs/operator/operations.md b/docs/operator/operations.md index 72a6bae0d..0d7430623 100644 --- a/docs/operator/operations.md +++ b/docs/operator/operations.md @@ -603,7 +603,11 @@ Do **not** wipe `store/` for mempool slot/full/schema errors. - **Discovery** queries Core DNS seeds for `NETWORK|WITNESS|P2P_V2` (`x809.` first; the bare seed name only if that returns nothing). Learned `addr` / `addrv2` is ingested only when the row advertises `P2P_V2` - (plus `NETWORK` or `NETWORK_LIMITED`). Dial ranking omits known-v1 + (plus `NETWORK` or `NETWORK_LIMITED`). Each peer, IBD or tip-follow, has + Core's address token bucket: one address at connect, then one per 10 s, up + to 1000. Addresses past it are dropped before the book and relay, and + `getpeerinfo` counts them in `addr_rate_limited` (`addr_processed` counts + the rest). A peer with the `addr` whitelist permission is not limited. Dial ranking omits known-v1 (`INCOMPATIBLE`) addresses while any better candidate remains. Seed host list lives in `dns_seeds()` (`crates/rbitcoin-net/src/seeds.rs`). - **Outbound diversity:** live IBD and tip-follow peers prefer unused From 952a0e6b724140d689859639df230bc2a7d5bee8 Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Thu, 15 Oct 2026 22:41:35 -0600 Subject: [PATCH 3/5] mempool: give a reloaded tx a fresh entry sequence Loaded entries were indexed with relay sequence 0. A new peer starts at last_inv_sequence 1, so is_relay_servable let it getdata a restored tx before any INV; before the reload fix such a tx had no sequence and was not servable. Core's load assigns a fresh sequence; do the same. Refs #1029 Co-Authored-By: Claude Opus 5.5 --- crates/rbitcoin-net/src/tx_relay.rs | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/crates/rbitcoin-net/src/tx_relay.rs b/crates/rbitcoin-net/src/tx_relay.rs index 29760580a..6a2f21e6d 100644 --- a/crates/rbitcoin-net/src/tx_relay.rs +++ b/crates/rbitcoin-net/src/tx_relay.rs @@ -879,10 +879,12 @@ impl MempoolHub { g.take_loaded_accept_times() }; // Stored accept times feed expiry and `getmempoolentry.time`. A loaded - // tx is not queued for INV, as before a restart dropped the maps. + // tx is not queued for INV. Like Core's load, it takes a fresh entry + // sequence, so a peer cannot getdata it before an INV. for (txid, wtxid, at) in loaded { if hub.contains(&txid) { - hub.insert_relay_maps(txid, wtxid, 0, at, false); + let seq = hub.next_relay_seq.fetch_add(1, Ordering::Relaxed); + hub.insert_relay_maps(txid, wtxid, seq, at, false); } } { @@ -4499,6 +4501,12 @@ mod tests { Some(t0), "accept time must survive a restart" ); + // Core gives a loaded tx a fresh entry sequence. A new peer starts at + // last_inv_sequence 1, so it cannot getdata the tx before an INV. + assert!( + !hub2.is_relay_servable(&tx.compute_wtxid(), 1), + "a restored tx is not servable to a fresh peer before an INV" + ); hub2.set_expiry_hours(1); hub2.note_mock_now(t0 + 3600 - 5); assert_eq!(hub2.expire_stale(), 0, "not yet an hour old"); From 3248e3094cf95f4f7e9a6e1615688628d2e22115 Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Fri, 16 Oct 2026 15:13:45 -0600 Subject: [PATCH 4/5] mempool: a restored replacement victim keeps its accept time A failed replacement puts its conflicts back through finish_commit, which stamped node time on the new sidecar record. The hub kept the original time in RAM, so the next open moved getmempoolentry.time and -mempoolexpiry for those entries. PreparedAdmit gains keep_accept_secs. The hub's restore sets it from its own accept_at. ActiveMempool reads each conflict's accept_secs from its slot before unlinking it, and carries those times through a failed commit (undone_accept_secs) and a package rollback (AcceptResult .replaced_accept_secs). Also bring the sidecar docs to schema 4: lib.rs, store.rs, packed.rs and crash-recovery.md, including the 1-3 upgrade rules. Co-Authored-By: Claude Opus 5.5 --- crates/rbitcoin-mempool/src/accept.rs | 147 +++++++++++++++++++++++++- crates/rbitcoin-mempool/src/lib.rs | 9 +- crates/rbitcoin-mempool/src/packed.rs | 7 +- crates/rbitcoin-mempool/src/store.rs | 20 +++- crates/rbitcoin-net/src/tx_relay.rs | 102 +++++++++++++++++- docs/crash-recovery.md | 2 +- 6 files changed, 273 insertions(+), 14 deletions(-) diff --git a/crates/rbitcoin-mempool/src/accept.rs b/crates/rbitcoin-mempool/src/accept.rs index 36e165ebe..60a561ee0 100644 --- a/crates/rbitcoin-mempool/src/accept.rs +++ b/crates/rbitcoin-mempool/src/accept.rs @@ -180,6 +180,9 @@ pub struct AcceptResult { pub replaced_scripthashes: Vec<[u8; 32]>, /// Replaced bodies, for package rollback to restore victims. pub replaced_txs: Vec, + /// Accept times of [`Self::replaced_txs`], so a rollback restores each + /// with its own time. + pub replaced_accept_secs: std::collections::HashMap, /// Txids dropped to free a slot or the weight budget while admitting this tx. /// Distinct from [`Self::replaced`]. The admitted tx is not in this list /// when the commit succeeds. @@ -433,6 +436,9 @@ pub struct PreparedAdmit { pub prevouts: Vec, pub chain_coins: Vec>, pub utxo_us: u64, + /// Accept time to persist instead of now. Set when a failed replacement + /// puts an already-admitted conflict back, so it keeps its own time. + pub keep_accept_secs: Option, } /// Graph spend edges for one tx (conflicts + mempool parents). @@ -489,6 +495,8 @@ pub struct ActiveMempool { /// Bodies removed for replacement when `finish_commit` later returns /// `Err`. Success moves them onto [`AcceptResult::replaced_txs`]. undone_replacements: Vec, + /// Accept times of `undone_replacements`, read before their slots died. + undone_accept_secs: std::collections::HashMap, /// `setmocktime` seconds, shared with the hub's clock (0 = wall). node_clock: Arc, /// Accept times of the records loaded at open, until the hub takes them. @@ -521,6 +529,7 @@ impl ActiveMempool { /// Conflicts a failed `finish_commit` already removed. Empty after a success. pub fn take_undone_replacements(&mut self) -> Vec { + self.undone_accept_secs.clear(); std::mem::take(&mut self.undone_replacements) } } @@ -633,6 +642,7 @@ impl ActiveMempool { extra_compact: VecDeque::new(), last_evicted: Vec::new(), undone_replacements: Vec::new(), + undone_accept_secs: std::collections::HashMap::new(), node_clock: Arc::new(AtomicU64::new(0)), loaded_accept: ingested.accept_secs, }) @@ -1024,6 +1034,7 @@ impl ActiveMempool { prevouts, chain_coins, utxo_us, + keep_accept_secs: None, }) } @@ -1111,6 +1122,7 @@ impl ActiveMempool { trim: bool, ) -> Result { self.undone_replacements.clear(); + self.undone_accept_secs.clear(); let (conflict_set, fee_sat, adj_weight) = self.plan_after_script(tx, &prep)?; let txid = prep.txid; let weight = prep.weight; @@ -1118,6 +1130,13 @@ impl ActiveMempool { let mut replaced_scripthashes: Vec<[u8; 32]> = Vec::new(); let mut replaced_txs: Vec = Vec::new(); for c in &conflict_set { + if let Some(at) = self + .graph + .get(c) + .and_then(|e| self.store.live_accept_secs(e.slot)) + { + self.undone_accept_secs.insert(*c, at); + } if let Some(old) = self.bodies.get(c) { let old_tx = (**old).clone(); self.note_extra(&old_tx); @@ -1143,7 +1162,9 @@ impl ActiveMempool { let mut evicted = self.ensure_free_slot(Some(txid))?; let aux = Self::vin_aux_from_prep(tx, &prep); - let accept_secs = self.node_now_secs(); + let accept_secs = prep + .keep_accept_secs + .unwrap_or_else(|| self.node_now_secs()); let t_dur = Instant::now(); let slot = match self.store.append_live_tx( tx, @@ -1202,6 +1223,7 @@ impl ActiveMempool { "a successful admit is not in its own eviction set" ); let replaced_txs = std::mem::take(&mut self.undone_replacements); + let replaced_accept_secs = std::mem::take(&mut self.undone_accept_secs); Ok(AcceptResult { txid, @@ -1212,6 +1234,7 @@ impl ActiveMempool { replaced: conflict_set.into_iter().collect(), replaced_scripthashes, replaced_txs, + replaced_accept_secs, evicted, }) } @@ -1233,6 +1256,7 @@ impl ActiveMempool { replaced: conflict_set.into_iter().collect(), replaced_scripthashes: Vec::new(), replaced_txs: Vec::new(), + replaced_accept_secs: std::collections::HashMap::new(), evicted: Vec::new(), }) } @@ -1594,11 +1618,15 @@ impl ActiveMempool { .iter() .flat_map(|r| r.replaced_txs.iter().cloned()) .collect(); + let accept_secs: std::collections::HashMap = accepted + .iter() + .flat_map(|r| r.replaced_accept_secs.iter().map(|(t, s)| (*t, *s))) + .collect(); for r in accepted.iter().rev() { let _ = self.unlink_tree(&r.txid); } let _ = self.store.sync_deaths(); - self.restore_victims(&victims, utxos, tip); + self.restore_victims(&victims, &accept_secs, utxos, tip); } /// Parents before children, so a restored descendant can see its in-set parent. @@ -1648,13 +1676,16 @@ impl ActiveMempool { /// Put an already-admitted body back. The fee floor does not apply again, /// and a missing parent is not parked. + /// It keeps `accept_secs`, its original accept time, when known. fn reaccept_already_admitted( &mut self, tx: &Transaction, + accept_secs: Option, utxos: &impl UtxoProvider, tip: ChainTipCtx, ) -> Result { - let prep = self.prepare_admit(tx, utxos, tip, 0, false, Some(0))?; + let mut prep = self.prepare_admit(tx, utxos, tip, 0, false, Some(0))?; + prep.keep_accept_secs = accept_secs; verify_tx_scripts(tx, prep.prevouts.clone())?; self.commit_after_script(tx, prep) } @@ -1662,11 +1693,13 @@ impl ActiveMempool { fn restore_victims( &mut self, victims: &[Transaction], + accept_secs: &std::collections::HashMap, utxos: &impl UtxoProvider, tip: ChainTipCtx, ) { for i in Self::ancestor_first_order(victims) { - match self.reaccept_already_admitted(&victims[i], utxos, tip) { + let at = accept_secs.get(&victims[i].compute_txid()).copied(); + match self.reaccept_already_admitted(&victims[i], at, utxos, tip) { Ok(r) => self.promote_orphans_of(r.txid, utxos, tip), Err(_) => { if !self.undone_replacements.is_empty() { @@ -1680,7 +1713,8 @@ impl ActiveMempool { /// Re-accept conflicts a failed commit already removed. fn restore_undone(&mut self, utxos: &impl UtxoProvider, tip: ChainTipCtx) { let victims = std::mem::take(&mut self.undone_replacements); - self.restore_victims(&victims, utxos, tip); + let accept_secs = std::mem::take(&mut self.undone_accept_secs); + self.restore_victims(&victims, &accept_secs, utxos, tip); } /// Drop one live tx from the graph and `pwrite` its slot DEAD. No `fdatasync`. @@ -3683,6 +3717,109 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// A conflict put back after a failed replacement keeps the accept time it + /// was admitted with. The sidecar must not take the restore instant. + #[test] + fn failed_replacement_restore_keeps_accept_time_on_disk() { + let value = 50_000_000u64; + let dir = tmp_dir(); + let utxos = coins(&[(0x11, value), (0x22, value)]); + let anchor_op = OutPoint { + txid: Txid::from_byte_array([0x11; 32]), + vout: 0, + }; + let low_op = OutPoint { + txid: Txid::from_byte_array([0x22; 32]), + vout: 0, + }; + let clock = Arc::new(AtomicU64::new(0)); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + mp.set_node_clock(Arc::clone(&clock)); + let anchor_wu = fat_spend(anchor_op, value, 0).weight().to_wu(); + let victim_wu = spend_tx_fee(low_op, value, 0).weight().to_wu(); + mp.max_weight = anchor_wu + victim_wu + 500; + let floor = mp.mempool_min_fee_sat_kvb(); + let anchor = fat_spend(anchor_op, value, fee_at_least(anchor_wu, floor)); + let victim = spend_tx_fee(low_op, value, fee_at_least(victim_wu, floor)); + let victim_id = victim.compute_txid(); + let t0 = 1_700_000_000u64; + clock.store(t0, Ordering::Relaxed); + mp.accept_tx(&anchor, &utxos, TIP_OK).expect("anchor"); + mp.accept_tx(&victim, &utxos, TIP_OK).expect("victim"); + clock.store(t0 + 600, Ordering::Relaxed); + let replacement = heavy_conflict( + low_op, + OutPoint { + txid: anchor.compute_txid(), + vout: 0, + }, + value + anchor.output[0].value.to_sat(), + 30_000, + victim_wu + 500, + ); + let err = mp + .accept_tx(&replacement, &utxos, TIP_OK) + .expect_err("replacement over budget"); + assert!(matches!(err, AcceptError::Policy("mempool full")), "{err}"); + assert!(mp.graph.contains(&victim_id), "conflict restored"); + mp.store.flush().unwrap(); + drop(mp); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + let loaded = mp.take_loaded_accept_times(); + let at = loaded + .iter() + .find(|(t, _, _)| *t == victim_id) + .map(|(_, _, s)| *s); + assert_eq!(at, Some(t0), "restored conflict keeps its accept time"); + let _ = std::fs::remove_dir_all(&dir); + } + + /// A package member that replaced a conflict, followed by a member that + /// fails, rolls back. The restored conflict keeps its own accept time. + #[test] + fn package_rollback_restore_keeps_accept_time_on_disk() { + let value = 50_000_000u64; + let dir = tmp_dir(); + let utxos = coins(&[(0x33, value)]); + let op = OutPoint { + txid: Txid::from_byte_array([0x33; 32]), + vout: 0, + }; + let clock = Arc::new(AtomicU64::new(0)); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + mp.set_node_clock(Arc::clone(&clock)); + let low = spend_tx_fee(op, value, 1_000); + let low_id = low.compute_txid(); + let high = spend_tx_fee(op, value, 100_000); + let mut bad_child = spend_tx_fee( + OutPoint { + txid: high.compute_txid(), + vout: 0, + }, + high.output[0].value.to_sat(), + 10_000, + ); + bad_child.input[0].witness = Witness::from_slice(&[vec![0x01], vec![0x50, 0x01]]); + let t0 = 1_700_000_000u64; + clock.store(t0, Ordering::Relaxed); + mp.accept_tx(&low, &utxos, TIP_OK).expect("low"); + clock.store(t0 + 600, Ordering::Relaxed); + mp.accept_package(&[high.clone(), bad_child], &utxos, TIP_OK) + .expect_err("annex child fails"); + assert!(!mp.graph.contains(&high.compute_txid())); + assert!(mp.graph.contains(&low_id), "victim restored"); + mp.store.flush().unwrap(); + drop(mp); + let mut mp = ActiveMempool::open_or_create(&dir).unwrap(); + let at = mp + .take_loaded_accept_times() + .iter() + .find(|(t, _, _)| *t == low_id) + .map(|(_, _, s)| *s); + assert_eq!(at, Some(t0), "restored victim keeps its accept time"); + let _ = std::fs::remove_dir_all(&dir); + } + fn coins(specs: &[(u8, u64)]) -> MapUtxoProvider { let mut map = HashMap::new(); for &(tag, value) in specs { diff --git a/crates/rbitcoin-mempool/src/lib.rs b/crates/rbitcoin-mempool/src/lib.rs index 77ab696b7..99d64dd2a 100644 --- a/crates/rbitcoin-mempool/src/lib.rs +++ b/crates/rbitcoin-mempool/src/lib.rs @@ -5,9 +5,9 @@ //! //! | File | Role | //! |------|------| -//! | `meta` | Magic, schema **3**, commit generation **G**, slot capacity, live count | +//! | `meta` | Magic, schema **4**, commit generation **G**, slot capacity, live count | //! | `slots` | Fixed-size slot records (status + body range + txid) | -//! | `tx.body` | Packed live records (fee, weight, sigop cost, txid, wtxid, packed tx, vin aux) | +//! | `tx.body` | Packed live records (fee, weight, sigop cost, accept time, txid, wtxid, packed tx, vin aux) | //! //! **Commit model:** body tail complete → slot LIVE → RAM graph. No fsync per //! tx. Admits persist on a 5 s timer ([`ActiveMempool::persist_due`]); that path @@ -18,7 +18,10 @@ //! slot dump, not one sync per tx). //! [`ActiveMempool::flush`] bumps `G`, rewrites slots, and `sync_data`s. Crash //! may lose ≤5 s of admits; never LIVE slots past durable `tx.body`. Leftover -//! schema 1 converts to packed on open (vin aux empty; SH reindex batch-fills). +//! schema 1–3 converts on open. Schema 1 has empty vin aux (SH reindex +//! batch-fills), schema ≤ 2 has unknown sigops until the hub recomputes them, +//! and every upgraded record takes the migrate's wall-clock second as its +//! accept time. //! //! Packed decode uses stored txid/wtxid (no SHA256d). Vin aux is hashed at //! admit from resolved prevouts. Compact copies packed payload ranges. diff --git a/crates/rbitcoin-mempool/src/packed.rs b/crates/rbitcoin-mempool/src/packed.rs index 442142b98..7f8ba5ed5 100644 --- a/crates/rbitcoin-mempool/src/packed.rs +++ b/crates/rbitcoin-mempool/src/packed.rs @@ -3,7 +3,8 @@ //! //! Schema 3 had no `accept_secs` and schema 2 had no `sigops` either; //! [`decode_packed_live_v3`] / [`decode_packed_live_v2`] read them for the soft -//! migrate, which stamps the migrate time and writes [`SIGOP_COST_UNKNOWN`] +//! migrate. Every upgraded record takes the migrate time as its accept time. +//! Schema 3 keeps its sigop cost; schema ≤ 2 writes [`SIGOP_COST_UNKNOWN`] //! until the hub recomputes. //! //! Vin codec mirrors Class A packed flags/witness but writes `prev_txid` when @@ -27,6 +28,8 @@ const VIN_HAS_SCRIPT_HASH: u8 = 1 << 4; const SIGOP_COST_UNKNOWN: u64 = u64::MAX; /// Byte offset of `sigops` in a schema-3 or schema-4 record. pub(crate) const SIGOP_COST_OFF: usize = 16; +/// Byte offset of `accept_secs` in a schema-4 record. +pub(crate) const ACCEPT_SECS_OFF: usize = SIGOP_COST_OFF + 8; /// Per-vin aux persisted with the packed body (SH index / tip-entry purge). #[derive(Debug, Clone, PartialEq, Eq)] @@ -510,7 +513,7 @@ mod tests { assert_eq!(decode_packed_live(&rec).unwrap().sigop_cost, want); } let mut v3 = v4.clone(); - v3.drain(SIGOP_COST_OFF + 8..SIGOP_COST_OFF + 16); + v3.drain(ACCEPT_SECS_OFF..ACCEPT_SECS_OFF + 8); let got = decode_packed_live_v3(&v3).unwrap(); assert_eq!((got.fee_sat, got.weight, got.sigop_cost), (7, 400, None)); assert_eq!(got.accept_secs, 0, "schema 3 has no accept time"); diff --git a/crates/rbitcoin-mempool/src/store.rs b/crates/rbitcoin-mempool/src/store.rs index fdcf86c8f..60bec1750 100644 --- a/crates/rbitcoin-mempool/src/store.rs +++ b/crates/rbitcoin-mempool/src/store.rs @@ -7,7 +7,8 @@ //! | `{datadir}/store/tx.body` | **Class A** confirmed archive — confirm commit sole writer | //! | `{datadir}/mempool/tx.body` | **This file** — unconfirmed live set only | //! -//! Schema **3** packed live records (schema 2 + per-record sigop cost). Body is append-only (`body_persisted_len`); +//! Schema **4** packed live records (`fee‖weight‖sigops‖accept_secs‖…`); schema 1–3 +//! recode on open. Body is append-only (`body_persisted_len`); //! `persist_due` syncs the dirty tail, then syncs new LIVE slots, then writes meta. //! Compact copies packed payload ranges. DEAD of a durable slot is a one-byte //! `pwrite`. A batch syncs `slots` once, then `meta`. @@ -24,7 +25,7 @@ use crate::error::MempoolError; use crate::packed::{ decode_packed_live, decode_packed_live_v2, decode_packed_live_v3, encode_packed_live, - PackedLive, VinAux, SIGOP_COST_OFF, + PackedLive, VinAux, ACCEPT_SECS_OFF, SIGOP_COST_OFF, }; use bitcoin::consensus::encode::deserialize; use bitcoin::hashes::Hash; @@ -591,6 +592,21 @@ impl Mempool { Ok(()) } + /// `accept_secs` of a LIVE slot's record, read from RAM. + pub(crate) fn live_accept_secs(&self, slot: u32) -> Option { + if slot >= self.slot_cap { + return None; + } + let off = SLOTS_HEADER + (slot as usize) * SLOT_REC; + if self.slots[off] != SLOT_LIVE { + return None; + } + let body_off = u64::from_le_bytes(self.slots[off + 4..off + 12].try_into().unwrap()); + let at = body_off as usize + ACCEPT_SECS_OFF; + let bytes = self.body.get(at..at + 8)?; + Some(u64::from_le_bytes(bytes.try_into().unwrap())) + } + /// Load LIVE txs whose packed bytes sit inside the logical body. /// /// A slot past that length is the issue 859 tear (slots reached disk, the diff --git a/crates/rbitcoin-net/src/tx_relay.rs b/crates/rbitcoin-net/src/tx_relay.rs index 6a2f21e6d..da3006050 100644 --- a/crates/rbitcoin-net/src/tx_relay.rs +++ b/crates/rbitcoin-net/src/tx_relay.rs @@ -2273,7 +2273,9 @@ impl MempoolHub { time_prepare_lock: false, min_relay: Some(0), }; - let prep = self.admit_staged(tx, utxo, spec, &mut stages, &mut lock_us)?; + let mut prep = self.admit_staged(tx, utxo, spec, &mut stages, &mut lock_us)?; + // Keep the time the hub already holds, so disk and RAM agree. + prep.keep_accept_secs = self.accept_time_txid(&prep.txid); let sample = self.sample_chain_coins(tx, utxo); let committed = { let mut g = self.lock_write(); @@ -3762,6 +3764,7 @@ impl MempoolHub { replaced: Vec::new(), replaced_scripthashes: Vec::new(), replaced_txs: Vec::new(), + replaced_accept_secs: Default::default(), evicted: Vec::new(), }) } @@ -7114,6 +7117,103 @@ mod tests { } } + /// A conflict restored after a failed replacement keeps its accept time + /// in RAM and on disk, so a restart moves neither `getmempoolentry.time` + /// nor its expiry. + #[test] + fn failed_replacement_restore_keeps_accept_time_across_restart() { + let (_store, owned_q, owned_cbs) = pad_cbs(3); + let q = &owned_q; + let cbs = owned_cbs.as_slice(); + for package in [false, true] { + let mut fat = Vec::with_capacity(5_000); + for _ in 0..9 { + fat.push(0x4d); + fat.extend_from_slice(&520u16.to_le_bytes()); + fat.extend(std::iter::repeat_n(0u8, 520)); + fat.push(0x75); + } + fat.push(0x51); + let parent = spend_true(cbs[0], 2_000, ScriptBuf::from_bytes(fat)); + let parent_id = parent.compute_txid(); + let victim = spend_true(cbs[1], 3_000, ScriptBuf::from_bytes(vec![0x51])); + let victim_id = victim.compute_txid(); + let parent_out = parent.output[0].value.to_sat(); + let repl_fee = 60_000u64; + let replacement = Transaction { + version: Version::TWO, + lock_time: LockTime::ZERO, + input: vec![ + TxIn { + previous_output: OutPoint { + txid: cbs[1], + vout: 0, + }, + script_sig: ScriptBuf::new(), + sequence: Sequence::ENABLE_RBF_NO_LOCKTIME, + witness: Witness::new(), + }, + TxIn { + previous_output: OutPoint { + txid: parent_id, + vout: 0, + }, + script_sig: ScriptBuf::new(), + sequence: Sequence::ENABLE_RBF_NO_LOCKTIME, + witness: Witness::new(), + }, + ], + output: vec![TxOut { + value: Amount::from_sat(50_0000_0000 + parent_out - repl_fee), + script_pubkey: ScriptBuf::from_bytes(vec![0x51; 800]), + }], + }; + let budget = parent.weight().to_wu() + victim.weight().to_wu() + 500; + let dir = tmp(); + let hub = MempoolHub::open_with_weight(&dir, Arc::clone(q), budget).unwrap(); + hub.set_relay_enabled(true); + let t0 = 1_700_000_000u64; + hub.note_mock_now(t0); + hub.accept_tx(&parent).expect("parent fits"); + hub.accept_tx(&victim).expect("victim fits"); + hub.note_mock_now(t0 + 600); + let err = if package { + hub.accept_package(std::slice::from_ref(&replacement)) + .map(|_| ()) + } else { + hub.accept_tx(&replacement).map(|_| ()) + }; + assert!( + matches!(err, Err(AcceptError::Policy("mempool full"))), + "package={package}: got {err:?}" + ); + assert!(hub.contains(&victim_id), "package={package}: restored"); + assert_eq!( + hub.accept_time_txid(&victim_id), + Some(t0), + "package={package}: RAM keeps the original accept time" + ); + hub.flush().expect("shutdown flush"); + drop(hub); + let hub2 = MempoolHub::open_with_weight(&dir, Arc::clone(q), budget).unwrap(); + assert!(hub2.contains(&victim_id), "package={package}: reloaded"); + assert_eq!( + hub2.accept_time_txid(&victim_id), + Some(t0), + "package={package}: disk keeps the original accept time" + ); + hub2.set_expiry_hours(1); + hub2.note_mock_now(t0 + 3600 + 5); + assert!( + hub2.expire_stale() >= 1, + "package={package}: expiry counts from the original accept" + ); + assert!(!hub2.contains(&victim_id)); + drop(hub2); + let _ = std::fs::remove_dir_all(&dir); + } + } + /// A conflict cheaper than the chunk the failed attempt evicted must stay, /// and must not be recent-rejected. #[test] diff --git a/docs/crash-recovery.md b/docs/crash-recovery.md index 07b546cab..ab778ae23 100644 --- a/docs/crash-recovery.md +++ b/docs/crash-recovery.md @@ -119,7 +119,7 @@ Private, **not** Class A. RAM graph is source of truth; files may lag. - **5 s admits:** one `fdatasync` of the dirty body, then the slot sync, per dirty interval, from the tip-follow perf tick. Admits that missed that sync can still be lost (relay re-fetch). Shutdown `flush` generation-bumps and uses the same sync order. - **Unreadable image:** bad magic, a schema this binary does not accept, or an in-range payload that does not decode or whose txid does not match the slot. Open moves `meta`, `slots`, `tx.body`, and their `.tmp` files under `mempool/torn-/` and starts empty. `fee_history` and `fee_history.log` stay. An IO error (the mempool path is not a directory, or the open returns IO rather than corrupt, magic, or schema) still fails startup. Class A is untouched. - **DEAD:** already-durable slots are a one-byte `pwrite` of the status. The owner of a removal (a block's confirmed txs and their conflict spends, an RBF, an expiry scan, a lone `remove_txid`) `fdatasync`s `slots` once, then writes `meta`. Open reloads from slot status and sets `live_count` from that scan, so a crash after the slots sync and before `meta` keeps the DEAD rows. A crash before the slots sync can resurrect those txs (same class as a lost admit: the body bytes are unchanged and still decode). An admit that never hit disk stays RAM-only (crash loses it). Do not dump the full slot table on strip — that would write LIVE rows whose body is still in the unpersisted tail. The many one-byte `pwrite`s dirty one 4K page per touched page; the following `fdatasync` writes those pages once. -- **Leftover schema 1:** convert on open (recode LIVE payloads, tmp+rename like compact). Vin aux is empty; SH reindex batch-fills. Unknown schema takes the unreadable-image path above. +- **Leftover schema 1–3:** convert on open (recode LIVE payloads, tmp+rename like compact). Every upgraded record takes the wall-clock second of that open as its accept time. Schema 1 has empty vin aux (SH reindex batch-fills); schema ≤ 2 has unknown sigops until the hub recomputes them; schema 3 keeps its sigop cost. Unknown schema takes the unreadable-image path above. ## Operator From b610c39928402bbcefe33108227bdd1dbf754c5d Mon Sep 17 00:00:00 2001 From: Benjamen Keroack Date: Fri, 16 Oct 2026 15:13:45 -0600 Subject: [PATCH 5/5] net: an addr-permission IBD peer is not rate limited Tip-follow lets a peer with the addr permission through the token bucket; IBD always applied it. IBD now reads the same whitelist (outbound, peer address in the bind slot, as for noban) and lets such a peer through, still spending its tokens. Co-Authored-By: Claude Opus 5.5 --- crates/rbitcoin-net/src/ibd/events/mod.rs | 48 +++++++++++++++++++---- 1 file changed, 40 insertions(+), 8 deletions(-) diff --git a/crates/rbitcoin-net/src/ibd/events/mod.rs b/crates/rbitcoin-net/src/ibd/events/mod.rs index 1936c0fdb..b9eb99352 100644 --- a/crates/rbitcoin-net/src/ibd/events/mod.rs +++ b/crates/rbitcoin-net/src/ibd/events/mod.rs @@ -873,18 +873,27 @@ fn apply_notfound(st: &mut IbdWorkState, peer: usize, hashes: Vec) { } /// How many of `want` addresses from `peer` may enter the dial book. The -/// bucket counts from the peer's connect on the IBD mono clock. +/// bucket counts from the peer's connect on the IBD mono clock. A peer with +/// the `addr` permission spends tokens but is never limited, as in tip-follow. fn take_addr_tokens(st: &mut IbdWorkState, peer: usize, want: usize) -> usize { let now = super::peer_io::ibd_mono_ms(); - let start = st - .slots - .iter() - .find(|s| s.id == peer) - .map_or(now, |s| s.connected_ms); - st.addr_buckets + let slot = st.slots.iter().find(|s| s.id == peer); + let start = slot.map_or(now, |s| s.connected_ms); + // IBD dials outbound only, so the peer address fills the bind slot. + let exempt = slot.zip(st.perms.as_ref()).is_some_and(|(s, p)| { + p.permission_flags(s.addr, false, s.addr) + .has(crate::net_permissions::NetPermissionFlags::ADDR) + }); + let took = st + .addr_buckets .entry(peer) .or_insert_with(|| crate::peer::AddrTokenBucket::new(start)) - .take(want, now) + .take(want, now); + if exempt { + want + } else { + took + } } fn apply_peer_dead(st: &mut IbdWorkState, peer_book: &mut AddrMan, peer: usize, reason: String) { @@ -1413,6 +1422,29 @@ mod addr_bucket_tests { "a dead peer's bucket is dropped" ); } + + /// An outbound `addr` whitelist exempts an IBD peer as it does a + /// tip-follow session. + #[test] + fn ibd_addr_permission_peer_is_not_rate_limited() { + let mut slot = dummy_slot(1); + slot.connected_ms = ibd_mono_ms(); + let ip = slot.addr.ip(); + let mut st = IbdWorkState::new(vec![slot], None, Some(0)); + let peers = crate::peers::PeerHub::new(); + let mut table = crate::net_permissions::NetPermTable::default(); + table + .whitelist + .push(crate::net_permissions::parse_whitelist(&format!("addr,out@{ip}")).unwrap()); + peers.set_net_perms(table); + st.perms = Some(peers); + assert_eq!(take_addr_tokens(&mut st, 1, 1_000), 1_000); + assert_eq!( + take_addr_tokens(&mut st, 1, 1_000), + 1_000, + "an addr-permission peer is never limited" + ); + } } #[cfg(test)]