Merge #308: Handle unconfirmed deposits that were replaced

3a573b695e07d357e844dd9d573d578e05ecebed qa: test we discard RBF'd deposits (Antoine Poinsot)
f1532f8afcceb5ded299f5ddc425c0473de49871 bitcoin: track expired unconfirmed deposits, remove them from DB (Antoine Poinsot)
ed156543c95db2e6e1fd4712ec7dcb0dfc025c41 database: permit to remove coins from DB (Antoine Poinsot)
62c4b9a01c16c60f66db99abab967efa4e0429dd bitcoind: cleanup gettx cache and conflict detection in spent_coins (Antoine Poinsot)
201c9f21b60dbc5ff0afd9aa73c3a011c08e4459 bitcoin: cache calls to 'gettransaction' when checking for coins confirmation (Antoine Poinsot)
13214c887f373fbe953c01aa8d3a31ae9b65ee98 qa: test we discard RBF'd deposits when replacement is confirmed (Antoine Poinsot)

Pull request description:

  Previously we would not detect whether the transaction for an unconfirmed deposit was still in our mempool. For instance a deposit that was RBF'd would result in us storing two coins, the replaced and the replacement. Keeping the unconfirmed replaced coin forever.

  Fix this by dropping unconfirmed coins from our database if their transaction isn't in our mempool (anymore).

  Fixes #72.

ACKs for top commit:
  edouardparis:
    utACK 3a573b695e07d357e844dd9d573d578e05ecebed

Tree-SHA512: 677bedbf8024eb16e5f97264c2e01d733732673548772fc3a97868b9ca42a0429ad8c0d888b05d3bb2db9b85d309c418b916c9af578f54e499b5d0689376a1b0

src/bitcoin/d/mod.rs

@@ -10,7 +10,13 @@ use crate::{
 use utils::{block_before_date, roundup_progress};
 
 use std::{
-    cmp, collections::HashSet, convert::TryInto, fs, io, str::FromStr, thread, time::Duration,
+    cmp,
+    collections::{HashMap, HashSet},
+    convert::TryInto,
+    fs, io,
+    str::FromStr,
+    thread,
+    time::Duration,
 };
 
 use jsonrpc::{
@@ -965,6 +971,22 @@ impl BitcoinD {
             |h| self.get_block_stats(h),
         )
     }
+
+    /// Whether this transaction is in the mempool.
+    pub fn is_in_mempool(&self, txid: &bitcoin::Txid) -> bool {
+        match self
+            .make_fallible_node_request("getmempoolentry", &params!(Json::String(txid.to_string())))
+        {
+            Ok(_) => true,
+            Err(BitcoindError::Server(jsonrpc::Error::Rpc(jsonrpc::error::RpcError {
+                code: -5,
+                ..
+            }))) => false,
+            Err(e) => {
+                panic!("Unexpected error returned by bitcoind {}", e);
+            }
+        }
+    }
 }
 
 /// An entry in the 'listdescriptors' result.
@@ -1127,3 +1149,33 @@ pub struct BlockStats {
     pub time: u32,
     pub median_time_past: u32,
 }
+
+/// Make cached calls to bitcoind's `gettransaction`. It's useful for instance when coins have been
+/// created or spent in a single transaction.
+pub struct CachedTxGetter<'a> {
+    bitcoind: &'a BitcoinD,
+    cache: HashMap<bitcoin::Txid, GetTxRes>,
+}
+
+impl<'a> CachedTxGetter<'a> {
+    pub fn new(bitcoind: &'a BitcoinD) -> Self {
+        Self {
+            bitcoind,
+            cache: HashMap::new(),
+        }
+    }
+
+    /// Query a transaction. Tries to get it from the cache and falls back to calling
+    /// `gettransaction` on bitcoind. If both fail, returns None.
+    pub fn get_transaction(&mut self, txid: &bitcoin::Txid) -> Option<GetTxRes> {
+        // TODO: work around the borrow checker to avoid having to clone.
+        if let Some(res) = self.cache.get(txid) {
+            Some(res.clone())
+        } else if let Some(res) = self.bitcoind.get_transaction(txid) {
+            self.cache.insert(*txid, res);
+            self.cache.get(txid).cloned()
+        } else {
+            None
+        }
+    }
+}

