wollet: fetch address histories concurrently

The address walk of the esplora full scan awaited one history request
per address sequentially, so wall-clock grew linearly with the number
of scanned addresses even though EsploraClientBuilder::concurrency
already parallelizes transaction and header downloads. Drive the same
batch through an ordered buffered stream honoring the configured
concurrency: buffered (not buffer_unordered) because callers map
results back to derivation indices positionally, and try_collect to
stop at the first error like the sequential loop did.

Default concurrency stays 1, so behavior is unchanged unless opted in.

Author: dadafros

lwk_wollet/CHANGELOG.md

@@ -4,6 +4,7 @@
 
 * Add Waterfalls descriptor subscriptions, returning `tip`, `mempool`, `block`, and `reorg` events that callers can use as wallet rescan hints.
 * Esplora client: address transaction history now follows esplora's confirmed-transactions paging — previously an address with more confirmed transactions than one esplora page (25 on Blockstream esplora) got a silently truncated history, which could also evict the missing transactions from the scan cache.
+* Esplora client: address history requests within a scan batch run concurrently, honoring `EsploraClientBuilder::concurrency` (default 1, so behavior is unchanged unless opted in).
 
 ## 0.18.0
 

lwk_wollet/src/clients/asyncr/esplora.rs

@@ -27,7 +27,7 @@ use futures::lock::Mutex;
 #[cfg(not(target_arch = "wasm32"))]
 use tokio::sync::Mutex;
 
-use futures::stream::{iter, StreamExt};
+use futures::stream::{iter, StreamExt, TryStreamExt};
 use reqwest::{Response, StatusCode};
 use serde::Deserialize;
 use std::sync::atomic::AtomicUsize;
@@ -225,11 +225,16 @@ impl EsploraClient {
         &self,
         addresses: &[Address],
     ) -> Result<Vec<Vec<History>>, Error> {
-        let mut result = vec![];
-        for address in addresses.iter() {
-            result.push(self.get_address_history(address).await?);
-        }
-        Ok(result)
+        // `buffered` (not `buffer_unordered`) so results keep the input
+        // order: callers map histories back to derivation indices
+        // positionally (see `get_history`). `try_collect` stops at the
+        // first error, dropping in-flight requests, matching the failure
+        // semantics of the previous sequential loop.
+        iter(addresses.iter())
+            .map(|address| self.get_address_history(address))
+            .buffered(self.concurrency)
+            .try_collect()
+            .await
     }
 
     /// Fetch the confirmed transaction history of a single address plus its

lwk_wollet/tests/e2e.rs

@@ -1044,11 +1044,15 @@ async fn test_esplora_address_history_paging() {
     // address (25 confirmed on Blockstream esplora); older confirmed
     // transactions must be fetched via `/txs/chain/{last_seen_txid}` pages.
     // Regression test: an address with more transactions than one page must
-    // sync its full history, not a silently truncated one.
+    // sync its full history, not a silently truncated one. concurrency(4)
+    // also exercises the ordered `buffered` address walk.
     let env = TestEnvBuilder::from_env().with_esplora().build();
     let url = env.esplora_url();
     let network = Network::default_regtest();
-    let mut client = clients::asyncr::EsploraClient::new(network, &url);
+    let mut client = clients::asyncr::EsploraClientBuilder::new(&url, network)
+        .concurrency(4)
+        .build()
+        .unwrap();
     let signer = generate_signer();
     let view_key = generate_view_key();
     let descriptor = format!("ct({},elwpkh({}/*))", view_key, signer.xpub());