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-mempool/src/accept.rs b/crates/rbitcoin-mempool/src/accept.rs index 7209d95c3..60a561ee0 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, @@ -177,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. @@ -430,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). @@ -447,6 +456,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 +495,33 @@ 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. + 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) @@ -494,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) } } @@ -606,6 +642,9 @@ 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, }) } @@ -695,8 +734,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 +760,7 @@ impl ActiveMempool { graph, bodies, vin_aux, + accept_secs, } } @@ -992,6 +1034,7 @@ impl ActiveMempool { prevouts, chain_coins, utxo_us, + keep_accept_secs: None, }) } @@ -1079,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; @@ -1086,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); @@ -1111,6 +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 = 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, @@ -1119,6 +1173,7 @@ impl ActiveMempool { fee_sat, weight, prep.sigop_cost, + accept_secs, &aux, ) { Ok(slot) => slot, @@ -1168,15 +1223,18 @@ 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, fee_sat, weight: adj_weight, slot, + accept_secs, replaced: conflict_set.into_iter().collect(), replaced_scripthashes, replaced_txs, + replaced_accept_secs, evicted, }) } @@ -1194,9 +1252,11 @@ 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(), + replaced_accept_secs: std::collections::HashMap::new(), evicted: Vec::new(), }) } @@ -1558,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. @@ -1612,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) } @@ -1626,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() { @@ -1644,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`. @@ -2815,7 +2885,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(); } @@ -3638,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 ffa1ac140..7f8ba5ed5 100644 --- a/crates/rbitcoin-mempool/src/packed.rs +++ b/crates/rbitcoin-mempool/src/packed.rs @@ -1,7 +1,11 @@ -//! 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. 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 //! there is no Class A `create_fk`. Outputs reuse [`OutputRecord`] unspent @@ -22,8 +26,10 @@ 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; +/// 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)] @@ -34,19 +40,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 +64,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 +75,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 +93,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 +126,11 @@ fn decode_packed(buf: &[u8], has_sigops: bool) -> Result Result Transaction { let mut witness = Witness::new(); witness.push([0x01, 0x02]); @@ -342,18 +373,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 +404,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 +420,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 +469,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 +478,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 +499,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(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"); + 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..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`. @@ -23,8 +24,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, ACCEPT_SECS_OFF, SIGOP_COST_OFF, }; use bitcoin::consensus::encode::deserialize; use bitcoin::hashes::Hash; @@ -37,7 +38,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 +326,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 +477,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 +520,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 +537,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; @@ -571,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 @@ -994,7 +1030,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 +1179,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 +1204,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 +1240,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 +1272,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 +1338,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 +1411,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 +1587,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 +1622,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 +1695,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 +1743,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 +1852,7 @@ pub(crate) mod tests { 10, 400, 0, + 1_700_000_000, &[], ) .unwrap(); @@ -1732,7 +1926,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 +1954,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 +1989,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 +2034,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 +2148,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/ibd/events/mod.rs b/crates/rbitcoin-net/src/ibd/events/mod.rs index fb1593fe0..b9eb99352 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,33 @@ 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. 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 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); + if exempt { + want + } else { + took + } +} + 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 +1394,59 @@ 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" + ); + } + + /// 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)] 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-net/src/tx_relay.rs b/crates/rbitcoin-net/src/tx_relay.rs index c7d2c7fe3..da3006050 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,21 @@ 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. 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) { + let seq = hub.next_relay_seq.fetch_add(1, Ordering::Relaxed); + hub.insert_relay_maps(txid, wtxid, seq, at, false); + } + } { let mut u = hub.unbroadcast.lock().unwrap(); u.retain(|t| hub.contains(t)); @@ -1028,8 +1042,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 +1061,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); } @@ -2256,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(); @@ -2333,7 +2352,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 +2778,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 +3084,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,16 +3743,28 @@ 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(), + replaced_accept_secs: Default::default(), evicted: Vec::new(), }) } @@ -4451,6 +4482,43 @@ 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" + ); + // 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"); + 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 +5161,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 +5193,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 +5223,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 +5947,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 +5962,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 +6327,7 @@ mod tests { 4_999_999_001, tx.weight().to_wu(), u64::MAX, + 1_700_000_000, &[], ) .unwrap(); @@ -7036,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/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/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/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 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..0d7430623 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. @@ -601,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