refactor(esplora): clear remaining panic paths

reez · reez · commit 053159381d39 · 2025-10-30T14:20:27.000-05:00
diff --git a/crates/esplora/src/async_ext.rs b/crates/esplora/src/async_ext.rs
@@ -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`].
 ///
@@ -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;
@@ -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
@@ -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 {
@@ -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;
         }
     }
@@ -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) => {{
@@ -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",
         );
 
diff --git a/crates/esplora/src/blocking_ext.rs b/crates/esplora/src/blocking_ext.rs
@@ -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`].
 ///
@@ -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)?;
 
     for (anchor, _) in anchors {
         let height = anchor.block_id.height;
@@ -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
@@ -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 {
@@ -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;
         }
     }
@@ -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))
                 })
             })
@@ -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>>>>();
@@ -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;
@@ -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",
         );
 
diff --git a/crates/esplora/src/lib.rs b/crates/esplora/src/lib.rs
@@ -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;
 
@@ -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>),
+}
+
+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>,
