feat(bitcoind_rpc)!: simplified tx graph update

evanlinjin · evanlinjin · commit 105d6c6013a7 · 2023-08-28T15:39:36.000+08:00
Replaced `into_tx_graph_update` with `indexed_tx_graph_update`. The
latter returns a vec of `(tx, anchors, last_seen)` to be passed into
`IndexedTxGraph::insert_relevant_txs`.
diff --git a/crates/bitcoind_rpc/src/lib.rs b/crates/bitcoind_rpc/src/lib.rs
@@ -1,8 +1,7 @@
 //! This crate is used for updating [`bdk_chain`] structures with data from the `bitcoind` RPC
-//! interface.
+//! interface (excluding the RPC wallet API).
 //!
-//! The main structure is [`Emitter`], which sources blockchain data from
-//! [`bitcoincore_rpc::Client`].
+//! [`Emitter`] is the main structure which sources blockchain data from [`bitcoincore_rpc::Client`].
 //!
 //! To only get block updates (exlude mempool transactions), the caller can use
 //! [`Emitter::emit_block`] until it returns `Ok(None)` (which means the chain tip is reached). A
@@ -43,9 +42,9 @@
 
 use bdk_chain::{
     bitcoin::{Block, Transaction},
-    indexed_tx_graph::Indexer,
+    indexed_tx_graph::TxItem,
     local_chain::{self, CheckPoint},
-    Append, BlockId, ConfirmationHeightAnchor, ConfirmationTimeAnchor, TxGraph,
+    BlockId, ConfirmationHeightAnchor, ConfirmationTimeAnchor,
 };
 pub use bitcoincore_rpc;
 use bitcoincore_rpc::{json::GetBlockResult, RpcApi};
@@ -92,24 +91,21 @@ impl EmittedUpdate {
         })
     }
 
-    /// Transforms the emitted update into a [`TxGraph`] update.
-    ///
-    /// The `tx_filter` parameter takes in a closure that filters out irrelevant transactions so
-    /// they do not get included in the [`TxGraph`] update. We have provided two closures;
-    /// [`empty_filter`] and [`indexer_filter`] for this purpose.
+    /// Return transaction items to be consumed by [`IndexedTxGraph::insert_relevant_txs`].
     ///
     /// The `anchor_map` parameter takes in a closure that creates anchors of a specific type.
     /// [`confirmation_height_anchor`] and [`confirmation_time_anchor`] are avaliable to create
     /// updates with [`ConfirmationHeightAnchor`] and [`ConfirmationTimeAnchor`] respectively.
-    pub fn into_tx_graph_update<F, M, A>(self, tx_filter: F, anchor_map: M) -> TxGraph<A>
+    ///
+    /// [`IndexedTxGraph::insert_relevant_txs`]: bdk_chain::IndexedTxGraph::insert_relevant_txs
+    pub fn indexed_tx_graph_update<M, A>(&self, anchor_map: M) -> Vec<TxItem<'_, Option<A>>>
     where
-        F: FnMut(&Transaction) -> bool,
         M: Fn(&CheckPoint, &Block, usize) -> A,
-        A: Clone + Ord + PartialOrd,
+        A: Clone + Ord + PartialEq,
     {
         match self {
-            EmittedUpdate::Block(e) => e.into_tx_graph_update(tx_filter, anchor_map),
-            EmittedUpdate::Mempool(e) => e.into_tx_graph_update(tx_filter),
+            EmittedUpdate::Block(e) => e.indexed_tx_graph_update(anchor_map).collect(),
+            EmittedUpdate::Mempool(e) => e.indexed_tx_graph_update().collect(),
         }
     }
 }
@@ -129,87 +125,63 @@ impl EmittedBlock {
         self.cp.clone()
     }
 
