[bitcoind_rpc] Introduce prune_and_apply_update for IndexedTxGraph

`prune_and_apply_update` first scans all txs contained in `update`
through the index, then filters out txs using `I::is_tx_relevant` before
applying the update. This is useful for block-by-block syncing.

`Wallet::apply_update` now has a second input; `prune: bool`. If `prune`
is set, irrelevant transactions of `update` will not be included.

Author: evanlinjin

crates/bdk/src/wallet/mod.rs

@@ -1694,14 +1694,16 @@ impl<D> Wallet<D> {
     }
 
     /// Applies an update to the wallet and stages the changes (but does not [`commit`] them).
+    /// Returns whether the `update` resulted in any changes.
     ///
-    /// This returns whether the `update` resulted in any changes.
+    /// If `prune` is set, irrelevant transactions are pruned. Relevant transactions change the UTXO
+    /// set of tracked script pubkeys (script pubkeys derived from tracked descriptors).
     ///
     /// Usually you create an `update` by interacting with some blockchain data source and inserting
     /// transactions related to your wallet into it.
     ///
     /// [`commit`]: Self::commit
-    pub fn apply_update(&mut self, update: Update) -> Result<bool, CannotConnectError>
+    pub fn apply_update(&mut self, update: Update, prune: bool) -> Result<bool, CannotConnectError>
     where
         D: PersistBackend<ChangeSet>,
     {
@@ -1711,7 +1713,14 @@ impl<D> Wallet<D> {
             .index
             .reveal_to_target_multi(&update.keychain);
         changeset.append(ChangeSet::from(IndexedAdditions::from(index_additions)));
-        changeset.append(self.indexed_graph.apply_update(update.graph).into());
+        changeset.append(
+            if prune {
+                self.indexed_graph.prune_and_apply_update(update.graph)
+            } else {
+                self.indexed_graph.apply_update(update.graph)
+            }
+            .into(),
+        );
 
         let changed = !changeset.is_empty();
         self.persist.stage(changeset);

crates/bitcoind_rpc/src/lib.rs

@@ -1,9 +1,10 @@
 use std::collections::HashSet;
 
 use bdk_chain::{
-    bitcoin::{Transaction, Txid},
+    bitcoin::{Block, Transaction, Txid},
+    keychain::LocalUpdate,
     local_chain::CheckPoint,
-    BlockId,
+    BlockId, ConfirmationHeightAnchor, ConfirmationTimeAnchor, TxGraph,
 };
 use bitcoincore_rpc::{bitcoincore_rpc_json::GetBlockResult, Client, RpcApi};
 
@@ -12,13 +13,78 @@ pub enum BitcoindRpcItem {
     Block {
         cp: CheckPoint,
         info: Box<GetBlockResult>,
+        block: Box<Block>,
     },
     Mempool {
         cp: CheckPoint,
         txs: Vec<(Transaction, u64)>,
     },
 }
 
+pub fn confirmation_height_anchor(
+    info: &GetBlockResult,
+    _txid: Txid,
+    _tx_pos: usize,
+) -> ConfirmationHeightAnchor {
+    ConfirmationHeightAnchor {
+        anchor_block: BlockId {
+            height: info.height as _,
+            hash: info.hash,
+        },
+        confirmation_height: info.height as _,
+    }
+}
+
+pub fn confirmation_time_anchor(
+    info: &GetBlockResult,
+    _txid: Txid,
+    _tx_pos: usize,
+) -> ConfirmationTimeAnchor {
+    ConfirmationTimeAnchor {
+        anchor_block: BlockId {
+            height: info.height as _,
+            hash: info.hash,
+        },
+        confirmation_height: info.height as _,
+        confirmation_time: info.time as _,
+    }
+}
+
+impl BitcoindRpcItem {
+    pub fn into_update<K, A, F>(self, anchor: F) -> LocalUpdate<K, A>
+    where
+        A: Clone + Ord + PartialOrd,
+        F: Fn(&GetBlockResult, Txid, usize) -> A,
+    {
+        match self {
+            BitcoindRpcItem::Block { cp, info, block } => LocalUpdate {
+                graph: {
+                    let mut g = TxGraph::<A>::new(block.txdata);
+                    for (tx_pos, &txid) in info.tx.iter().enumerate() {
+                        let _ = g.insert_anchor(txid, anchor(&info, txid, tx_pos));
+                    }
+                    g
+                },
+                ..LocalUpdate::new(cp)
+            },
+            BitcoindRpcItem::Mempool { cp, txs } => LocalUpdate {
+                graph: {
+                    let mut last_seens = Vec::<(Txid, u64)>::with_capacity(txs.len());
+                    let mut g = TxGraph::<A>::new(txs.into_iter().map(|(tx, last_seen)| {
+                        last_seens.push((tx.txid(), last_seen));
+                        tx
+                    }));
+                    for (txid, seen_at) in last_seens {
+                        let _ = g.insert_seen_at(txid, seen_at);
+                    }
+                    g
+                },
+                ..LocalUpdate::new(cp)
+            },
+        }
+    }
+}
+
 pub struct BitcoindRpcIter<'a> {
     client: &'a Client,
     fallback_height: u32,
@@ -57,6 +123,7 @@ impl<'a> BitcoindRpcIter<'a> {
                     // get first item at fallback_height
                     let info = client
                         .get_block_info(&client.get_block_hash(self.fallback_height as _)?)?;
+                    let block = self.client.get_block(&info.hash)?;
                     let cp = CheckPoint::new(BlockId {
                         height: info.height as _,
                         hash: info.hash,
@@ -66,6 +133,7 @@ impl<'a> BitcoindRpcIter<'a> {
                     return Ok(Some(BitcoindRpcItem::Block {
                         cp,
                         info: Box::new(info),
+                        block: Box::new(block),
                     }));
                 }
                 (last_cp @ Some(_), last_info @ None) => {
@@ -98,6 +166,7 @@ impl<'a> BitcoindRpcIter<'a> {
                                 continue 'main_loop;
                             }
 
+                            let block = self.client.get_block(&info.hash)?;
                             let cp = CheckPoint::new_with_prev(
                                 BlockId {
                                     height: info.height as _,
@@ -113,6 +182,7 @@ impl<'a> BitcoindRpcIter<'a> {
                             return Ok(Some(BitcoindRpcItem::Block {
                                 cp,
                                 info: Box::new(info),
+                                block: Box::new(block),
                             }));
                         }
                         None => {

crates/chain/src/indexed_tx_graph.rs

@@ -90,6 +90,38 @@ where
         }
     }
 
+    /// Apply `update`, but filters out irrelevant transactions.
+    ///
+    /// Relevancy is determined by the [`Indexer::is_tx_relevant`] implementation of `I`.
+    pub fn prune_and_apply_update(
+        &mut self,
+        update: TxGraph<A>,
+    ) -> IndexedAdditions<A, I::Additions> {
+        let mut additions = IndexedAdditions::<A, I::Additions>::default();
+
+        // index all transactions first
+        for tx_node in update.full_txs() {
+            additions
+                .index_additions
+                .append(self.index.index_tx(&tx_node));
+        }
+
+        let update = update
+            .full_txs()
+            .filter(|tx_node| self.index.is_tx_relevant(tx_node))
+            .fold(TxGraph::default(), |mut g, tx_node| -> TxGraph<A> {
+                let _ = g.insert_tx(tx_node.tx.clone());
+                for anchor in tx_node.anchors {
+                    let _ = g.insert_anchor(tx_node.txid, anchor.clone());
+                }
+                let _ = g.insert_seen_at(tx_node.txid, tx_node.last_seen_unconfirmed);
+                g
+            });
+
+        additions.append(self.apply_update(update));
+        additions
+    }
+
     /// Insert a floating `txout` of given `outpoint`.
     pub fn insert_txout(
         &mut self,
@@ -146,14 +178,12 @@ where
         // 2. decide whether to insert them into the graph depending on whether `is_tx_relevant`
         //    returns true or not. (in a second loop).
         let mut additions = IndexedAdditions::<A, I::Additions>::default();
-        let mut transactions = Vec::new();
-        for (tx, anchors) in txs.into_iter() {
-            additions.index_additions.append(self.index.index_tx(tx));
-            transactions.push((tx, anchors));
-        }
+        let txs = txs
+            .into_iter()
+            .inspect(|(tx, _)| additions.index_additions.append(self.index.index_tx(tx)))
+            .collect::<Vec<_>>();
         additions.append(
-            transactions
-                .into_iter()
+            txs.into_iter()
                 .filter_map(|(tx, anchors)| match self.index.is_tx_relevant(tx) {
                     true => Some(self.insert_tx(tx, anchors, seen_at)),
                     false => None,

example-crates/wallet_electrum/src/main.rs

@@ -59,7 +59,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let missing = electrum_update.missing_full_txs(wallet.as_ref());
     let update = electrum_update.finalize_as_confirmation_time(&client, None, missing)?;
 
-    wallet.apply_update(update)?;
+    wallet.apply_update(update, false)?;
     wallet.commit()?;
 
     let balance = wallet.get_balance();

example-crates/wallet_esplora/src/main.rs

@@ -61,7 +61,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         PARALLEL_REQUESTS,
     )?;
     println!();
-    wallet.apply_update(update)?;
+    wallet.apply_update(update, false)?;
     wallet.commit()?;
 
     let balance = wallet.get_balance();

example-crates/wallet_esplora_async/src/main.rs

@@ -57,7 +57,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .scan(prev_cp, keychain_spks, [], [], STOP_GAP, PARALLEL_REQUESTS)
         .await?;
     println!();
-    wallet.apply_update(update)?;
+    wallet.apply_update(update, false)?;
     wallet.commit()?;
 
     let balance = wallet.get_balance();