src/bitcoin/mod.rs

@@ -5,11 +5,11 @@ pub mod d;
 pub mod poller;
 
 use crate::{
-    bitcoin::d::{BitcoindError, LSBlockEntry},
+    bitcoin::d::{BitcoindError, CachedTxGetter, LSBlockEntry},
     descriptors,
 };
 
-use std::{collections::HashMap, fmt, sync};
+use std::{fmt, sync};
 
 use miniscript::bitcoin;
 
@@ -58,11 +58,12 @@ pub trait BitcoinInterface: Send {
         descs: &[descriptors::InheritanceDescriptor],
     ) -> Vec<UTxO>;
 
-    /// Get all coins that were confirmed, and at what height and time.
+    /// Get all coins that were confirmed, and at what height and time. Along with "expired"
+    /// unconfirmed coins (for instance whose creating transaction may have been replaced).
     fn confirmed_coins(
         &self,
         outpoints: &[bitcoin::OutPoint],
-    ) -> Vec<(bitcoin::OutPoint, i32, u32)>;
+    ) -> (Vec<(bitcoin::OutPoint, i32, u32)>, Vec<bitcoin::OutPoint>);
 
     /// Get all coins that are being spent, and the spending txid.
     fn spending_coins(
@@ -166,21 +167,34 @@ impl BitcoinInterface for d::BitcoinD {
     fn confirmed_coins(
         &self,
         outpoints: &[bitcoin::OutPoint],
-    ) -> Vec<(bitcoin::OutPoint, i32, u32)> {
+    ) -> (Vec<(bitcoin::OutPoint, i32, u32)>, Vec<bitcoin::OutPoint>) {
+        // The confirmed and expired coins to be returned.
         let mut confirmed = Vec::with_capacity(outpoints.len());
+        let mut expired = Vec::new();
+        // Cached calls to `gettransaction`.
+        let mut tx_getter = CachedTxGetter::new(self);
 
         for op in outpoints {
-            // TODO: batch those calls to gettransaction
-            if let Some(res) = self.get_transaction(&op.txid) {
-                if let Some(block) = res.block {
-                    confirmed.push((*op, block.height, block.time));
-                }
+            let res = if let Some(res) = tx_getter.get_transaction(&op.txid) {
+                res
             } else {
                 log::error!("Transaction not in wallet for coin '{}'.", op);
+                continue;
+            };
+
+            // If the transaction was confirmed, mark the coin as such.
+            if let Some(block) = res.block {
+                confirmed.push((*op, block.height, block.time));
+                continue;
+            }
+
+            // If the transaction was dropped from the mempool, discard the coin.
+            if !self.is_in_mempool(&op.txid) {
+                expired.push(*op);
             }
         }
 
-        confirmed
+        (confirmed, expired)
     }
 
     fn spending_coins(
@@ -213,47 +227,33 @@ impl BitcoinInterface for d::BitcoinD {
         &self,
         outpoints: &[(bitcoin::OutPoint, bitcoin::Txid)],
     ) -> Vec<(bitcoin::OutPoint, bitcoin::Txid, Block)> {
+        // Spend coins to be returned.
         let mut spent = Vec::with_capacity(outpoints.len());
+        // Cached calls to `gettransaction`.
+        let mut tx_getter = CachedTxGetter::new(self);
 
-        let mut cache: HashMap<bitcoin::Txid, Option<d::GetTxRes>> = HashMap::new();
         for (op, txid) in outpoints {
-            let tx: Option<&d::GetTxRes> = match cache.get(txid) {
-                Some(tx) => tx.as_ref(),
-                None => {
-                    let tx = self.get_transaction(txid);
-                    cache.insert(*txid, tx);
-                    cache.get(txid).unwrap().as_ref()
-                }
+            let res = if let Some(res) = tx_getter.get_transaction(txid) {
+                res
+            } else {
+                log::error!("Could not get tx {} spending coin {}.", txid, op);
+                continue;
             };
 
-            // There is an immutable borrow on the cache, these txs will be added once it is
-            // dropped.
-            let mut txs_to_cache: Vec<(bitcoin::Txid, Option<d::GetTxRes>)> = Vec::new();
-
-            if let Some(tx) = tx {
-                if let Some(block) = tx.block {
-                    spent.push((*op, *txid, block));
-                } else if !tx.conflicting_txs.is_empty() {
-                    for txid in &tx.conflicting_txs {
-                        let tx: Option<&d::GetTxRes> = match cache.get(txid) {
-                            Some(tx) => tx.as_ref(),
-                            None => {
-                                let tx = self.get_transaction(txid);
-                                txs_to_cache.push((*txid, tx));
-                                txs_to_cache.last().unwrap().1.as_ref()
-                            }
-                        };
-                        if let Some(tx) = tx {
-                            if let Some(block) = tx.block {
-                                spent.push((*op, *txid, block))
-                            }
-                        }
-                    }
-                }
+            // If the transaction was confirmed, mark it as such.
+            if let Some(block) = res.block {
+                spent.push((*op, *txid, block));
+                continue;
             }
 
-            for (txid, res) in txs_to_cache {
-                cache.insert(txid, res);
+            // If a conflicting transaction was confirmed instead, replace the txid of the
+            // spender for this coin with it and mark it as confirmed.
+            for txid in &res.conflicting_txs {
+                if let Some(tx) = tx_getter.get_transaction(txid) {
+                    if let Some(block) = tx.block {
+                        spent.push((*op, *txid, block))
+                    }
+                }
             }
         }
 
@@ -347,7 +347,7 @@ impl BitcoinInterface for sync::Arc<sync::Mutex<dyn BitcoinInterface + 'static>>
     fn confirmed_coins(
         &self,
         outpoints: &[bitcoin::OutPoint],
-    ) -> Vec<(bitcoin::OutPoint, i32, u32)> {
+    ) -> (Vec<(bitcoin::OutPoint, i32, u32)>, Vec<bitcoin::OutPoint>) {
         self.lock().unwrap().confirmed_coins(outpoints)
     }
 

src/bitcoin/poller/looper.rs

@@ -15,6 +15,7 @@ use miniscript::bitcoin::{self, secp256k1};
 struct UpdatedCoins {
     pub received: Vec<Coin>,
     pub confirmed: Vec<(bitcoin::OutPoint, i32, u32)>,
+    pub expired: Vec<bitcoin::OutPoint>,
     pub spending: Vec<(bitcoin::OutPoint, bitcoin::Txid)>,
     pub spent: Vec<(bitcoin::OutPoint, bitcoin::Txid, i32, u32)>,
 }
@@ -91,8 +92,9 @@ fn update_coins(
             }
         })
         .collect();
-    let confirmed = bit.confirmed_coins(&to_be_confirmed);
+    let (confirmed, expired) = bit.confirmed_coins(&to_be_confirmed);
     log::debug!("Newly confirmed coins: {:?}", confirmed);
+    log::debug!("Expired coins: {:?}", expired);
 
     // We need to take the newly received ones into account as well, as they may have been
     // spent within the previous tip and the current one, and we may not poll this chunk of the
@@ -131,6 +133,7 @@ fn update_coins(
     UpdatedCoins {
         received,
         confirmed,
+        expired,
         spending,
         spent,
     }
@@ -219,6 +222,7 @@ fn updates(
     // Having the tip in database means that, as far as the chain is concerned, we've got all
     // updates up to this block. But not more.
     db_conn.new_unspent_coins(&updated_coins.received);
+    db_conn.remove_coins(&updated_coins.expired);
     db_conn.confirm_coins(&updated_coins.confirmed);
     db_conn.spend_coins(&updated_coins.spending);
     db_conn.confirm_spend(&updated_coins.spent);

src/database/mod.rs

@@ -89,6 +89,9 @@ pub trait DatabaseConnection {
     /// Store new UTxOs. Coins must not already be in database.
     fn new_unspent_coins(&mut self, coins: &[Coin]);
 
+    /// Remove some UTxOs from the database.
+    fn remove_coins(&mut self, coins: &[bitcoin::OutPoint]);
+
     /// Mark a set of coins as being confirmed at a specified height and block time.
     fn confirm_coins(&mut self, outpoints: &[(bitcoin::OutPoint, i32, u32)]);
 
@@ -196,6 +199,10 @@ impl DatabaseConnection for SqliteConn {
         self.new_unspent_coins(coins)
     }
 
+    fn remove_coins(&mut self, outpoints: &[bitcoin::OutPoint]) {
+        self.remove_coins(outpoints)
+    }
+
     fn confirm_coins<'a>(&mut self, outpoints: &[(bitcoin::OutPoint, i32, u32)]) {
         self.confirm_coins(outpoints)
     }

src/database/sqlite/mod.rs

@@ -363,6 +363,21 @@ impl SqliteConn {
         .expect("Database must be available")
     }
 
+    /// Remove a set of coins from the database.
+    pub fn remove_coins(&mut self, outpoints: &[bitcoin::OutPoint]) {
+        db_exec(&mut self.conn, |db_tx| {
+            for outpoint in outpoints {
+                db_tx.execute(
+                    "DELETE FROM coins WHERE txid = ?1 AND vout = ?2",
+                    rusqlite::params![outpoint.txid.to_vec(), outpoint.vout,],
+                )?;
+            }
+
+            Ok(())
+        })
+        .expect("Database must be available")
+    }
+
     /// Mark a set of coins as confirmed.
     pub fn confirm_coins<'a>(
         &mut self,
@@ -705,6 +720,13 @@ mod tests {
             conn.new_unspent_coins(&[coin_a]);
             assert_eq!(conn.coins(CoinType::All)[0].outpoint, coin_a.outpoint);
 
+            // We can also remove it. Say the unconfirmed tx that created it got replaced.
+            conn.remove_coins(&[coin_a.outpoint]);
+            assert!(conn.coins(CoinType::All).is_empty());
+
+            // Add it back for the rest of the test.
+            conn.new_unspent_coins(&[coin_a]);
+
             // We can query it by its outpoint
             let coins = conn.db_coins(&[coin_a.outpoint]);
             assert_eq!(coins.len(), 1);

src/testutils.rs

@@ -65,8 +65,11 @@ impl BitcoinInterface for DummyBitcoind {
         Vec::new()
     }
 
-    fn confirmed_coins(&self, _: &[bitcoin::OutPoint]) -> Vec<(bitcoin::OutPoint, i32, u32)> {
-        Vec::new()
+    fn confirmed_coins(
+        &self,
+        _: &[bitcoin::OutPoint],
+    ) -> (Vec<(bitcoin::OutPoint, i32, u32)>, Vec<bitcoin::OutPoint>) {
+        (Vec::new(), Vec::new())
     }
 
     fn spending_coins(&self, _: &[bitcoin::OutPoint]) -> Vec<(bitcoin::OutPoint, bitcoin::Txid)> {
@@ -220,6 +223,12 @@ impl DatabaseConnection for DummyDatabase {
         }
     }
 
+    fn remove_coins(&mut self, outpoints: &[bitcoin::OutPoint]) {
+        for op in outpoints {
+            self.db.write().unwrap().coins.remove(op);
+        }
+    }
+
     fn confirm_coins<'a>(&mut self, outpoints: &[(bitcoin::OutPoint, i32, u32)]) {
         for (op, height, time) in outpoints {
             let mut db = self.db.write().unwrap();

tests/test_chain.py

@@ -1,14 +1,10 @@
-import time
-
 from fixtures import *
 from test_framework.utils import wait_for, get_txid, spend_coins
 
 
 def get_coin(lianad, outpoint_or_txid):
     return next(
-        c
-        for c in lianad.rpc.listcoins()["coins"]
-        if outpoint_or_txid in c["outpoint"]
+        c for c in lianad.rpc.listcoins()["coins"] if outpoint_or_txid in c["outpoint"]
     )
 
 
@@ -80,13 +76,10 @@ def test_reorg_exclusion(lianad, bitcoind):
     bitcoind.simple_reorg(initial_height, shift=-1)
     wait_for(lambda: lianad.rpc.getinfo()["block_height"] == current_height + 1)
 
-    # They must all be marked as unconfirmed.
-    new_coin_a = get_coin(lianad, coin_a["outpoint"])
-    assert new_coin_a["block_height"] is None
-    new_coin_b = get_coin(lianad, coin_b["outpoint"])
-    assert new_coin_b["block_height"] is None
-    new_coin_c = get_coin(lianad, coin_c["outpoint"])
-    assert new_coin_c["block_height"] is None
+    # For a too deep reorg bitcoind doesn't update the mempool. The deposit transactions were
+    # dropped. And we discard the unconfirmed coins whose deposit tx isn't part of our mempool
+    # anymore: the coins must have been marked as unconfirmed and subsequently discarded.
+    wait_for(lambda: len(lianad.rpc.listcoins()["coins"]) == 0)
 
     # And if we now confirm everything, they'll be marked as such. The one that was 'spending'
     # will now be spent (its spending transaction will be confirmed) and the one that was spent
@@ -257,3 +250,50 @@ def test_rescan_edge_cases(lianad, bitcoind):
     wait_for(lambda: lianad.rpc.getinfo()["rescan_progress"] is None)
     assert len(sorted_coins()) == len(coins_before)
     assert all(c["outpoint"] in outpoints_before for c in list_coins())
+
+
+def test_deposit_replacement(lianad, bitcoind):
+    """Test we discard an unconfirmed deposit that was replaced."""
+    # Get some more coins.
+    bitcoind.generate_block(1)
+
+    # Create a new unconfirmed deposit.
+    addr = lianad.rpc.getnewaddress()["address"]
+    txid = bitcoind.rpc.sendtoaddress(addr, 1)
+
+    # Create a transaction conflicting with the deposit that pays more fee.
+    deposit_tx = bitcoind.rpc.gettransaction(txid, False, True)["decoded"]
+    bitcoind.rpc.lockunspent(
+        False,
+        [
+            {"txid": deposit_tx["txid"], "vout": i}
+            for i in range(len(deposit_tx["vout"]))
+        ],
+    )
+    res = bitcoind.rpc.walletcreatefundedpsbt(
+        [
+            {"txid": txin["txid"], "vout": txin["vout"], "sequence": 0xFF_FF_FF_FD}
+            for txin in deposit_tx["vin"]
+        ],
+        [
+            {bitcoind.rpc.getnewaddress(): txout["value"]}
+            for txout in deposit_tx["vout"]
+        ],
+        0,
+        {"fee_rate": 10, "add_inputs": True},
+    )
+    res = bitcoind.rpc.walletprocesspsbt(res["psbt"])
+    assert res["complete"]
+    conflicting_tx = bitcoind.rpc.finalizepsbt(res["psbt"])["hex"]
+
+    # Make sure we registered the unconfirmed coin. Then RBF the deposit tx.
+    wait_for(lambda: len(lianad.rpc.listcoins()["coins"]) == 1)
+    txid = bitcoind.rpc.sendrawtransaction(conflicting_tx)
+
+    # We must forget about the deposit.
+    wait_for(lambda: len(lianad.rpc.listcoins()["coins"]) == 0)
+
+    # Send a new one, it'll be detected.
+    addr = lianad.rpc.getnewaddress()["address"]
+    bitcoind.rpc.sendtoaddress(addr, 2)
+    wait_for(lambda: len(lianad.rpc.listcoins()["coins"]) == 1)