From 4356819e33c057f2ec6dbf9b94bae215d77b5807 Mon Sep 17 00:00:00 2001 From: Edward Houston Date: Thu, 27 Aug 2026 15:40:14 +0200 Subject: [PATCH 1/2] fix(mempool): reconcile conflicts before insertion --- src/new_index/mempool.rs | 65 +++++++++++++++++++++++++++++++++++++ tests/rest.rs | 70 ++++++++++++++++++++++++---------------- 2 files changed, 107 insertions(+), 28 deletions(-) diff --git a/src/new_index/mempool.rs b/src/new_index/mempool.rs index ddb91fe9c..49a16906b 100644 --- a/src/new_index/mempool.rs +++ b/src/new_index/mempool.rs @@ -373,6 +373,19 @@ impl Mempool { // Fails if any are missing. txos.extend(self.lookup_txos(remain_prevouts)?); + // Transactions submitted through broadcast_raw()/submit_package() are inserted into the + // local view immediately, before the next periodic sync has a chance to remove transactions + // they replaced in bitcoind. Reconcile those conflicts here so the indexes never contain two + // spenders for one outpoint. Descendants of a replaced transaction are no longer valid either. + 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()); + } + // Add to txstore and indexes for (txid, tx) in txs_map { self.txstore.insert(txid, tx); @@ -464,6 +477,58 @@ impl Mempool { Ok(()) } + fn conflicts_and_descendants( + &self, + txs_map: &HashMap, + ) -> Result> { + 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::>(); + 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 { self.txstore .get(&outpoint.txid) diff --git a/tests/rest.rs b/tests/rest.rs index 91100cf9c..55dd3d8de 100644 --- a/tests/rest.rs +++ b/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::(&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. From 7e60e9a50c3a9aa949b19653c4f65326ee6e8c67 Mon Sep 17 00:00:00 2001 From: Edward Houston Date: Fri, 28 Aug 2026 15:33:05 +0200 Subject: [PATCH 2/2] refactor(mempool): reconcile only submitted transactions --- src/new_index/mempool.rs | 33 +++++++++++++++++++-------------- 1 file changed, 19 insertions(+), 14 deletions(-) diff --git a/src/new_index/mempool.rs b/src/new_index/mempool.rs index 49a16906b..570960bba 100644 --- a/src/new_index/mempool.rs +++ b/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) -> 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) } @@ -373,19 +391,6 @@ impl Mempool { // Fails if any are missing. txos.extend(self.lookup_txos(remain_prevouts)?); - // Transactions submitted through broadcast_raw()/submit_package() are inserted into the - // local view immediately, before the next periodic sync has a chance to remove transactions - // they replaced in bitcoind. Reconcile those conflicts here so the indexes never contain two - // spenders for one outpoint. Descendants of a replaced transaction are no longer valid either. - 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()); - } - // Add to txstore and indexes for (txid, tx) in txs_map { self.txstore.insert(txid, tx);