fix(mempool): reconcile conflicts before insertion

src/new_index/mempool.rs

@@ -313,7 +313,7 @@ impl Mempool {
             if let Ok(tx) = daemon.getmempooltx(&txid) {
                 let mut txs_map = HashMap::new();
                 txs_map.insert(txid, tx);
-                self.add(txs_map)
+                self.add_submitted(txs_map)
             } else {
                 bail!("add_by_txid cannot find {}", txid);
             }
@@ -341,6 +341,24 @@ impl Mempool {
         if txs_map.is_empty() {
             return Ok(());
         }
+        self.add_submitted(txs_map)
+    }
+
+    /// Add transactions submitted through broadcast_raw()/submit_package().
+    ///
+    /// Unlike the periodic full-snapshot update, manual insertion can race with
+    /// bitcoind removing transactions that the submission replaced. Reconcile
+    /// those conflicts before inserting the submitted transactions locally.
+    fn add_submitted(&mut self, txs_map: HashMap<Txid, Transaction>) -> Result<()> {
+        let conflicts = self.conflicts_and_descendants(&txs_map)?;
+        if !conflicts.is_empty() {
+            debug!(
+                "removing {} conflicting mempool transactions before insertion",
+                conflicts.len()
+            );
+            self.remove(conflicts.iter().collect());
+        }
+
         self.add(txs_map)
     }
 
@@ -464,6 +482,58 @@ impl Mempool {
         Ok(())
     }
 
+    fn conflicts_and_descendants(
+        &self,
+        txs_map: &HashMap<Txid, Transaction>,
+    ) -> Result<HashSet<Txid>> {
+        let mut incoming_spends = HashMap::new();
+        let mut to_remove = HashSet::new();
+
+        for (txid, tx) in txs_map {
+            for txin in &tx.input {
+                if let Some(other_txid) = incoming_spends.insert(txin.previous_output, *txid) {
+                    if other_txid != *txid {
+                        bail!(
+                            "incoming mempool transactions {} and {} both spend outpoint {}:{}",
+                            other_txid,
+                            txid,
+                            txin.previous_output.txid,
+                            txin.previous_output.vout
+                        );
+                    }
+                }
+
+                if let Some((indexed_txid, _)) = self.edges.get(&txin.previous_output) {
+                    if indexed_txid != txid && !txs_map.contains_key(indexed_txid) {
+                        to_remove.insert(*indexed_txid);
+                    }
+                }
+            }
+        }
+
+        // Follow spend edges from every output of each conflict to collect the full descendant
+        // closure without scanning the whole mempool.
+        let mut pending = to_remove.iter().copied().collect::<Vec<_>>();
+        while let Some(txid) = pending.pop() {
+            let Some(tx) = self.txstore.get(&txid) else {
+                continue;
+            };
+            for vout in 0..tx.output.len() {
+                let outpoint = OutPoint {
+                    txid,
+                    vout: vout as u32,
+                };
+                if let Some((descendant_txid, _)) = self.edges.get(&outpoint) {
+                    if to_remove.insert(*descendant_txid) {
+                        pending.push(*descendant_txid);
+                    }
+                }
+            }
+        }
+
+        Ok(to_remove)
+    }
+
     fn lookup_txo(&self, outpoint: &OutPoint) -> Option<TxOut> {
         self.txstore
             .get(&outpoint.txid)

tests/rest.rs

@@ -1576,13 +1576,22 @@ fn test_rest_liquid_block() -> Result<()> {
 
 #[cfg(not(feature = "liquid"))]
 #[test]
-fn test_rest_mempool_rbf_eviction() -> Result<()> {
-    // Regression test for the mempool eviction panic ("missing mempool edge
-    // for outpoint"): tx A and its RBF replacement B spend the same outpoint
-    // and transiently coexist in the local mempool view when B is injected
-    // through the broadcast endpoint (add_by_txid) while A is still indexed.
-    // B's add() clobbers A's `edges` entry; the next sync round evicts A and
-    // must tolerate the missing/foreign edge instead of panicking.
+fn test_rest_mempool_rbf_reconciled_on_broadcast() -> Result<()> {
+    test_rest_mempool_rbf_reconciliation(false)
+}
+
+#[cfg(not(feature = "liquid"))]
+#[test]
+fn test_rest_mempool_rbf_reconciled_on_package_broadcast() -> Result<()> {
+    test_rest_mempool_rbf_reconciliation(true)
+}
+
+#[cfg(not(feature = "liquid"))]
+fn test_rest_mempool_rbf_reconciliation(use_package: bool) -> Result<()> {
+    // Tx A and its RBF replacement B spend the same outpoint. B is inserted
+    // immediately through add_by_txid(s), before the periodic sync can evict A.
+    // The insertion must reconcile the conflict instead of exposing both
+    // transactions in the local indexes.
     let (rest_handle, rest_addr, mut tester) = common::init_rest_tester().unwrap();
 
     // Broadcast tx A via the node wallet, explicitly BIP125-replaceable,
@@ -1616,22 +1625,28 @@ fn test_rest_mempool_rbf_eviction() -> Result<()> {
         .node_client()
         .call("finalizepsbt", &[processed["psbt"].clone()])?;
     let b_hex = finalized["hex"].as_str().expect("finalized tx hex");
+    let b_bytes = Vec::from_hex(b_hex).expect("valid finalized tx hex");
+    let txid_b = bitcoin::consensus::deserialize::<bitcoin::Transaction>(&b_bytes)
+        .expect("valid finalized transaction")
+        .compute_txid()
+        .to_string();
 
-    // Inject B through the electrs broadcast endpoint: the node accepts the
-    // replacement (evicting A node-side), and add_by_txid() indexes B locally
-    // while A is still present - clobbering A's edges entry for the shared
-    // outpoint.
-    let broadcast_resp = ureq::post(&format!("http://{}/tx", rest_addr)).send(b_hex)?;
-    assert_eq!(broadcast_resp.status(), 200);
-    let txid_b = broadcast_resp.into_body().read_to_string()?;
-
-    // The next sync evicts A from the local view. The unfixed code passes the
-    // eviction assert here - but only by STEALING B's edge entry (any Some()
-    // satisfied it), which is the actual arming step of the crash.
-    tester.sync()?;
+    // Inject B through either immediate insertion path. bitcoind accepts the
+    // replacement and evicts A node-side; electrs must do the same locally as
+    // part of this request, without waiting for tester.sync().
+    if use_package {
+        let response = ureq::post(&format!("http://{}/txs/package", rest_addr))
+            .send_json([b_hex])?;
+        assert_eq!(response.status(), 200);
+    } else {
+        let response = ureq::post(&format!("http://{}/tx", rest_addr)).send(b_hex)?;
+        assert_eq!(response.status(), 200);
+        assert_eq!(response.into_body().read_to_string()?.trim(), txid_b);
+    }
 
-    // B remains queryable in the mempool; A is gone.
-    let res = get_json(rest_addr, &format!("/tx/{}", txid_b.trim()))?;
+    // B is immediately queryable and A is immediately absent. Before #236,
+    // this assertion observed both conflicting transactions until the poll.
+    let res = get_json(rest_addr, &format!("/tx/{}", txid_b))?;
     assert_eq!(res["status"]["confirmed"].as_bool(), Some(false));
     let gone = ureq::get(&format!("http://{}/tx/{}", rest_addr, txid_a))
         .config()
@@ -1640,16 +1655,15 @@ fn test_rest_mempool_rbf_eviction() -> Result<()> {
         .call()?;
     assert_eq!(gone.status(), 404);
 
-    // Now B itself leaves the mempool (confirmed here; RBF-of-B or expiry are
-    // equivalent). Evicting B finds its edge entry gone - stolen by A's
-    // eviction above - and the unfixed code panics with "missing mempool edge
-    // for outpoint", killing the sync loop. The fixed code only removes an
-    // edge its evicted tx still owns, so B's edge survived A's eviction and
-    // this round stays clean.
+    // A periodic sync should preserve the already-reconciled state.
+    tester.sync()?;
+
+    // B can subsequently leave the mempool without missing-edge warnings or
+    // a panic, and the server remains fully alive.
     tester.mine()?;
     tester.sync()?;
 
-    let res = get_json(rest_addr, &format!("/tx/{}", txid_b.trim()))?;
+    let res = get_json(rest_addr, &format!("/tx/{}", txid_b))?;
     assert_eq!(res["status"]["confirmed"].as_bool(), Some(true));
 
     // And the server is still fully alive.