Introduce evicted-at/last-evicted timestamps

crates/bitcoind_rpc/src/lib.rs

@@ -10,8 +10,9 @@
 #![warn(missing_docs)]
 
 use bdk_core::{BlockId, CheckPoint};
-use bitcoin::{block::Header, Block, BlockHash, Transaction};
+use bitcoin::{block::Header, Block, BlockHash, Transaction, Txid};
 use bitcoincore_rpc::bitcoincore_rpc_json;
+use std::collections::HashSet;
 
 pub mod bip158;
 
@@ -64,17 +65,19 @@ impl<'c, C: bitcoincore_rpc::RpcApi> Emitter<'c, C> {
         }
     }
 
-    /// Emit mempool transactions, alongside their first-seen unix timestamps.
+    /// Emit mempool transactions and capture the initial snapshot of all mempool [`Txid`]s.
     ///
-    /// This method emits each transaction only once, unless we cannot guarantee the transaction's
-    /// ancestors are already emitted.
+    /// This method returns a [`MempoolEvent`] containing the full transactions (with their
+    /// first-seen unix timestamps) that were emitted, and the set of all [`Txid`]s present from the
+    /// initial mempool query. Each transaction is emitted only once, unless we cannot guarantee the
+    /// transaction's ancestors are already emitted.
     ///
     /// To understand why, consider a receiver which filters transactions based on whether it
     /// alters the UTXO set of tracked script pubkeys. If an emitted mempool transaction spends a
     /// tracked UTXO which is confirmed at height `h`, but the receiver has only seen up to block
     /// of height `h-1`, we want to re-emit this transaction until the receiver has seen the block
     /// at height `h`.