-    /// Transforms the emitted update into a [`TxGraph`] update.
+    /// Convenience method to get [`local_chain::Update`].
+    pub fn chain_update(&self) -> local_chain::Update {
+        local_chain::Update {
+            tip: self.cp.clone(),
+            introduce_older_blocks: false,
+        }
+    }
+
+    /// Return transaction items to be consumed by [`IndexedTxGraph::insert_relevant_txs`].
     ///
-    /// The `tx_filter` parameter takes in a closure that filters out irrelevant transactions so
-    /// they do not get included in the [`TxGraph`] update. We have provided two closures;
-    /// [`empty_filter`] and [`indexer_filter`] for this purpose.
+    /// Refer to [`EmittedUpdate::indexed_tx_graph_update`] for more.
     ///
-    /// The `anchor_map` parameter takes in a closure that creates anchors of a specific type.
-    /// [`confirmation_height_anchor`] and [`confirmation_time_anchor`] are avaliable to create
-    /// updates with [`ConfirmationHeightAnchor`] and [`ConfirmationTimeAnchor`] respectively.
-    pub fn into_tx_graph_update<F, M, A>(self, mut tx_filter: F, anchor_map: M) -> TxGraph<A>
+    /// [`IndexedTxGraph::insert_relevant_txs`]: bdk_chain::IndexedTxGraph::insert_relevant_txs
+    pub fn indexed_tx_graph_update<M, A>(
+        &self,
+        anchor_map: M,
+    ) -> impl Iterator<Item = TxItem<'_, Option<A>>>
     where
-        F: FnMut(&Transaction) -> bool,
         M: Fn(&CheckPoint, &Block, usize) -> A,
-        A: Clone + Ord + PartialOrd,
+        A: Clone + Ord + PartialEq,
     {
-        let mut tx_graph = TxGraph::default();
-        let tx_iter = self
-            .block
+        self.block
             .txdata
             .iter()
             .enumerate()
-            .filter(move |(_, tx)| tx_filter(tx));
-        for (tx_pos, tx) in tx_iter {
-            let txid = tx.txid();
-            let _ = tx_graph.insert_anchor(txid, anchor_map(&self.cp, &self.block, tx_pos));
-            let _ = tx_graph.insert_tx(tx.clone());
-        }
-        tx_graph
+            .map(move |(i, tx)| (tx, Some(anchor_map(&self.cp, &self.block, i)), None))
     }
 }
 
 /// An emitted subset of mempool transactions.
 #[derive(Debug, Clone)]
 pub struct EmittedMempool {
-    /// Subset of mempool transactions.
+    /// Subset of mempool transactions as tuples of `(tx, seen_at)`.
+    ///
+    /// `seen_at` is the unix timestamp of when the transaction was first seen in the mempool.
     pub txs: Vec<(Transaction, u64)>,
 }
 
 impl EmittedMempool {
-    /// Transforms the emitted mempool into a [`TxGraph`] update.
+    /// Return transaction items to be consumed by [`IndexedTxGraph::insert_relevant_txs`].
+    ///
+    /// Refer to [`EmittedUpdate::indexed_tx_graph_update`] for more.
     ///
-    /// The `tx_filter` parameter takes in a closure that filters out irrelevant transactions so
-    /// they do not get included in the [`TxGraph`] update. We have provided two closures;
-    /// [`empty_filter`] and [`indexer_filter`] for this purpose.
-    pub fn into_tx_graph_update<F, A>(self, mut tx_filter: F) -> TxGraph<A>
+    /// [`IndexedTxGraph::insert_relevant_txs`]: bdk_chain::IndexedTxGraph::insert_relevant_txs
+    pub fn indexed_tx_graph_update<A>(&self) -> impl Iterator<Item = TxItem<'_, Option<A>>>
     where
-        F: FnMut(&Transaction) -> bool,
-        A: Clone + Ord + PartialOrd,
+        A: Clone + Ord + PartialEq,
     {
-        let mut tx_graph = TxGraph::default();
-        let tx_iter = self.txs.into_iter().filter(move |(tx, _)| tx_filter(tx));
-        for (tx, seen_at) in tx_iter {
-            let _ = tx_graph.insert_seen_at(tx.txid(), seen_at);
-            let _ = tx_graph.insert_tx(tx);
-        }
-        tx_graph
-    }
-}
-
-/// Creates a closure that filters transactions based on an [`Indexer`] implementation.
-pub fn indexer_filter<'i, I: Indexer>(
-    indexer: &'i mut I,
-    changeset: &'i mut I::ChangeSet,
-) -> impl FnMut(&Transaction) -> bool + 'i
-where
-    I::ChangeSet: bdk_chain::Append,
-{
-    |tx| {
-        changeset.append(indexer.index_tx(tx));
-        indexer.is_tx_relevant(tx)
+        self.txs
+            .iter()
+            .map(|(tx, seen_at)| (tx, None, Some(*seen_at)))
     }
 }
 
