mempool: keep accept time across restart; net: rate limit addrv2 before AddrMan

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).

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<Txid>,
     /// 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<Transaction>,
+    /// Accept times of [`Self::replaced_txs`], so a rollback restores each
+    /// with its own time.
+    pub replaced_accept_secs: std::collections::HashMap<Txid, u64>,
     /// 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<TxOut>,
     pub chain_coins: Vec<Option<Coin>>,
     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<u64>,
 }
 
 /// Graph spend edges for one tx (conflicts + mempool parents).
@@ -447,6 +456,8 @@ struct IngestedLive {
     graph: TxGraph,
     bodies: std::collections::HashMap<Txid, Arc<Transaction>>,
     vin_aux: std::collections::HashMap<Txid, Vec<VinAux>>,
+    /// `(txid, wtxid, accept_secs)` per loaded record.
+    accept_secs: Vec<(Txid, Wtxid, u64)>,
 }
 
 /// Mempool with RAM TxGraph layered on durable store.
@@ -484,16 +495,41 @@ pub struct ActiveMempool {
     /// Bodies removed for replacement when `finish_commit` later returns
     /// `Err`. Success moves them onto [`AcceptResult::replaced_txs`].
     undone_replacements: Vec<Transaction>,
+    /// Accept times of `undone_replacements`, read before their slots died.
+    undone_accept_secs: std::collections::HashMap<Txid, u64>,
+    /// `setmocktime` seconds, shared with the hub's clock (0 = wall).
+    node_clock: Arc<AtomicU64>,
+    /// 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<AtomicU64>) {
+        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<Txid> {
         std::mem::take(&mut self.last_evicted)
     }
 
     /// Conflicts a failed `finish_commit` already removed. Empty after a success.
     pub fn take_undone_replacements(&mut self) -> Vec<Transaction> {
+        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,13 +1122,21 @@ impl ActiveMempool {
         trim: bool,
     ) -> Result<AcceptResult, AcceptError> {
         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;
 
         let mut replaced_scripthashes: Vec<[u8; 32]> = Vec::new();
         let mut replaced_txs: Vec<Transaction> = 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<Txid, u64> = 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,25 +1676,30 @@ 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<u64>,
         utxos: &impl UtxoProvider,
         tip: ChainTipCtx,
     ) -> Result<AcceptResult, AcceptError> {
-        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)
     }
 
     fn restore_victims(
         &mut self,
         victims: &[Transaction],
+        accept_secs: &std::collections::HashMap<Txid, u64>,
         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 {