feat(electrum): batched Headers and script_get_history

Author: LagginTimes

crates/electrum/benches/test_sync.rs

@@ -56,7 +56,15 @@ where
 
 pub fn test_sync_performance(c: &mut Criterion) {
     let env = TestEnv::new().unwrap();
-    let electrum_client = electrum_client::Client::new(env.electrsd.electrum_url.as_str()).unwrap();
+
+    // Test local server:
+    // let electrum_client =
+    // electrum_client::Client::new(env.electrsd.electrum_url.as_str()).unwrap();
+
+    // Test remote server:
+    let electrum_client =
+        electrum_client::Client::new("ssl://electrum.blockstream.info:50002").unwrap();
+
     let client = BdkElectrumClient::new(electrum_client);
 
     const NUM_BLOCKS: usize = 200;

crates/electrum/src/bdk_electrum_client.rs

@@ -70,33 +70,6 @@ impl<E: ElectrumApi> BdkElectrumClient<E> {
         Ok(tx)
     }
 
-    /// Fetch block header of given `height`.
-    ///
-    /// If it hits the cache it will return the cached version and avoid making the request.
-    fn fetch_header(&self, height: u32) -> Result<Header, Error> {
-        let block_header_cache = self.block_header_cache.lock().unwrap();
-
-        if let Some(header) = block_header_cache.get(&height) {
-            return Ok(*header);
-        }
-
-        drop(block_header_cache);
-
-        self.update_header(height)
-    }
-
-    /// Update a block header at given `height`. Returns the updated header.
-    fn update_header(&self, height: u32) -> Result<Header, Error> {
-        let header = self.inner.block_header(height as usize)?;
-
-        self.block_header_cache
-            .lock()
-            .unwrap()
-            .insert(height, header);
-
-        Ok(header)
-    }
-
     /// Broadcasts a transaction to the network.
     ///
     /// This is a re-export of [`ElectrumApi::transaction_broadcast`].
@@ -359,27 +332,29 @@ impl<E: ElectrumApi> BdkElectrumClient<E> {
         outpoints: impl IntoIterator<Item = OutPoint>,
         pending_anchors: &mut Vec<(Txid, usize)>,
     ) -> Result<(), Error> {
-        for outpoint in outpoints {
-            let op_txid = outpoint.txid;
-            let op_tx = self.fetch_tx(op_txid)?;
-            let op_txout = match op_tx.output.get(outpoint.vout as usize) {
-                Some(txout) => txout,
-                None => continue,
-            };
-            debug_assert_eq!(op_tx.compute_txid(), op_txid);
+        let ops_txs = outpoints
+            .into_iter()
+            .filter_map(|op| self.fetch_tx(op.txid).ok().map(|tx| (op, tx)))
+            .collect::<Vec<_>>();
+
+        let scripts = ops_txs
+            .iter()
+            .map(|(op, tx)| tx.output[op.vout as usize].script_pubkey.as_script());
 
-            // attempt to find the following transactions (alongside their chain positions), and
-            // add to our sparsechain `update`:
+        let spk_histories = self.inner.batch_script_get_history(scripts)?;
+
+        for ((outpoint, tx), spk_history) in ops_txs.into_iter().zip(spk_histories) {
             let mut has_residing = false; // tx in which the outpoint resides
             let mut has_spending = false; // tx that spends the outpoint
-            for res in self.inner.script_get_history(&op_txout.script_pubkey)? {
+
+            for res in spk_history {
                 if has_residing && has_spending {
                     break;
                 }
 
-                if !has_residing && res.tx_hash == op_txid {
+                if !has_residing && res.tx_hash == outpoint.txid {
                     has_residing = true;
-                    tx_update.txs.push(Arc::clone(&op_tx));
+                    tx_update.txs.push(Arc::clone(&tx));
                     match res.height.try_into() {
                         // Returned heights 0 & -1 are reserved for unconfirmed txs.
                         Ok(height) if height > 0 => {
@@ -391,7 +366,7 @@ impl<E: ElectrumApi> BdkElectrumClient<E> {
                     }
                 }
 
-                if !has_spending && res.tx_hash != op_txid {
+                if !has_spending && res.tx_hash != outpoint.txid {
                     let res_tx = self.fetch_tx(res.tx_hash)?;
                     // we exclude txs/anchors that do not spend our specified outpoint(s)
                     has_spending = res_tx
@@ -426,34 +401,45 @@ impl<E: ElectrumApi> BdkElectrumClient<E> {
         txids: impl IntoIterator<Item = Txid>,
         pending_anchors: &mut Vec<(Txid, usize)>,
     ) -> Result<(), Error> {
+        let mut txs = Vec::new();
+        let mut scripts = Vec::new();
         for txid in txids {
-            let tx = match self.fetch_tx(txid) {
-                Ok(tx) => tx,
-                Err(electrum_client::Error::Protocol(_)) => continue,
-                Err(other_err) => return Err(other_err),
-            };
+            match self.fetch_tx(txid) {
+                Ok(tx) => {
+                    let spk = tx
+                        .output
+                        .first()
+                        .map(|txo| &txo.script_pubkey)
+                        .expect("tx must have an output")
+                        .clone();
+                    txs.push(tx);
+                    scripts.push(spk);
+                }
+                Err(electrum_client::Error::Protocol(_)) => {
+                    continue;
+                }
+                Err(e) => return Err(e),
+            }
+        }
 
-            let spk = tx
-                .output
-                .first()
-                .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
+        let spk_histories = self
+            .inner
+            .batch_script_get_history(scripts.iter().map(|spk| spk.as_script()))?;
 
-            // 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)?
+        for (tx, spk_history) in txs.into_iter().zip(spk_histories) {
+            if let Some(res) = spk_history
                 .into_iter()
-                .find(|r| r.tx_hash == txid)
+                .find(|res| res.tx_hash == tx.compute_txid())
             {
-                match r.height.try_into() {
+                match res.height.try_into() {
                     // Returned heights 0 & -1 are reserved for unconfirmed txs.
                     Ok(height) if height > 0 => {
-                        pending_anchors.push((txid, height));
+                        pending_anchors.push((tx.compute_txid(), height));
                     }
                     _ => {
-                        tx_update.seen_ats.insert((r.tx_hash, start_time));
+                        tx_update.seen_ats.insert((res.tx_hash, start_time));
                     }
                 }
             }
@@ -472,19 +458,37 @@ impl<E: ElectrumApi> BdkElectrumClient<E> {
         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<u32, BlockHash> = 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);
+        // Figure out which block heights we need headers for.
+        let mut needed_heights: Vec<u32> =
+            txs_with_heights.iter().map(|&(_, h)| h as u32).collect();
+        needed_heights.sort_unstable();
+        needed_heights.dedup();
+
+        // Filter out any heights already in our cache.
+        let mut cache = self.block_header_cache.lock().unwrap();
+        let missing_heights: Vec<u32> = needed_heights
+            .into_iter()
+            .filter(|h| !cache.contains_key(h))
+            .collect();
+
+        // Fetch all missing headers in one RPC call.
+        if !missing_heights.is_empty() {
+            let headers = self.inner.batch_block_header(missing_heights.clone())?;
+            for (h, hdr) in missing_heights.into_iter().zip(headers.into_iter()) {
+                cache.insert(h, hdr);
             }
         }
 
-        // Check cache.
+        let height_to_hash: HashMap<u32, BlockHash> = txs_with_heights
+            .iter()
+            .map(|&(_, h)| {
+                let h = h as u32;
+                (h, cache[&h].block_hash())
+            })
+            .collect();
+        drop(cache);
+
+        // Check our anchor cache and queue up any proofs we still need.
         {
             let anchor_cache = self.anchor_cache.lock().unwrap();
             for &(txid, height) in txs_with_heights {
@@ -505,14 +509,22 @@ impl<E: ElectrumApi> BdkElectrumClient<E> {
                 let proof = self.inner.transaction_get_merkle(&txid, height)?;
 
                 // Validate against header, retrying once on stale header.
-                let mut header = self.fetch_header(height as u32)?;
+                let mut header = {
+                    let cache = self.block_header_cache.lock().unwrap();
+                    cache[&(height as u32)]
+                };
                 let mut valid = electrum_client::utils::validate_merkle_proof(
                     &txid,
                     &header.merkle_root,
                     &proof,
                 );
                 if !valid {
-                    header = self.update_header(height as u32)?;
+                    let new_header = self.inner.block_header(height)?;
+                    self.block_header_cache
+                        .lock()
+                        .unwrap()
+                        .insert(height as u32, new_header);
+                    header = new_header;
                     valid = electrum_client::utils::validate_merkle_proof(
                         &txid,
                         &header.merkle_root,