fix(mempool): make initial-sync tx insertion idempotent

lib/mempool/sync/task.rs

@@ -636,6 +636,16 @@ where
     Enforcer: BorrowMut<BorrowedEnforcer>,
     BorrowedEnforcer: CusfEnforcer,
 {
+    // The initial-sync RPC snapshot and the live ZMQ `sequence` stream are
+    // reconciled together, so the same txid can be queued for insertion via
+    // both paths (e.g. the snapshot `InsertTx` and a buffered `Added`
+    // sequence message). Inserting a tx already in the mempool would make
+    // `Mempool::insert` fail with `TxAlreadyExists` and abort the whole sync
+    // task, so treat an already-present tx as a successfully-applied no-op.
+    if inner.mempool.txs.0.contains_key(&txid) {
+        inner.unfiltered_mempool.txs.insert(txid);
+        return Ok(true);
+    }
     let Some(tx) = sync_state.tx_cache.get(&txid) else {
         if sync_state.unavailable_txs.contains(&txid) {
             inner.unfiltered_mempool.txs.insert(txid);
@@ -1295,3 +1305,123 @@ where
         Some(res)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use bitcoin::{
+        Amount, BlockHash, Network, OutPoint, ScriptBuf, Sequence, Transaction,
+        TxIn, TxOut, Txid, Witness, absolute::LockTime, hashes::Hash as _,
+        transaction::Version,
+    };
+    use hashlink::LinkedHashSet;
+
+    use super::*;
+    use crate::{cusf_enforcer::DefaultEnforcer, mempool::Mempool};
+
+    fn make_tx(inputs: &[OutPoint], output_values: &[u64]) -> Transaction {
+        Transaction {
+            version: Version::TWO,
+            lock_time: LockTime::ZERO,
+            input: inputs
+                .iter()
+                .map(|prev| TxIn {
+                    previous_output: *prev,
+                    sequence: Sequence::MAX,
+                    script_sig: ScriptBuf::new(),
+                    witness: Witness::new(),
+                })
+                .collect(),
+            output: output_values
+                .iter()
+                .map(|value| TxOut {
+                    value: Amount::from_sat(*value),
+                    script_pubkey: ScriptBuf::new(),
+                })
+                .collect(),
+        }
+    }
+
+    /// Regression test for the initial-sync crash reported in
+    /// LayerTwo-Labs/bip300301_enforcer#406.
+    ///
+    /// During initial sync the RPC mempool snapshot and the live ZMQ
+    /// `sequence` stream are reconciled together, so the same txid can be
+    /// queued for insertion twice (e.g. once as the snapshot `InsertTx` and
+    /// once as a buffered `Added` sequence message). The second application
+    /// must be a no-op success rather than aborting the entire sync task with
+    /// `MempoolInsertError::TxAlreadyExists`.
+    #[test]
+    fn add_tx_from_caches_is_idempotent() {
+        let genesis = BlockHash::all_zeros();
+        let mut inner = MempoolSyncInner {
+            abandoned_pool: AbandonedPool::default(),
+            enforcer: DefaultEnforcer,
+            mempool: Mempool::new(Network::Regtest, genesis),
+            unfiltered_mempool: UnfilteredMempool {
+                tip: genesis,
+                txs: HashSet::new(),
+            },
+        };
+
+        // A parent tx supplying a spendable output, available only via the tx
+        // cache (it is not itself a mempool tx).
+        let parent =
+            make_tx(&[OutPoint::new(Txid::all_zeros(), 0)], &[100_000]);
+        let parent_txid = parent.compute_txid();
+        // The tx under test, spending the parent's output.
+        let tx = make_tx(&[OutPoint::new(parent_txid, 0)], &[90_000]);
+        let txid = tx.compute_txid();
+
+        let mut tx_cache = HashMap::new();
+        tx_cache.insert(parent_txid, parent);
+        tx_cache.insert(txid, tx);
+
+        let mut blocks_needed = LinkedHashSet::new();
+        let mut rejected_txs = HashSet::new();
+        let request_queue = RequestQueue::default();
+        let mut txs_needed = LinkedHashSet::new();
+        let mut unavailable_txs = HashSet::new();
+
+        // First insertion adds the tx to the mempool.
+        {
+            let sync_state = SyncStateBorrowedMut {
+                blocks_needed: &mut blocks_needed,
+                rejected_txs: &mut rejected_txs,
+                request_queue: &request_queue,
+                tx_cache: &mut tx_cache,
+                txs_needed: &mut txs_needed,
+                unavailable_txs: &mut unavailable_txs,
+            };
+            let added = try_add_tx_from_caches::<_, DefaultEnforcer>(
+                &mut inner, sync_state, txid,
+            )
+            .expect("first add should succeed");
+            assert!(added, "tx should be added to the mempool");
+        }
+        assert!(
+            inner.mempool.txs.0.contains_key(&txid),
+            "tx should be present in the mempool after the first add"
+        );
+
+        // Re-applying the same txid must be an idempotent no-op success, not a
+        // fatal `TxAlreadyExists`.
+        {
+            let sync_state = SyncStateBorrowedMut {
+                blocks_needed: &mut blocks_needed,
+                rejected_txs: &mut rejected_txs,
+                request_queue: &request_queue,
+                tx_cache: &mut tx_cache,
+                txs_needed: &mut txs_needed,
+                unavailable_txs: &mut unavailable_txs,
+            };
+            let added = try_add_tx_from_caches::<_, DefaultEnforcer>(
+                &mut inner, sync_state, txid,
+            )
+            .expect(
+                "re-adding an already-present tx must not error \
+                 (idempotent reconciliation)",
+            );
+            assert!(added, "re-add should report success");
+        }
+    }
+}