diff --git a/lwk_wollet/CHANGELOG.md b/lwk_wollet/CHANGELOG.md index d839df988..6fda4fcd0 100644 --- a/lwk_wollet/CHANGELOG.md +++ b/lwk_wollet/CHANGELOG.md @@ -6,6 +6,7 @@ * Add the `electrum_oidc` feature: `TokenProvider::Blockstream` support for `ElectrumClient` (automatic OAuth2 token fetch, plus invalidate and retry once when the server denies a call with an authentication error). Not available on wasm. * `Wollet::assets_owned()` returns all assets ever owned instead of only unspent ones. * `Contract` no longer contains public fields. To build a `Contract` for issuance, use `Contract::builder`. +* 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 diff --git a/lwk_wollet/src/clients/asyncr/esplora.rs b/lwk_wollet/src/clients/asyncr/esplora.rs index 27dfeb09e..caf169a10 100644 --- a/lwk_wollet/src/clients/asyncr/esplora.rs +++ b/lwk_wollet/src/clients/asyncr/esplora.rs @@ -28,7 +28,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; @@ -217,11 +217,16 @@ impl EsploraClient { &self, addresses: &[Address], ) -> Result>, Error> { - let mut result = vec![]; - for address in addresses.iter() { - result.push(self.get_address_history(address).await?); - } - Ok(result) + // `buffered` (not `buffer_unordered`): callers map results back to + // derivation indices positionally (see `get_history`). + // Futures are created eagerly (inert until polled): an `&Address`- + // borrowing closure in the stream hits rust-lang/rust#89976 when + // async_trait consumers (e.g. lwk_boltz) need the future to be `Send`. + let futures: Vec<_> = addresses + .iter() + .map(|address| self.get_address_history(address)) + .collect(); + iter(futures).buffered(self.concurrency).try_collect().await } /// Fetch an address' unconfirmed transactions plus its full confirmed history. diff --git a/lwk_wollet/tests/e2e.rs b/lwk_wollet/tests/e2e.rs index a0ad9101a..db7b37a34 100644 --- a/lwk_wollet/tests/e2e.rs +++ b/lwk_wollet/tests/e2e.rs @@ -1077,7 +1077,11 @@ async fn test_esplora_address_history_paging() { first_page.len() ); - let mut client = clients::asyncr::EsploraClient::new(network, &url); + // concurrency(4) also exercises the ordered concurrent address walk. + let mut client = clients::asyncr::EsploraClientBuilder::new(&url, network) + .concurrency(4) + .build() + .unwrap(); for _ in 0..50 { if let Some(update) = client.full_scan(&wollet).await.unwrap() { wollet.apply_update(update).unwrap();