refactor(bitcoind_rpc)!: store conversion closure in FilterIter and Emitter

evanlinjin · claude · evanlinjin · commit 25e8feb864e9 · 2026-03-04T07:38:59.000Z
Store `Fn(Header) -&gt; D` in `FilterIter` and `Fn(&amp;Block) -&gt; D` in
`Emitter`, enabling `Iterator` for any `D` without a `FromBlockHeader`
bound. Add `new_with` constructors for custom `D` types; `new` delegates
to `new_with` for the common `FromBlockHeader` case.

Remove `FilterIter::next_with` and `Emitter::next_block_with` — callers
should use `new_with` at construction time instead.

Co-Authored-By: Claude Sonnet 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/crates/bitcoind_rpc/src/bip158.rs b/crates/bitcoind_rpc/src/bip158.rs
@@ -13,6 +13,8 @@ use bdk_core::CheckPoint;
 use bdk_core::FromBlockHeader;
 use bdk_core::ToBlockHash;
 use bitcoin::block::Header;
+use bitcoin::hashes::Hash;
+use bitcoin::pow::CompactTarget;
 use bitcoin::BlockHash;
 use bitcoin::{bip158::BlockFilter, Block, ScriptBuf};
 use bitcoincore_rpc;
@@ -33,7 +35,6 @@ use bitcoincore_rpc::{json::GetBlockHeaderResult, RpcApi};
 ///   occur. `FilterIter` will continue to yield events until it reaches the latest chain tip.
 ///   Events contain the updated checkpoint `cp` which may be incorporated into the local chain
 ///   state to stay in sync with the tip.
-#[derive(Debug)]
 pub struct FilterIter<'a, D = BlockHash> {
     /// RPC client
     client: &'a bitcoincore_rpc::Client,
@@ -43,26 +44,56 @@ pub struct FilterIter<'a, D = BlockHash> {
     cp: CheckPoint<D>,
     /// Header info, contains the prev and next hashes for each header.
     header: Option<GetBlockHeaderResult>,
+    /// Closure to convert a block header into checkpoint data `D`.
+    to_data: Box<dyn Fn(Header) -> D + Send + Sync>,
 }
 
-impl<'a, D> FilterIter<'a, D>
-where
-    D: ToBlockHash + Clone + Debug,
-{
-    /// Construct [`FilterIter`] with checkpoint, RPC client and SPKs.
-    pub fn new(
+impl<D: Debug> core::fmt::Debug for FilterIter<'_, D> {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        f.debug_struct("FilterIter")
+            .field("spks", &self.spks)
+            .field("cp", &self.cp)
+            .field("header", &self.header)
+            .finish_non_exhaustive()
+    }
+}
+
+impl<'a, D: 'static> FilterIter<'a, D> {
+    /// Construct [`FilterIter`] with a custom closure to convert block headers into checkpoint
+    /// data.
+    ///
+    /// Use [`new`](Self::new) for the common case where `D` implements [`FromBlockHeader`].
+    pub fn new_with(
         client: &'a bitcoincore_rpc::Client,
         cp: CheckPoint<D>,
         spks: impl IntoIterator<Item = ScriptBuf>,
+        to_data: impl Fn(Header) -> D + Send + Sync + 'static,
     ) -> Self {
         Self {
             client,
             spks: spks.into_iter().collect(),
             cp,
             header: None,
+            to_data: Box::new(to_data),
         }
     }
+}
 
+impl<'a, D: FromBlockHeader + 'static> FilterIter<'a, D> {
+    /// Construct [`FilterIter`] with checkpoint, RPC client and SPKs.
+    pub fn new(
+        client: &'a bitcoincore_rpc::Client,
+        cp: CheckPoint<D>,
+        spks: impl IntoIterator<Item = ScriptBuf>,
+    ) -> Self {
+        Self::new_with(client, cp, spks, D::from_blockheader)
+    }
+}
+
+impl<'a, D> FilterIter<'a, D>
+where
+    D: ToBlockHash + Clone + Debug,
+{
     /// Return the agreement header with the remote node.
     ///
     /// Error if no agreement header is found.
@@ -78,10 +109,7 @@ where
         Err(Error::ReorgDepthExceeded)
     }
 
