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

Merge bitcoindevkit/bdk#1380: Simplified EsploraExt API

Browse Source 
96a9aa6e63 feat(chain): refactor `merge_chains` (志宇)
2f22987c9e chore(chain): fix comment (志宇)
daf588f016 feat(chain): optimize `merge_chains` (志宇)
77d35954c1 feat(chain)!: rm `local_chain::Update` (志宇)
1269b0610e test(chain): fix incorrect test case (志宇)
72fe65b65f feat(esplora)!: simplify chain update logic (志宇)
eded1a7ea0 feat(chain): introduce `CheckPoint::insert` (志宇)
519cd75d23 test(esplora): move esplora tests into src files (志宇)
a6e613e6b9 test(esplora): add `test_finalize_chain_update` (志宇)
494d253493 feat(testenv): add `genesis_hash` method (志宇)
886d72e3d5 chore(chain)!: rm `missing_heights` and `missing_heights_from` methods (志宇)
bd62aa0fe1 feat(esplora)!: remove `EsploraExt::update_local_chain` (志宇)
1e99793983 feat(testenv): add `make_checkpoint_tip` (志宇)

Pull request description:

  Fixes #1354

  ### Description

  Built on top of both #1369 and #1373, we simplify the `EsploraExt` API by removing the `update_local_chain` method and having `full_scan` and `sync` update the local chain in the same call. The `full_scan` and `sync` methods now takes in an additional input (`local_tip`) which provides us with the view of the `LocalChain` before the update. These methods now return structs `FullScanUpdate` and `SyncUpdate`.

  The examples are updated to use this new API. `TxGraph::missing_heights` and `tx_graph::ChangeSet::missing_heights_from` are no longer needed, therefore they are removed.

  Additionally, we used this opportunity to simplify the logic which updates `LocalChain`. We got rid of the `local_chain::Update` struct (which contained the update `CheckPoint` tip and a `bool` which signaled whether we want to introduce blocks below point of agreement). It turns out we can use something like `CheckPoint::insert` so the chain source can craft an update based on the old tip. This way, we can make better use of `merge_chains`' optimization that compares the `Arc` pointers of the local and update chain (before we were crafting the update chain NOT based on top of the previous local chain). With this, we no longer need the `Update::introduce_older_block` field since the logic will naturally break when we reach a matching `Arc` pointer.

  ### Notes to the reviewers

  * Obtaining the `LocalChain`'s update now happens within `EsploraExt::full_scan` and `EsploraExt::sync`. Creating the `LocalChain` update is now split into two methods (`fetch_latest_blocks` and `chain_update`) that are called before and after fetching transactions and anchors.
  * We need to duplicate code for `bdk_esplora`. One for blocking and one for async.

  ### Changelog notice

  * Changed `EsploraExt` API so that sync only requires one round of fetching data. The `local_chain_update` method is removed and the `local_tip` parameter is added to the `full_scan` and `sync` methods.
  * Removed `TxGraph::missing_heights` and `tx_graph::ChangeSet::missing_heights_from` methods.
  * Introduced `CheckPoint::insert` which allows convenient checkpoint-insertion. This is intended for use by chain-sources when crafting an update.
  * Refactored `merge_chains` to also return the resultant `CheckPoint` tip.
  * Optimized the update `LocalChain` logic - use the update `CheckPoint` as the new `CheckPoint` tip when possible.

  ### Checklists

  #### All Submissions:

  * [x] I've signed all my commits
  * [x] I followed the [contribution guidelines](https://github.com/bitcoindevkit/bdk/blob/master/CONTRIBUTING.md)
  * [x] I ran `cargo fmt` and `cargo clippy` before committing

  #### New Features:

  * [x] I've added tests for the new feature
  * [x] I've added docs for the new feature

ACKs for top commit:
  LLFourn:
    ACK 96a9aa6e63

Tree-SHA512: 3d4f2eab08a1fe94eb578c594126e99679f72e231680b2edd4bfb018ba1d998ca123b07acb2d19c644d5887fc36b8e42badba91cd09853df421ded04de45bf69
This commit is contained in:
志宇
2024-04-22 17:44:44 +08:00
parent 9800f8d88e 96a9aa6e63
commit 8e73998cfa
19 changed files with 1609 additions and 1120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/esplora/Cargo.toml
View File

@@ -25,6 +25,7 @@ miniscript = { version = "11.0.0", optional = true, default-features = false }
bdk_testenv = { path = "../testenv", default_features = false }
electrsd = { version= "0.27.1", features = ["bitcoind_25_0", "esplora_a33e97e1", "legacy"] }
tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros"] }
anyhow = "1"
[features]
default = ["std", "async-https", "blocking-https-rustls"]

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

@@ -1,15 +1,17 @@
use std::collections::BTreeSet;
use async_trait::async_trait;
use bdk_chain::collections::btree_map;
use bdk_chain::Anchor;
use bdk_chain::{
bitcoin::{Amount, BlockHash, OutPoint, ScriptBuf, TxOut, Txid},
bitcoin::{BlockHash, OutPoint, ScriptBuf, TxOut, Txid},
collections::BTreeMap,
local_chain::{self, CheckPoint},
local_chain::CheckPoint,
BlockId, ConfirmationTimeHeightAnchor, TxGraph,
};
use esplora_client::TxStatus;
use esplora_client::{Amount, TxStatus};
use futures::{stream::FuturesOrdered, TryStreamExt};
use crate::anchor_from_status;
use crate::{anchor_from_status, FullScanUpdate, SyncUpdate};
/// [`esplora_client::Error`]
type Error = Box<esplora_client::Error>;
@@ -22,36 +24,15 @@ type Error = Box<esplora_client::Error>;
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait EsploraAsyncExt {
/// Prepare a [`LocalChain`] update with blocks fetched from Esplora.
///
/// * `local_tip` is the previous tip of [`LocalChain::tip`].
/// * `request_heights` is the block heights that we are interested in fetching from Esplora.
///
/// The result of this method can be applied to [`LocalChain::apply_update`].
///
/// ## Consistency
///
/// The chain update returned is guaranteed to be consistent as long as there is not a *large* re-org
/// during the call. The size of re-org we can tollerate is server dependent but will be at
/// least 10.
///
/// [`LocalChain`]: bdk_chain::local_chain::LocalChain
/// [`LocalChain::tip`]: bdk_chain::local_chain::LocalChain::tip
/// [`LocalChain::apply_update`]: bdk_chain::local_chain::LocalChain::apply_update
async fn update_local_chain(
&self,
local_tip: CheckPoint,
request_heights: impl IntoIterator<IntoIter = impl Iterator<Item = u32> + Send> + Send,
) -> Result<local_chain::Update, Error>;
/// Full scan the keychain scripts specified with the blockchain (via an Esplora client) and
/// returns a [`TxGraph`] and a map of last active indices.
/// Scan keychain scripts for transactions against Esplora, returning an update that can be
/// applied to the receiving structures.
///
/// * `local_tip`: the previously seen tip from [`LocalChain::tip`].
/// * `keychain_spks`: keychains that we want to scan transactions for
///
/// The full 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.
/// The full 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.
///
/// ## Note
///
@@ -65,19 +46,23 @@ pub trait EsploraAsyncExt {
/// and [Sparrow](https://www.sparrowwallet.com/docs/faq.html#ive-restored-my-wallet-but-some-of-my-funds-are-missing).
///
/// A `stop_gap` of 0 will be treated as a `stop_gap` of 1.
///
/// [`LocalChain::tip`]: bdk_chain::local_chain::LocalChain::tip
async fn full_scan<K: Ord + Clone + Send>(
&self,
local_tip: CheckPoint,
keychain_spks: BTreeMap<
K,
impl IntoIterator<IntoIter = impl Iterator<Item = (u32, ScriptBuf)> + Send> + Send,
>,
stop_gap: usize,
parallel_requests: usize,
) -> Result<(TxGraph<ConfirmationTimeHeightAnchor>, BTreeMap<K, u32>), Error>;
) -> Result<FullScanUpdate<K>, Error>;
/// Sync a set of scripts with the blockchain (via an Esplora client) for the data
/// specified and return a [`TxGraph`].
///
/// * `local_tip`: the previously seen tip from [`LocalChain::tip`].
/// * `misc_spks`: scripts that we want to sync transactions for
/// * `txids`: transactions for which we want updated [`ConfirmationTimeHeightAnchor`]s
/// * `outpoints`: transactions associated with these outpoints (residing, spending) that we
@@ -86,210 +71,203 @@ pub trait EsploraAsyncExt {
/// If the scripts to sync are unknown, such as when restoring or importing a keychain that
/// may include scripts that have been used, use [`full_scan`] with the keychain.
///
/// [`LocalChain::tip`]: bdk_chain::local_chain::LocalChain::tip
/// [`full_scan`]: EsploraAsyncExt::full_scan
async fn sync(
&self,
local_tip: CheckPoint,
misc_spks: impl IntoIterator<IntoIter = impl Iterator<Item = ScriptBuf> + 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<TxGraph<ConfirmationTimeHeightAnchor>, Error>;
) -> Result<SyncUpdate, Error>;
}
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl EsploraAsyncExt for esplora_client::AsyncClient {
async fn update_local_chain(
&self,
local_tip: CheckPoint,
request_heights: impl IntoIterator<IntoIter = impl Iterator<Item = u32> + Send> + Send,
) -> Result<local_chain::Update, Error> {
// Fetch latest N (server dependent) blocks from Esplora. The server guarantees these are
// consistent.
let mut fetched_blocks = self
.get_blocks(None)
.await?
.into_iter()
.map(|b| (b.time.height, b.id))
.collect::<BTreeMap<u32, BlockHash>>();
let new_tip_height = fetched_blocks
.keys()
.last()
.copied()
.expect("must have atleast one block");
// Fetch blocks of heights that the caller is interested in, skipping blocks that are
// already fetched when constructing `fetched_blocks`.
for height in request_heights {
// do not fetch blocks higher than remote tip
if height > new_tip_height {
continue;
}
// only fetch what is missing
if let btree_map::Entry::Vacant(entry) = fetched_blocks.entry(height) {
// ❗The return value of `get_block_hash` is not strictly guaranteed to be consistent
// with the chain at the time of `get_blocks` above (there could have been a deep
// re-org). Since `get_blocks` returns 10 (or so) blocks we are assuming that it's
// not possible to have a re-org deeper than that.
entry.insert(self.get_block_hash(height).await?);
}
}
// Ensure `fetched_blocks` can create an update that connects with the original chain by
// finding a "Point of Agreement".
for (height, local_hash) in local_tip.iter().map(|cp| (cp.height(), cp.hash())) {
if height > new_tip_height {
continue;
}
let fetched_hash = match fetched_blocks.entry(height) {
btree_map::Entry::Occupied(entry) => *entry.get(),
btree_map::Entry::Vacant(entry) => {
*entry.insert(self.get_block_hash(height).await?)
}
};
// We have found point of agreement so the update will connect!
if fetched_hash == local_hash {
break;
}
}
Ok(local_chain::Update {
tip: CheckPoint::from_block_ids(fetched_blocks.into_iter().map(BlockId::from))
.expect("must be in height order"),
introduce_older_blocks: true,
})
}
async fn full_scan<K: Ord + Clone + Send>(
&self,
local_tip: CheckPoint,
keychain_spks: BTreeMap<
K,
impl IntoIterator<IntoIter = impl Iterator<Item = (u32, ScriptBuf)> + Send> + Send,
>,
stop_gap: usize,
parallel_requests: usize,
) -> Result<(TxGraph<ConfirmationTimeHeightAnchor>, BTreeMap<K, u32>), Error> {
type TxsOfSpkIndex = (u32, Vec<esplora_client::Tx>);
let parallel_requests = Ord::max(parallel_requests, 1);
let mut graph = TxGraph::<ConfirmationTimeHeightAnchor>::default();
let mut last_active_indexes = BTreeMap::<K, u32>::new();
let stop_gap = Ord::max(stop_gap, 1);
for (keychain, spks) in keychain_spks {
let mut spks = spks.into_iter();
let mut last_index = Option::<u32>::None;
let mut last_active_index = Option::<u32>::None;
loop {
let handles = spks
.by_ref()
.take(parallel_requests)
.map(|(spk_index, spk)| {
let client = self.clone();
async move {
let mut last_seen = None;
let mut spk_txs = Vec::new();
loop {
let txs = client.scripthash_txs(&spk, last_seen).await?;
let tx_count = txs.len();
last_seen = txs.last().map(|tx| tx.txid);
spk_txs.extend(txs);
if tx_count < 25 {
break Result::<_, Error>::Ok((spk_index, spk_txs));
}
}
}
})
.collect::<FuturesOrdered<_>>();
if handles.is_empty() {
break;
}
for (index, txs) in handles.try_collect::<Vec<TxsOfSpkIndex>>().await? {
last_index = Some(index);
if !txs.is_empty() {
last_active_index = Some(index);
}
for tx in txs {
let _ = graph.insert_tx(tx.to_tx());
if let Some(anchor) = anchor_from_status(&tx.status) {
let _ = graph.insert_anchor(tx.txid, anchor);
}
let previous_outputs = tx.vin.iter().filter_map(|vin| {
let prevout = vin.prevout.as_ref()?;
Some((
OutPoint {
txid: vin.txid,
vout: vin.vout,
},
TxOut {
script_pubkey: prevout.scriptpubkey.clone(),
value: Amount::from_sat(prevout.value),
},
))
});
for (outpoint, txout) in previous_outputs {
let _ = graph.insert_txout(outpoint, txout);
}
}
}
let last_index = last_index.expect("Must be set since handles wasn't empty.");
let gap_limit_reached = if let Some(i) = last_active_index {
last_index >= i.saturating_add(stop_gap as u32)
} else {
last_index + 1 >= stop_gap as u32
};
if gap_limit_reached {
break;
}
}
if let Some(last_active_index) = last_active_index {
last_active_indexes.insert(keychain, last_active_index);
}
}
Ok((graph, last_active_indexes))
) -> Result<FullScanUpdate<K>, Error> {
let latest_blocks = fetch_latest_blocks(self).await?;
let (tx_graph, last_active_indices) =
full_scan_for_index_and_graph(self, keychain_spks, stop_gap, parallel_requests).await?;
let local_chain =
chain_update(self, &latest_blocks, &local_tip, tx_graph.all_anchors()).await?;
Ok(FullScanUpdate {
local_chain,
tx_graph,
last_active_indices,
})
}
async fn sync(
&self,
local_tip: CheckPoint,
misc_spks: impl IntoIterator<IntoIter = impl Iterator<Item = ScriptBuf> + 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<TxGraph<ConfirmationTimeHeightAnchor>, Error> {
let mut graph = self
.full_scan(
[(
(),
misc_spks
.into_iter()
.enumerate()
.map(|(i, spk)| (i as u32, spk)),
)]
.into(),
usize::MAX,
parallel_requests,
)
.await
.map(|(g, _)| g)?;
) -> Result<SyncUpdate, Error> {
let latest_blocks = fetch_latest_blocks(self).await?;
let tx_graph =
sync_for_index_and_graph(self, misc_spks, txids, outpoints, parallel_requests).await?;
let local_chain =
chain_update(self, &latest_blocks, &local_tip, tx_graph.all_anchors()).await?;
Ok(SyncUpdate {
tx_graph,
local_chain,
})
}
}
/// Fetch latest blocks from Esplora in an atomic call.
///
/// We want to do this before fetching transactions and anchors as we cannot fetch latest blocks AND
/// transactions atomically, and the checkpoint tip is used to determine last-scanned block (for
/// block-based chain-sources). Therefore it's better to be conservative when setting the tip (use
/// an earlier tip rather than a later tip) otherwise the caller may accidentally skip blocks when
/// alternating between chain-sources.
async fn fetch_latest_blocks(
client: &esplora_client::AsyncClient,
) -> Result<BTreeMap<u32, BlockHash>, Error> {
Ok(client
.get_blocks(None)
.await?
.into_iter()
.map(|b| (b.time.height, b.id))
.collect())
}
/// Used instead of [`esplora_client::BlockingClient::get_block_hash`].
///
/// This first checks the previously fetched `latest_blocks` before fetching from Esplora again.
async fn fetch_block(
client: &esplora_client::AsyncClient,
latest_blocks: &BTreeMap<u32, BlockHash>,
height: u32,
) -> Result<Option<BlockHash>, Error> {
if let Some(&hash) = latest_blocks.get(&height) {
return Ok(Some(hash));
}
// We avoid fetching blocks higher than previously fetched `latest_blocks` as the local chain
// tip is used to signal for the last-synced-up-to-height.
let &tip_height = latest_blocks
.keys()
.last()
.expect("must have atleast one entry");
if height > tip_height {
return Ok(None);
}
Ok(Some(client.get_block_hash(height).await?))
}
/// Create the [`local_chain::Update`].
///
/// We want to have a corresponding checkpoint per anchor height. However, checkpoints fetched
/// should not surpass `latest_blocks`.
async fn chain_update<A: Anchor>(
client: &esplora_client::AsyncClient,
latest_blocks: &BTreeMap<u32, BlockHash>,
local_tip: &CheckPoint,
anchors: &BTreeSet<(A, Txid)>,
) -> Result<CheckPoint, Error> {
let mut point_of_agreement = None;
let mut conflicts = vec![];
for local_cp in local_tip.iter() {
let remote_hash = match fetch_block(client, latest_blocks, local_cp.height()).await? {
Some(hash) => hash,
None => continue,
};
if remote_hash == local_cp.hash() {
point_of_agreement = Some(local_cp.clone());
break;
} else {
// it is not strictly necessary to include all the conflicted heights (we do need the
// first one) but it seems prudent to make sure the updated chain's heights are a
// superset of the existing chain after update.
conflicts.push(BlockId {
height: local_cp.height(),
hash: remote_hash,
});
}
}
let mut tip = point_of_agreement.expect("remote esplora should have same genesis block");
tip = tip
.extend(conflicts.into_iter().rev())
.expect("evicted are in order");
for anchor in anchors {
let height = anchor.0.anchor_block().height;
if tip.get(height).is_none() {
let hash = match fetch_block(client, latest_blocks, height).await? {
Some(hash) => hash,
None => continue,
};
tip = tip.insert(BlockId { height, hash });
}
}
// insert the most recent blocks at the tip to make sure we update the tip and make the update
// robust.
for (&height, &hash) in latest_blocks.iter() {
tip = tip.insert(BlockId { height, hash });
}
Ok(tip)
}
/// This performs a full scan to get an update for the [`TxGraph`] and
/// [`KeychainTxOutIndex`](bdk_chain::keychain::KeychainTxOutIndex).
async fn full_scan_for_index_and_graph<K: Ord + Clone + Send>(
client: &esplora_client::AsyncClient,
keychain_spks: BTreeMap<
K,
impl IntoIterator<IntoIter = impl Iterator<Item = (u32, ScriptBuf)> + Send> + Send,
>,
stop_gap: usize,
parallel_requests: usize,
) -> Result<(TxGraph<ConfirmationTimeHeightAnchor>, BTreeMap<K, u32>), Error> {
type TxsOfSpkIndex = (u32, Vec<esplora_client::Tx>);
let parallel_requests = Ord::max(parallel_requests, 1);
let mut graph = TxGraph::<ConfirmationTimeHeightAnchor>::default();
let mut last_active_indexes = BTreeMap::<K, u32>::new();
for (keychain, spks) in keychain_spks {
let mut spks = spks.into_iter();
let mut last_index = Option::<u32>::None;
let mut last_active_index = Option::<u32>::None;
let mut txids = txids.into_iter();
loop {
let handles = txids
let handles = spks
.by_ref()
.take(parallel_requests)
.filter(|&txid| graph.get_tx(txid).is_none())
.map(|txid| {
let client = self.clone();
async move { client.get_tx_status(&txid).await.map(|s| (txid, s)) }
.map(|(spk_index, spk)| {
let client = client.clone();
async move {
let mut last_seen = None;
let mut spk_txs = Vec::new();
loop {
let txs = client.scripthash_txs(&spk, last_seen).await?;
let tx_count = txs.len();
last_seen = txs.last().map(|tx| tx.txid);
spk_txs.extend(txs);
if tx_count < 25 {
break Result::<_, Error>::Ok((spk_index, spk_txs));
}
}
}
})
.collect::<FuturesOrdered<_>>();
@@ -297,38 +275,315 @@ impl EsploraAsyncExt for esplora_client::AsyncClient {
break;
}
for (txid, status) in handles.try_collect::<Vec<(Txid, TxStatus)>>().await? {
if let Some(anchor) = anchor_from_status(&status) {
let _ = graph.insert_anchor(txid, anchor);
for (index, txs) in handles.try_collect::<Vec<TxsOfSpkIndex>>().await? {
last_index = Some(index);
if !txs.is_empty() {
last_active_index = Some(index);
}
for tx in txs {
let _ = graph.insert_tx(tx.to_tx());
if let Some(anchor) = anchor_from_status(&tx.status) {
let _ = graph.insert_anchor(tx.txid, anchor);
}
let previous_outputs = tx.vin.iter().filter_map(|vin| {
let prevout = vin.prevout.as_ref()?;
Some((
OutPoint {
txid: vin.txid,
vout: vin.vout,
},
TxOut {
script_pubkey: prevout.scriptpubkey.clone(),
value: Amount::from_sat(prevout.value),
},
))
});
for (outpoint, txout) in previous_outputs {
let _ = graph.insert_txout(outpoint, txout);
}
}
}
let last_index = last_index.expect("Must be set since handles wasn't empty.");
let gap_limit_reached = if let Some(i) = last_active_index {
last_index >= i.saturating_add(stop_gap as u32)
} else {
last_index + 1 >= stop_gap as u32
};
if gap_limit_reached {
break;
}
}
for op in outpoints.into_iter() {
if graph.get_tx(op.txid).is_none() {
if let Some(tx) = self.get_tx(&op.txid).await? {
let _ = graph.insert_tx(tx);
}
let status = self.get_tx_status(&op.txid).await?;
if let Some(anchor) = anchor_from_status(&status) {
let _ = graph.insert_anchor(op.txid, anchor);
}
}
if let Some(last_active_index) = last_active_index {
last_active_indexes.insert(keychain, last_active_index);
}
}
if let Some(op_status) = self.get_output_status(&op.txid, op.vout as _).await? {
if let Some(txid) = op_status.txid {
if graph.get_tx(txid).is_none() {
if let Some(tx) = self.get_tx(&txid).await? {
let _ = graph.insert_tx(tx);
}
let status = self.get_tx_status(&txid).await?;
if let Some(anchor) = anchor_from_status(&status) {
let _ = graph.insert_anchor(txid, anchor);
}
Ok((graph, last_active_indexes))
}
async fn sync_for_index_and_graph(
client: &esplora_client::AsyncClient,
misc_spks: impl IntoIterator<IntoIter = impl Iterator<Item = ScriptBuf> + 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<TxGraph<ConfirmationTimeHeightAnchor>, Error> {
let mut graph = full_scan_for_index_and_graph(
client,
[(
(),
misc_spks
.into_iter()
.enumerate()
.map(|(i, spk)| (i as u32, spk)),
)]
.into(),
usize::MAX,
parallel_requests,
)
.await
.map(|(g, _)| g)?;
let mut txids = txids.into_iter();
loop {
let handles = txids
.by_ref()
.take(parallel_requests)
.filter(|&txid| graph.get_tx(txid).is_none())
.map(|txid| {
let client = client.clone();
async move { client.get_tx_status(&txid).await.map(|s| (txid, s)) }
})
.collect::<FuturesOrdered<_>>();
if handles.is_empty() {
break;
}
for (txid, status) in handles.try_collect::<Vec<(Txid, TxStatus)>>().await? {
if let Some(anchor) = anchor_from_status(&status) {
let _ = graph.insert_anchor(txid, anchor);
}
}
}
for op in outpoints.into_iter() {
if graph.get_tx(op.txid).is_none() {
if let Some(tx) = client.get_tx(&op.txid).await? {
let _ = graph.insert_tx(tx);
}
let status = client.get_tx_status(&op.txid).await?;
if let Some(anchor) = anchor_from_status(&status) {
let _ = graph.insert_anchor(op.txid, anchor);
}
}
if let Some(op_status) = client.get_output_status(&op.txid, op.vout as _).await? {
if let Some(txid) = op_status.txid {
if graph.get_tx(txid).is_none() {
if let Some(tx) = client.get_tx(&txid).await? {
let _ = graph.insert_tx(tx);
}
let status = client.get_tx_status(&txid).await?;
if let Some(anchor) = anchor_from_status(&status) {
let _ = graph.insert_anchor(txid, anchor);
}
}
}
}
Ok(graph)
}
Ok(graph)
}
#[cfg(test)]
mod test {
use std::{collections::BTreeSet, time::Duration};
use bdk_chain::{
bitcoin::{hashes::Hash, Txid},
local_chain::LocalChain,
BlockId,
};
use bdk_testenv::TestEnv;
use electrsd::bitcoind::bitcoincore_rpc::RpcApi;
use esplora_client::Builder;
use crate::async_ext::{chain_update, fetch_latest_blocks};
macro_rules! h {
($index:literal) => {{
bdk_chain::bitcoin::hashes::Hash::hash($index.as_bytes())
}};
}
/// Ensure that update does not remove heights (from original), and all anchor heights are included.
#[tokio::test]
pub async fn test_finalize_chain_update() -> anyhow::Result<()> {
struct TestCase<'a> {
name: &'a str,
/// Initial blockchain height to start the env with.
initial_env_height: u32,
/// Initial checkpoint heights to start with.
initial_cps: &'a [u32],
/// The final blockchain height of the env.
final_env_height: u32,
/// The anchors to test with: `(height, txid)`. Only the height is provided as we can fetch
/// the blockhash from the env.
anchors: &'a [(u32, Txid)],
}
let test_cases = [
TestCase {
name: "chain_extends",
initial_env_height: 60,
initial_cps: &[59, 60],
final_env_height: 90,
anchors: &[],
},
TestCase {
name: "introduce_older_heights",
initial_env_height: 50,
initial_cps: &[10, 15],
final_env_height: 50,
anchors: &[(11, h!("A")), (14, h!("B"))],
},
TestCase {
name: "introduce_older_heights_after_chain_extends",
initial_env_height: 50,
initial_cps: &[10, 15],
final_env_height: 100,
anchors: &[(11, h!("A")), (14, h!("B"))],
},
];
for (i, t) in test_cases.into_iter().enumerate() {
println!("[{}] running test case: {}", i, t.name);
let env = TestEnv::new()?;
let base_url = format!("http://{}", &env.electrsd.esplora_url.clone().unwrap());
let client = Builder::new(base_url.as_str()).build_async()?;
// set env to `initial_env_height`
if let Some(to_mine) = t
.initial_env_height
.checked_sub(env.make_checkpoint_tip().height())
{
env.mine_blocks(to_mine as _, None)?;
}
while client.get_height().await? < t.initial_env_height {
std::thread::sleep(Duration::from_millis(10));
}
// craft initial `local_chain`
let local_chain = {
let (mut chain, _) = LocalChain::from_genesis_hash(env.genesis_hash()?);
// force `chain_update_blocking` to add all checkpoints in `t.initial_cps`
let anchors = t
.initial_cps
.iter()
.map(|&height| -> anyhow::Result<_> {
Ok((
BlockId {
height,
hash: env.bitcoind.client.get_block_hash(height as _)?,
},
Txid::all_zeros(),
))
})
.collect::<anyhow::Result<BTreeSet<_>>>()?;
let update = chain_update(
&client,
&fetch_latest_blocks(&client).await?,
&chain.tip(),
&anchors,
)
.await?;
chain.apply_update(update)?;
chain
};
println!("local chain height: {}", local_chain.tip().height());
// extend env chain
if let Some(to_mine) = t
.final_env_height
.checked_sub(env.make_checkpoint_tip().height())
{
env.mine_blocks(to_mine as _, None)?;
}
while client.get_height().await? < t.final_env_height {
std::thread::sleep(Duration::from_millis(10));
}
// craft update
let update = {
let anchors = t
.anchors
.iter()
.map(|&(height, txid)| -> anyhow::Result<_> {
Ok((
BlockId {
height,
hash: env.bitcoind.client.get_block_hash(height as _)?,
},
txid,
))
})
.collect::<anyhow::Result<_>>()?;
chain_update(
&client,
&fetch_latest_blocks(&client).await?,
&local_chain.tip(),
&anchors,
)
.await?
};
// apply update
let mut updated_local_chain = local_chain.clone();
updated_local_chain.apply_update(update)?;
println!(
"updated local chain height: {}",
updated_local_chain.tip().height()
);
assert!(
{
let initial_heights = local_chain
.iter_checkpoints()
.map(|cp| cp.height())
.collect::<BTreeSet<_>>();
let updated_heights = updated_local_chain
.iter_checkpoints()
.map(|cp| cp.height())
.collect::<BTreeSet<_>>();
updated_heights.is_superset(&initial_heights)
},
"heights from the initial chain must all be in the updated chain",
);
assert!(
{
let exp_anchor_heights = t
.anchors
.iter()
.map(|(h, _)| *h)
.chain(t.initial_cps.iter().copied())
.collect::<BTreeSet<_>>();
let anchor_heights = updated_local_chain
.iter_checkpoints()
.map(|cp| cp.height())
.collect::<BTreeSet<_>>();
anchor_heights.is_superset(&exp_anchor_heights)
},
"anchor heights must all be in updated chain",
);
}
Ok(())
}
}

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

File diff suppressed because it is too large Load Diff

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

@@ -16,7 +16,9 @@
//! [`TxGraph`]: bdk_chain::tx_graph::TxGraph
//! [`example_esplora`]: https://github.com/bitcoindevkit/bdk/tree/master/example-crates/example_esplora
use bdk_chain::{BlockId, ConfirmationTimeHeightAnchor};
use std::collections::BTreeMap;
use bdk_chain::{local_chain::CheckPoint, BlockId, ConfirmationTimeHeightAnchor, TxGraph};
use esplora_client::TxStatus;
pub use esplora_client;
@@ -48,3 +50,21 @@ fn anchor_from_status(status: &TxStatus) -> Option<ConfirmationTimeHeightAnchor>
None
}
}
/// Update returns from a full scan.
pub struct FullScanUpdate<K> {
/// The update to apply to the receiving [`LocalChain`](bdk_chain::local_chain::LocalChain).
pub local_chain: CheckPoint,
/// The update to apply to the receiving [`TxGraph`].
pub tx_graph: TxGraph<ConfirmationTimeHeightAnchor>,
/// Last active indices for the corresponding keychains (`K`).
pub last_active_indices: BTreeMap<K, u32>,
}
/// Update returned from a sync.
pub struct SyncUpdate {
/// The update to apply to the receiving [`LocalChain`](bdk_chain::local_chain::LocalChain).
pub local_chain: CheckPoint,
/// The update to apply to the receiving [`TxGraph`].
pub tx_graph: TxGraph<ConfirmationTimeHeightAnchor>,
}

69
crates/esplora/tests/async_ext.rs
View File

@@ -2,7 +2,7 @@ use bdk_esplora::EsploraAsyncExt;
use electrsd::bitcoind::anyhow;
use electrsd::bitcoind::bitcoincore_rpc::RpcApi;
use esplora_client::{self, Builder};
use std::collections::{BTreeMap, HashSet};
use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::str::FromStr;
use std::thread::sleep;
use std::time::Duration;
@@ -52,8 +52,12 @@ pub async fn test_update_tx_graph_without_keychain() -> anyhow::Result<()> {
sleep(Duration::from_millis(10))
}
let graph_update = client
// use a full checkpoint linked list (since this is not what we are testing)
let cp_tip = env.make_checkpoint_tip();
let sync_update = client
.sync(
cp_tip.clone(),
misc_spks.into_iter(),
vec![].into_iter(),
vec![].into_iter(),
@@ -61,6 +65,23 @@ pub async fn test_update_tx_graph_without_keychain() -> anyhow::Result<()> {
)
.await?;
assert!(
{
let update_cps = sync_update
.local_chain
.iter()
.map(|cp| cp.block_id())
.collect::<BTreeSet<_>>();
let superset_cps = cp_tip
.iter()
.map(|cp| cp.block_id())
.collect::<BTreeSet<_>>();
superset_cps.is_superset(&update_cps)
},
"update should not alter original checkpoint tip since we already started with all checkpoints",
);
let graph_update = sync_update.tx_graph;
// Check to see if we have the floating txouts available from our two created transactions'
// previous outputs in order to calculate transaction fees.
for tx in graph_update.full_txs() {
@@ -140,14 +161,24 @@ pub async fn test_async_update_tx_graph_stop_gap() -> anyhow::Result<()> {
sleep(Duration::from_millis(10))
}
// use a full checkpoint linked list (since this is not what we are testing)
let cp_tip = env.make_checkpoint_tip();
// A scan with a gap limit of 3 won't find the transaction, but a scan with a gap limit of 4
// will.
let (graph_update, active_indices) = client.full_scan(keychains.clone(), 3, 1).await?;
assert!(graph_update.full_txs().next().is_none());
assert!(active_indices.is_empty());
let (graph_update, active_indices) = client.full_scan(keychains.clone(), 4, 1).await?;
assert_eq!(graph_update.full_txs().next().unwrap().txid, txid_4th_addr);
assert_eq!(active_indices[&0], 3);
let full_scan_update = client
.full_scan(cp_tip.clone(), keychains.clone(), 3, 1)
.await?;
assert!(full_scan_update.tx_graph.full_txs().next().is_none());
assert!(full_scan_update.last_active_indices.is_empty());
let full_scan_update = client
.full_scan(cp_tip.clone(), keychains.clone(), 4, 1)
.await?;
assert_eq!(
full_scan_update.tx_graph.full_txs().next().unwrap().txid,
txid_4th_addr
);
assert_eq!(full_scan_update.last_active_indices[&0], 3);
// Now receive a coin on the last address.
let txid_last_addr = env.bitcoind.client.send_to_address(
@@ -167,16 +198,26 @@ pub async fn test_async_update_tx_graph_stop_gap() -> anyhow::Result<()> {
// A scan with gap limit 5 won't find the second transaction, but a scan with gap limit 6 will.
// The last active indice won't be updated in the first case but will in the second one.
let (graph_update, active_indices) = client.full_scan(keychains.clone(), 5, 1).await?;
let txs: HashSet<_> = graph_update.full_txs().map(|tx| tx.txid).collect();
let full_scan_update = client
.full_scan(cp_tip.clone(), keychains.clone(), 5, 1)
.await?;
let txs: HashSet<_> = full_scan_update
.tx_graph
.full_txs()
.map(|tx| tx.txid)
.collect();
assert_eq!(txs.len(), 1);
assert!(txs.contains(&txid_4th_addr));
assert_eq!(active_indices[&0], 3);
let (graph_update, active_indices) = client.full_scan(keychains, 6, 1).await?;
let txs: HashSet<_> = graph_update.full_txs().map(|tx| tx.txid).collect();
assert_eq!(full_scan_update.last_active_indices[&0], 3);
let full_scan_update = client.full_scan(cp_tip, keychains, 6, 1).await?;
let txs: HashSet<_> = full_scan_update
.tx_graph
.full_txs()
.map(|tx| tx.txid)
.collect();
assert_eq!(txs.len(), 2);
assert!(txs.contains(&txid_4th_addr) && txs.contains(&txid_last_addr));
assert_eq!(active_indices[&0], 9);
assert_eq!(full_scan_update.last_active_indices[&0], 9);
Ok(())
}

255
crates/esplora/tests/blocking_ext.rs
View File

@@ -1,5 +1,3 @@
use bdk_chain::local_chain::LocalChain;
use bdk_chain::BlockId;
use bdk_esplora::EsploraExt;
use electrsd::bitcoind::anyhow;
use electrsd::bitcoind::bitcoincore_rpc::RpcApi;
@@ -12,20 +10,6 @@ use std::time::Duration;
use bdk_chain::bitcoin::{Address, Amount, Txid};
use bdk_testenv::TestEnv;
macro_rules! h {
($index:literal) => {{
bdk_chain::bitcoin::hashes::Hash::hash($index.as_bytes())
}};
}
macro_rules! local_chain {
[ $(($height:expr, $block_hash:expr)), * ] => {{
#[allow(unused_mut)]
bdk_chain::local_chain::LocalChain::from_blocks([$(($height, $block_hash).into()),*].into_iter().collect())
.expect("chain must have genesis block")
}};
}
#[test]
pub fn test_update_tx_graph_without_keychain() -> anyhow::Result<()> {
let env = TestEnv::new()?;
@@ -68,13 +52,34 @@ pub fn test_update_tx_graph_without_keychain() -> anyhow::Result<()> {
sleep(Duration::from_millis(10))
}
let graph_update = client.sync(
// use a full checkpoint linked list (since this is not what we are testing)
let cp_tip = env.make_checkpoint_tip();
let sync_update = client.sync(
cp_tip.clone(),
misc_spks.into_iter(),
vec![].into_iter(),
vec![].into_iter(),
1,
)?;
assert!(
{
let update_cps = sync_update
.local_chain
.iter()
.map(|cp| cp.block_id())
.collect::<BTreeSet<_>>();
let superset_cps = cp_tip
.iter()
.map(|cp| cp.block_id())
.collect::<BTreeSet<_>>();
superset_cps.is_superset(&update_cps)
},
"update should not alter original checkpoint tip since we already started with all checkpoints",
);
let graph_update = sync_update.tx_graph;
// Check to see if we have the floating txouts available from our two created transactions'
// previous outputs in order to calculate transaction fees.
for tx in graph_update.full_txs() {
@@ -155,14 +160,20 @@ pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> {
sleep(Duration::from_millis(10))
}
// use a full checkpoint linked list (since this is not what we are testing)
let cp_tip = env.make_checkpoint_tip();
// A scan with a stop_gap of 3 won't find the transaction, but a scan with a gap limit of 4
// will.
let (graph_update, active_indices) = client.full_scan(keychains.clone(), 3, 1)?;
assert!(graph_update.full_txs().next().is_none());
assert!(active_indices.is_empty());
let (graph_update, active_indices) = client.full_scan(keychains.clone(), 4, 1)?;
assert_eq!(graph_update.full_txs().next().unwrap().txid, txid_4th_addr);
assert_eq!(active_indices[&0], 3);
let full_scan_update = client.full_scan(cp_tip.clone(), keychains.clone(), 3, 1)?;
assert!(full_scan_update.tx_graph.full_txs().next().is_none());
assert!(full_scan_update.last_active_indices.is_empty());
let full_scan_update = client.full_scan(cp_tip.clone(), keychains.clone(), 4, 1)?;
assert_eq!(
full_scan_update.tx_graph.full_txs().next().unwrap().txid,
txid_4th_addr
);
assert_eq!(full_scan_update.last_active_indices[&0], 3);
// Now receive a coin on the last address.
let txid_last_addr = env.bitcoind.client.send_to_address(
@@ -182,194 +193,24 @@ pub fn test_update_tx_graph_stop_gap() -> anyhow::Result<()> {
// A scan with gap limit 5 won't find the second transaction, but a scan with gap limit 6 will.
// The last active indice won't be updated in the first case but will in the second one.
let (graph_update, active_indices) = client.full_scan(keychains.clone(), 5, 1)?;
let txs: HashSet<_> = graph_update.full_txs().map(|tx| tx.txid).collect();
let full_scan_update = client.full_scan(cp_tip.clone(), keychains.clone(), 5, 1)?;
let txs: HashSet<_> = full_scan_update
.tx_graph
.full_txs()
.map(|tx| tx.txid)
.collect();
assert_eq!(txs.len(), 1);
assert!(txs.contains(&txid_4th_addr));
assert_eq!(active_indices[&0], 3);
let (graph_update, active_indices) = client.full_scan(keychains, 6, 1)?;
let txs: HashSet<_> = graph_update.full_txs().map(|tx| tx.txid).collect();
assert_eq!(full_scan_update.last_active_indices[&0], 3);
let full_scan_update = client.full_scan(cp_tip.clone(), keychains, 6, 1)?;
let txs: HashSet<_> = full_scan_update
.tx_graph
.full_txs()
.map(|tx| tx.txid)
.collect();
assert_eq!(txs.len(), 2);
assert!(txs.contains(&txid_4th_addr) && txs.contains(&txid_last_addr));
assert_eq!(active_indices[&0], 9);
Ok(())
}
#[test]
fn update_local_chain() -> anyhow::Result<()> {
const TIP_HEIGHT: u32 = 50;
let env = TestEnv::new()?;
let blocks = {
let bitcoind_client = &env.bitcoind.client;
assert_eq!(bitcoind_client.get_block_count()?, 1);
[
(0, bitcoind_client.get_block_hash(0)?),
(1, bitcoind_client.get_block_hash(1)?),
]
.into_iter()
.chain((2..).zip(env.mine_blocks((TIP_HEIGHT - 1) as usize, None)?))
.collect::<BTreeMap<_, _>>()
};
// so new blocks can be seen by Electrs
let env = env.reset_electrsd()?;
let base_url = format!("http://{}", &env.electrsd.esplora_url.clone().unwrap());
let client = Builder::new(base_url.as_str()).build_blocking();
struct TestCase {
name: &'static str,
chain: LocalChain,
request_heights: &'static [u32],
exp_update_heights: &'static [u32],
}
let test_cases = [
TestCase {
name: "request_later_blocks",
chain: local_chain![(0, blocks[&0]), (21, blocks[&21])],
request_heights: &[22, 25, 28],
exp_update_heights: &[21, 22, 25, 28],
},
TestCase {
name: "request_prev_blocks",
chain: local_chain![(0, blocks[&0]), (1, blocks[&1]), (5, blocks[&5])],
request_heights: &[4],
exp_update_heights: &[4, 5],
},
TestCase {
name: "request_prev_blocks_2",
chain: local_chain![(0, blocks[&0]), (1, blocks[&1]), (10, blocks[&10])],
request_heights: &[4, 6],
exp_update_heights: &[4, 6, 10],
},
TestCase {
name: "request_later_and_prev_blocks",
chain: local_chain![(0, blocks[&0]), (7, blocks[&7]), (11, blocks[&11])],
request_heights: &[8, 9, 15],
exp_update_heights: &[8, 9, 11, 15],
},
TestCase {
name: "request_tip_only",
chain: local_chain![(0, blocks[&0]), (5, blocks[&5]), (49, blocks[&49])],
request_heights: &[TIP_HEIGHT],
exp_update_heights: &[49],
},
TestCase {
name: "request_nothing",
chain: local_chain![(0, blocks[&0]), (13, blocks[&13]), (23, blocks[&23])],
request_heights: &[],
exp_update_heights: &[23],
},
TestCase {
name: "request_nothing_during_reorg",
chain: local_chain![(0, blocks[&0]), (13, blocks[&13]), (23, h!("23"))],
request_heights: &[],
exp_update_heights: &[13, 23],
},
TestCase {
name: "request_nothing_during_reorg_2",
chain: local_chain![
(0, blocks[&0]),
(21, blocks[&21]),
(22, h!("22")),
(23, h!("23"))
],
request_heights: &[],
exp_update_heights: &[21, 22, 23],
},
TestCase {
name: "request_prev_blocks_during_reorg",
chain: local_chain![
(0, blocks[&0]),
(21, blocks[&21]),
(22, h!("22")),
(23, h!("23"))
],
request_heights: &[17, 20],
exp_update_heights: &[17, 20, 21, 22, 23],
},
TestCase {
name: "request_later_blocks_during_reorg",
chain: local_chain![
(0, blocks[&0]),
(9, blocks[&9]),
(22, h!("22")),
(23, h!("23"))
],
request_heights: &[25, 27],
exp_update_heights: &[9, 22, 23, 25, 27],
},
TestCase {
name: "request_later_blocks_during_reorg_2",
chain: local_chain![(0, blocks[&0]), (9, h!("9"))],
request_heights: &[10],
exp_update_heights: &[0, 9, 10],
},
TestCase {
name: "request_later_and_prev_blocks_during_reorg",
chain: local_chain![(0, blocks[&0]), (1, blocks[&1]), (9, h!("9"))],
request_heights: &[8, 11],
exp_update_heights: &[1, 8, 9, 11],
},
];
for (i, t) in test_cases.into_iter().enumerate() {
println!("Case {}: {}", i, t.name);
let mut chain = t.chain;
let update = client
.update_local_chain(chain.tip(), t.request_heights.iter().copied())
.map_err(|err| {
anyhow::format_err!("[{}:{}] `update_local_chain` failed: {}", i, t.name, err)
})?;
let update_blocks = update
.tip
.iter()
.map(|cp| cp.block_id())
.collect::<BTreeSet<_>>();
let exp_update_blocks = t
.exp_update_heights
.iter()
.map(|&height| {
let hash = blocks[&height];
BlockId { height, hash }
})
.chain(
// Electrs Esplora `get_block` call fetches 10 blocks which is included in the
// update
blocks
.range(TIP_HEIGHT - 9..)
.map(|(&height, &hash)| BlockId { height, hash }),
)
.collect::<BTreeSet<_>>();
assert_eq!(
update_blocks, exp_update_blocks,
"[{}:{}] unexpected update",
i, t.name
);
let _ = chain
.apply_update(update)
.unwrap_or_else(|err| panic!("[{}:{}] update failed to apply: {}", i, t.name, err));
// all requested heights must exist in the final chain
for height in t.request_heights {
let exp_blockhash = blocks.get(height).expect("block must exist in bitcoind");
assert_eq!(
chain.get(*height).map(|cp| cp.hash()),
Some(*exp_blockhash),
"[{}:{}] block {}:{} must exist in final chain",
i,
t.name,
height,
exp_blockhash
);
}
}
assert_eq!(full_scan_update.last_active_indices[&0], 9);
Ok(())
}