From 576f31ad5b7e5ddc69fc32ca9f835d3d882960b5 Mon Sep 17 00:00:00 2001 From: Wei Chen Date: Tue, 13 Aug 2024 23:43:39 +0800 Subject: [PATCH] feat(chain): Introduce `tx_graph::Update` WIP --- crates/chain/src/spk_client.rs | 11 +- crates/chain/src/tx_graph.rs | 185 +++++++++++++++--- crates/electrum/src/bdk_electrum_client.rs | 37 ++-- crates/electrum/tests/test_electrum.rs | 13 +- crates/esplora/src/async_ext.rs | 69 ++++--- crates/esplora/src/blocking_ext.rs | 61 +++--- crates/esplora/src/lib.rs | 12 +- crates/esplora/tests/async_ext.rs | 12 +- crates/esplora/tests/blocking_ext.rs | 12 +- crates/wallet/src/wallet/mod.rs | 4 +- example-crates/example_electrum/src/main.rs | 12 +- example-crates/example_esplora/src/main.rs | 14 +- example-crates/wallet_electrum/src/main.rs | 6 +- .../wallet_esplora_async/src/main.rs | 4 +- .../wallet_esplora_blocking/src/main.rs | 4 +- 15 files changed, 305 insertions(+), 151 deletions(-) diff --git a/crates/chain/src/spk_client.rs b/crates/chain/src/spk_client.rs index 567a8f0a95..f6b2a8b681 100644 --- a/crates/chain/src/spk_client.rs +++ b/crates/chain/src/spk_client.rs @@ -3,7 +3,8 @@ use crate::{ alloc::{boxed::Box, collections::VecDeque, vec::Vec}, collections::BTreeMap, local_chain::CheckPoint, - ConfirmationBlockTime, Indexed, TxGraph, + tx_graph::Update, + ConfirmationBlockTime, Indexed, }; use bitcoin::{OutPoint, Script, ScriptBuf, Txid}; @@ -345,8 +346,8 @@ impl SyncRequest { #[must_use] #[derive(Debug)] pub struct SyncResult { - /// The update to apply to the receiving [`TxGraph`]. - pub graph_update: TxGraph, + /// The update to apply to the receiving [`Update`]. + pub graph_update: Update, /// The update to apply to the receiving [`LocalChain`](crate::local_chain::LocalChain). pub chain_update: Option, } @@ -497,8 +498,8 @@ impl FullScanRequest { #[derive(Debug)] pub struct FullScanResult { /// The update to apply to the receiving [`LocalChain`](crate::local_chain::LocalChain). - pub graph_update: TxGraph, - /// The update to apply to the receiving [`TxGraph`]. + pub graph_update: Update, + /// The update to apply to the receiving [`Update`]. pub chain_update: Option, /// Last active indices for the corresponding keychains (`K`). pub last_active_indices: BTreeMap, diff --git a/crates/chain/src/tx_graph.rs b/crates/chain/src/tx_graph.rs index 0eab93867b..03e6152007 100644 --- a/crates/chain/src/tx_graph.rs +++ b/crates/chain/src/tx_graph.rs @@ -89,7 +89,8 @@ //! [`insert_txout`]: TxGraph::insert_txout use crate::{ - collections::*, Anchor, Balance, BlockId, ChainOracle, ChainPosition, FullTxOut, Merge, + collections::*, Anchor, Balance, BlockId, ChainOracle, ChainPosition, ConfirmationBlockTime, + FullTxOut, Merge, }; use alloc::collections::vec_deque::VecDeque; use alloc::sync::Arc; @@ -640,15 +641,15 @@ impl TxGraph { /// /// The returned [`ChangeSet`] is the set difference between `update` and `self` (transactions that /// exist in `update` but not in `self`). - pub fn apply_update(&mut self, update: TxGraph) -> ChangeSet { - let changeset = self.determine_changeset(update); + pub fn apply_update(&mut self, update: impl Into>) -> ChangeSet { + let changeset = self.determine_changeset(update.into()); self.apply_changeset(changeset.clone()); changeset } /// Determines the [`ChangeSet`] between `self` and an empty [`TxGraph`]. pub fn initial_changeset(&self) -> ChangeSet { - Self::default().determine_changeset(self.clone()) + Self::default().determine_changeset(Update::from(self.clone())) } /// Applies [`ChangeSet`] to [`TxGraph`]. @@ -715,36 +716,28 @@ impl TxGraph { /// /// The [`ChangeSet`] would be the set difference between `update` and `self` (transactions that /// exist in `update` but not in `self`). - pub(crate) fn determine_changeset(&self, update: TxGraph) -> ChangeSet { + pub(crate) fn determine_changeset(&self, update: Update) -> ChangeSet { let mut changeset = ChangeSet::::default(); - for (&txid, (update_tx_node, _)) in &update.txs { - match (self.txs.get(&txid), update_tx_node) { - (None, TxNodeInternal::Whole(update_tx)) => { - changeset.txs.insert(update_tx.clone()); + for (txid, tx) in update.whole_txs { + match self.txs.get(&txid) { + None | Some((TxNodeInternal::Partial(_), _)) => { + changeset.txs.insert(tx); } - (None, TxNodeInternal::Partial(update_txos)) => { - changeset.txouts.extend( - update_txos - .iter() - .map(|(&vout, txo)| (OutPoint::new(txid, vout), txo.clone())), - ); - } - (Some((TxNodeInternal::Whole(_), _)), _) => {} - (Some((TxNodeInternal::Partial(_), _)), TxNodeInternal::Whole(update_tx)) => { - changeset.txs.insert(update_tx.clone()); - } - ( - Some((TxNodeInternal::Partial(txos), _)), - TxNodeInternal::Partial(update_txos), - ) => { - changeset.txouts.extend( - update_txos - .iter() - .filter(|(vout, _)| !txos.contains_key(*vout)) - .map(|(&vout, txo)| (OutPoint::new(txid, vout), txo.clone())), - ); + Some((TxNodeInternal::Whole(old_tx), _)) if *old_tx != tx => { + // Update the `tx` in graph if does not match up with the tx from `update`. + changeset.txs.insert(tx); } + _ => {} + } + } + + for (op, txout) in update.partial_txs { + if matches!( + self.txs.get(&op.txid), + None | Some((TxNodeInternal::Partial(_), _)) + ) { + changeset.txouts.insert(op, txout); } } @@ -761,6 +754,138 @@ impl TxGraph { } } +/// An update to [`TxGraph`]. +#[derive(Clone, Debug)] +pub struct Update { + whole_txs: HashMap>, + partial_txs: HashMap, + last_seen: HashMap, + anchors: BTreeSet<(A, Txid)>, +} + +impl Default for Update { + fn default() -> Self { + Update { + whole_txs: Default::default(), + partial_txs: Default::default(), + last_seen: Default::default(), + anchors: Default::default(), + } + } +} + +impl Update { + /// Iterate over all full transactions in the graph. + pub fn whole_txs(&self) -> impl Iterator)> { + self.whole_txs.clone().into_iter() + } + + /// Get a transaction by txid. This only returns `Some` for full transactions. + pub fn get_tx(&self, txid: Txid) -> Option> { + self.whole_txs.get(&txid).cloned() + } + + /// Inserts the given transaction into [`Update`]. + pub fn insert_tx>>(&mut self, tx: T) { + let tx = tx.into(); + let txid = tx.compute_txid(); + + // Remove any floating txouts with the full transaction's txid to enforce invariance. + self.partial_txs.retain(|op, _| op.txid != txid); + + self.whole_txs.insert(txid, tx); + } + + /// Inserts the given [`TxOut`] at [`OutPoint`] into [`Update`]. + /// + /// Inserting floating txouts are useful for determining fee/feerate of transactions we care + /// about. + pub fn insert_txout(&mut self, outpoint: OutPoint, txout: TxOut) { + self.partial_txs.insert(outpoint, txout); + } + + /// Inserts the given `seen_at` for `txid` into [`Update`]. + pub fn insert_seen_at(&mut self, txid: Txid, seen_at: u64) { + self.last_seen.insert(txid, seen_at); + } +} + +impl Update { + /// Get all transaction anchors known by [`Update`]. + pub fn all_anchors(&self) -> &BTreeSet<(A, Txid)> { + &self.anchors + } + + /// Returns the [`Update`] as a `TxGraph` with `ConfirmationBlockTime` anchors. + pub fn into_tx_graph(self) -> TxGraph { + let mut txs = HashMap::new(); + let mut conf_anchors = BTreeSet::new(); + + for (txid, tx) in self.whole_txs { + txs.insert(txid, (TxNodeInternal::Whole(tx), BTreeSet::new())); + } + for (op, txout) in self.partial_txs { + txs.insert( + op.txid, + ( + TxNodeInternal::Partial([(op.vout, txout)].into()), + BTreeSet::new(), + ), + ); + } + for (anchor, txid) in self.anchors { + conf_anchors.insert(( + ConfirmationBlockTime { + block_id: anchor.anchor_block(), + confirmation_time: anchor.confirmation_height_upper_bound() as u64, + }, + txid, + )); + } + + TxGraph { + txs, + spends: BTreeMap::new(), + anchors: conf_anchors, + last_seen: self.last_seen, + empty_outspends: HashSet::new(), + } + } +} + +impl Update { + /// Inserts the given `anchor` into [`Update`]. + pub fn insert_anchor(&mut self, txid: Txid, anchor: A) { + self.anchors.insert((anchor, txid)); + } + + /// Extends this [`Update`] with another so that `self` becomes the union of the two sets of + /// [`Update`]s. + pub fn extend(&mut self, update: Update) { + self.whole_txs.extend(update.whole_txs); + self.partial_txs.extend(update.partial_txs); + self.last_seen.extend(update.last_seen); + self.anchors.extend(update.anchors); + } +} + +impl From> for Update { + fn from(graph: TxGraph) -> Self { + Update { + whole_txs: graph + .full_txs() + .map(|value| (value.txid, value.tx)) + .collect::>(), + partial_txs: graph + .floating_txouts() + .map(|(op, txout)| (op, txout.clone())) + .collect::>(), + last_seen: graph.last_seen, + anchors: graph.anchors, + } + } +} + impl TxGraph { /// Get the position of the transaction in `chain` with tip `chain_tip`. /// diff --git a/crates/electrum/src/bdk_electrum_client.rs b/crates/electrum/src/bdk_electrum_client.rs index 1458e2bd96..dc8c0cdbcf 100644 --- a/crates/electrum/src/bdk_electrum_client.rs +++ b/crates/electrum/src/bdk_electrum_client.rs @@ -3,7 +3,7 @@ use bdk_chain::{ collections::{BTreeMap, HashMap}, local_chain::CheckPoint, spk_client::{FullScanRequest, FullScanResult, SyncRequest, SyncResult}, - tx_graph::TxGraph, + tx_graph::Update, Anchor, BlockId, ConfirmationBlockTime, }; use electrum_client::{ElectrumApi, Error, HeaderNotification}; @@ -39,11 +39,8 @@ impl BdkElectrumClient { /// Inserts transactions into the transaction cache so that the client will not fetch these /// transactions. - pub fn populate_tx_cache(&self, tx_graph: impl AsRef>) { - let txs = tx_graph - .as_ref() - .full_txs() - .map(|tx_node| (tx_node.txid, tx_node.tx)); + pub fn populate_tx_cache(&self, update: impl Into>) { + let txs = update.into().whole_txs(); let mut tx_cache = self.tx_cache.lock().unwrap(); for (txid, tx) in txs { @@ -138,7 +135,7 @@ impl BdkElectrumClient { None => None, }; - let mut graph_update = TxGraph::::default(); + let mut graph_update = Update::::default(); let mut last_active_indices = BTreeMap::::default(); for keychain in request.keychains() { let spks = request.iter_spks(keychain.clone()); @@ -205,7 +202,7 @@ impl BdkElectrumClient { None => None, }; - let mut graph_update = TxGraph::::default(); + let mut graph_update = Update::::default(); self.populate_with_spks( &mut graph_update, request @@ -245,7 +242,7 @@ impl BdkElectrumClient { /// also included. fn populate_with_spks( &self, - graph_update: &mut TxGraph, + graph_update: &mut Update, mut spks: impl Iterator, stop_gap: usize, batch_size: usize, @@ -278,7 +275,7 @@ impl BdkElectrumClient { } for tx_res in spk_history { - let _ = graph_update.insert_tx(self.fetch_tx(tx_res.tx_hash)?); + graph_update.insert_tx(self.fetch_tx(tx_res.tx_hash)?); self.validate_merkle_for_anchor(graph_update, tx_res.tx_hash, tx_res.height)?; } } @@ -291,7 +288,7 @@ impl BdkElectrumClient { /// included. Anchors of the aforementioned transactions are included. fn populate_with_outpoints( &self, - graph_update: &mut TxGraph, + graph_update: &mut Update, outpoints: impl IntoIterator, ) -> Result<(), Error> { for outpoint in outpoints { @@ -314,7 +311,7 @@ impl BdkElectrumClient { if !has_residing && res.tx_hash == op_txid { has_residing = true; - let _ = graph_update.insert_tx(Arc::clone(&op_tx)); + graph_update.insert_tx(Arc::clone(&op_tx)); self.validate_merkle_for_anchor(graph_update, res.tx_hash, res.height)?; } @@ -328,7 +325,7 @@ impl BdkElectrumClient { if !has_spending { continue; } - let _ = graph_update.insert_tx(Arc::clone(&res_tx)); + graph_update.insert_tx(Arc::clone(&res_tx)); self.validate_merkle_for_anchor(graph_update, res.tx_hash, res.height)?; } } @@ -339,7 +336,7 @@ impl BdkElectrumClient { /// Populate the `graph_update` with transactions/anchors of the provided `txids`. fn populate_with_txids( &self, - graph_update: &mut TxGraph, + graph_update: &mut Update, txids: impl IntoIterator, ) -> Result<(), Error> { for txid in txids { @@ -366,7 +363,7 @@ impl BdkElectrumClient { self.validate_merkle_for_anchor(graph_update, txid, r.height)?; } - let _ = graph_update.insert_tx(tx); + graph_update.insert_tx(tx); } Ok(()) } @@ -375,7 +372,7 @@ impl BdkElectrumClient { // An anchor is inserted if the transaction is validated to be in a confirmed block. fn validate_merkle_for_anchor( &self, - graph_update: &mut TxGraph, + graph_update: &mut Update, txid: Txid, confirmation_height: i32, ) -> Result<(), Error> { @@ -402,7 +399,7 @@ impl BdkElectrumClient { } if is_confirmed_tx { - let _ = graph_update.insert_anchor( + graph_update.insert_anchor( txid, ConfirmationBlockTime { confirmation_time: header.time as u64, @@ -421,17 +418,17 @@ impl BdkElectrumClient { // which we do not have by default. This data is needed to calculate the transaction fee. fn fetch_prev_txout( &self, - graph_update: &mut TxGraph, + graph_update: &mut Update, ) -> Result<(), Error> { let full_txs: Vec> = - graph_update.full_txs().map(|tx_node| tx_node.tx).collect(); + graph_update.whole_txs().map(|(_txid, tx)| tx).collect(); for tx in full_txs { for vin in &tx.input { let outpoint = vin.previous_output; let vout = outpoint.vout; let prev_tx = self.fetch_tx(outpoint.txid)?; let txout = prev_tx.output[vout as usize].clone(); - let _ = graph_update.insert_txout(outpoint, txout); + graph_update.insert_txout(outpoint, txout); } } Ok(()) diff --git a/crates/electrum/tests/test_electrum.rs b/crates/electrum/tests/test_electrum.rs index 63e91081b6..78842ae391 100644 --- a/crates/electrum/tests/test_electrum.rs +++ b/crates/electrum/tests/test_electrum.rs @@ -43,20 +43,22 @@ where BATCH_SIZE, true, )?; + let mut tx_graph = update.graph_update.into_tx_graph(); // Update `last_seen` to be able to calculate balance for unconfirmed transactions. let now = std::time::UNIX_EPOCH .elapsed() .expect("must get time") .as_secs(); - let _ = update.graph_update.update_last_seen_unconfirmed(now); + let _ = tx_graph.update_last_seen_unconfirmed(now); if let Some(chain_update) = update.chain_update.clone() { let _ = chain .apply_update(chain_update) .map_err(|err| anyhow::anyhow!("LocalChain update error: {:?}", err))?; } - let _ = graph.apply_update(update.graph_update.clone()); + let _ = graph.apply_update(tx_graph.clone()); + update.graph_update = graph.graph().clone().into(); Ok(update) } @@ -127,7 +129,7 @@ pub fn test_update_tx_graph_without_keychain() -> anyhow::Result<()> { "update should not alter original checkpoint tip since we already started with all checkpoints", ); - let graph_update = sync_update.graph_update; + let graph_update = sync_update.graph_update.into_tx_graph(); // Check to see if we have the floating txouts available from our two created transactions' // previous outputs in order to calculate transaction fees. for tx in graph_update.full_txs() { @@ -216,7 +218,7 @@ pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> { .spks_for_keychain(0, spks.clone()); client.full_scan(request, 3, 1, false)? }; - assert!(full_scan_update.graph_update.full_txs().next().is_none()); + assert!(full_scan_update.graph_update.whole_txs().next().is_none()); assert!(full_scan_update.last_active_indices.is_empty()); let full_scan_update = { let request = FullScanRequest::builder() @@ -227,6 +229,7 @@ pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> { assert_eq!( full_scan_update .graph_update + .into_tx_graph() .full_txs() .next() .unwrap() @@ -259,6 +262,7 @@ pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> { }; let txs: HashSet<_> = full_scan_update .graph_update + .into_tx_graph() .full_txs() .map(|tx| tx.txid) .collect(); @@ -273,6 +277,7 @@ pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> { }; let txs: HashSet<_> = full_scan_update .graph_update + .into_tx_graph() .full_txs() .map(|tx| tx.txid) .collect(); diff --git a/crates/esplora/src/async_ext.rs b/crates/esplora/src/async_ext.rs index 066b91e170..da0f129a31 100644 --- a/crates/esplora/src/async_ext.rs +++ b/crates/esplora/src/async_ext.rs @@ -6,7 +6,8 @@ use bdk_chain::{ bitcoin::{BlockHash, OutPoint, ScriptBuf, Txid}, collections::BTreeMap, local_chain::CheckPoint, - BlockId, ConfirmationBlockTime, TxGraph, + tx_graph::Update, + BlockId, ConfirmationBlockTime, }; use bdk_chain::{Anchor, Indexed}; use esplora_client::{Tx, TxStatus}; @@ -72,14 +73,14 @@ impl EsploraAsyncExt for esplora_client::AsyncClient { None }; - let mut graph_update = TxGraph::default(); + let mut graph_update = Update::default(); let mut last_active_indices = BTreeMap::::new(); for keychain in keychains { let keychain_spks = request.iter_spks(keychain.clone()); - let (tx_graph, last_active_index) = + let (update, last_active_index) = fetch_txs_with_keychain_spks(self, keychain_spks, stop_gap, parallel_requests) .await?; - let _ = graph_update.apply_update(tx_graph); + graph_update.extend(update); if let Some(last_active_index) = last_active_index { last_active_indices.insert(keychain, last_active_index); } @@ -113,13 +114,12 @@ impl EsploraAsyncExt for esplora_client::AsyncClient { None }; - let mut graph_update = TxGraph::::default(); - let _ = graph_update - .apply_update(fetch_txs_with_spks(self, request.iter_spks(), parallel_requests).await?); - let _ = graph_update.apply_update( - fetch_txs_with_txids(self, request.iter_txids(), parallel_requests).await?, - ); - let _ = graph_update.apply_update( + let mut graph_update = Update::::default(); + graph_update + .extend(fetch_txs_with_spks(self, request.iter_spks(), parallel_requests).await?); + graph_update + .extend(fetch_txs_with_txids(self, request.iter_txids(), parallel_requests).await?); + graph_update.extend( fetch_txs_with_outpoints(self, request.iter_outpoints(), parallel_requests).await?, ); @@ -245,7 +245,7 @@ async fn chain_update( /// scripts with no transaction history is reached. `parallel_requests` specifies the maximum /// number of HTTP requests to make in parallel. /// -/// A [`TxGraph`] (containing the fetched transactions and anchors) and the last active +/// A [`Update`] (containing the fetched transactions and anchors) and the last active /// keychain index (if any) is returned. The last active keychain index is the keychain's last /// script pubkey that contains a non-empty transaction history. /// @@ -255,10 +255,10 @@ async fn fetch_txs_with_keychain_spks> + S mut keychain_spks: I, stop_gap: usize, parallel_requests: usize, -) -> Result<(TxGraph, Option), Error> { +) -> Result<(Update, Option), Error> { type TxsOfSpkIndex = (u32, Vec); - let mut tx_graph = TxGraph::default(); + let mut update = Update::default(); let mut last_index = Option::::None; let mut last_active_index = Option::::None; @@ -294,9 +294,9 @@ async fn fetch_txs_with_keychain_spks> + S last_active_index = Some(index); } for tx in txs { - let _ = tx_graph.insert_tx(tx.to_tx()); - insert_anchor_from_status(&mut tx_graph, tx.txid, tx.status); - insert_prevouts(&mut tx_graph, tx.vin); + update.insert_tx(tx.to_tx()); + insert_anchor_from_status(&mut update, tx.txid, tx.status); + insert_prevouts(&mut update, tx.vin); } } @@ -311,7 +311,7 @@ async fn fetch_txs_with_keychain_spks> + S } } - Ok((tx_graph, last_active_index)) + Ok((update, last_active_index)) } /// Fetch transactions and associated [`ConfirmationBlockTime`]s by scanning `spks` @@ -326,7 +326,7 @@ async fn fetch_txs_with_spks + Send>( client: &esplora_client::AsyncClient, spks: I, parallel_requests: usize, -) -> Result, Error> +) -> Result, Error> where I::IntoIter: Send, { @@ -337,7 +337,7 @@ where parallel_requests, ) .await - .map(|(tx_graph, _)| tx_graph) + .map(|(update, _)| update) } /// Fetch transactions and associated [`ConfirmationBlockTime`]s by scanning `txids` @@ -350,7 +350,7 @@ async fn fetch_txs_with_txids + Send>( client: &esplora_client::AsyncClient, txids: I, parallel_requests: usize, -) -> Result, Error> +) -> Result, Error> where I::IntoIter: Send, { @@ -359,7 +359,7 @@ where Tx(Option), } - let mut tx_graph = TxGraph::default(); + let mut update = Update::default(); let mut txids = txids.into_iter(); loop { let handles = txids @@ -367,7 +367,7 @@ where .take(parallel_requests) .map(|txid| { let client = client.clone(); - let tx_already_exists = tx_graph.get_tx(txid).is_some(); + let tx_already_exists = update.get_tx(txid).is_some(); async move { if tx_already_exists { client @@ -391,18 +391,18 @@ where for (txid, resp) in handles.try_collect::>().await? { match resp { EsploraResp::TxStatus(status) => { - insert_anchor_from_status(&mut tx_graph, txid, status); + insert_anchor_from_status(&mut update, txid, status); } EsploraResp::Tx(Some(tx_info)) => { - let _ = tx_graph.insert_tx(tx_info.to_tx()); - insert_anchor_from_status(&mut tx_graph, txid, tx_info.status); - insert_prevouts(&mut tx_graph, tx_info.vin); + update.insert_tx(tx_info.to_tx()); + insert_anchor_from_status(&mut update, txid, tx_info.status); + insert_prevouts(&mut update, tx_info.vin); } _ => continue, } } } - Ok(tx_graph) + Ok(update) } /// Fetch transactions and [`ConfirmationBlockTime`]s that contain and spend the provided @@ -415,7 +415,7 @@ async fn fetch_txs_with_outpoints + Send>( client: &esplora_client::AsyncClient, outpoints: I, parallel_requests: usize, -) -> Result, Error> +) -> Result, Error> where I::IntoIter: Send, { @@ -423,7 +423,7 @@ where // make sure txs exists in graph and tx statuses are updated // TODO: We should maintain a tx cache (like we do with Electrum). - let mut tx_graph = fetch_txs_with_txids( + let mut update = fetch_txs_with_txids( client, outpoints.iter().copied().map(|op| op.txid), parallel_requests, @@ -452,18 +452,17 @@ where Some(txid) => txid, None => continue, }; - if tx_graph.get_tx(spend_txid).is_none() { + if update.get_tx(spend_txid).is_none() { missing_txs.push(spend_txid); } if let Some(spend_status) = op_status.status { - insert_anchor_from_status(&mut tx_graph, spend_txid, spend_status); + insert_anchor_from_status(&mut update, spend_txid, spend_status); } } } - let _ = - tx_graph.apply_update(fetch_txs_with_txids(client, missing_txs, parallel_requests).await?); - Ok(tx_graph) + update.extend(fetch_txs_with_txids(client, missing_txs, parallel_requests).await?); + Ok(update) } #[cfg(test)] diff --git a/crates/esplora/src/blocking_ext.rs b/crates/esplora/src/blocking_ext.rs index 6e3e25afe7..1711fa6845 100644 --- a/crates/esplora/src/blocking_ext.rs +++ b/crates/esplora/src/blocking_ext.rs @@ -6,7 +6,8 @@ use bdk_chain::spk_client::{FullScanRequest, FullScanResult, SyncRequest, SyncRe use bdk_chain::{ bitcoin::{BlockHash, OutPoint, ScriptBuf, Txid}, local_chain::CheckPoint, - BlockId, ConfirmationBlockTime, TxGraph, + tx_graph::Update, + BlockId, ConfirmationBlockTime, }; use bdk_chain::{Anchor, Indexed}; use esplora_client::{OutputStatus, Tx, TxStatus}; @@ -66,13 +67,13 @@ impl EsploraExt for esplora_client::BlockingClient { None }; - let mut graph_update = TxGraph::default(); + let mut graph_update = Update::default(); let mut last_active_indices = BTreeMap::::new(); for keychain in request.keychains() { let keychain_spks = request.iter_spks(keychain.clone()); - let (tx_graph, last_active_index) = + let (update, last_active_index) = fetch_txs_with_keychain_spks(self, keychain_spks, stop_gap, parallel_requests)?; - let _ = graph_update.apply_update(tx_graph); + graph_update.extend(update); if let Some(last_active_index) = last_active_index { last_active_indices.insert(keychain, last_active_index); } @@ -109,18 +110,18 @@ impl EsploraExt for esplora_client::BlockingClient { None }; - let mut graph_update = TxGraph::default(); - let _ = graph_update.apply_update(fetch_txs_with_spks( + let mut graph_update = Update::default(); + graph_update.extend(fetch_txs_with_spks( self, request.iter_spks(), parallel_requests, )?); - let _ = graph_update.apply_update(fetch_txs_with_txids( + graph_update.extend(fetch_txs_with_txids( self, request.iter_txids(), parallel_requests, )?); - let _ = graph_update.apply_update(fetch_txs_with_outpoints( + graph_update.extend(fetch_txs_with_outpoints( self, request.iter_outpoints(), parallel_requests, @@ -247,10 +248,10 @@ fn fetch_txs_with_keychain_spks>>( mut keychain_spks: I, stop_gap: usize, parallel_requests: usize, -) -> Result<(TxGraph, Option), Error> { +) -> Result<(Update, Option), Error> { type TxsOfSpkIndex = (u32, Vec); - let mut tx_graph = TxGraph::default(); + let mut update = Update::default(); let mut last_index = Option::::None; let mut last_active_index = Option::::None; @@ -289,9 +290,9 @@ fn fetch_txs_with_keychain_spks>>( last_active_index = Some(index); } for tx in txs { - let _ = tx_graph.insert_tx(tx.to_tx()); - insert_anchor_from_status(&mut tx_graph, tx.txid, tx.status); - insert_prevouts(&mut tx_graph, tx.vin); + update.insert_tx(tx.to_tx()); + insert_anchor_from_status(&mut update, tx.txid, tx.status); + insert_prevouts(&mut update, tx.vin); } } @@ -306,7 +307,7 @@ fn fetch_txs_with_keychain_spks>>( } } - Ok((tx_graph, last_active_index)) + Ok((update, last_active_index)) } /// Fetch transactions and associated [`ConfirmationBlockTime`]s by scanning `spks` @@ -321,14 +322,14 @@ fn fetch_txs_with_spks>( client: &esplora_client::BlockingClient, spks: I, parallel_requests: usize, -) -> Result, Error> { +) -> Result, Error> { fetch_txs_with_keychain_spks( client, spks.into_iter().enumerate().map(|(i, spk)| (i as u32, spk)), usize::MAX, parallel_requests, ) - .map(|(tx_graph, _)| tx_graph) + .map(|(update, _)| update) } /// Fetch transactions and associated [`ConfirmationBlockTime`]s by scanning `txids` @@ -341,13 +342,13 @@ fn fetch_txs_with_txids>( client: &esplora_client::BlockingClient, txids: I, parallel_requests: usize, -) -> Result, Error> { +) -> Result, Error> { enum EsploraResp { TxStatus(TxStatus), Tx(Option), } - let mut tx_graph = TxGraph::default(); + let mut update = Update::default(); let mut txids = txids.into_iter(); loop { let handles = txids @@ -355,7 +356,7 @@ fn fetch_txs_with_txids>( .take(parallel_requests) .map(|txid| { let client = client.clone(); - let tx_already_exists = tx_graph.get_tx(txid).is_some(); + let tx_already_exists = update.get_tx(txid).is_some(); std::thread::spawn(move || { if tx_already_exists { client @@ -380,18 +381,18 @@ fn fetch_txs_with_txids>( let (txid, resp) = handle.join().expect("thread must not panic")?; match resp { EsploraResp::TxStatus(status) => { - insert_anchor_from_status(&mut tx_graph, txid, status); + insert_anchor_from_status(&mut update, txid, status); } EsploraResp::Tx(Some(tx_info)) => { - let _ = tx_graph.insert_tx(tx_info.to_tx()); - insert_anchor_from_status(&mut tx_graph, txid, tx_info.status); - insert_prevouts(&mut tx_graph, tx_info.vin); + update.insert_tx(tx_info.to_tx()); + insert_anchor_from_status(&mut update, txid, tx_info.status); + insert_prevouts(&mut update, tx_info.vin); } _ => continue, } } } - Ok(tx_graph) + Ok(update) } /// Fetch transactions and [`ConfirmationBlockTime`]s that contain and spend the provided @@ -404,12 +405,12 @@ fn fetch_txs_with_outpoints>( client: &esplora_client::BlockingClient, outpoints: I, parallel_requests: usize, -) -> Result, Error> { +) -> Result, Error> { let outpoints = outpoints.into_iter().collect::>(); // make sure txs exists in graph and tx statuses are updated // TODO: We should maintain a tx cache (like we do with Electrum). - let mut tx_graph = fetch_txs_with_txids( + let mut update = fetch_txs_with_txids( client, outpoints.iter().map(|op| op.txid), parallel_requests, @@ -442,22 +443,22 @@ fn fetch_txs_with_outpoints>( Some(txid) => txid, None => continue, }; - if tx_graph.get_tx(spend_txid).is_none() { + if update.get_tx(spend_txid).is_none() { missing_txs.push(spend_txid); } if let Some(spend_status) = op_status.status { - insert_anchor_from_status(&mut tx_graph, spend_txid, spend_status); + insert_anchor_from_status(&mut update, spend_txid, spend_status); } } } } - let _ = tx_graph.apply_update(fetch_txs_with_txids( + update.extend(fetch_txs_with_txids( client, missing_txs, parallel_requests, )?); - Ok(tx_graph) + Ok(update) } #[cfg(test)] diff --git a/crates/esplora/src/lib.rs b/crates/esplora/src/lib.rs index 7db6967b65..8bdf62c4f1 100644 --- a/crates/esplora/src/lib.rs +++ b/crates/esplora/src/lib.rs @@ -20,13 +20,13 @@ //! [`esplora_client::BlockingClient`], [`EsploraAsyncExt`] is the async version which extends //! [`esplora_client::AsyncClient`]. //! -//! [`TxGraph`]: bdk_chain::tx_graph::TxGraph +//! [`Update`]: bdk_chain::tx_graph::Update //! [`LocalChain`]: bdk_chain::local_chain::LocalChain //! [`ChainOracle`]: bdk_chain::ChainOracle //! [`example_esplora`]: https://github.com/bitcoindevkit/bdk/tree/master/example-crates/example_esplora use bdk_chain::bitcoin::{Amount, OutPoint, TxOut, Txid}; -use bdk_chain::{BlockId, ConfirmationBlockTime, TxGraph}; +use bdk_chain::{tx_graph::Update, BlockId, ConfirmationBlockTime}; use esplora_client::TxStatus; pub use esplora_client; @@ -42,7 +42,7 @@ mod async_ext; pub use async_ext::*; fn insert_anchor_from_status( - tx_graph: &mut TxGraph, + update: &mut Update, txid: Txid, status: TxStatus, ) { @@ -57,21 +57,21 @@ fn insert_anchor_from_status( block_id: BlockId { height, hash }, confirmation_time: time, }; - let _ = tx_graph.insert_anchor(txid, anchor); + update.insert_anchor(txid, anchor); } } /// Inserts floating txouts into `tx_graph` using [`Vin`](esplora_client::api::Vin)s returned by /// Esplora. fn insert_prevouts( - tx_graph: &mut TxGraph, + update: &mut Update, esplora_inputs: impl IntoIterator, ) { let prevouts = esplora_inputs .into_iter() .filter_map(|vin| Some((vin.txid, vin.vout, vin.prevout?))); for (prev_txid, prev_vout, prev_txout) in prevouts { - let _ = tx_graph.insert_txout( + update.insert_txout( OutPoint::new(prev_txid, prev_vout), TxOut { script_pubkey: prev_txout.scriptpubkey, diff --git a/crates/esplora/tests/async_ext.rs b/crates/esplora/tests/async_ext.rs index 70d4641941..4e9d44fb79 100644 --- a/crates/esplora/tests/async_ext.rs +++ b/crates/esplora/tests/async_ext.rs @@ -77,7 +77,7 @@ pub async fn test_update_tx_graph_without_keychain() -> anyhow::Result<()> { "update should not alter original checkpoint tip since we already started with all checkpoints", ); - let graph_update = sync_update.graph_update; + let graph_update = sync_update.graph_update.into_tx_graph(); // Check to see if we have the floating txouts available from our two created transactions' // previous outputs in order to calculate transaction fees. for tx in graph_update.full_txs() { @@ -167,7 +167,12 @@ pub async fn test_async_update_tx_graph_stop_gap() -> anyhow::Result<()> { .spks_for_keychain(0, spks.clone()); client.full_scan(request, 3, 1).await? }; - assert!(full_scan_update.graph_update.full_txs().next().is_none()); + assert!(full_scan_update + .graph_update + .into_tx_graph() + .full_txs() + .next() + .is_none()); assert!(full_scan_update.last_active_indices.is_empty()); let full_scan_update = { let request = FullScanRequest::builder() @@ -178,6 +183,7 @@ pub async fn test_async_update_tx_graph_stop_gap() -> anyhow::Result<()> { assert_eq!( full_scan_update .graph_update + .into_tx_graph() .full_txs() .next() .unwrap() @@ -212,6 +218,7 @@ pub async fn test_async_update_tx_graph_stop_gap() -> anyhow::Result<()> { }; let txs: HashSet<_> = full_scan_update .graph_update + .into_tx_graph() .full_txs() .map(|tx| tx.txid) .collect(); @@ -226,6 +233,7 @@ pub async fn test_async_update_tx_graph_stop_gap() -> anyhow::Result<()> { }; let txs: HashSet<_> = full_scan_update .graph_update + .into_tx_graph() .full_txs() .map(|tx| tx.txid) .collect(); diff --git a/crates/esplora/tests/blocking_ext.rs b/crates/esplora/tests/blocking_ext.rs index 818f1f5fb6..0dddc6abc9 100644 --- a/crates/esplora/tests/blocking_ext.rs +++ b/crates/esplora/tests/blocking_ext.rs @@ -77,7 +77,7 @@ pub fn test_update_tx_graph_without_keychain() -> anyhow::Result<()> { "update should not alter original checkpoint tip since we already started with all checkpoints", ); - let graph_update = sync_update.graph_update; + let graph_update = sync_update.graph_update.into_tx_graph(); // Check to see if we have the floating txouts available from our two created transactions' // previous outputs in order to calculate transaction fees. for tx in graph_update.full_txs() { @@ -168,7 +168,12 @@ pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> { .spks_for_keychain(0, spks.clone()); client.full_scan(request, 3, 1)? }; - assert!(full_scan_update.graph_update.full_txs().next().is_none()); + assert!(full_scan_update + .graph_update + .into_tx_graph() + .full_txs() + .next() + .is_none()); assert!(full_scan_update.last_active_indices.is_empty()); let full_scan_update = { let request = FullScanRequest::builder() @@ -179,6 +184,7 @@ pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> { assert_eq!( full_scan_update .graph_update + .into_tx_graph() .full_txs() .next() .unwrap() @@ -213,6 +219,7 @@ pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> { }; let txs: HashSet<_> = full_scan_update .graph_update + .into_tx_graph() .full_txs() .map(|tx| tx.txid) .collect(); @@ -227,6 +234,7 @@ pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> { }; let txs: HashSet<_> = full_scan_update .graph_update + .into_tx_graph() .full_txs() .map(|tx| tx.txid) .collect(); diff --git a/crates/wallet/src/wallet/mod.rs b/crates/wallet/src/wallet/mod.rs index f98b16e911..a8f4519434 100644 --- a/crates/wallet/src/wallet/mod.rs +++ b/crates/wallet/src/wallet/mod.rs @@ -153,7 +153,7 @@ impl From> for Update { fn from(value: FullScanResult) -> Self { Self { last_active_indices: value.last_active_indices, - graph: value.graph_update, + graph: value.graph_update.into_tx_graph(), chain: value.chain_update, } } @@ -163,7 +163,7 @@ impl From for Update { fn from(value: SyncResult) -> Self { Self { last_active_indices: BTreeMap::new(), - graph: value.graph_update, + graph: value.graph_update.into_tx_graph(), chain: value.chain_update, } } diff --git a/example-crates/example_electrum/src/main.rs b/example-crates/example_electrum/src/main.rs index 49608fbf15..09e4e88b22 100644 --- a/example-crates/example_electrum/src/main.rs +++ b/example-crates/example_electrum/src/main.rs @@ -5,7 +5,7 @@ use bdk_chain::{ collections::BTreeSet, indexed_tx_graph, spk_client::{FullScanRequest, SyncRequest}, - ConfirmationBlockTime, Merge, + ConfirmationBlockTime, Merge, TxGraph, }; use bdk_electrum::{ electrum_client::{self, Client, ElectrumApi}, @@ -127,9 +127,9 @@ fn main() -> anyhow::Result<()> { let client = BdkElectrumClient::new(electrum_cmd.electrum_args().client(network)?); // Tell the electrum client about the txs we've already got locally so it doesn't re-download them - client.populate_tx_cache(&*graph.lock().unwrap()); + client.populate_tx_cache(graph.lock().unwrap().graph().clone()); - let (chain_update, mut graph_update, keychain_update) = match electrum_cmd.clone() { + let (chain_update, graph_update, keychain_update) = match electrum_cmd.clone() { ElectrumCommands::Scan { stop_gap, scan_options, @@ -248,11 +248,13 @@ fn main() -> anyhow::Result<()> { } }; + let mut tx_graph: TxGraph = graph_update.into_tx_graph(); + let now = std::time::UNIX_EPOCH .elapsed() .expect("must get time") .as_secs(); - let _ = graph_update.update_last_seen_unconfirmed(now); + let _ = tx_graph.update_last_seen_unconfirmed(now); let db_changeset = { let mut chain = chain.lock().unwrap(); @@ -266,7 +268,7 @@ fn main() -> anyhow::Result<()> { let keychain_changeset = graph.index.reveal_to_target_multi(&keychain_update); indexed_tx_graph_changeset.merge(keychain_changeset.into()); } - indexed_tx_graph_changeset.merge(graph.apply_update(graph_update)); + indexed_tx_graph_changeset.merge(graph.apply_update(tx_graph)); ChangeSet { local_chain: chain_changeset, diff --git a/example-crates/example_esplora/src/main.rs b/example-crates/example_esplora/src/main.rs index b07a6697d9..cfcb7deacc 100644 --- a/example-crates/example_esplora/src/main.rs +++ b/example-crates/example_esplora/src/main.rs @@ -166,13 +166,14 @@ fn main() -> anyhow::Result<()> { // is reached. It returns a `TxGraph` update (`graph_update`) and a structure that // represents the last active spk derivation indices of keychains // (`keychain_indices_update`). - let mut update = client + let update = client .full_scan(request, *stop_gap, scan_options.parallel_requests) .context("scanning for transactions")?; + let mut tx_graph = update.graph_update.into_tx_graph(); // We want to keep track of the latest time a transaction was seen unconfirmed. let now = std::time::UNIX_EPOCH.elapsed().unwrap().as_secs(); - let _ = update.graph_update.update_last_seen_unconfirmed(now); + let _ = tx_graph.update_last_seen_unconfirmed(now); let mut graph = graph.lock().expect("mutex must not be poisoned"); let mut chain = chain.lock().expect("mutex must not be poisoned"); @@ -186,7 +187,7 @@ fn main() -> anyhow::Result<()> { let index_changeset = graph .index .reveal_to_target_multi(&update.last_active_indices); - let mut indexed_tx_graph_changeset = graph.apply_update(update.graph_update); + let mut indexed_tx_graph_changeset = graph.apply_update(tx_graph); indexed_tx_graph_changeset.merge(index_changeset.into()); indexed_tx_graph_changeset }, @@ -265,18 +266,19 @@ fn main() -> anyhow::Result<()> { } } - let mut update = client.sync(request, scan_options.parallel_requests)?; + let update = client.sync(request, scan_options.parallel_requests)?; + let mut tx_graph = update.graph_update.into_tx_graph(); // Update last seen unconfirmed let now = std::time::UNIX_EPOCH.elapsed().unwrap().as_secs(); - let _ = update.graph_update.update_last_seen_unconfirmed(now); + let _ = tx_graph.update_last_seen_unconfirmed(now); ( chain .lock() .unwrap() .apply_update(update.chain_update.expect("request has chain tip"))?, - graph.lock().unwrap().apply_update(update.graph_update), + graph.lock().unwrap().apply_update(tx_graph), ) } }; diff --git a/example-crates/wallet_electrum/src/main.rs b/example-crates/wallet_electrum/src/main.rs index f4596ce18c..e900be1fd2 100644 --- a/example-crates/wallet_electrum/src/main.rs +++ b/example-crates/wallet_electrum/src/main.rs @@ -50,7 +50,7 @@ fn main() -> Result<(), anyhow::Error> { // Populate the electrum client's transaction cache so it doesn't redownload transaction we // already have. - client.populate_tx_cache(wallet.tx_graph()); + client.populate_tx_cache(wallet.tx_graph().clone()); let request = wallet.start_full_scan().inspect({ let mut stdout = std::io::stdout(); @@ -65,9 +65,11 @@ fn main() -> Result<(), anyhow::Error> { }); let mut update = client.full_scan(request, STOP_GAP, BATCH_SIZE, false)?; + let mut tx_graph = update.graph_update.into_tx_graph(); let now = std::time::UNIX_EPOCH.elapsed().unwrap().as_secs(); - let _ = update.graph_update.update_last_seen_unconfirmed(now); + let _ = tx_graph.update_last_seen_unconfirmed(now); + update.graph_update = tx_graph.into(); println!(); diff --git a/example-crates/wallet_esplora_async/src/main.rs b/example-crates/wallet_esplora_async/src/main.rs index f81f8101ca..63589da4d7 100644 --- a/example-crates/wallet_esplora_async/src/main.rs +++ b/example-crates/wallet_esplora_async/src/main.rs @@ -60,8 +60,10 @@ async fn main() -> Result<(), anyhow::Error> { let mut update = client .full_scan(request, STOP_GAP, PARALLEL_REQUESTS) .await?; + let mut tx_graph = update.graph_update.clone().into_tx_graph(); let now = std::time::UNIX_EPOCH.elapsed().unwrap().as_secs(); - let _ = update.graph_update.update_last_seen_unconfirmed(now); + let _ = tx_graph.update_last_seen_unconfirmed(now); + update.graph_update = tx_graph.into(); wallet.apply_update(update)?; wallet.persist(&mut conn)?; diff --git a/example-crates/wallet_esplora_blocking/src/main.rs b/example-crates/wallet_esplora_blocking/src/main.rs index bec3956114..35b24eb14f 100644 --- a/example-crates/wallet_esplora_blocking/src/main.rs +++ b/example-crates/wallet_esplora_blocking/src/main.rs @@ -60,8 +60,10 @@ fn main() -> Result<(), anyhow::Error> { }); let mut update = client.full_scan(request, STOP_GAP, PARALLEL_REQUESTS)?; + let mut tx_graph = update.graph_update.clone().into_tx_graph(); let now = std::time::UNIX_EPOCH.elapsed().unwrap().as_secs(); - let _ = update.graph_update.update_last_seen_unconfirmed(now); + let _ = tx_graph.update_last_seen_unconfirmed(now); + update.graph_update = tx_graph.into(); wallet.apply_update(update)?; if let Some(changeset) = wallet.take_staged() {