mempool: save and reload parents before their children

CHANGELOG.md

@@ -41,6 +41,7 @@ item below is (or will be) written up in full in the in-development
 - Address announcements are rate-limited per peer as in Core (`addr_processed`, `addr_rate_limited` in `getpeerinfo`); dials in flight count against the outbound limit; manual connections sit outside it; `setban` of an onion host disconnects it (#952).
 - A full mempool checks an incoming transaction's scripts before evicting for it and never evicts its parent; eviction scales with what it removes; lowering `maxmempool` on reload trims at once (#942).
 - `getmempoolentry`, `getrawmempool`, `getmempoolinfo`, `gettxout` and `sendrawtransaction` answer as Core does; a restart keeps entry times and loads a large mempool in linear time (#946).
+- A restart re-admits the whole saved mempool: `mempool.dat` is written parents first, as Core writes it, and a file saved by 0.6.0 is reordered on load, where a child saved in the same second as its parent could be tried first and dropped with all its descendants (#949).
 - RPC numbers use Core's formatting (16 significant digits, 8-decimal fee amounts); `submitblock` answers `duplicate-invalid` and throws -22/-25; `getblockfilter` uses Core's errors; a failed RPC password is answered after 250 ms (#955).
 - Electrum scripthash status hashes the history in `get_history` order, which stops Electrum desktop reconnecting in a loop on addresses with pending transactions (#938).
 - Esplora honours `?after_txid=` paging, answers 404/400 for out-of-range or misaligned block-txs pages, drops empty `witness` arrays, and `/fee-estimates` omits targets with no estimate (#943).

docs/release-notes/0.6.1-pre.md

@@ -327,6 +327,18 @@ Descendants of a transaction that conflicts with a new block now leave the
 mempool with it, and confirmed transactions' priority deltas are cleared
 (#946).
 
+### A restart keeps the whole mempool
+
+A clean restart could drop most of the saved mempool. satd re-admitted the
+transactions in `mempool.dat` in order of entry time, which has one-second
+resolution, and the file was written in no particular order, so a child that
+entered in the same second as its parent was often tried first, refused for
+missing inputs, and took every descendant behind it along. Each restart made
+it worse, as the transactions it did re-admit then shared even fewer seconds.
+satd now writes the file parents first, as Bitcoin Core does, and on load
+puts every transaction after any parent in the same file, so a `mempool.dat`
+saved by 0.6.0 also reloads in full. The file format is unchanged (#949).
+
 ## Electrum and Esplora
 
 ### Electrum status hashes follow history order

node/src/mempool/persist.rs

@@ -59,7 +59,13 @@ pub struct LoadStats {
 /// torn file that breaks the next startup. Returns the number of
 /// transactions written.
 pub fn dump_mempool(mempool: &Mempool, net_datadir: &Path) -> std::io::Result<usize> {
-    let entries = mempool.get_all_entries();
+    // Parents before children, as Core writes its file (`DumpMempool` takes
+    // `infoAll()`, which is in topological order), so a reload that admits
+    // records in file order never meets a child before its parent.
+    let mut entries = mempool.get_all_entries();
+    entries.sort_by_key(|(txid, e)| (e.time, *txid));
+    let order = parent_first_order(entries.iter().map(|(_, e)| &e.tx));
+    let entries = permute(entries, &order);
     let unbroadcast: std::collections::HashSet<bitcoin::Txid> =
         mempool.unbroadcast_txids().into_iter().collect();
 
@@ -160,10 +166,16 @@ pub fn load_mempool(
         tracing::warn!(%note, "persisted mempool body was truncated/corrupt; loaded the valid prefix");
     }
 
-    // Re-admit oldest-first so a parent that entered before its child is
-    // accepted first; a child whose parent fails still just gets skipped.
+    // Re-admit oldest-first, but never a child before a parent that is in
+    // the same file. `time` has one-second resolution, so a parent and its
+    // child often tie, and a file written by an older satd is in no
+    // particular order; the dependency pass puts every in-file parent ahead
+    // of its children whatever the times say. A child whose parent fails
+    // still just gets skipped.
     let mut records = parsed.records;
     records.sort_by_key(|r| r.time);
+    let order = parent_first_order(records.iter().map(|r| &r.tx));
+    let records = permute(records, &order);
 
     // Core's `LoadMempool` (src/node/mempool_persist.cpp): the delta first,
     // so a transaction that met the floor only through
@@ -201,6 +213,60 @@ pub fn load_mempool(
     Ok(stats)
 }
 
+/// A stable topological order of `txs`: indices into the input, every
+/// transaction after each of its parents that is also in the input, and
+/// otherwise in input order. One depth-first pass over the spend edges, so
+/// linear in the number of inputs. A duplicate txid links its children to
+/// the first copy; a cycle cannot occur between real transactions (a txid
+/// commits to its inputs) and would just be broken where it was entered.
+fn parent_first_order<'a>(txs: impl Iterator<Item = &'a bitcoin::Transaction>) -> Vec<usize> {
+    let txs: Vec<&bitcoin::Transaction> = txs.collect();
+    let mut index: std::collections::HashMap<bitcoin::Txid, usize> =
+        std::collections::HashMap::with_capacity(txs.len());
+    for (i, tx) in txs.iter().enumerate() {
+        index.entry(tx.compute_txid()).or_insert(i);
+    }
+
+    const UNSEEN: u8 = 0;
+    const OPEN: u8 = 1;
+    const DONE: u8 = 2;
+    let mut state = vec![UNSEEN; txs.len()];
+    let mut order = Vec::with_capacity(txs.len());
+    // (transaction, next input to look at)
+    let mut stack: Vec<(usize, usize)> = Vec::new();
+    for root in 0..txs.len() {
+        if state[root] != UNSEEN {
+            continue;
+        }
+        state[root] = OPEN;
+        stack.push((root, 0));
+        while let Some((node, next)) = stack.last_mut() {
+            let node = *node;
+            if let Some(input) = txs[node].input.get(*next) {
+                *next += 1;
+                if let Some(&parent) = index.get(&input.previous_output.txid)
+                    && state[parent] == UNSEEN
+                {
+                    state[parent] = OPEN;
+                    stack.push((parent, 0));
+                }
+            } else {
+                state[node] = DONE;
+                order.push(node);
+                stack.pop();
+            }
+        }
+    }
+    order
+}
+
+/// `items` rearranged so that position `k` holds the item at `order[k]`.
+/// `order` must be a permutation of `0..items.len()`.
+fn permute<T>(items: Vec<T>, order: &[usize]) -> Vec<T> {
+    let mut slots: Vec<Option<T>> = items.into_iter().map(Some).collect();
+    order.iter().filter_map(|&i| slots[i].take()).collect()
+}
+
 struct Record {
     time: u64,
     fee_delta: i64,
@@ -328,6 +394,10 @@ impl<'a> Cursor<'a> {
     }
 }
 
+#[cfg(test)]
+#[path = "persist_reload_order_tests.rs"]
+mod reload_order_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -371,7 +441,7 @@ mod tests {
         .unwrap()
     }
 
-    fn dummy_tx(nonce: u8) -> bitcoin::Transaction {
+    pub(super) fn dummy_tx(nonce: u8) -> bitcoin::Transaction {
         bitcoin::Transaction {
             version: transaction::Version(2),
             lock_time: LockTime::ZERO,
@@ -395,7 +465,7 @@ mod tests {
 
     /// Frame records exactly as `dump_mempool` does (current version), for
     /// load-side tests.
-    fn frame(records: &[(u64, i64, bitcoin::Transaction)]) -> Vec<u8> {
+    pub(super) fn frame(records: &[(u64, i64, bitcoin::Transaction)]) -> Vec<u8> {
         let mut buf = Vec::new();
         buf.extend_from_slice(MAGIC);
         buf.extend_from_slice(&VERSION.to_le_bytes());
@@ -516,7 +586,7 @@ mod tests {
 
     /// A regtest chain holding one coin per `tag`, and an empty pool under
     /// `min_fee_rate`.
-    fn funded(tags: &[u8], min_fee_rate: u64) -> (ChainState, Mempool, std::path::PathBuf) {
+    pub(super) fn funded(tags: &[u8], min_fee_rate: u64) -> (ChainState, Mempool, std::path::PathBuf) {
         let coins: Vec<(OutPoint, crate::storage::coinview::Coin)> = tags
             .iter()
             .map(|t| {
@@ -553,14 +623,14 @@ mod tests {
         }
     }
 
-    fn p2wpkh(tag: u8) -> ScriptBuf {
+    pub(super) fn p2wpkh(tag: u8) -> ScriptBuf {
         let mut spk = vec![0x00, 0x14];
         spk.extend_from_slice(&[tag; 20]);
         ScriptBuf::from_bytes(spk)
     }
 
     /// A standard spend of the coin `funded` made for `tag`, paying `fee`.
-    fn funded_spend(tag: u8, fee: u64) -> bitcoin::Transaction {
+    pub(super) fn funded_spend(tag: u8, fee: u64) -> bitcoin::Transaction {
         let mut tx = dummy_tx(tag);
         tx.input[0].previous_output = funded_prev(tag);
         tx.output[0] = TxOut { value: Amount::from_sat(100_000 - fee), script_pubkey: p2wpkh(tag) };

node/src/mempool/persist_reload_order_tests.rs

@@ -0,0 +1,211 @@
+//! Reload order: a `mempool.dat` record is never re-admitted before a parent
+//! that is in the same file, whatever the records' one-second `time` values
+//! say and whatever order the file was written in.
+
+use super::tests::{dummy_tx, frame, funded, funded_spend, p2wpkh};
+use super::*;
+use crate::validation::script::NoopVerifier;
+use bitcoin::{Amount, OutPoint, ScriptBuf, Sequence, Transaction, TxIn, TxOut, Witness};
+
+/// A spend of every listed outpoint of `parents`, paying `fee` and splitting
+/// what is left over `outputs` outputs.
+fn spend(parents: &[(&Transaction, u32)], fee: u64, outputs: usize) -> Transaction {
+    let value: u64 = parents.iter().map(|(p, v)| p.output[*v as usize].value.to_sat()).sum();
+    let each = (value - fee) / outputs as u64;
+    Transaction {
+        version: bitcoin::transaction::Version(2),
+        lock_time: bitcoin::absolute::LockTime::ZERO,
+        input: parents
+            .iter()
+            .map(|(p, v)| TxIn {
+                previous_output: OutPoint { txid: p.compute_txid(), vout: *v },
+                script_sig: ScriptBuf::new(),
+                sequence: Sequence::MAX,
+                witness: Witness::new(),
+            })
+            .collect(),
+        output: (0..outputs)
+            .map(|i| TxOut { value: Amount::from_sat(each), script_pubkey: p2wpkh(0x40 + i as u8) })
+            .collect(),
+    }
+}
+
+/// Eight generations off the coin for tag 1: a two-output root, two
+/// children of it, a grandchild spending both children, then a chain of
+/// four more. Returned parent-first.
+fn family() -> Vec<Transaction> {
+    let mut root = funded_spend(1, 1_000);
+    let half = root.output[0].value.to_sat() / 2;
+    root.output = vec![
+        TxOut { value: Amount::from_sat(half), script_pubkey: p2wpkh(0x21) },
+        TxOut { value: Amount::from_sat(half), script_pubkey: p2wpkh(0x22) },
+    ];
+    let a = spend(&[(&root, 0)], 1_000, 1);
+    let b = spend(&[(&root, 1)], 1_000, 1);
+    let mut txs = vec![root.clone(), a.clone(), b.clone()];
+    let mut last = spend(&[(&a, 0), (&b, 0)], 1_000, 1);
+    txs.push(last.clone());
+    for _ in 0..4 {
+        last = spend(&[(&last, 0)], 1_000, 1);
+        txs.push(last.clone());
+    }
+    txs
+}
+
+/// A fresh pool under the same policy `funded` sets, standing in for the
+/// pool of the restarted node.
+fn restarted_pool() -> Mempool {
+    let mp = Mempool::new(1_000_000, 0);
+    mp.reload_policy(crate::mempool::pool::MempoolConfig {
+        max_size_bytes: 1_000_000,
+        min_fee_rate: 0,
+        ..Default::default()
+    });
+    mp
+}
+
+/// Panics unless every record whose parent is also in `txs` comes after it.
+fn assert_parent_first(txs: &[Transaction]) {
+    let pos: std::collections::HashMap<bitcoin::Txid, usize> =
+        txs.iter().enumerate().map(|(i, t)| (t.compute_txid(), i)).collect();
+    for (i, tx) in txs.iter().enumerate() {
+        for input in &tx.input {
+            if let Some(&p) = pos.get(&input.previous_output.txid) {
+                assert!(p < i, "record {i} comes before its parent at record {p}");
+            }
+        }
+    }
+}
+
+#[test]
+fn dump_writes_same_second_parents_before_children_and_all_reload() {
+    // Every entry shares one second, so `time` cannot order them. The file
+    // must still be parent-first, and a restart must keep the whole family.
+    let (cs, mp, dir) = funded(&[1], 0);
+    let then = crate::time::now_secs() - 60;
+    let txs = family();
+    for tx in &txs {
+        mp.accept_persisted_transaction(tx.clone(), &cs, &NoopVerifier, then)
+            .expect("the family is admissible parent-first");
+    }
+    assert_eq!(dump_mempool(&mp, &dir).unwrap(), txs.len());
+
+    let written = parse(&std::fs::read(dir.join("mempool.dat")).unwrap()).unwrap();
+    let written: Vec<Transaction> = written.records.into_iter().map(|r| r.tx).collect();
+    assert_eq!(written.len(), txs.len());
+    assert_parent_first(&written);
+
+    let fresh = restarted_pool();
+    let stats = load_mempool(&fresh, &cs, &NoopVerifier, &dir).unwrap();
+    assert_eq!((stats.accepted, stats.skipped, stats.expired), (txs.len(), 0, 0));
+    for tx in &txs {
+        assert_eq!(fresh.get(&tx.compute_txid()).expect("re-admitted").time, then);
+    }
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+#[test]
+fn load_admits_a_file_written_children_first() {
+    // The shape an older satd could write: children ahead of their parents,
+    // all in the same second. Every record is still re-admitted.
+    let (cs, mp, dir) = funded(&[1], 0);
+    let then = crate::time::now_secs() - 60;
+    let txs = family();
+    let records: Vec<(u64, i64, Transaction)> =
+        txs.iter().rev().map(|tx| (then, 0, tx.clone())).collect();
+    std::fs::write(dir.join("mempool.dat"), frame(&records)).unwrap();
+
+    let stats = load_mempool(&mp, &cs, &NoopVerifier, &dir).unwrap();
+    assert_eq!((stats.accepted, stats.skipped, stats.expired), (txs.len(), 0, 0));
+    for tx in &txs {
+        assert!(mp.get(&tx.compute_txid()).is_some(), "every generation is re-admitted");
+    }
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+#[test]
+fn load_admits_a_parent_stamped_later_than_its_child() {
+    // A dependency in the file outranks the times: the child's earlier
+    // stamp does not put it ahead of its parent.
+    let (cs, mp, dir) = funded(&[1], 0);
+    let now = crate::time::now_secs();
+    let parent = funded_spend(1, 1_000);
+    let child = spend(&[(&parent, 0)], 1_000, 1);
+    std::fs::write(
+        dir.join("mempool.dat"),
+        frame(&[(now - 50, 0, parent.clone()), (now - 90, 0, child.clone())]),
+    )
+    .unwrap();
+
+    let stats = load_mempool(&mp, &cs, &NoopVerifier, &dir).unwrap();
+    assert_eq!((stats.accepted, stats.skipped), (2, 0));
+    assert_eq!(mp.get(&parent.compute_txid()).expect("parent").time, now - 50);
+    assert_eq!(mp.get(&child.compute_txid()).expect("child").time, now - 90);
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+#[test]
+fn load_still_skips_a_child_whose_parent_is_missing() {
+    // The reordering only moves records; it admits nothing new. A child of
+    // a transaction that is neither in the file nor confirmed is skipped,
+    // and an unrelated record beside it is admitted.
+    let (cs, mp, dir) = funded(&[1, 2], 0);
+    let then = crate::time::now_secs() - 60;
+    let absent_parent = funded_spend(1, 1_000);
+    let orphan = spend(&[(&absent_parent, 0)], 1_000, 1);
+    let other = funded_spend(2, 1_000);
+    std::fs::write(
+        dir.join("mempool.dat"),
+        frame(&[(then, 0, orphan.clone()), (then, 0, other.clone())]),
+    )
+    .unwrap();
+
+    let stats = load_mempool(&mp, &cs, &NoopVerifier, &dir).unwrap();
+    assert_eq!((stats.accepted, stats.skipped), (1, 1));
+    assert!(mp.get(&orphan.compute_txid()).is_none());
+    assert!(mp.get(&other.compute_txid()).is_some());
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
+#[test]
+fn parent_first_order_keeps_unrelated_records_in_place() {
+    // No in-set links: the order is the input order, so the oldest-first
+    // sort ahead of it survives.
+    let txs: Vec<Transaction> = (1..=5).map(dummy_tx).collect();
+    assert_eq!(parent_first_order(txs.iter()), vec![0, 1, 2, 3, 4]);
+}
+
+#[test]
+fn parent_first_order_reorders_a_reversed_family() {
+    let mut txs = family();
+    txs.reverse();
+    let order = parent_first_order(txs.iter());
+    let mut sorted = order.clone();
+    sorted.sort_unstable();
+    assert_eq!(sorted, (0..txs.len()).collect::<Vec<_>>(), "a permutation");
+    assert_parent_first(&permute(txs, &order));
+}
+
+#[test]
+fn parent_first_order_handles_a_long_reversed_chain_and_duplicates() {
+    // Depth is walked on the heap, not the call stack, and a repeated
+    // record is kept rather than dropped.
+    let mut txs = vec![dummy_tx(7)];
+    for _ in 0..20_000 {
+        let last = txs.last().unwrap().clone();
+        let mut next = dummy_tx(7);
+        next.input[0].previous_output = OutPoint { txid: last.compute_txid(), vout: 0 };
+        txs.push(next);
+    }
+    txs.push(txs[3].clone());
+    txs.reverse();
+    let order = parent_first_order(txs.iter());
+    assert_eq!(order.len(), txs.len());
+    let ordered = permute(txs, &order);
+    // The duplicate links to nothing new; drop it before the parent check.
+    let mut seen = std::collections::HashSet::new();
+    let unique: Vec<Transaction> =
+        ordered.into_iter().filter(|t| seen.insert(t.compute_txid())).collect();
+    assert_eq!(unique.len(), 20_001);
+    assert_parent_first(&unique);
+}