-    pub fn mempool(&mut self) -> Result<Vec<(Transaction, u64)>, bitcoincore_rpc::Error> {
+    pub fn mempool(&mut self) -> Result<MempoolEvent, bitcoincore_rpc::Error> {
         let client = self.client;
 
         // This is the emitted tip height during the last mempool emission.
@@ -91,8 +94,11 @@ impl<'c, C: bitcoincore_rpc::RpcApi> Emitter<'c, C> {
         let prev_mempool_time = self.last_mempool_time;
         let mut latest_time = prev_mempool_time;
 
-        let txs_to_emit = client
-            .get_raw_mempool_verbose()?
+        // Get the raw mempool result from the RPC client.
+        let raw_mempool = client.get_raw_mempool_verbose()?;
+        let raw_mempool_txids = raw_mempool.keys().copied().collect::<HashSet<_>>();
+
+        let emitted_txs = raw_mempool
             .into_iter()
             .filter_map({
                 let latest_time = &mut latest_time;
@@ -128,7 +134,11 @@ impl<'c, C: bitcoincore_rpc::RpcApi> Emitter<'c, C> {
         self.last_mempool_time = latest_time;
         self.last_mempool_tip = Some(self.last_cp.height());
 
-        Ok(txs_to_emit)
+        Ok(MempoolEvent {
+            emitted_txs,
+            raw_mempool_txids,
+            last_seen: latest_time as u64,
+        })
     }
 
     /// Emit the next block height and header (if any).
@@ -144,6 +154,37 @@ impl<'c, C: bitcoincore_rpc::RpcApi> Emitter<'c, C> {
     }
 }
 
+/// A new emission from mempool.
+#[derive(Debug)]
+pub struct MempoolEvent {
+    /// Emitted mempool transactions with their first‐seen unix timestamps.
+    pub emitted_txs: Vec<(Transaction, u64)>,
+
+    /// Set of all [`Txid`]s from the raw mempool result, including transactions that may have been
+    /// confirmed or evicted during processing. This is used to determine which expected
+    /// transactions are missing.
+    pub raw_mempool_txids: HashSet<Txid>,
+
+    /// The latest first-seen epoch of emitted mempool transactions.
+    pub last_seen: u64,
+}
+
+impl MempoolEvent {
+    /// Given an iterator of expected [`Txid`]s, return those that are missing from the mempool.
+    pub fn evicted_txids(
+        &self,
+        expected_unconfirmed_txids: impl IntoIterator<Item = Txid>,
+    ) -> HashSet<Txid> {
+        let expected_set = expected_unconfirmed_txids
+            .into_iter()
+            .collect::<HashSet<_>>();
+        expected_set
+            .difference(&self.raw_mempool_txids)
+            .copied()
+            .collect()
+    }
+}
+
 /// A newly emitted block from [`Emitter`].
 #[derive(Debug)]
 pub struct BlockEvent<B> {

crates/bitcoind_rpc/tests/test_emitter.rs

@@ -189,7 +189,7 @@ fn test_into_tx_graph() -> anyhow::Result<()> {
         assert!(emitter.next_block()?.is_none());
 
         let mempool_txs = emitter.mempool()?;
-        let indexed_additions = indexed_tx_graph.batch_insert_unconfirmed(mempool_txs);
+        let indexed_additions = indexed_tx_graph.batch_insert_unconfirmed(mempool_txs.emitted_txs);
         assert_eq!(
             indexed_additions
                 .tx_graph
@@ -437,6 +437,7 @@ fn mempool_avoids_re_emission() -> anyhow::Result<()> {
     // the first emission should include all transactions
     let emitted_txids = emitter
         .mempool()?
+        .emitted_txs
         .into_iter()
         .map(|(tx, _)| tx.compute_txid())
         .collect::<BTreeSet<Txid>>();
@@ -447,7 +448,7 @@ fn mempool_avoids_re_emission() -> anyhow::Result<()> {
 
     // second emission should be empty
     assert!(
-        emitter.mempool()?.is_empty(),
+        emitter.mempool()?.emitted_txs.is_empty(),
         "second emission should be empty"
     );
 
@@ -457,7 +458,7 @@ fn mempool_avoids_re_emission() -> anyhow::Result<()> {
     }
     while emitter.next_header()?.is_some() {}
     assert!(
-        emitter.mempool()?.is_empty(),
+        emitter.mempool()?.emitted_txs.is_empty(),
         "third emission, after chain tip is extended, should also be empty"
     );
 
@@ -506,6 +507,7 @@ fn mempool_re_emits_if_tx_introduction_height_not_reached() -> anyhow::Result<()
     assert_eq!(
         emitter
             .mempool()?
+            .emitted_txs
             .into_iter()
             .map(|(tx, _)| tx.compute_txid())
             .collect::<BTreeSet<_>>(),
@@ -515,6 +517,7 @@ fn mempool_re_emits_if_tx_introduction_height_not_reached() -> anyhow::Result<()
     assert_eq!(
         emitter
             .mempool()?
+            .emitted_txs
             .into_iter()
             .map(|(tx, _)| tx.compute_txid())
             .collect::<BTreeSet<_>>(),
@@ -535,6 +538,7 @@ fn mempool_re_emits_if_tx_introduction_height_not_reached() -> anyhow::Result<()
                 .collect::<BTreeSet<_>>();
             let emitted_txids = emitter
                 .mempool()?
+                .emitted_txs
                 .into_iter()
                 .map(|(tx, _)| tx.compute_txid())
                 .collect::<BTreeSet<_>>();
@@ -593,6 +597,7 @@ fn mempool_during_reorg() -> anyhow::Result<()> {
     assert_eq!(
         emitter
             .mempool()?
+            .emitted_txs
             .into_iter()
             .map(|(tx, _)| tx.compute_txid())
             .collect::<BTreeSet<_>>(),
@@ -628,6 +633,7 @@ fn mempool_during_reorg() -> anyhow::Result<()> {
             // include mempool txs introduced at reorg height or greater
             let mempool = emitter
                 .mempool()?
+                .emitted_txs
                 .into_iter()
                 .map(|(tx, _)| tx.compute_txid())
                 .collect::<BTreeSet<_>>();
@@ -643,6 +649,7 @@ fn mempool_during_reorg() -> anyhow::Result<()> {
 
             let mempool = emitter
                 .mempool()?
+                .emitted_txs
                 .into_iter()
                 .map(|(tx, _)| tx.compute_txid())
                 .collect::<BTreeSet<_>>();
@@ -731,3 +738,107 @@ fn no_agreement_point() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn test_expect_tx_evicted() -> anyhow::Result<()> {
+    use bdk_bitcoind_rpc::bitcoincore_rpc::bitcoin;
+    use bdk_bitcoind_rpc::bitcoincore_rpc::bitcoincore_rpc_json::CreateRawTransactionInput;
+    use bdk_chain::miniscript;
+    use bdk_chain::spk_txout::SpkTxOutIndex;
+    use bdk_chain::ConfirmationBlockTime;
+    use bitcoin::constants::genesis_block;
+    use bitcoin::secp256k1::Secp256k1;
+    use bitcoin::Network;
+    use std::collections::HashMap;
+    let env = TestEnv::new()?;
+
+    let s = bdk_testenv::utils::DESCRIPTORS[0];
+    let desc = miniscript::Descriptor::parse_descriptor(&Secp256k1::new(), s)
+        .unwrap()
+        .0;
+    let spk = desc.at_derivation_index(0)?.script_pubkey();
+
+    let chain = LocalChain::from_genesis_hash(genesis_block(Network::Regtest).block_hash()).0;
+    let chain_tip = chain.tip().block_id();
+
+    let mut index = SpkTxOutIndex::default();
+    index.insert_spk(("external", 0u32), spk.clone());
+    let mut graph = IndexedTxGraph::<ConfirmationBlockTime, _>::new(index);
+
+    // Receive tx1.
+    let _ = env.mine_blocks(100, None)?;
+    let txid_1 = env.send(
+        &Address::from_script(&spk, Network::Regtest)?,
+        Amount::ONE_BTC,
+    )?;
+
+    let mut emitter = Emitter::new(env.rpc_client(), chain.tip(), 1);
+    let changeset = graph.batch_insert_unconfirmed(emitter.mempool()?.emitted_txs);
+    assert!(changeset
+        .tx_graph
+        .txs
+        .iter()
+        .any(|tx| tx.compute_txid() == txid_1));
+    let seen_at = graph
+        .graph()
+        .get_tx_node(txid_1)
+        .unwrap()
+        .last_seen_unconfirmed
+        .unwrap();
+
+    // Double spend tx1.
+
+    // Get `prevout` from core.
+    let core = env.rpc_client();
+    let tx1 = &core.get_raw_transaction(&txid_1, None)?;
+    let txin = &tx1.input[0];
+    let op = txin.previous_output;
+
+    // Create `tx1b` using the previous output from tx1.
+    let utxo = CreateRawTransactionInput {
+        txid: op.txid,
+        vout: op.vout,
+        sequence: None,
+    };
+    let addr = core.get_new_address(None, None)?.assume_checked();
+    let tx = core.create_raw_transaction(
+        &[utxo],
+        &HashMap::from([(addr.to_string(), Amount::from_btc(49.99)?)]),
+        None,
+        None,
+    )?;
+    let res = core.sign_raw_transaction_with_wallet(&tx, None, None)?;
+    let tx1b = res.transaction()?;
+
+    // Send the tx.
+    let txid_2 = core.send_raw_transaction(&tx1b)?;
+
+    // Retrieve the expected unconfirmed txids and spks from the graph.
+    let exp_spk_txids = graph.expected_unconfirmed_spk_txids(&chain, chain_tip, ..)?;
+    assert_eq!(exp_spk_txids, vec![(txid_1, spk)]);
+
+    // Check that mempool emission contains evicted txid.
+    let mempool_event = emitter.mempool()?;
+    let unseen_txids: Vec<Txid> = mempool_event
+        .emitted_txs
+        .iter()
+        .map(|(tx, _)| tx.compute_txid())
+        .collect();
+    assert!(unseen_txids.contains(&txid_2));
+
+    // Update graph with evicted tx.
+    let exp_txids = exp_spk_txids.into_iter().map(|(txid, _)| txid);
+    let evicted_txids = mempool_event.evicted_txids(exp_txids);
+    for txid in evicted_txids {
+        let _ = graph.insert_evicted_at(txid, seen_at);
+    }
+
+    // tx1 should no longer be canonical.
+    assert!(graph
+        .graph()
+        .list_canonical_txs(&chain, chain_tip)
+        .next()
+        .is_none());
+
+    Ok(())
+}

crates/chain/benches/canonicalization.rs

@@ -132,10 +132,8 @@ pub fn many_conflicting_unconfirmed(c: &mut Criterion) {
                 }],
                 ..new_tx(i)
             };
-            let update = TxUpdate {
-                txs: vec![Arc::new(tx)],
-                ..Default::default()
-            };
+            let mut update = TxUpdate::default();
+            update.txs = vec![Arc::new(tx)];
             let _ = tx_graph.apply_update_at(update, Some(i as u64));
         }
     }));
@@ -169,10 +167,8 @@ pub fn many_chained_unconfirmed(c: &mut Criterion) {
                 ..new_tx(i)
             };
             let txid = tx.compute_txid();
-            let update = TxUpdate {
-                txs: vec![Arc::new(tx)],
-                ..Default::default()
-            };
+            let mut update = TxUpdate::default();
+            update.txs = vec![Arc::new(tx)];
             let _ = tx_graph.apply_update_at(update, Some(i as u64));
             // Store the next prevout.
             previous_output = OutPoint::new(txid, 0);