diff --git a/crates/electrum/src/bdk_electrum_client.rs b/crates/electrum/src/bdk_electrum_client.rs index fb387bb39..f8fddbc6d 100644 --- a/crates/electrum/src/bdk_electrum_client.rs +++ b/crates/electrum/src/bdk_electrum_client.rs @@ -6,12 +6,15 @@ use bdk_core::{ }, BlockId, CheckPoint, ConfirmationBlockTime, TxUpdate, }; -use electrum_client::{ElectrumApi, Error, HeaderNotification}; +use electrum_client::{ElectrumApi, Error, GetMerkleRes, HeaderNotification}; use std::sync::{Arc, Mutex}; /// We include a chain suffix of a certain length for the purpose of robustness. const CHAIN_SUFFIX_LENGTH: u32 = 8; +/// Maximum batch size for Merkle proof requests +const MAX_MERKLE_BATCH_SIZE: usize = 100; + /// Wrapper around an [`electrum_client::ElectrumApi`] which includes an internal in-memory /// transaction cache to avoid re-fetching already downloaded transactions. #[derive(Debug)] @@ -22,6 +25,8 @@ pub struct BdkElectrumClient { tx_cache: Mutex>>, /// The header cache block_header_cache: Mutex>, + /// The Merkle proof cache + merkle_cache: Mutex>, } impl BdkElectrumClient { @@ -31,6 +36,7 @@ impl BdkElectrumClient { inner: client, tx_cache: Default::default(), block_header_cache: Default::default(), + merkle_cache: Default::default(), } } @@ -254,13 +260,14 @@ impl BdkElectrumClient { ) -> Result, Error> { let mut unused_spk_count = 0_usize; let mut last_active_index = Option::::None; + let mut txs_to_validate = Vec::new(); loop { let spks = (0..batch_size) .map_while(|_| spks_with_expected_txids.next()) .collect::>(); if spks.is_empty() { - return Ok(last_active_index); + break; } let spk_histories = self @@ -271,7 +278,7 @@ impl BdkElectrumClient { if spk_history.is_empty() { unused_spk_count = unused_spk_count.saturating_add(1); if unused_spk_count >= stop_gap { - return Ok(last_active_index); + break; } } else { last_active_index = Some(spk_index); @@ -294,7 +301,7 @@ impl BdkElectrumClient { match tx_res.height.try_into() { // Returned heights 0 & -1 are reserved for unconfirmed txs. Ok(height) if height > 0 => { - self.validate_merkle_for_anchor(tx_update, tx_res.tx_hash, height)?; + txs_to_validate.push((tx_res.tx_hash, height)); } _ => { tx_update.seen_ats.insert((tx_res.tx_hash, start_time)); @@ -302,7 +309,19 @@ impl BdkElectrumClient { } } } + + if unused_spk_count >= stop_gap { + break; + } + } + + // Batch validate all collected transactions. + if !txs_to_validate.is_empty() { + let proofs = self.batch_fetch_merkle_proofs(&txs_to_validate)?; + self.batch_validate_merkle_proofs(tx_update, proofs)?; } + + Ok(last_active_index) } /// Populate the `tx_update` with associated transactions/anchors of `outpoints`. @@ -315,6 +334,8 @@ impl BdkElectrumClient { tx_update: &mut TxUpdate, outpoints: impl IntoIterator, ) -> Result<(), Error> { + let mut txs_to_validate = Vec::new(); + for outpoint in outpoints { let op_txid = outpoint.txid; let op_tx = self.fetch_tx(op_txid)?; @@ -324,7 +345,7 @@ impl BdkElectrumClient { }; debug_assert_eq!(op_tx.compute_txid(), op_txid); - // attempt to find the following transactions (alongside their chain positions), and + // Attempt to find the following transactions (alongside their chain positions), and // add to our sparsechain `update`: let mut has_residing = false; // tx in which the outpoint resides let mut has_spending = false; // tx that spends the outpoint @@ -339,7 +360,7 @@ impl BdkElectrumClient { match res.height.try_into() { // Returned heights 0 & -1 are reserved for unconfirmed txs. Ok(height) if height > 0 => { - self.validate_merkle_for_anchor(tx_update, res.tx_hash, height)?; + txs_to_validate.push((res.tx_hash, height)); } _ => { tx_update.seen_ats.insert((res.tx_hash, start_time)); @@ -349,7 +370,7 @@ impl BdkElectrumClient { if !has_spending && res.tx_hash != op_txid { let res_tx = self.fetch_tx(res.tx_hash)?; - // we exclude txs/anchors that do not spend our specified outpoint(s) + // We exclude txs/anchors that do not spend our specified outpoint(s). has_spending = res_tx .input .iter() @@ -361,7 +382,7 @@ impl BdkElectrumClient { match res.height.try_into() { // Returned heights 0 & -1 are reserved for unconfirmed txs. Ok(height) if height > 0 => { - self.validate_merkle_for_anchor(tx_update, res.tx_hash, height)?; + txs_to_validate.push((res.tx_hash, height)); } _ => { tx_update.seen_ats.insert((res.tx_hash, start_time)); @@ -370,6 +391,13 @@ impl BdkElectrumClient { } } } + + // Batch validate all collected transactions. + if !txs_to_validate.is_empty() { + let proofs = self.batch_fetch_merkle_proofs(&txs_to_validate)?; + self.batch_validate_merkle_proofs(tx_update, proofs)?; + } + Ok(()) } @@ -380,6 +408,8 @@ impl BdkElectrumClient { tx_update: &mut TxUpdate, txids: impl IntoIterator, ) -> Result<(), Error> { + let mut txs_to_validate = Vec::new(); + for txid in txids { let tx = match self.fetch_tx(txid) { Ok(tx) => tx, @@ -393,8 +423,8 @@ impl BdkElectrumClient { .map(|txo| &txo.script_pubkey) .expect("tx must have an output"); - // because of restrictions of the Electrum API, we have to use the `script_get_history` - // call to get confirmation status of our transaction + // Because of restrictions of the Electrum API, we have to use the `script_get_history` + // call to get confirmation status of our transaction. if let Some(r) = self .inner .script_get_history(spk)? @@ -404,7 +434,7 @@ impl BdkElectrumClient { match r.height.try_into() { // Returned heights 0 & -1 are reserved for unconfirmed txs. Ok(height) if height > 0 => { - self.validate_merkle_for_anchor(tx_update, txid, height)?; + txs_to_validate.push((txid, height)); } _ => { tx_update.seen_ats.insert((r.tx_hash, start_time)); @@ -414,45 +444,114 @@ impl BdkElectrumClient { tx_update.txs.push(tx); } + + // Batch validate all collected transactions. + if !txs_to_validate.is_empty() { + let proofs = self.batch_fetch_merkle_proofs(&txs_to_validate)?; + self.batch_validate_merkle_proofs(tx_update, proofs)?; + } + Ok(()) } - // Helper function which checks if a transaction is confirmed by validating the merkle proof. - // An anchor is inserted if the transaction is validated to be in a confirmed block. - fn validate_merkle_for_anchor( + /// Batch fetch Merkle proofs for multiple transactions. + fn batch_fetch_merkle_proofs( + &self, + txs_with_heights: &[(Txid, usize)], + ) -> Result, Error> { + // Evict proofs whose block hash is no longer on our current chain. + self.clear_stale_proofs()?; + + let mut results = Vec::with_capacity(txs_with_heights.len()); + let mut to_fetch = Vec::new(); + + // Build a map for height to block hash conversions. This is for obtaining block hash data + // with minimum `fetch_header` calls. + let mut height_to_hash: HashMap = HashMap::new(); + for &(_, height) in txs_with_heights { + let h = height as u32; + if !height_to_hash.contains_key(&h) { + // Try to obtain hash from the header cache, or fetch the header if absent. + let hash = self.fetch_header(h)?.block_hash(); + height_to_hash.insert(h, hash); + } + } + + // Check cache. + { + let merkle_cache = self.merkle_cache.lock().unwrap(); + for &(txid, height) in txs_with_heights { + let h = height as u32; + let hash = height_to_hash[&h]; + if let Some(proof) = merkle_cache.get(&(txid, hash)) { + results.push((txid, proof.clone())); + } else { + to_fetch.push((txid, height, hash)); + } + } + } + + // Fetch missing proofs in batches. + for chunk in to_fetch.chunks(MAX_MERKLE_BATCH_SIZE) { + for &(txid, height, hash) in chunk { + let merkle_res = self.inner.transaction_get_merkle(&txid, height)?; + self.merkle_cache + .lock() + .unwrap() + .insert((txid, hash), merkle_res.clone()); + results.push((txid, merkle_res)); + } + } + + Ok(results) + } + + /// Batch validate Merkle proofs. + fn batch_validate_merkle_proofs( &self, tx_update: &mut TxUpdate, - txid: Txid, - confirmation_height: usize, + proofs: Vec<(Txid, GetMerkleRes)>, ) -> Result<(), Error> { - if let Ok(merkle_res) = self - .inner - .transaction_get_merkle(&txid, confirmation_height) - { - let mut header = self.fetch_header(merkle_res.block_height as u32)?; + // Pre-fetch all required headers. + let heights: HashSet = proofs + .iter() + .map(|(_, proof)| proof.block_height as u32) + .collect(); + + let mut headers = HashMap::new(); + for height in heights { + headers.insert(height, self.fetch_header(height)?); + } + + // Validate proofs. + for (txid, merkle_res) in proofs { + let height = merkle_res.block_height as u32; + let header = headers.get(&height).unwrap(); + let mut is_confirmed_tx = electrum_client::utils::validate_merkle_proof( &txid, &header.merkle_root, &merkle_res, ); - // Merkle validation will fail if the header in `block_header_cache` is outdated, so we - // want to check if there is a new header and validate against the new one. + // Retry with updated header if validation fails. if !is_confirmed_tx { - header = self.update_header(merkle_res.block_height as u32)?; + let updated_header = self.update_header(height)?; + headers.insert(height, updated_header); is_confirmed_tx = electrum_client::utils::validate_merkle_proof( &txid, - &header.merkle_root, + &updated_header.merkle_root, &merkle_res, ); } if is_confirmed_tx { + let header = headers.get(&height).unwrap(); tx_update.anchors.insert(( ConfirmationBlockTime { confirmation_time: header.time as u64, block_id: BlockId { - height: merkle_res.block_height as u32, + height, hash: header.block_hash(), }, }, @@ -460,9 +559,51 @@ impl BdkElectrumClient { )); } } + Ok(()) } + /// Remove any proofs for blocks that may have been re-orged out. + /// + /// Checks if the latest cached block hash matches the current chain tip. If not, evicts proofs + /// for blocks that were re-orged out, stopping at the fork point. + fn clear_stale_proofs(&self) -> Result<(), Error> { + let mut cache = self.merkle_cache.lock().unwrap(); + + // Collect one (height, old_hash) pair per proof. + let mut entries: Vec<(u32, BlockHash)> = cache + .iter() + .map(|((_, old_hash), res)| (res.block_height as u32, *old_hash)) + .collect(); + + // Sort descending and dedup so we only check each height once. + entries.sort_unstable_by(|a, b| b.0.cmp(&a.0)); + entries.dedup(); + + // Evict any stale proofs until fork point is found. + for (height, old_hash) in entries { + let current_hash = self.fetch_header(height)?.block_hash(); + if current_hash == old_hash { + break; + } + cache.retain(|&(_txid, bh), _| bh != old_hash); + } + Ok(()) + } + + /// Validate the Merkle proof for a single transaction using the batch APIs. + #[allow(dead_code)] + fn validate_merkle_for_anchor( + &self, + tx_update: &mut TxUpdate, + txid: Txid, + confirmation_height: usize, + ) -> Result<(), Error> { + // Use the batch processing functions even for single tx. + let proofs = self.batch_fetch_merkle_proofs(&[(txid, confirmation_height)])?; + self.batch_validate_merkle_proofs(tx_update, proofs) + } + // Helper function which fetches the `TxOut`s of our relevant transactions' previous transactions, // which we do not have by default. This data is needed to calculate the transaction fee. fn fetch_prev_txout( diff --git a/crates/electrum/tests/test_electrum.rs b/crates/electrum/tests/test_electrum.rs index 3c1d11803..cc1672443 100644 --- a/crates/electrum/tests/test_electrum.rs +++ b/crates/electrum/tests/test_electrum.rs @@ -19,6 +19,7 @@ use core::time::Duration; use electrum_client::ElectrumApi; use std::collections::{BTreeSet, HashMap, HashSet}; use std::str::FromStr; +use std::time::Instant; // Batch size for `sync_with_electrum`. const BATCH_SIZE: usize = 5; @@ -876,3 +877,51 @@ fn test_check_fee_calculation() -> anyhow::Result<()> { } Ok(()) } + +#[test] +pub fn test_sync_performance() -> anyhow::Result<()> { + const EXPECTED_MAX_SYNC_TIME: Duration = Duration::from_secs(15); + const NUM_ADDRESSES: usize = 1000; + + let env = TestEnv::new()?; + let electrum_client = electrum_client::Client::new(env.electrsd.electrum_url.as_str())?; + let client = BdkElectrumClient::new(electrum_client); + + // Generate test addresses + let mut spks = Vec::with_capacity(NUM_ADDRESSES); + for _ in 0..NUM_ADDRESSES { + spks.push(get_test_spk()); + } + + // Mine some blocks and send transactions + env.mine_blocks(101, None)?; + for spk in spks.iter().take(10) { + let addr = Address::from_script(spk, Network::Regtest)?; + env.send(&addr, Amount::from_sat(10_000))?; + } + env.mine_blocks(1, None)?; + + // Setup receiver + let (mut recv_chain, _) = LocalChain::from_genesis_hash(env.bitcoind.client.get_block_hash(0)?); + let mut recv_graph = IndexedTxGraph::::new({ + let mut recv_index = SpkTxOutIndex::default(); + for spk in spks.iter() { + recv_index.insert_spk((), spk.clone()); + } + recv_index + }); + + // Measure sync time + let start = Instant::now(); + let _ = sync_with_electrum(&client, spks.clone(), &mut recv_chain, &mut recv_graph)?; + let sync_duration = start.elapsed(); + + assert!( + sync_duration <= EXPECTED_MAX_SYNC_TIME, + "Sync took {:?}, which is longer than expected {:?}", + sync_duration, + EXPECTED_MAX_SYNC_TIME + ); + + Ok(()) +}