-    fn try_next_with<F>(&mut self, to_data: F) -> Result<Option<Event<D>>, Error>
-    where
-        F: Fn(Header) -> D,
-    {
+    fn try_next(&mut self) -> Result<Option<Event<D>>, Error> {
         let mut cp = self.cp.clone();
 
         let header = match self.header.take() {
@@ -111,10 +139,21 @@ where
         next_hash = next_header.hash;
         let next_height: u32 = next_header.height.try_into()?;
 
-        cp = cp.insert(
-            next_height,
-            to_data(self.client.get_block_header(&next_hash)?),
-        );
+        // Reconstruct the block header from the already-fetched GetBlockHeaderResult,
+        // avoiding an extra `get_block_header` RPC call.
+        let block_header = Header {
+            version: next_header.version,
+            prev_blockhash: next_header
+                .previous_block_hash
+                .unwrap_or_else(BlockHash::all_zeros),
+            merkle_root: next_header.merkle_root,
+            time: next_header.time as u32,
+            bits: CompactTarget::from_unprefixed_hex(&next_header.bits)
+                .map_err(|_| Error::InvalidBits(next_header.bits.clone()))?,
+            nonce: next_header.nonce,
+        };
+
+        cp = cp.insert(next_height, (self.to_data)(block_header));
 
         let mut block = None;
         let filter = BlockFilter::new(self.client.get_block_filter(&next_hash)?.filter.as_slice());
@@ -132,14 +171,6 @@ where
 
         Ok(Some(Event { cp, block }))
     }
-
-    /// Get the next event with a custom checkpoint data type.
-    pub fn next_with<F>(&mut self, to_data: F) -> Option<Result<Event<D>, Error>>
-    where
-        F: Fn(Header) -> D,
-    {
-        self.try_next_with(to_data).transpose()
-    }
 }
 
 /// Event returned by [`FilterIter`].
@@ -165,12 +196,12 @@ impl Event {
 
 impl<D> Iterator for FilterIter<'_, D>
 where
-    D: ToBlockHash + FromBlockHeader + Clone + Debug,
+    D: ToBlockHash + Clone + Debug,
 {
     type Item = Result<Event<D>, Error>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        self.try_next_with(D::from_blockheader).transpose()
+        self.try_next().transpose()
     }
 }
 
@@ -185,6 +216,8 @@ pub enum Error {
     ReorgDepthExceeded,
     /// Error converting an integer
     TryFromInt(core::num::TryFromIntError),
+    /// Invalid bits string from RPC
+    InvalidBits(String),
 }
 
 impl core::fmt::Display for Error {
@@ -194,6 +227,7 @@ impl core::fmt::Display for Error {
             Self::Bip158(e) => write!(f, "{e}"),
             Self::ReorgDepthExceeded => write!(f, "maximum reorg depth exceeded"),
             Self::TryFromInt(e) => write!(f, "{e}"),
+            Self::InvalidBits(s) => write!(f, "invalid bits string: {s}"),
         }
     }
 }
diff --git a/crates/bitcoind_rpc/src/lib.rs b/crates/bitcoind_rpc/src/lib.rs
@@ -55,6 +55,9 @@ pub struct Emitter<C, D = BlockHash> {
     /// sure the tip block is already emitted. When a block is emitted, the transactions in the
     /// block are removed from this field.
     mempool_snapshot: HashMap<Txid, Arc<Transaction>>,
+
+    /// Closure to convert a block into checkpoint data `D`.
+    to_cp_data: Box<dyn Fn(&Block) -> D + Send + Sync>,
 }
 
 /// Indicates that there are no initially-expected mempool transactions.
@@ -67,24 +70,17 @@ impl<C, D> Emitter<C, D>
 where
     C: Deref,
     C::Target: RpcApi,
-    D: ToBlockHash + Clone + Debug,
+    D: ToBlockHash + Clone + Debug + 'static,
 {
-    /// Construct a new [`Emitter`].
+    /// Construct a new [`Emitter`] with a custom closure to convert blocks into checkpoint data.
     ///
-    /// `last_cp` informs the emitter of the chain we are starting off with. This way, the emitter
-    /// can start emission from a block that connects to the original chain.
-    ///
-    /// `start_height` starts emission from a given height (if there are no conflicts with the
-    /// original chain).
-    ///
-    /// `expected_mempool_txs` is the initial set of unconfirmed transactions provided by the
-    /// wallet. This allows the [`Emitter`] to inform the wallet about relevant mempool evictions.
-    /// If it is known that the wallet is empty, [`NO_EXPECTED_MEMPOOL_TXS`] can be used.
-    pub fn new(
+    /// Use [`new`](Self::new) for the common case where `D` implements [`FromBlockHeader`].
+    pub fn new_with(
         client: C,
         last_cp: CheckPoint<D>,
         start_height: u32,
         expected_mempool_txs: impl IntoIterator<Item = impl Into<Arc<Transaction>>>,
+        to_cp_data: impl Fn(&Block) -> D + Send + Sync + 'static,
     ) -> Self {
         Self {
             client,
@@ -98,15 +94,56 @@ where
                     (tx.compute_txid(), tx)
                 })
                 .collect(),
+            to_cp_data: Box::new(to_cp_data),
         }
     }
+}
 
