Skip to content

esplora: fetch address histories concurrently #37

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
albertodeluigi wants to merge 1 commit into
base: sequentia
Choose a base branch
from
Open

esplora: fetch address histories concurrently #37

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 28 additions & 18 deletions lwk_wollet/src/clients/asyncr/esplora.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,25 +211,35 @@ impl EsploraClient {
&self,
addresses: &[Address],
) -> Result<Vec<Vec<History>>, 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?;
// SEQUENTIA: up to `concurrency` address histories in flight at once, as get_transactions
// and get_headers already do. Upstream walks the batch one request after another, so a
// first scan (every used address plus the 20-address gap, per chain) costs one round trip
// per address: ~44 sequential requests, about 7 s for a fresh testnet wallet behind a
// 120 ms round trip, whatever concurrency the caller asked for. `buffered` (not
// `buffer_unordered`) keeps the results in address order, which get_history relies on to
// find the last used index.
let stream = iter(addresses.iter())
.map(|address| async move {
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<EsploraTx> = 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<History> = json.into_iter().map(Into::into).collect();
result.push(history)
}
Ok(result)
// 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<EsploraTx> = 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<History> = json.into_iter().map(Into::into).collect();
Ok::<Vec<History>, Error>(history)
})
.buffered(self.concurrency);

let results: Vec<Result<Vec<History>, Error>> = stream.collect().await;
results.into_iter().collect()
}

async fn get_scripts_history_waterfalls(
Expand Down
Loading