From b72bf485898433b2285c953e7c22204cf4ebedfb Mon Sep 17 00:00:00 2001 From: dadafros Date: Sat, 18 Jul 2026 15:16:01 +0200 Subject: [PATCH 1/2] wollet: handle esplora address history paging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The esplora client read only the first page of an address's transaction history: implementations bound it (25 confirmed transactions on Blockstream esplora), so busier addresses got a silently truncated history, and the scan cache then evicted the missing transactions (txid_height_delete) — an incomplete view that re-added and deleted the same transactions on every scan. Follow the /txs/chain/{last_seen_txid} pages from the last confirmed txid until a page with no confirmed transactions arrives, the only termination condition every esplora implementation guarantees (page sizes are configurable in the mempool/electrs fork, and its first page shares one budget between mempool and confirmed entries, which can crowd confirmed transactions out entirely). A repeated cursor is rejected so a server that ignores it cannot loop the client forever. Resolves the "TODO must handle paging" in get_scripts_history_esplora. --- lwk_wollet/CHANGELOG.md | 1 + lwk_wollet/src/clients/asyncr/esplora.rs | 121 ++++++++++++++++++++--- lwk_wollet/tests/e2e.rs | 41 ++++++++ 3 files changed, 148 insertions(+), 15 deletions(-) diff --git a/lwk_wollet/CHANGELOG.md b/lwk_wollet/CHANGELOG.md index daaf67fa4..ac58e1345 100644 --- a/lwk_wollet/CHANGELOG.md +++ b/lwk_wollet/CHANGELOG.md @@ -3,6 +3,7 @@ ## Unreleased * 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. ## 0.18.0 diff --git a/lwk_wollet/src/clients/asyncr/esplora.rs b/lwk_wollet/src/clients/asyncr/esplora.rs index 288e1ae43..171c74c98 100644 --- a/lwk_wollet/src/clients/asyncr/esplora.rs +++ b/lwk_wollet/src/clients/asyncr/esplora.rs @@ -212,31 +212,99 @@ impl EsploraClient { } } + /// The txid of the oldest confirmed transaction in an esplora history + /// page (pages are sorted newest first), used as the paging cursor. + /// `None` when the page has no confirmed transactions. Unconfirmed + /// entries carry a non-positive height (see + /// [`crate::clients::History::height`]). + fn last_confirmed_txid(page: &[History]) -> Option { + page.iter().rev().find(|h| h.height > 0).map(|h| h.txid) + } + async fn get_scripts_history_esplora( &self, addresses: &[Address], ) -> Result>, Error> { let mut result = vec![]; for address in addresses.iter() { - let url = format!("{}/address/{}/txs", self.base_url, address); - // TODO must handle paging -> https://github.com/blockstream/esplora/blob/master/API.md#addresses - let response = self.get_with_retry(&url).await?; - - // TODO going through string and then json is not as efficient as it could be but we prioritize debugging for now - let text = response.text().await?; - let json: Vec = match serde_json::from_str(&text) { - Ok(e) => e, - Err(e) => { - log::warn!("error {e:?} in converting following text:\n{text}"); - return Err(e.into()); - } - }; - let history: Vec = json.into_iter().map(Into::into).collect(); - result.push(history) + result.push(self.get_address_history(address).await?); } Ok(result) } + /// Fetch the confirmed transaction history of a single address plus its + /// first page of mempool transactions, following esplora's + /// confirmed-transactions paging. + /// + /// The first page returns mempool transactions plus the newest confirmed + /// ones, with implementation-dependent page sizes (Blockstream esplora + /// returns up to 25 confirmed plus up to 50 mempool; other + /// implementations share a single budget between the two). Older + /// confirmed transactions require paging via + /// `/txs/chain/{last_seen_txid}`, so the loop keeps requesting from the + /// last confirmed txid until a page with no confirmed transactions + /// arrives — the only termination condition every implementation + /// guarantees. Without the paging, an address with more confirmed + /// transactions than one page holds gets a silently truncated history. + /// Mempool transactions beyond esplora's first-page window are still + /// not returned (the API offers no mempool paging). + /// + async fn get_address_history(&self, address: &Address) -> Result, Error> { + let url = format!("{}/address/{}/txs", self.base_url, address); + let mut history = self.get_history_page(&url).await?; + + let mut cursor = match Self::last_confirmed_txid(&history) { + Some(txid) => Some(txid), + None if !history.is_empty() => { + // The first page contains transactions but none confirmed: + // implementations sharing one first-page budget can fill it + // entirely with mempool entries, so ask for the newest + // confirmed page explicitly before concluding there is no + // confirmed history. + let url = format!("{}/address/{}/txs/chain", self.base_url, address); + let page = self.get_history_page(&url).await?; + let cursor = Self::last_confirmed_txid(&page); + history.extend(page); + cursor + } + None => None, + }; + + // Guards the loop against a server that ignores the cursor or + // cycles pages: a repeated cursor means no progress was made. + let mut seen_cursors = HashSet::new(); + while let Some(last_seen) = cursor { + if !seen_cursors.insert(last_seen) { + return Err(Error::Generic(format!( + "address history paging did not advance past txid {last_seen}" + ))); + } + let url = format!( + "{}/address/{}/txs/chain/{}", + self.base_url, address, last_seen + ); + let page = self.get_history_page(&url).await?; + cursor = Self::last_confirmed_txid(&page); + history.extend(page); + } + Ok(history) + } + + async fn get_history_page(&self, url: &str) -> Result, Error> { + let response = self.get_with_retry(url).await?; + + // TODO going through string and then json is not as efficient as it could be but we prioritize debugging for now + let text = response.text().await?; + let json: Vec = match serde_json::from_str(&text) { + Ok(e) => e, + Err(e) => { + log::warn!("error {e:?} in converting following text:\n{text}"); + return Err(e.into()); + } + }; + Ok(json.into_iter().map(Into::into).collect()) + } + async fn get_scripts_history_waterfalls( &self, addresses: &[Address], @@ -1321,6 +1389,29 @@ mod tests { elements::Block::consensus_decode(&response.bytes().await.unwrap()[..]).unwrap() } + #[test] + fn test_last_confirmed_txid() { + use elements::Txid; + let txid = |n: u8| Txid::from_str(&format!("{n:064x}")).unwrap(); + let h = |height: i32, n: u8| History { + txid: txid(n), + height, + block_hash: None, + block_timestamp: None, + v: 0, + }; + // Pages are sorted newest first: the cursor must be the OLDEST + // confirmed entry, skipping unconfirmed ones (height -1/0) anywhere + // in the page. + let page = [h(-1, 1), h(0, 2), h(100, 3), h(90, 4), h(-1, 5)]; + assert_eq!(EsploraClient::last_confirmed_txid(&page), Some(txid(4))); + assert_eq!( + EsploraClient::last_confirmed_txid(&[h(-1, 1), h(0, 2)]), + None + ); + assert_eq!(EsploraClient::last_confirmed_txid(&[]), None); + } + #[ignore = "Should be integration test, but it is testing private function"] #[tokio::test] async fn esplora_wasm_local() { diff --git a/lwk_wollet/tests/e2e.rs b/lwk_wollet/tests/e2e.rs index 579cecbec..1eef389a2 100644 --- a/lwk_wollet/tests/e2e.rs +++ b/lwk_wollet/tests/e2e.rs @@ -1037,6 +1037,47 @@ async fn wait_update_with_txs( panic!("update didn't arrive"); } +#[cfg(feature = "esplora")] +#[tokio::test] +async fn test_esplora_address_history_paging() { + // Esplora returns a bounded first page of transaction history per + // 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. + 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 signer = generate_signer(); + let view_key = generate_view_key(); + let descriptor = format!("ct({},elwpkh({}/*))", view_key, signer.xpub()); + let descriptor: WolletDescriptor = descriptor.parse().unwrap(); + let mut wollet = WolletBuilder::new(network, descriptor).build().unwrap(); + + let address = wollet.address(None).unwrap(); + const TXS: usize = 30; // more than one 25-tx confirmed page + for i in 0..TXS { + env.elementsd_sendtoaddress(address.address(), 10_000, None); + // Confirm in batches so the sends never hit the node's unconfirmed + // descendant-chain limit (25). + if i % 10 == 9 { + env.elementsd_generate(1); + } + } + + for _ in 0..50 { + if let Some(update) = client.full_scan(&wollet).await.unwrap() { + wollet.apply_update(update).unwrap(); + } + if wollet.transactions().unwrap().len() == TXS { + break; + } + std::thread::sleep(std::time::Duration::from_millis(200)); + } + assert_eq!(wollet.transactions().unwrap().len(), TXS); +} + #[cfg(feature = "esplora")] #[tokio::test] async fn test_esplora_requests_counter() { From b3df2c7d8801c660ed997cb880616144deab2245 Mon Sep 17 00:00:00 2001 From: dadafros Date: Sat, 18 Jul 2026 15:16:37 +0200 Subject: [PATCH 2/2] 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. The per-address futures are instantiated eagerly (they stay inert until polled; buffered still caps concurrent polling): holding an &Address-borrowing closure inside the stream trips rustc's higher-ranked FnOnce limitation (rust-lang/rust#89976) as soon as a downstream async_trait consumer such as lwk_boltz must prove the resulting future Send. Default concurrency stays 1, so behavior is unchanged unless opted in. --- lwk_wollet/CHANGELOG.md | 1 + lwk_wollet/src/clients/asyncr/esplora.rs | 24 ++++++++++++++++++------ lwk_wollet/tests/e2e.rs | 8 ++++++-- 3 files changed, 25 insertions(+), 8 deletions(-) diff --git a/lwk_wollet/CHANGELOG.md b/lwk_wollet/CHANGELOG.md index ac58e1345..7d644d529 100644 --- a/lwk_wollet/CHANGELOG.md +++ b/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 diff --git a/lwk_wollet/src/clients/asyncr/esplora.rs b/lwk_wollet/src/clients/asyncr/esplora.rs index 171c74c98..fe7048857 100644 --- a/lwk_wollet/src/clients/asyncr/esplora.rs +++ b/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,23 @@ 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`) 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. + // + // The futures are instantiated eagerly (they stay inert until + // polled; `buffered` still caps concurrent polling) so the stream + // holds concrete futures rather than an `&Address`-borrowing + // closure — the closure form trips rustc's higher-ranked `FnOnce` + // limitation (rust-lang/rust#89976) when downstream `async_trait` + // users such as lwk_boltz must prove the resulting future `Send`. + let futures: Vec<_> = addresses + .iter() + .map(|address| self.get_address_history(address)) + .collect(); + iter(futures).buffered(self.concurrency).try_collect().await } /// Fetch the confirmed transaction history of a single address plus its diff --git a/lwk_wollet/tests/e2e.rs b/lwk_wollet/tests/e2e.rs index 1eef389a2..6b76c340b 100644 --- a/lwk_wollet/tests/e2e.rs +++ b/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());