Skip to content

refactor(esplora): clear remaining panic paths #2053

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
wants to merge 1 commit into
base: master
Choose a base branch
from
+65 −44
Open

refactor(esplora): clear remaining panic paths #2053

Show file tree
Hide file tree
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
37 changes: 16 additions & 21 deletions crates/esplora/src/async_ext.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,7 @@ use bdk_core::{
use esplora_client::Sleeper;
use futures::{stream::FuturesOrdered, TryStreamExt};

use crate::{insert_anchor_or_seen_at_from_status, insert_prevouts};

/// [`esplora_client::Error`]
type Error = Box<esplora_client::Error>;
use crate::{insert_anchor_or_seen_at_from_status, insert_prevouts, Error};

/// Trait to extend the functionality of [`esplora_client::AsyncClient`].
///
Expand Down Expand Up @@ -256,15 +253,13 @@ async fn chain_update<S: Sleeper>(
let mut tip = match point_of_agreement {
Some(tip) => tip,
None => {
return Err(Box::new(esplora_client::Error::HeaderHashNotFound(
local_cp_hash,
)));
return Err(esplora_client::Error::HeaderHashNotFound(local_cp_hash).into());
}
};

tip = tip
.extend(conflicts.into_iter().rev().map(|b| (b.height, b.hash)))
.expect("evicted are in order");
.map_err(Error::Checkpoint)?;

for (anchor, _txid) in anchors {
let height = anchor.block_id.height;
Expand Down Expand Up @@ -314,8 +309,9 @@ where
type TxsOfSpkIndex = (u32, Vec<esplora_client::Tx>, HashSet<Txid>);

let mut update = TxUpdate::<ConfirmationBlockTime>::default();
let mut last_index = Option::<u32>::None;
let mut last_active_index = Option::<u32>::None;
let mut consecutive_unused = 0usize;
let gap_limit = stop_gap.max(parallel_requests.max(1));

loop {
let handles = keychain_spks
Expand Down Expand Up @@ -352,8 +348,10 @@ where
}

for (index, txs, evicted) in handles.try_collect::<Vec<TxsOfSpkIndex>>().await? {
last_index = Some(index);
if !txs.is_empty() {
if txs.is_empty() {
consecutive_unused = consecutive_unused.saturating_add(1);
} else {
consecutive_unused = 0;
last_active_index = Some(index);
}
for tx in txs {
Expand All @@ -368,13 +366,7 @@ where
.extend(evicted.into_iter().map(|txid| (txid, start_time)));
}

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 {
if consecutive_unused >= gap_limit {
break;
}
}
Expand Down Expand Up @@ -563,7 +555,10 @@ mod test {
use bdk_testenv::{anyhow, bitcoincore_rpc::RpcApi, TestEnv};
use esplora_client::Builder;

use crate::async_ext::{chain_update, fetch_latest_blocks};
use crate::{
async_ext::{chain_update, fetch_latest_blocks},
Error as EsploraError,
};

macro_rules! h {
($index:literal) => {{
Expand Down Expand Up @@ -594,9 +589,9 @@ mod test {

let anchors = BTreeSet::new();
let res = chain_update(&client, &latest_blocks, &cp, &anchors).await;
use esplora_client::Error;
use esplora_client::Error as ClientError;
assert!(
matches!(*res.unwrap_err(), Error::HeaderHashNotFound(hash) if hash == genesis_hash),
matches!(res.unwrap_err(), EsploraError::Client(ClientError::HeaderHashNotFound(hash)) if hash == genesis_hash),
"`chain_update` should error if it can't connect to the local CP",
);

Expand Down
35 changes: 14 additions & 21 deletions crates/esplora/src/blocking_ext.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,7 @@ use bdk_core::{
use esplora_client::{OutputStatus, Tx};
use std::thread::JoinHandle;

use crate::{insert_anchor_or_seen_at_from_status, insert_prevouts};

/// [`esplora_client::Error`]
pub type Error = Box<esplora_client::Error>;
use crate::{insert_anchor_or_seen_at_from_status, insert_prevouts, Error};

/// Trait to extend the functionality of [`esplora_client::BlockingClient`].
///
Expand Down Expand Up @@ -241,15 +238,13 @@ fn chain_update(
let mut tip = match point_of_agreement {
Some(tip) => tip,
None => {
return Err(Box::new(esplora_client::Error::HeaderHashNotFound(
local_cp_hash,
)));
return Err(esplora_client::Error::HeaderHashNotFound(local_cp_hash).into());
}
};

tip = tip
.extend(conflicts.into_iter().rev().map(|b| (b.height, b.hash)))
.expect("evicted are in order");
.map_err(Error::Checkpoint)?;
Comment on lines 245 to +247
Copy link
Member

@evanlinjin evanlinjin Nov 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I still think panic/expect here is more appropriate.

  1. CheckPoint::extend only errors if a block is not higher than the previous.
  2. The logic to populate conflicts uses heights from local_tip. local_tip is guaranteed to have ordered heights (since it is a CheckPoint type).


for (anchor, _) in anchors {
let height = anchor.block_id.height;
Expand Down Expand Up @@ -282,8 +277,9 @@ fn fetch_txs_with_keychain_spks<I: Iterator<Item = Indexed<SpkWithExpectedTxids>
type TxsOfSpkIndex = (u32, Vec<esplora_client::Tx>, HashSet<Txid>);

let mut update = TxUpdate::<ConfirmationBlockTime>::default();
let mut last_index = Option::<u32>::None;
let mut last_active_index = Option::<u32>::None;
let mut consecutive_unused = 0usize;
let gap_limit = stop_gap.max(1);

loop {
let handles = keychain_spks
Expand Down Expand Up @@ -321,8 +317,10 @@ fn fetch_txs_with_keychain_spks<I: Iterator<Item = Indexed<SpkWithExpectedTxids>

for handle in handles {
let (index, txs, evicted) = handle.join().expect("thread must not panic")?;
last_index = Some(index);
if !txs.is_empty() {
if txs.is_empty() {
consecutive_unused = consecutive_unused.saturating_add(1);
} else {
consecutive_unused = 0;
last_active_index = Some(index);
}
for tx in txs {
Expand All @@ -337,13 +335,7 @@ fn fetch_txs_with_keychain_spks<I: Iterator<Item = Indexed<SpkWithExpectedTxids>
.extend(evicted.into_iter().map(|txid| (txid, start_time)));
}

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 {
if consecutive_unused >= gap_limit {
break;
}
}
Expand Down Expand Up @@ -406,7 +398,7 @@ fn fetch_txs_with_txids<I: IntoIterator<Item = Txid>>(
std::thread::spawn(move || {
client
.get_tx_info(&txid)
.map_err(Box::new)
.map_err(Error::Client)
.map(|t| (txid, t))
})
})
Expand Down Expand Up @@ -468,7 +460,7 @@ fn fetch_txs_with_outpoints<I: IntoIterator<Item = OutPoint>>(
std::thread::spawn(move || {
client
.get_output_status(&op.txid, op.vout as _)
.map_err(Box::new)
.map_err(Error::Client)
})
})
.collect::<Vec<JoinHandle<Result<Option<OutputStatus>, Error>>>>();
Expand Down Expand Up @@ -512,6 +504,7 @@ fn fetch_txs_with_outpoints<I: IntoIterator<Item = OutPoint>>(
#[cfg_attr(coverage_nightly, coverage(off))]
mod test {
use crate::blocking_ext::{chain_update, fetch_latest_blocks};
use crate::Error as EsploraError;
use bdk_chain::bitcoin;
use bdk_chain::bitcoin::hashes::Hash;
use bdk_chain::bitcoin::Txid;
Expand Down Expand Up @@ -562,7 +555,7 @@ mod test {
let res = chain_update(&client, &latest_blocks, &cp, &anchors);
use esplora_client::Error;
assert!(
matches!(*res.unwrap_err(), Error::HeaderHashNotFound(hash) if hash == genesis_hash),
matches!(res.unwrap_err(), EsploraError::Client(Error::HeaderHashNotFound(hash)) if hash == genesis_hash),
"`chain_update` should error if it can't connect to the local CP",
);

Expand Down
37 changes: 35 additions & 2 deletions crates/esplora/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,10 @@
//! [`esplora_client::AsyncClient`].
#![cfg_attr(coverage_nightly, feature(coverage_attribute))]

use bdk_core::bitcoin::{Amount, OutPoint, TxOut, Txid};
use bdk_core::{BlockId, ConfirmationBlockTime, TxUpdate};
use bdk_core::bitcoin::{Amount, BlockHash, OutPoint, TxOut, Txid};
use bdk_core::{BlockId, CheckPoint, ConfirmationBlockTime, TxUpdate};
use esplora_client::TxStatus;
use std::fmt;

pub use esplora_client;

Expand All @@ -37,6 +38,38 @@ mod async_ext;
#[cfg(feature = "async")]
pub use async_ext::*;

#[derive(Debug)]
pub enum Error {
Client(esplora_client::Error),
Checkpoint(CheckPoint<BlockHash>),
}
Comment on lines +42 to +45
Copy link
Member

@evanlinjin evanlinjin Nov 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We're missing doc-comments for the Error type and it's variants!

We should make it obvious to the reader when the Error::CheckPoint variant occurs.

Edit: I don't think the CheckPoint variant is needed, so this Error type is not needed either.


impl From<esplora_client::Error> for Error {
fn from(err: esplora_client::Error) -> Self {
Self::Client(err)
}
}

impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Client(err) => write!(f, "{err}"),
Self::Checkpoint(cp) => {
write!(f, "checkpoint ordering error at height {}", cp.height())
}
}
}
}

impl std::error::Error for Error {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Client(err) => Some(err),
Self::Checkpoint(_) => None,
}
}
}

#[allow(dead_code)]
fn insert_anchor_or_seen_at_from_status(
update: &mut TxUpdate<ConfirmationBlockTime>,
Expand Down
Loading