+impl<C, D> Emitter<C, D>
+where
+    C: Deref,
+    C::Target: RpcApi,
+    D: ToBlockHash + FromBlockHeader + Clone + Debug + 'static,
+{
+    /// Construct a new [`Emitter`].
+    ///
+    /// `last_cp` informs the emitter of the chain we are starting off with. This way, the emitter
+    /// can start emission from a block that connects to the original chain.
+    ///
+    /// `start_height` starts emission from a given height (if there are no conflicts with the
+    /// original chain).
+    ///
+    /// `expected_mempool_txs` is the initial set of unconfirmed transactions provided by the
+    /// wallet. This allows the [`Emitter`] to inform the wallet about relevant mempool evictions.
+    /// If it is known that the wallet is empty, [`NO_EXPECTED_MEMPOOL_TXS`] can be used.
+    pub fn new(
+        client: C,
+        last_cp: CheckPoint<D>,
+        start_height: u32,
+        expected_mempool_txs: impl IntoIterator<Item = impl Into<Arc<Transaction>>>,
+    ) -> Self {
+        Self::new_with(
+            client,
+            last_cp,
+            start_height,
+            expected_mempool_txs,
+            |block| D::from_blockheader(block.header),
+        )
+    }
+}
+
+impl<C, D> Emitter<C, D>
+where
+    C: Deref,
+    C::Target: RpcApi,
+    D: ToBlockHash + Clone + Debug + 'static,
+{
     /// Emit mempool transactions and any evicted [`Txid`]s.
     ///
     /// This method returns a [`MempoolEvent`] containing the full transactions (with their
     /// first-seen unix timestamps) that were emitted, and [`MempoolEvent::evicted`] which are
     /// any [`Txid`]s which were previously seen in the mempool and are now missing. Evicted txids
-    /// are only reported once the emitter’s checkpoint matches the RPC’s best block in both height
+    /// are only reported once the emitter's checkpoint matches the RPC's best block in both height
     /// and hash. Until `next_block()` advances the checkpoint to tip, `mempool()` will always
     /// return an empty `evicted` set.
     #[cfg(feature = "std")]
@@ -200,18 +237,9 @@ where
         Ok(mempool_event)
     }
 
-    /// Emit the next block, using `to_data` to construct checkpoint data from the block.
-    ///
-    /// This is the alternative to [`next_block`](Self::next_block) when [`FromBlockHeader`] isn't
-    /// implemented for `D`.
-    pub fn next_block_with<F>(
-        &mut self,
-        to_data: F,
-    ) -> Result<Option<BlockEvent<Block, D>>, bitcoincore_rpc::Error>
-    where
-        F: Fn(&Block) -> D,
-    {
-        if let Some((checkpoint, block)) = poll(self, to_data)? {
+    /// Emit the next block height and block (if any).
+    pub fn next_block(&mut self) -> Result<Option<BlockEvent<Block, D>>, bitcoincore_rpc::Error> {
+        if let Some((checkpoint, block)) = poll(self)? {
             // Stop tracking unconfirmed transactions that have been confirmed in this block.
             for tx in &block.txdata {
                 self.mempool_snapshot.remove(&tx.compute_txid());
@@ -220,14 +248,6 @@ where
         }
         Ok(None)
     }
-
-    /// Emit the next block height and block (if any).
-    pub fn next_block(&mut self) -> Result<Option<BlockEvent<Block, D>>, bitcoincore_rpc::Error>
-    where
-        D: FromBlockHeader,
-    {
-        self.next_block_with(|block| D::from_blockheader(block.header))
-    }
 }
 
 /// A new emission from mempool.
@@ -347,23 +367,21 @@ where
     Ok(PollResponse::AgreementPointNotFound(genesis_hash))
 }
 
-fn poll<C, D, F>(
+fn poll<C, D>(
     emitter: &mut Emitter<C, D>,
-    to_cp_data: F,
 ) -> Result<Option<(CheckPoint<D>, Block)>, bitcoincore_rpc::Error>
 where
     C: Deref,
     C::Target: RpcApi,
-    D: ToBlockHash + Clone + Debug,
-    F: Fn(&Block) -> D,
+    D: ToBlockHash + Clone + Debug + 'static,
 {
     let client = &emitter.client;
     loop {
         match poll_once(emitter)? {
             PollResponse::Block(res) => {
                 let height = res.height as u32;
                 let block = client.get_block(&res.hash)?;
-                let cp_data = to_cp_data(&block);
+                let cp_data = (emitter.to_cp_data)(&block);
 
                 let new_cp = emitter
                     .last_cp
@@ -390,7 +408,7 @@ where
             }
             PollResponse::AgreementPointNotFound(genesis_hash) => {
                 let block = client.get_block(&genesis_hash)?;
-                let cp_data = to_cp_data(&block);
+                let cp_data = (emitter.to_cp_data)(&block);
                 emitter.last_cp = CheckPoint::new(0, cp_data);
                 emitter.last_block = None;
                 continue;
