Don't store mempool txs before all prevouts are available

The indexing process was adding transactions into the store so that
prevouts funded & spent within the same batch could be looked up via
Mempool::lookup_txos(). If the indexing process later failed for any
reason, these transactions would remain in the store.

With this change, we instead explicitly look for prevouts funded within
the same batch, then look for the rest in the chain/mempool indexes and
fail if any are missing, without keeping the transactions in the store.

Author: shesek

src/new_index/mempool.rs

@@ -21,7 +21,7 @@ use crate::new_index::{
     SpendingInfo, SpendingInput, TxHistoryInfo, Utxo,
 };
 use crate::util::fees::{make_fee_histogram, TxFeeInfo};
-use crate::util::{extract_tx_prevouts, full_hash, has_prevout, is_spendable, Bytes};
+use crate::util::{extract_tx_prevouts, full_hash, get_prev_outpoints, is_spendable, Bytes};
 
 #[cfg(feature = "liquid")]
 use crate::elements::asset;
@@ -288,40 +288,57 @@ impl Mempool {
         self.backlog_stats = (BacklogStats::new(&self.feeinfo), Instant::now());
     }
 
-    pub fn add_by_txid(&mut self, daemon: &Daemon, txid: &Txid) {
+    pub fn add_by_txid(&mut self, daemon: &Daemon, txid: &Txid) -> Result<()> {
         if self.txstore.get(txid).is_none() {
             if let Ok(tx) = daemon.getmempooltx(&txid) {
                 self.add(vec![tx])
+            } else {
+                bail!("add_by_txid cannot find {}", txid);
             }
+        } else {
+            Ok(())
         }
     }
 
-    fn add(&mut self, txs: Vec<Transaction>) {
+    fn add(&mut self, txs: Vec<Transaction>) -> Result<()> {
         self.delta
             .with_label_values(&["add"])
             .observe(txs.len() as f64);
         let _timer = self.latency.with_label_values(&["add"]).start_timer();
 
-        let mut txids = vec![];
-        // Phase 1: add to txstore
-        for tx in txs {
-            let txid = tx.txid();
-            txids.push(txid);
+        let spent_prevouts = get_prev_outpoints(&txs);
+        let txs_map = txs
+            .into_iter()
+            .map(|tx| (tx.txid(), tx))
+            .collect::<HashMap<_, _>>();
+
+        // Lookup spent prevouts that were funded within the same `add` batch
+        let mut txos = HashMap::new();
+        let remain_prevouts = spent_prevouts
+            .into_iter()
+            .filter(|prevout| {
+                if let Some(prevtx) = txs_map.get(&prevout.txid) {
+                    if let Some(out) = prevtx.output.get(prevout.vout as usize) {
+                        txos.insert(prevout.clone(), out.clone());
+                        // remove from the list of remaining `prevouts`
+                        return false;
+                    }
+                }
+                true
+            })
+            .collect();
+
+        // Lookup remaining spent prevouts in mempool & on-chain
+        // Fails if any are missing.
+        txos.extend(self.lookup_txos(remain_prevouts)?);
+
+        // Add to txstore and indexes
+        for (txid, tx) in txs_map {
             self.txstore.insert(txid, tx);
-        }
-        // Phase 2: index history and spend edges (can fail if some txos cannot be found)
-        let txos = match self.lookup_txos(self.get_prevouts(&txids)) {
-            Ok(txos) => txos,
-            Err(err) => {
-                warn!("lookup txouts failed: {}", err);
-                // TODO: should we remove txids from txstore?
-                return;
-            }
-        };
-        for txid in txids {
-            let tx = self.txstore.get(&txid).expect("missing mempool tx");
-            let txid_bytes = full_hash(&txid[..]);
+            let tx = self.txstore.get(&txid).expect("was just added");
+
             let prevouts = extract_tx_prevouts(&tx, &txos, false);
+            let txid_bytes = full_hash(&txid[..]);
 
             // Get feeinfo for caching and recent tx overview
             let feeinfo = TxFeeInfo::new(&tx, &prevouts, self.config.network_type);
@@ -395,6 +412,8 @@ impl Mempool {
                 &mut self.asset_issuance,
             );
         }
+
+        Ok(())
     }
 
     fn lookup_txo(&self, outpoint: &OutPoint) -> Option<TxOut> {
@@ -423,24 +442,6 @@ impl Mempool {
         Ok(txos)
     }
 
-    fn get_prevouts(&self, txids: &[Txid]) -> BTreeSet<OutPoint> {
-        let _timer = self
-            .latency
-            .with_label_values(&["get_prevouts"])
-            .start_timer();
-
-        txids
-            .iter()
-            .map(|txid| self.txstore.get(txid).expect("missing mempool tx"))
-            .flat_map(|tx| {
-                tx.input
-                    .iter()
-                    .filter(|txin| has_prevout(txin))
-                    .map(|txin| txin.previous_output)
-            })
-            .collect()
-    }
-
     fn remove(&mut self, to_remove: HashSet<&Txid>) {
         self.delta
             .with_label_values(&["remove"])
@@ -510,7 +511,7 @@ impl Mempool {
         {
             let mut mempool = mempool.write().unwrap();
             // Add new transactions
-            mempool.add(txs_to_add);
+            mempool.add(txs_to_add)?;
 
             mempool
                 .count

src/new_index/query.rs

@@ -71,7 +71,8 @@ impl Query {
 
     pub fn broadcast_raw(&self, txhex: &str) -> Result<Txid> {
         let txid = self.daemon.broadcast_raw(txhex)?;
-        self.mempool
+        let _ = self
+            .mempool
             .write()
             .unwrap()
             .add_by_txid(&self.daemon, &txid);

src/util/mod.rs

@@ -12,8 +12,8 @@ pub use self::block::{
 pub use self::fees::get_tx_fee;
 pub use self::script::{get_innerscripts, ScriptToAddr, ScriptToAsm};
 pub use self::transaction::{
-    extract_tx_prevouts, has_prevout, is_coinbase, is_spendable, serialize_outpoint,
-    TransactionStatus, TxInput,
+    extract_tx_prevouts, get_prev_outpoints, has_prevout, is_coinbase, is_spendable,
+    serialize_outpoint, TransactionStatus, TxInput,
 };
 
 use std::collections::HashMap;

src/util/transaction.rs

@@ -1,7 +1,7 @@
 use crate::chain::{BlockHash, OutPoint, Transaction, TxIn, TxOut, Txid};
 use crate::util::BlockId;
 
-use std::collections::HashMap;
+use std::collections::{BTreeSet, HashMap};
 
 #[cfg(feature = "liquid")]
 lazy_static! {
@@ -96,6 +96,17 @@ pub fn extract_tx_prevouts<'a>(
         .collect()
 }
 
+pub fn get_prev_outpoints(txs: &[Transaction]) -> BTreeSet<OutPoint> {
+    txs.iter()
+        .flat_map(|tx| {
+            tx.input
+                .iter()
+                .filter(|txin| has_prevout(txin))
+                .map(|txin| txin.previous_output)
+        })
+        .collect()
+}
+
 pub fn serialize_outpoint<S>(outpoint: &OutPoint, serializer: S) -> Result<S::Ok, S::Error>
 where
     S: serde::ser::Serializer,