-/// Returns an empty filter-closure.
-pub fn empty_filter() -> impl FnMut(&Transaction) -> bool {
-    |_| true
-}
-
 /// A closure that transforms a [`EmittedUpdate`] into a [`ConfirmationHeightAnchor`].
 ///
-/// This is to be used as an input to [`EmittedUpdate::into_tx_graph_update`].
+/// This is to be used as an input to [`EmittedUpdate::indexed_tx_graph_update`].
 pub fn confirmation_height_anchor(
     cp: &CheckPoint,
     _block: &Block,
@@ -224,7 +196,7 @@ pub fn confirmation_height_anchor(
 
 /// A closure that transforms a [`EmittedUpdate`] into a [`ConfirmationTimeAnchor`].
 ///
-/// This is to be used as an input to [`EmittedUpdate::into_tx_graph_update`].
+/// This is to be used as an input to [`EmittedUpdate::indexed_tx_graph_update`].
 pub fn confirmation_time_anchor(
     cp: &CheckPoint,
     block: &Block,
diff --git a/crates/bitcoind_rpc/tests/test_emitter.rs b/crates/bitcoind_rpc/tests/test_emitter.rs
@@ -215,15 +215,10 @@ fn test_into_tx_graph() -> anyhow::Result<()> {
             let _ = chain.apply_update(chain_update)?;
         }
 
-        let tx_graph_update = update.into_tx_graph_update(
-            bdk_bitcoind_rpc::indexer_filter(&mut indexed_tx_graph.index, &mut BTreeSet::new()),
-            bdk_bitcoind_rpc::confirmation_height_anchor,
-        );
-        assert_eq!(tx_graph_update.full_txs().count(), 0);
-        assert_eq!(tx_graph_update.all_txouts().count(), 0);
-        assert_eq!(tx_graph_update.all_anchors().len(), 0);
+        let tx_graph_update =
+            update.indexed_tx_graph_update(bdk_bitcoind_rpc::confirmation_height_anchor);
 
-        let indexed_additions = indexed_tx_graph.apply_update(tx_graph_update);
+        let indexed_additions = indexed_tx_graph.insert_relevant_txs(tx_graph_update);
         assert!(indexed_additions.is_empty());
     }
 
@@ -250,20 +245,10 @@ fn test_into_tx_graph() -> anyhow::Result<()> {
         let update = Emitter::new(&env.client, 0, chain.tip()).emit_update()?;
         assert!(update.is_mempool());
 
-        let tx_graph_update = update.into_tx_graph_update(
-            bdk_bitcoind_rpc::indexer_filter(&mut indexed_tx_graph.index, &mut BTreeSet::new()),
-            bdk_bitcoind_rpc::confirmation_height_anchor,
-        );
-        assert_eq!(
-            tx_graph_update
-                .full_txs()
-                .map(|tx| tx.txid)
-                .collect::<BTreeSet<Txid>>(),
-            exp_txids,
-            "the mempool update should have 3 relevant transactions",
-        );
+        let tx_graph_update =
+            update.indexed_tx_graph_update(bdk_bitcoind_rpc::confirmation_height_anchor);
 
-        let indexed_additions = indexed_tx_graph.apply_update(tx_graph_update);
+        let indexed_additions = indexed_tx_graph.insert_relevant_txs(tx_graph_update);
         assert_eq!(
             indexed_additions
                 .graph
@@ -302,25 +287,10 @@ fn test_into_tx_graph() -> anyhow::Result<()> {
             let _ = chain.apply_update(chain_update)?;
         }
 
-        let tx_graph_update = update.into_tx_graph_update(
-            bdk_bitcoind_rpc::indexer_filter(&mut indexed_tx_graph.index, &mut BTreeSet::new()),
-            bdk_bitcoind_rpc::confirmation_height_anchor,
-        );
-        assert_eq!(
-            tx_graph_update
-                .full_txs()
-                .map(|tx| tx.txid)
-                .collect::<BTreeSet<Txid>>(),
-            exp_txids,
-            "block update should have 3 relevant transactions",
-        );
-        assert_eq!(
-            tx_graph_update.all_anchors(),
-            &exp_anchors,
-            "the block update should introduce anchors",
-        );
+        let tx_graph_update =
+            update.indexed_tx_graph_update(bdk_bitcoind_rpc::confirmation_height_anchor);
 
-        let indexed_additions = indexed_tx_graph.apply_update(tx_graph_update);
+        let indexed_additions = indexed_tx_graph.insert_relevant_txs(tx_graph_update);
         assert!(indexed_additions.graph.txs.is_empty());
         assert!(indexed_additions.graph.txouts.is_empty());
         assert_eq!(indexed_additions.graph.anchors, exp_anchors);
diff --git a/crates/chain/src/indexed_tx_graph.rs b/crates/chain/src/indexed_tx_graph.rs
@@ -135,8 +135,7 @@ where
     /// timestamp of when the transactions are last seen.
     pub fn insert_relevant_txs<'t>(
         &mut self,
-        txs: impl IntoIterator<Item = (&'t Transaction, impl IntoIterator<Item = A>)>,
-        seen_at: Option<u64>,
+        txs: impl IntoIterator<Item = TxItem<'t, impl IntoIterator<Item = A>>>,
     ) -> ChangeSet<A, I::ChangeSet> {
         // The algorithm below allows for non-topologically ordered transactions by using two loops.
         // This is achieved by:
@@ -146,17 +145,19 @@ where
         //    returns true or not. (in a second loop).
         let mut changeset = ChangeSet::<A, I::ChangeSet>::default();
         let mut transactions = Vec::new();
-        for (tx, anchors) in txs.into_iter() {
+        for (tx, anchors, seen_at) in txs.into_iter() {
             changeset.indexer.append(self.index.index_tx(tx));
-            transactions.push((tx, anchors));
+            transactions.push((tx, anchors, seen_at));
         }
         changeset.append(
             transactions
                 .into_iter()
-                .filter_map(|(tx, anchors)| match self.index.is_tx_relevant(tx) {
-                    true => Some(self.insert_tx(tx, anchors, seen_at)),
-                    false => None,
-                })
+                .filter_map(
+                    |(tx, anchors, seen_at)| match self.index.is_tx_relevant(tx) {
+                        true => Some(self.insert_tx(tx, anchors, seen_at)),
+                        false => None,
+                    },
+                )
                 .fold(Default::default(), |mut acc, other| {
                     acc.append(other);
                     acc
@@ -166,6 +167,9 @@ where
     }
 }
 
+/// Represents a single transaction update.
+pub type TxItem<'t, A> = (&'t Transaction, A, Option<u64>);
+
 /// A structure that represents changes to an [`IndexedTxGraph`].
 #[derive(Clone, Debug, PartialEq)]
 #[cfg_attr(
diff --git a/crates/chain/tests/test_indexed_tx_graph.rs b/crates/chain/tests/test_indexed_tx_graph.rs
@@ -74,7 +74,7 @@ fn insert_relevant_txs() {
     };
 
     assert_eq!(
-        graph.insert_relevant_txs(txs.iter().map(|tx| (tx, None)), None),
+        graph.insert_relevant_txs(txs.iter().map(|tx| (tx, None, None))),
         changeset,
     );
 
@@ -211,8 +211,8 @@ fn test_list_owned_txouts() {
     // Insert transactions into graph with respective anchors
     // For unconfirmed txs we pass in `None`.
 
-    let _ = graph.insert_relevant_txs(
-        [&tx1, &tx2, &tx3, &tx6].iter().enumerate().map(|(i, tx)| {
+    let _ =
+        graph.insert_relevant_txs([&tx1, &tx2, &tx3, &tx6].iter().enumerate().map(|(i, tx)| {
             let height = i as u32;
             (
                 *tx,
@@ -225,12 +225,11 @@ fn test_list_owned_txouts() {
                         anchor_block,
                         confirmation_height: anchor_block.height,
                     }),
+                None,
             )
-        }),
-        None,
-    );
+        }));
 
-    let _ = graph.insert_relevant_txs([&tx4, &tx5].iter().map(|tx| (*tx, None)), Some(100));
+    let _ = graph.insert_relevant_txs([&tx4, &tx5].iter().map(|tx| (*tx, None, Some(100))));
 
     // A helper lambda to extract and filter data from the graph.
     let fetch =
diff --git a/example-crates/example_rpc/src/main.rs b/example-crates/example_rpc/src/main.rs
@@ -64,9 +64,9 @@ impl From<RpcArgs> for Auth {
 
 #[derive(Subcommand, Debug, Clone)]
 enum RpcCommands {
-    /// Scans blocks via RPC (starting from last point of agreement) and stores/indexes relevant
-    /// transactions
-    Scan {
+    /// Syncs local state with remote state via RPC (starting from last point of agreement) and
+    /// stores/indexes relevant transactions
+    Sync {
         /// Starting block height to fallback to if no point of agreement if found
         #[clap(env = "FALLBACK_HEIGHT", long, default_value = "0")]
         fallback_height: u32,
@@ -93,7 +93,7 @@ enum RpcCommands {
 impl RpcCommands {
     fn rpc_args(&self) -> &RpcArgs {
         match self {
-            RpcCommands::Scan { rpc_args, .. } => rpc_args,
+            RpcCommands::Sync { rpc_args, .. } => rpc_args,
             RpcCommands::Tx { rpc_args, .. } => rpc_args,
         }
     }
@@ -145,7 +145,7 @@ fn main() -> anyhow::Result<()> {
     };
 
     match rpc_cmd {
-        RpcCommands::Scan {
+        RpcCommands::Sync {
             fallback_height,
             lookahead,
             live,
@@ -200,15 +200,9 @@ fn main() -> anyhow::Result<()> {
                     let mut chain = chain.lock().unwrap();
                     let mut graph = graph.lock().unwrap();
 
-                    let graph_update = {
-                        let tx_filter = bdk_bitcoind_rpc::indexer_filter(
-                            &mut graph.index,
-                            &mut indexed_changeset.indexer,
-                        );
-                        let anchor_map = bdk_bitcoind_rpc::confirmation_time_anchor;
-                        item.into_tx_graph_update(tx_filter, anchor_map)
-                    };
-                    indexed_changeset.append(graph.apply_update(graph_update));
+                    let graph_update =
+                        item.indexed_tx_graph_update(bdk_bitcoind_rpc::confirmation_time_anchor);
+                    indexed_changeset.append(graph.insert_relevant_txs(graph_update));
 
                     let chain_changeset = match chain_update {
                         Some(update) => chain.apply_update(update)?,
