frost/bdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[chain_redesign] Remove old structures

Browse Source 
Other changes:

* The `async-https` feature of `bdk_esplora` is no longer default.
* Rename `ObservedAs` to `ChainPosition`.
* Set temporary MSRV to 1.60.0 to compile all workspace members will all
  features.
This commit is contained in:
志宇
2023-05-24 11:37:26 +08:00
parent 5860704b2d
commit 1c3cbefa4d
48 changed files with 310 additions and 7161 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/esplora/Cargo.toml
View File

@@ -13,12 +13,12 @@ readme = "README.md"
[dependencies]
bdk_chain = { path = "../chain", version = "0.4.0", features = ["serde", "miniscript"] }
esplora-client = { version = "0.3", default-features = false }
esplora-client = { version = "0.5", default-features = false }
async-trait = { version = "0.1.66", optional = true }
futures = { version = "0.3.26", optional = true }
[features]
default = ["async-https", "blocking"]
default = ["blocking"]
async = ["async-trait", "futures", "esplora-client/async"]
async-https = ["async", "esplora-client/async-https"]
blocking = ["esplora-client/blocking"]

2
crates/esplora/README.md
View File

@@ -27,7 +27,7 @@ To use the extension traits:
// for blocking
use bdk_esplora::EsploraExt;
// for async
use bdk_esplora::EsploraAsyncExt;
// use bdk_esplora::EsploraAsyncExt;
```
For full examples, refer to [`example-crates/wallet_esplora`](https://github.com/bitcoindevkit/bdk/tree/master/example-crates/wallet_esplora) (blocking) and [`example-crates/wallet_esplora_async`](https://github.com/bitcoindevkit/bdk/tree/master/example-crates/wallet_esplora_async).

236
crates/esplora/src/async_ext.rs
View File

@@ -1,16 +1,14 @@
use std::collections::BTreeMap;
use async_trait::async_trait;
use bdk_chain::{
bitcoin::{BlockHash, OutPoint, Script, Txid},
chain_graph::ChainGraph,
keychain::KeychainScan,
sparse_chain, BlockId, ConfirmationTime,
collections::BTreeMap,
keychain::LocalUpdate,
BlockId, ConfirmationTimeAnchor,
};
use esplora_client::{Error, OutputStatus};
use futures::stream::{FuturesOrdered, TryStreamExt};
use esplora_client::{Error, OutputStatus, TxStatus};
use futures::{stream::FuturesOrdered, TryStreamExt};
use crate::map_confirmation_time;
use crate::map_confirmation_time_anchor;
/// Trait to extend [`esplora_client::AsyncClient`] functionality.
///
@@ -22,19 +20,18 @@ use crate::map_confirmation_time;
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait EsploraAsyncExt {
/// Scan the blockchain (via esplora) for the data specified and returns a [`KeychainScan`].
/// Scan the blockchain (via esplora) for the data specified and returns a
/// [`LocalUpdate<K, ConfirmationTimeAnchor>`].
///
/// - `local_chain`: the most recent block hashes present locally
/// - `keychain_spks`: keychains that we want to scan transactions for
/// - `txids`: transactions for which we want updated [`ChainPosition`]s
/// - `txids`: transactions for which we want updated [`ConfirmationTimeAnchor`]s
/// - `outpoints`: transactions associated with these outpoints (residing, spending) that we
/// want to included in the update
///
/// The scan for each keychain stops after a gap of `stop_gap` script pubkeys with no associated
/// transactions. `parallel_requests` specifies the max number of HTTP requests to make in
/// parallel.
///
/// [`ChainPosition`]: bdk_chain::sparse_chain::ChainPosition
#[allow(clippy::result_large_err)] // FIXME
async fn scan<K: Ord + Clone + Send>(
&self,
@@ -47,7 +44,7 @@ pub trait EsploraAsyncExt {
outpoints: impl IntoIterator<IntoIter = impl Iterator<Item = OutPoint> + Send> + Send,
stop_gap: usize,
parallel_requests: usize,
) -> Result<KeychainScan<K, ConfirmationTime>, Error>;
) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error>;
/// Convenience method to call [`scan`] without requiring a keychain.
///
@@ -60,26 +57,23 @@ pub trait EsploraAsyncExt {
txids: impl IntoIterator<IntoIter = impl Iterator<Item = Txid> + Send> + Send,
outpoints: impl IntoIterator<IntoIter = impl Iterator<Item = OutPoint> + Send> + Send,
parallel_requests: usize,
) -> Result<ChainGraph<ConfirmationTime>, Error> {
let wallet_scan = self
.scan(
local_chain,
[(
(),
misc_spks
.into_iter()
.enumerate()
.map(|(i, spk)| (i as u32, spk)),
)]
.into(),
txids,
outpoints,
usize::MAX,
parallel_requests,
)
.await?;
Ok(wallet_scan.update)
) -> Result<LocalUpdate<(), ConfirmationTimeAnchor>, Error> {
self.scan(
local_chain,
[(
(),
misc_spks
.into_iter()
.enumerate()
.map(|(i, spk)| (i as u32, spk)),
)]
.into(),
txids,
outpoints,
usize::MAX,
parallel_requests,
)
.await
}
}
@@ -98,47 +92,35 @@ impl EsploraAsyncExt for esplora_client::AsyncClient {
outpoints: impl IntoIterator<IntoIter = impl Iterator<Item = OutPoint> + Send> + Send,
stop_gap: usize,
parallel_requests: usize,
) -> Result<KeychainScan<K, ConfirmationTime>, Error> {
let txids = txids.into_iter();
let outpoints = outpoints.into_iter();
) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error> {
let parallel_requests = Ord::max(parallel_requests, 1);
let mut scan = KeychainScan::default();
let update = &mut scan.update;
let last_active_indices = &mut scan.last_active_indices;
for (&height, &original_hash) in local_chain.iter().rev() {
let update_block_id = BlockId {
height,
hash: self.get_block_hash(height).await?,
};
let _ = update
.insert_checkpoint(update_block_id)
.expect("cannot repeat height here");
if update_block_id.hash == original_hash {
break;
}
}
let tip_at_start = BlockId {
height: self.get_height().await?,
hash: self.get_tip_hash().await?,
};
if let Err(failure) = update.insert_checkpoint(tip_at_start) {
match failure {
sparse_chain::InsertCheckpointError::HashNotMatching { .. } => {
// there was a re-org before we started scanning. We haven't consumed any iterators, so calling this function recursively is safe.
return EsploraAsyncExt::scan(
self,
local_chain,
keychain_spks,
txids,
outpoints,
stop_gap,
parallel_requests,
)
.await;
let (mut update, tip_at_start) = loop {
let mut update = LocalUpdate::<K, ConfirmationTimeAnchor>::default();
for (&height, &original_hash) in local_chain.iter().rev() {
let update_block_id = BlockId {
height,
hash: self.get_block_hash(height).await?,
};
let _ = update
.chain
.insert_block(update_block_id)
.expect("cannot repeat height here");
if update_block_id.hash == original_hash {
break;
}
}
}
let tip_at_start = BlockId {
height: self.get_height().await?,
hash: self.get_tip_hash().await?,
};
if update.chain.insert_block(tip_at_start).is_ok() {
break (update, tip_at_start);
}
};
for (keychain, spks) in keychain_spks {
let mut spks = spks.into_iter();
@@ -147,7 +129,7 @@ impl EsploraAsyncExt for esplora_client::AsyncClient {
type IndexWithTxs = (u32, Vec<esplora_client::Tx>);
loop {
let futures: FuturesOrdered<_> = (0..parallel_requests)
let futures = (0..parallel_requests)
.filter_map(|_| {
let (index, script) = spks.next()?;
let client = self.clone();
@@ -178,13 +160,11 @@ impl EsploraAsyncExt for esplora_client::AsyncClient {
Result::<_, esplora_client::Error>::Ok((index, related_txs))
})
})
.collect();
.collect::<FuturesOrdered<_>>();
let n_futures = futures.len();
let idx_with_tx: Vec<IndexWithTxs> = futures.try_collect().await?;
for (index, related_txs) in idx_with_tx {
for (index, related_txs) in futures.try_collect::<Vec<IndexWithTxs>>().await? {
if related_txs.is_empty() {
empty_scripts += 1;
} else {
@@ -192,22 +172,11 @@ impl EsploraAsyncExt for esplora_client::AsyncClient {
empty_scripts = 0;
}
for tx in related_txs {
let confirmation_time =
map_confirmation_time(&tx.status, tip_at_start.height);
let anchor = map_confirmation_time_anchor(&tx.status, tip_at_start);
if let Err(failure) = update.insert_tx(tx.to_tx(), confirmation_time) {
use bdk_chain::{
chain_graph::InsertTxError, sparse_chain::InsertTxError::*,
};
match failure {
InsertTxError::Chain(TxTooHigh { .. }) => {
unreachable!("chain position already checked earlier")
}
InsertTxError::Chain(TxMovedUnexpectedly { .. })
| InsertTxError::UnresolvableConflict(_) => {
/* implies reorg during a scan. We deal with that below */
}
}
let _ = update.graph.insert_tx(tx.to_tx());
if let Some(anchor) = anchor {
let _ = update.graph.insert_anchor(tx.txid, anchor);
}
}
}
@@ -218,36 +187,37 @@ impl EsploraAsyncExt for esplora_client::AsyncClient {
}
if let Some(last_active_index) = last_active_index {
last_active_indices.insert(keychain, last_active_index);
update.keychain.insert(keychain, last_active_index);
}
}
for txid in txids {
let (tx, tx_status) =
match (self.get_tx(&txid).await?, self.get_tx_status(&txid).await?) {
(Some(tx), Some(tx_status)) => (tx, tx_status),
_ => continue,
};
let confirmation_time = map_confirmation_time(&tx_status, tip_at_start.height);
if let Err(failure) = update.insert_tx(tx, confirmation_time) {
use bdk_chain::{chain_graph::InsertTxError, sparse_chain::InsertTxError::*};
match failure {
InsertTxError::Chain(TxTooHigh { .. }) => {
unreachable!("chain position already checked earlier")
}
InsertTxError::Chain(TxMovedUnexpectedly { .. })
| InsertTxError::UnresolvableConflict(_) => {
/* implies reorg during a scan. We deal with that below */
for txid in txids.into_iter() {
if update.graph.get_tx(txid).is_none() {
match self.get_tx(&txid).await? {
Some(tx) => {
let _ = update.graph.insert_tx(tx);
}
None => continue,
}
}
match self.get_tx_status(&txid).await? {
tx_status if tx_status.confirmed => {
if let Some(anchor) = map_confirmation_time_anchor(&tx_status, tip_at_start) {
let _ = update.graph.insert_anchor(txid, anchor);
}
}
_ => continue,
}
}
for op in outpoints {
for op in outpoints.into_iter() {
let mut op_txs = Vec::with_capacity(2);
if let (Some(tx), Some(tx_status)) = (
if let (
Some(tx),
tx_status @ TxStatus {
confirmed: true, ..
},
) = (
self.get_tx(&op.txid).await?,
self.get_tx_status(&op.txid).await?,
) {
@@ -265,39 +235,24 @@ impl EsploraAsyncExt for esplora_client::AsyncClient {
}
for (tx, status) in op_txs {
let confirmation_time = map_confirmation_time(&status, tip_at_start.height);
let txid = tx.txid();
let anchor = map_confirmation_time_anchor(&status, tip_at_start);
if let Err(failure) = update.insert_tx(tx, confirmation_time) {
use bdk_chain::{chain_graph::InsertTxError, sparse_chain::InsertTxError::*};
match failure {
InsertTxError::Chain(TxTooHigh { .. }) => {
unreachable!("chain position already checked earlier")
}
InsertTxError::Chain(TxMovedUnexpectedly { .. })
| InsertTxError::UnresolvableConflict(_) => {
/* implies reorg during a scan. We deal with that below */
}
}
let _ = update.graph.insert_tx(tx);
if let Some(anchor) = anchor {
let _ = update.graph.insert_anchor(txid, anchor);
}
}
}
let reorg_occurred = {
if let Some(checkpoint) = ChainGraph::chain(update).latest_checkpoint() {
self.get_block_hash(checkpoint.height).await? != checkpoint.hash
} else {
false
}
};
if reorg_occurred {
// A reorg occurred, so let's find out where all the txids we found are in the chain now.
// XXX: collect required because of weird type naming issues
let txids_found = ChainGraph::chain(update)
.txids()
.map(|(_, txid)| *txid)
if tip_at_start.hash != self.get_block_hash(tip_at_start.height).await? {
// A reorg occurred, so let's find out where all the txids we found are now in the chain
let txids_found = update
.graph
.full_txs()
.map(|tx_node| tx_node.txid)
.collect::<Vec<_>>();
scan.update = EsploraAsyncExt::scan_without_keychain(
update.chain = EsploraAsyncExt::scan_without_keychain(
self,
local_chain,
[],
@@ -305,9 +260,10 @@ impl EsploraAsyncExt for esplora_client::AsyncClient {
[],
parallel_requests,
)
.await?;
.await?
.chain;
}
Ok(scan)
Ok(update)
}
}

194
crates/esplora/src/blocking_ext.rs
View File

@@ -1,14 +1,10 @@
use std::collections::BTreeMap;
use bdk_chain::bitcoin::{BlockHash, OutPoint, Script, Txid};
use bdk_chain::collections::BTreeMap;
use bdk_chain::BlockId;
use bdk_chain::{keychain::LocalUpdate, ConfirmationTimeAnchor};
use esplora_client::{Error, OutputStatus, TxStatus};
use bdk_chain::{
bitcoin::{BlockHash, OutPoint, Script, Txid},
chain_graph::ChainGraph,
keychain::KeychainScan,
sparse_chain, BlockId, ConfirmationTime,
};
use esplora_client::{Error, OutputStatus};
use crate::map_confirmation_time;
use crate::map_confirmation_time_anchor;
/// Trait to extend [`esplora_client::BlockingClient`] functionality.
///
@@ -16,19 +12,18 @@ use crate::map_confirmation_time;
///
/// [crate-level documentation]: crate
pub trait EsploraExt {
/// Scan the blockchain (via esplora) for the data specified and returns a [`KeychainScan`].
/// Scan the blockchain (via esplora) for the data specified and returns a
/// [`LocalUpdate<K, ConfirmationTimeAnchor>`].
///
/// - `local_chain`: the most recent block hashes present locally
/// - `keychain_spks`: keychains that we want to scan transactions for
/// - `txids`: transactions for which we want updated [`ChainPosition`]s
/// - `txids`: transactions for which we want updated [`ConfirmationTimeAnchor`]s
/// - `outpoints`: transactions associated with these outpoints (residing, spending) that we
/// want to included in the update
///
/// The scan for each keychain stops after a gap of `stop_gap` script pubkeys with no associated
/// transactions. `parallel_requests` specifies the max number of HTTP requests to make in
/// parallel.
///
/// [`ChainPosition`]: bdk_chain::sparse_chain::ChainPosition
#[allow(clippy::result_large_err)] // FIXME
fn scan<K: Ord + Clone>(
&self,
@@ -38,7 +33,7 @@ pub trait EsploraExt {
outpoints: impl IntoIterator<Item = OutPoint>,
stop_gap: usize,
parallel_requests: usize,
) -> Result<KeychainScan<K, ConfirmationTime>, Error>;
) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error>;
/// Convenience method to call [`scan`] without requiring a keychain.
///
@@ -51,8 +46,8 @@ pub trait EsploraExt {
txids: impl IntoIterator<Item = Txid>,
outpoints: impl IntoIterator<Item = OutPoint>,
parallel_requests: usize,
) -> Result<ChainGraph<ConfirmationTime>, Error> {
let wallet_scan = self.scan(
) -> Result<LocalUpdate<(), ConfirmationTimeAnchor>, Error> {
self.scan(
local_chain,
[(
(),
@@ -66,9 +61,7 @@ pub trait EsploraExt {
outpoints,
usize::MAX,
parallel_requests,
)?;
Ok(wallet_scan.update)
)
}
}
@@ -81,44 +74,35 @@ impl EsploraExt for esplora_client::BlockingClient {
outpoints: impl IntoIterator<Item = OutPoint>,
stop_gap: usize,
parallel_requests: usize,
) -> Result<KeychainScan<K, ConfirmationTime>, Error> {
) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error> {
let parallel_requests = Ord::max(parallel_requests, 1);
let mut scan = KeychainScan::default();
let update = &mut scan.update;
let last_active_indices = &mut scan.last_active_indices;
for (&height, &original_hash) in local_chain.iter().rev() {
let update_block_id = BlockId {
height,
hash: self.get_block_hash(height)?,
};
let _ = update
.insert_checkpoint(update_block_id)
.expect("cannot repeat height here");
if update_block_id.hash == original_hash {
break;
}
}
let tip_at_start = BlockId {
height: self.get_height()?,
hash: self.get_tip_hash()?,
};
if let Err(failure) = update.insert_checkpoint(tip_at_start) {
match failure {
sparse_chain::InsertCheckpointError::HashNotMatching { .. } => {
// there was a re-org before we started scanning. We haven't consumed any iterators, so calling this function recursively is safe.
return EsploraExt::scan(
self,
local_chain,
keychain_spks,
txids,
outpoints,
stop_gap,
parallel_requests,
);
let (mut update, tip_at_start) = loop {
let mut update = LocalUpdate::<K, ConfirmationTimeAnchor>::default();
for (&height, &original_hash) in local_chain.iter().rev() {
let update_block_id = BlockId {
height,
hash: self.get_block_hash(height)?,
};
let _ = update
.chain
.insert_block(update_block_id)
.expect("cannot repeat height here");
if update_block_id.hash == original_hash {
break;
}
}
}
let tip_at_start = BlockId {
height: self.get_height()?,
hash: self.get_tip_hash()?,
};
if update.chain.insert_block(tip_at_start).is_ok() {
break (update, tip_at_start);
}
};
for (keychain, spks) in keychain_spks {
let mut spks = spks.into_iter();
@@ -171,22 +155,11 @@ impl EsploraExt for esplora_client::BlockingClient {
empty_scripts = 0;
}
for tx in related_txs {
let confirmation_time =
map_confirmation_time(&tx.status, tip_at_start.height);
let anchor = map_confirmation_time_anchor(&tx.status, tip_at_start);
if let Err(failure) = update.insert_tx(tx.to_tx(), confirmation_time) {
use bdk_chain::{
chain_graph::InsertTxError, sparse_chain::InsertTxError::*,
};
match failure {
InsertTxError::Chain(TxTooHigh { .. }) => {
unreachable!("chain position already checked earlier")
}
InsertTxError::Chain(TxMovedUnexpectedly { .. })
| InsertTxError::UnresolvableConflict(_) => {
/* implies reorg during a scan. We deal with that below */
}
}
let _ = update.graph.insert_tx(tx.to_tx());
if let Some(anchor) = anchor {
let _ = update.graph.insert_anchor(tx.txid, anchor);
}
}
}
@@ -197,36 +170,39 @@ impl EsploraExt for esplora_client::BlockingClient {
}
if let Some(last_active_index) = last_active_index {
last_active_indices.insert(keychain, last_active_index);
update.keychain.insert(keychain, last_active_index);
}
}
for txid in txids.into_iter() {
let (tx, tx_status) = match (self.get_tx(&txid)?, self.get_tx_status(&txid)?) {
(Some(tx), Some(tx_status)) => (tx, tx_status),
_ => continue,
};
let confirmation_time = map_confirmation_time(&tx_status, tip_at_start.height);
if let Err(failure) = update.insert_tx(tx, confirmation_time) {
use bdk_chain::{chain_graph::InsertTxError, sparse_chain::InsertTxError::*};
match failure {
InsertTxError::Chain(TxTooHigh { .. }) => {
unreachable!("chain position already checked earlier")
if update.graph.get_tx(txid).is_none() {
match self.get_tx(&txid)? {
Some(tx) => {
let _ = update.graph.insert_tx(tx);
}
InsertTxError::Chain(TxMovedUnexpectedly { .. })
| InsertTxError::UnresolvableConflict(_) => {
/* implies reorg during a scan. We deal with that below */
None => continue,
}
}
match self.get_tx_status(&txid)? {
tx_status @ TxStatus {
confirmed: true, ..
} => {
if let Some(anchor) = map_confirmation_time_anchor(&tx_status, tip_at_start) {
let _ = update.graph.insert_anchor(txid, anchor);
}
}
_ => continue,
}
}
for op in outpoints.into_iter() {
let mut op_txs = Vec::with_capacity(2);
if let (Some(tx), Some(tx_status)) =
(self.get_tx(&op.txid)?, self.get_tx_status(&op.txid)?)
if let (
Some(tx),
tx_status @ TxStatus {
confirmed: true, ..
},
) = (self.get_tx(&op.txid)?, self.get_tx_status(&op.txid)?)
{
op_txs.push((tx, tx_status));
if let Some(OutputStatus {
@@ -242,48 +218,34 @@ impl EsploraExt for esplora_client::BlockingClient {
}
for (tx, status) in op_txs {
let confirmation_time = map_confirmation_time(&status, tip_at_start.height);
let txid = tx.txid();
let anchor = map_confirmation_time_anchor(&status, tip_at_start);
if let Err(failure) = update.insert_tx(tx, confirmation_time) {
use bdk_chain::{chain_graph::InsertTxError, sparse_chain::InsertTxError::*};
match failure {
InsertTxError::Chain(TxTooHigh { .. }) => {
unreachable!("chain position already checked earlier")
}
InsertTxError::Chain(TxMovedUnexpectedly { .. })
| InsertTxError::UnresolvableConflict(_) => {
/* implies reorg during a scan. We deal with that below */
}
}
let _ = update.graph.insert_tx(tx);
if let Some(anchor) = anchor {
let _ = update.graph.insert_anchor(txid, anchor);
}
}
}
let reorg_occurred = {
if let Some(checkpoint) = ChainGraph::chain(update).latest_checkpoint() {
self.get_block_hash(checkpoint.height)? != checkpoint.hash
} else {
false
}
};
if reorg_occurred {
// A reorg occurred, so let's find out where all the txids we found are now in the chain.
// XXX: collect required because of weird type naming issues
let txids_found = ChainGraph::chain(update)
.txids()
.map(|(_, txid)| *txid)
if tip_at_start.hash != self.get_block_hash(tip_at_start.height)? {
// A reorg occurred, so let's find out where all the txids we found are now in the chain
let txids_found = update
.graph
.full_txs()
.map(|tx_node| tx_node.txid)
.collect::<Vec<_>>();
scan.update = EsploraExt::scan_without_keychain(
update.chain = EsploraExt::scan_without_keychain(
self,
local_chain,
[],
txids_found,
[],
parallel_requests,
)?;
)?
.chain;
}
Ok(scan)
Ok(update)
}
}

15
crates/esplora/src/lib.rs
View File

@@ -1,9 +1,8 @@
#![doc = include_str!("../README.md")]
use bdk_chain::{BlockId, ConfirmationTime, ConfirmationTimeAnchor};
use bdk_chain::{BlockId, ConfirmationTimeAnchor};
use esplora_client::TxStatus;
pub use esplora_client;
pub mod v2;
#[cfg(feature = "blocking")]
mod blocking_ext;
@@ -15,18 +14,6 @@ mod async_ext;
#[cfg(feature = "async")]
pub use async_ext::*;
pub(crate) fn map_confirmation_time(
tx_status: &TxStatus,
height_at_start: u32,
) -> ConfirmationTime {
match (tx_status.block_time, tx_status.block_height) {
(Some(time), Some(height)) if height <= height_at_start => {
ConfirmationTime::Confirmed { height, time }
}
_ => ConfirmationTime::Unconfirmed { last_seen: 0 },
}
}
pub(crate) fn map_confirmation_time_anchor(
tx_status: &TxStatus,
tip_at_start: BlockId,

266
crates/esplora/src/v2/async_ext.rs
View File

@@ -1,266 +0,0 @@
use async_trait::async_trait;
use bdk_chain::{
bitcoin::{BlockHash, OutPoint, Script, Txid},
collections::BTreeMap,
keychain::LocalUpdate,
BlockId, ConfirmationTimeAnchor,
};
use esplora_client::{Error, OutputStatus};
use futures::{stream::FuturesOrdered, TryStreamExt};
use crate::map_confirmation_time_anchor;
/// Trait to extend [`esplora_client::AsyncClient`] functionality.
///
/// This is the async version of [`EsploraExt`]. Refer to
/// [crate-level documentation] for more.
///
/// [`EsploraExt`]: crate::EsploraExt
/// [crate-level documentation]: crate
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait EsploraAsyncExt {
/// Scan the blockchain (via esplora) for the data specified and returns a
/// [`LocalUpdate<K, ConfirmationTimeAnchor>`].
///
/// - `local_chain`: the most recent block hashes present locally
/// - `keychain_spks`: keychains that we want to scan transactions for
/// - `txids`: transactions for which we want updated [`ChainPosition`]s
/// - `outpoints`: transactions associated with these outpoints (residing, spending) that we
/// want to included in the update
///
/// The scan for each keychain stops after a gap of `stop_gap` script pubkeys with no associated
/// transactions. `parallel_requests` specifies the max number of HTTP requests to make in
/// parallel.
///
/// [`ChainPosition`]: bdk_chain::sparse_chain::ChainPosition
#[allow(clippy::result_large_err)] // FIXME
async fn scan<K: Ord + Clone + Send>(
&self,
local_chain: &BTreeMap<u32, BlockHash>,
keychain_spks: BTreeMap<
K,
impl IntoIterator<IntoIter = impl Iterator<Item = (u32, Script)> + Send> + Send,
>,
txids: impl IntoIterator<IntoIter = impl Iterator<Item = Txid> + Send> + Send,
outpoints: impl IntoIterator<IntoIter = impl Iterator<Item = OutPoint> + Send> + Send,
stop_gap: usize,
parallel_requests: usize,
) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error>;
/// Convenience method to call [`scan`] without requiring a keychain.
///
/// [`scan`]: EsploraAsyncExt::scan
#[allow(clippy::result_large_err)] // FIXME
async fn scan_without_keychain(
&self,
local_chain: &BTreeMap<u32, BlockHash>,
misc_spks: impl IntoIterator<IntoIter = impl Iterator<Item = Script> + Send> + Send,
txids: impl IntoIterator<IntoIter = impl Iterator<Item = Txid> + Send> + Send,
outpoints: impl IntoIterator<IntoIter = impl Iterator<Item = OutPoint> + Send> + Send,
parallel_requests: usize,
) -> Result<LocalUpdate<(), ConfirmationTimeAnchor>, Error> {
self.scan(
local_chain,
[(
(),
misc_spks
.into_iter()
.enumerate()
.map(|(i, spk)| (i as u32, spk)),
)]
.into(),
txids,
outpoints,
usize::MAX,
parallel_requests,
)
.await
}
}
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl EsploraAsyncExt for esplora_client::AsyncClient {
#[allow(clippy::result_large_err)] // FIXME
async fn scan<K: Ord + Clone + Send>(
&self,
local_chain: &BTreeMap<u32, BlockHash>,
keychain_spks: BTreeMap<
K,
impl IntoIterator<IntoIter = impl Iterator<Item = (u32, Script)> + Send> + Send,
>,
txids: impl IntoIterator<IntoIter = impl Iterator<Item = Txid> + Send> + Send,
outpoints: impl IntoIterator<IntoIter = impl Iterator<Item = OutPoint> + Send> + Send,
stop_gap: usize,
parallel_requests: usize,
) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error> {
let parallel_requests = Ord::max(parallel_requests, 1);
let (mut update, tip_at_start) = loop {
let mut update = LocalUpdate::<K, ConfirmationTimeAnchor>::default();
for (&height, &original_hash) in local_chain.iter().rev() {
let update_block_id = BlockId {
height,
hash: self.get_block_hash(height).await?,
};
let _ = update
.chain
.insert_block(update_block_id)
.expect("cannot repeat height here");
if update_block_id.hash == original_hash {
break;
}
}
let tip_at_start = BlockId {
height: self.get_height().await?,
hash: self.get_tip_hash().await?,
};
if update.chain.insert_block(tip_at_start).is_ok() {
break (update, tip_at_start);
}
};
for (keychain, spks) in keychain_spks {
let mut spks = spks.into_iter();
let mut last_active_index = None;
let mut empty_scripts = 0;
type IndexWithTxs = (u32, Vec<esplora_client::Tx>);
loop {
let futures = (0..parallel_requests)
.filter_map(|_| {
let (index, script) = spks.next()?;
let client = self.clone();
Some(async move {
let mut related_txs = client.scripthash_txs(&script, None).await?;
let n_confirmed =
related_txs.iter().filter(|tx| tx.status.confirmed).count();
// esplora pages on 25 confirmed transactions. If there are 25 or more we
// keep requesting to see if there's more.
if n_confirmed >= 25 {
loop {
let new_related_txs = client
.scripthash_txs(
&script,
Some(related_txs.last().unwrap().txid),
)
.await?;
let n = new_related_txs.len();
related_txs.extend(new_related_txs);
// we've reached the end
if n < 25 {
break;
}
}
}
Result::<_, esplora_client::Error>::Ok((index, related_txs))
})
})
.collect::<FuturesOrdered<_>>();
let n_futures = futures.len();
for (index, related_txs) in futures.try_collect::<Vec<IndexWithTxs>>().await? {
if related_txs.is_empty() {
empty_scripts += 1;
} else {
last_active_index = Some(index);
empty_scripts = 0;
}
for tx in related_txs {
let anchor = map_confirmation_time_anchor(&tx.status, tip_at_start);
let _ = update.graph.insert_tx(tx.to_tx());
if let Some(anchor) = anchor {
let _ = update.graph.insert_anchor(tx.txid, anchor);
}
}
}
if n_futures == 0 || empty_scripts >= stop_gap {
break;
}
}
if let Some(last_active_index) = last_active_index {
update.keychain.insert(keychain, last_active_index);
}
}
for txid in txids.into_iter() {
if update.graph.get_tx(txid).is_none() {
match self.get_tx(&txid).await? {
Some(tx) => {
let _ = update.graph.insert_tx(tx);
}
None => continue,
}
}
match self.get_tx_status(&txid).await? {
Some(tx_status) => {
if let Some(anchor) = map_confirmation_time_anchor(&tx_status, tip_at_start) {
let _ = update.graph.insert_anchor(txid, anchor);
}
}
None => continue,
}
}
for op in outpoints.into_iter() {
let mut op_txs = Vec::with_capacity(2);
if let (Some(tx), Some(tx_status)) = (
self.get_tx(&op.txid).await?,
self.get_tx_status(&op.txid).await?,
) {
op_txs.push((tx, tx_status));
if let Some(OutputStatus {
txid: Some(txid),
status: Some(spend_status),
..
}) = self.get_output_status(&op.txid, op.vout as _).await?
{
if let Some(spend_tx) = self.get_tx(&txid).await? {
op_txs.push((spend_tx, spend_status));
}
}
}
for (tx, status) in op_txs {
let txid = tx.txid();
let anchor = map_confirmation_time_anchor(&status, tip_at_start);
let _ = update.graph.insert_tx(tx);
if let Some(anchor) = anchor {
let _ = update.graph.insert_anchor(txid, anchor);
}
}
}
if tip_at_start.hash != self.get_block_hash(tip_at_start.height).await? {
// A reorg occurred, so let's find out where all the txids we found are now in the chain
let txids_found = update
.graph
.full_txs()
.map(|tx_node| tx_node.txid)
.collect::<Vec<_>>();
update.chain = EsploraAsyncExt::scan_without_keychain(
self,
local_chain,
[],
txids_found,
[],
parallel_requests,
)
.await?
.chain;
}
Ok(update)
}
}

247
crates/esplora/src/v2/blocking_ext.rs
View File

@@ -1,247 +0,0 @@
use bdk_chain::bitcoin::{BlockHash, OutPoint, Script, Txid};
use bdk_chain::collections::BTreeMap;
use bdk_chain::BlockId;
use bdk_chain::{keychain::LocalUpdate, ConfirmationTimeAnchor};
use esplora_client::{Error, OutputStatus};
use crate::map_confirmation_time_anchor;
/// Trait to extend [`esplora_client::BlockingClient`] functionality.
///
/// Refer to [crate-level documentation] for more.
///
/// [crate-level documentation]: crate
pub trait EsploraExt {
/// Scan the blockchain (via esplora) for the data specified and returns a
/// [`LocalUpdate<K, ConfirmationTimeAnchor>`].
///
/// - `local_chain`: the most recent block hashes present locally
/// - `keychain_spks`: keychains that we want to scan transactions for
/// - `txids`: transactions for which we want updated [`ChainPosition`]s
/// - `outpoints`: transactions associated with these outpoints (residing, spending) that we
/// want to included in the update
///
/// The scan for each keychain stops after a gap of `stop_gap` script pubkeys with no associated
/// transactions. `parallel_requests` specifies the max number of HTTP requests to make in
/// parallel.
///
/// [`ChainPosition`]: bdk_chain::sparse_chain::ChainPosition
#[allow(clippy::result_large_err)] // FIXME
fn scan<K: Ord + Clone>(
&self,
local_chain: &BTreeMap<u32, BlockHash>,
keychain_spks: BTreeMap<K, impl IntoIterator<Item = (u32, Script)>>,
txids: impl IntoIterator<Item = Txid>,
outpoints: impl IntoIterator<Item = OutPoint>,
stop_gap: usize,
parallel_requests: usize,
) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error>;
/// Convenience method to call [`scan`] without requiring a keychain.
///
/// [`scan`]: EsploraExt::scan
#[allow(clippy::result_large_err)] // FIXME
fn scan_without_keychain(
&self,
local_chain: &BTreeMap<u32, BlockHash>,
misc_spks: impl IntoIterator<Item = Script>,
txids: impl IntoIterator<Item = Txid>,
outpoints: impl IntoIterator<Item = OutPoint>,
parallel_requests: usize,
) -> Result<LocalUpdate<(), ConfirmationTimeAnchor>, Error> {
self.scan(
local_chain,
[(
(),
misc_spks
.into_iter()
.enumerate()
.map(|(i, spk)| (i as u32, spk)),
)]
.into(),
txids,
outpoints,
usize::MAX,
parallel_requests,
)
}
}
impl EsploraExt for esplora_client::BlockingClient {
fn scan<K: Ord + Clone>(
&self,
local_chain: &BTreeMap<u32, BlockHash>,
keychain_spks: BTreeMap<K, impl IntoIterator<Item = (u32, Script)>>,
txids: impl IntoIterator<Item = Txid>,
outpoints: impl IntoIterator<Item = OutPoint>,
stop_gap: usize,
parallel_requests: usize,
) -> Result<LocalUpdate<K, ConfirmationTimeAnchor>, Error> {
let parallel_requests = Ord::max(parallel_requests, 1);
let (mut update, tip_at_start) = loop {
let mut update = LocalUpdate::<K, ConfirmationTimeAnchor>::default();
for (&height, &original_hash) in local_chain.iter().rev() {
let update_block_id = BlockId {
height,
hash: self.get_block_hash(height)?,
};
let _ = update
.chain
.insert_block(update_block_id)
.expect("cannot repeat height here");
if update_block_id.hash == original_hash {
break;
}
}
let tip_at_start = BlockId {
height: self.get_height()?,
hash: self.get_tip_hash()?,
};
if update.chain.insert_block(tip_at_start).is_ok() {
break (update, tip_at_start);
}
};
for (keychain, spks) in keychain_spks {
let mut spks = spks.into_iter();
let mut last_active_index = None;
let mut empty_scripts = 0;
type IndexWithTxs = (u32, Vec<esplora_client::Tx>);
loop {
let handles = (0..parallel_requests)
.filter_map(
|_| -> Option<std::thread::JoinHandle<Result<IndexWithTxs, _>>> {
let (index, script) = spks.next()?;
let client = self.clone();
Some(std::thread::spawn(move || {
let mut related_txs = client.scripthash_txs(&script, None)?;
let n_confirmed =
related_txs.iter().filter(|tx| tx.status.confirmed).count();
// esplora pages on 25 confirmed transactions. If there are 25 or more we
// keep requesting to see if there's more.
if n_confirmed >= 25 {
loop {
let new_related_txs = client.scripthash_txs(
&script,
Some(related_txs.last().unwrap().txid),
)?;
let n = new_related_txs.len();
related_txs.extend(new_related_txs);
// we've reached the end
if n < 25 {
break;
}
}
}
Result::<_, esplora_client::Error>::Ok((index, related_txs))
}))
},
)
.collect::<Vec<_>>();
let n_handles = handles.len();
for handle in handles {
let (index, related_txs) = handle.join().unwrap()?; // TODO: don't unwrap
if related_txs.is_empty() {
empty_scripts += 1;
} else {
last_active_index = Some(index);
empty_scripts = 0;
}
for tx in related_txs {
let anchor = map_confirmation_time_anchor(&tx.status, tip_at_start);
let _ = update.graph.insert_tx(tx.to_tx());
if let Some(anchor) = anchor {
let _ = update.graph.insert_anchor(tx.txid, anchor);
}
}
}
if n_handles == 0 || empty_scripts >= stop_gap {
break;
}
}
if let Some(last_active_index) = last_active_index {
update.keychain.insert(keychain, last_active_index);
}
}
for txid in txids.into_iter() {
if update.graph.get_tx(txid).is_none() {
match self.get_tx(&txid)? {
Some(tx) => {
let _ = update.graph.insert_tx(tx);
}
None => continue,
}
}
match self.get_tx_status(&txid)? {
Some(tx_status) => {
if let Some(anchor) = map_confirmation_time_anchor(&tx_status, tip_at_start) {
let _ = update.graph.insert_anchor(txid, anchor);
}
}
None => continue,
}
}
for op in outpoints.into_iter() {
let mut op_txs = Vec::with_capacity(2);
if let (Some(tx), Some(tx_status)) =
(self.get_tx(&op.txid)?, self.get_tx_status(&op.txid)?)
{
op_txs.push((tx, tx_status));
if let Some(OutputStatus {
txid: Some(txid),
status: Some(spend_status),
..
}) = self.get_output_status(&op.txid, op.vout as _)?
{
if let Some(spend_tx) = self.get_tx(&txid)? {
op_txs.push((spend_tx, spend_status));
}
}
}
for (tx, status) in op_txs {
let txid = tx.txid();
let anchor = map_confirmation_time_anchor(&status, tip_at_start);
let _ = update.graph.insert_tx(tx);
if let Some(anchor) = anchor {
let _ = update.graph.insert_anchor(txid, anchor);
}
}
}
if tip_at_start.hash != self.get_block_hash(tip_at_start.height)? {
// A reorg occurred, so let's find out where all the txids we found are now in the chain
let txids_found = update
.graph
.full_txs()
.map(|tx_node| tx_node.txid)
.collect::<Vec<_>>();
update.chain = EsploraExt::scan_without_keychain(
self,
local_chain,
[],
txids_found,
[],
parallel_requests,
)?
.chain;
}
Ok(update)
}
}

9
crates/esplora/src/v2/mod.rs
View File

@@ -1,9 +0,0 @@
#[cfg(feature = "blocking")]
mod blocking_ext;
#[cfg(feature = "blocking")]
pub use blocking_ext::*;
#[cfg(feature = "async")]
mod async_ext;
#[cfg(feature = "async")]
pub use async_ext::*;