From aca98c3d951230360fdee5745315b1d854d0650a Mon Sep 17 00:00:00 2001 From: valued mammal Date: Sat, 11 Jul 2026 13:28:42 -0400 Subject: [PATCH 1/7] refactor(rpc)!: Replace bitcoincore-rpc with bdk_bitcoind_client bitcoincore-rpc is effectively unmaintained. Swap it out for bdk_bitcoind_client (bitreq backend + corepc_types), which is maintained under the bitcoindevkit org. The Emitter and FilterIter now borrow a concrete `&Client` instead of being generic over `C: Deref`, since the new client does not implement the old RpcApi trait. All RPC call sites and response types are migrated to the new client while preserving emission logic unchanged. Core v28 is the default (`28_0`); `29_0`/`30_0` passthrough features let callers target newer nodes. --- crates/bitcoind_rpc/Cargo.toml | 7 +- crates/bitcoind_rpc/examples/filter_iter.rs | 6 +- crates/bitcoind_rpc/src/bip158.rs | 49 ++----- crates/bitcoind_rpc/src/lib.rs | 137 +++++------------- crates/bitcoind_rpc/tests/common/mod.rs | 16 +- crates/bitcoind_rpc/tests/test_emitter.rs | 13 +- crates/bitcoind_rpc/tests/test_filter_iter.rs | 7 +- 7 files changed, 78 insertions(+), 157 deletions(-) diff --git a/crates/bitcoind_rpc/Cargo.toml b/crates/bitcoind_rpc/Cargo.toml index 572d1f5230..0cf58e8e4e 100644 --- a/crates/bitcoind_rpc/Cargo.toml +++ b/crates/bitcoind_rpc/Cargo.toml @@ -17,11 +17,11 @@ workspace = true [dependencies] bitcoin = { version = "0.32.0", default-features = false } -bitcoincore-rpc = { version = "0.19.0" } +bitcoind_client = { package = "bdk_bitcoind_client", version = "0.2.0", default-features = false, features = ["bitreq", "28_0"] } bdk_core = { path = "../core", version = "0.6.1", default-features = false } [dev-dependencies] -bdk_bitcoind_rpc = { path = "." } +bdk_bitcoind_rpc = { path = ".", features = ["28_0"] } bdk_testenv = { path = "../testenv" } bdk_chain = { path = "../chain" } @@ -29,6 +29,9 @@ bdk_chain = { path = "../chain" } default = ["std"] std = ["bitcoin/std", "bdk_core/std"] serde = ["bitcoin/serde", "bdk_core/serde"] +28_0 = ["bitcoind_client/28_0"] +29_0 = ["bitcoind_client/29_0"] +30_0 = ["bitcoind_client/30_0"] [[example]] name = "filter_iter" diff --git a/crates/bitcoind_rpc/examples/filter_iter.rs b/crates/bitcoind_rpc/examples/filter_iter.rs index a346a2db0e..03ceb17613 100644 --- a/crates/bitcoind_rpc/examples/filter_iter.rs +++ b/crates/bitcoind_rpc/examples/filter_iter.rs @@ -44,8 +44,10 @@ fn main() -> anyhow::Result<()> { // Configure RPC client let url = std::env::var("RPC_URL").context("must set RPC_URL")?; let cookie = std::env::var("RPC_COOKIE").context("must set RPC_COOKIE")?; - let rpc_client = - bitcoincore_rpc::Client::new(&url, bitcoincore_rpc::Auth::CookieFile(cookie.into()))?; + let rpc_client = bitcoind_client::bitreq::Client::with_auth( + &url, + bitcoind_client::bitreq::Auth::CookieFile(cookie.into()), + )?; // Initialize `FilterIter` let mut spks = vec![]; diff --git a/crates/bitcoind_rpc/src/bip158.rs b/crates/bitcoind_rpc/src/bip158.rs index 4caf59fcdc..82a3cfa2ec 100644 --- a/crates/bitcoind_rpc/src/bip158.rs +++ b/crates/bitcoind_rpc/src/bip158.rs @@ -10,14 +10,15 @@ use bdk_core::bitcoin; use bdk_core::CheckPoint; use bitcoin::BlockHash; use bitcoin::{bip158::BlockFilter, Block, ScriptBuf}; -use bitcoincore_rpc; -use bitcoincore_rpc::{json::GetBlockHeaderResult, RpcApi}; +use bitcoind_client::bitreq::Client; + +use crate::corepc_types::model::GetBlockHeaderVerbose; /// Type that returns Bitcoin blocks by matching a list of script pubkeys (SPKs) against a /// [`bip158::BlockFilter`](bitcoin::bip158::BlockFilter). /// /// * `FilterIter` talks to bitcoind via JSON-RPC interface, which is handled by the -/// [`bitcoincore_rpc::Client`]. +/// [`bitcoind_client::bitreq::Client`]. /// * Collect the script pubkeys (SPKs) you want to watch. These will usually correspond to wallet /// addresses that have been handed out for receiving payments. /// * Construct `FilterIter` with the RPC client, SPKs, and [`CheckPoint`]. The checkpoint tip @@ -31,19 +32,19 @@ use bitcoincore_rpc::{json::GetBlockHeaderResult, RpcApi}; #[derive(Debug)] pub struct FilterIter<'a> { /// RPC client - client: &'a bitcoincore_rpc::Client, + client: &'a Client, /// SPK inventory spks: Vec, /// checkpoint cp: CheckPoint, /// Header info, contains the prev and next hashes for each header. - header: Option, + header: Option, } impl<'a> FilterIter<'a> { /// Construct [`FilterIter`] with checkpoint, RPC client and SPKs. pub fn new( - client: &'a bitcoincore_rpc::Client, + client: &'a Client, cp: CheckPoint, spks: impl IntoIterator, ) -> Self { @@ -58,10 +59,10 @@ impl<'a> FilterIter<'a> { /// Return the agreement header with the remote node. /// /// Error if no agreement header is found. - fn find_base(&self) -> Result { + fn find_base(&self) -> Result { for cp in self.cp.iter() { - match self.client.get_block_header_info(&cp.hash()) { - Err(e) if is_not_found(&e) => continue, + match self.client.get_block_header_verbose(&cp.hash()) { + Err(e) if e.is_not_found_error() => continue, Ok(header) if header.confirmations <= 0 => continue, Ok(header) => return Ok(header), Err(e) => return Err(Error::Rpc(e)), @@ -111,7 +112,7 @@ impl Iterator for FilterIter<'_> { None => return Ok(None), }; - let mut next_header = self.client.get_block_header_info(&next_hash)?; + let mut next_header = self.client.get_block_header_verbose(&next_hash)?; // In case of a reorg, rewind by fetching headers of previous hashes until we find // one with enough confirmations. @@ -119,12 +120,12 @@ impl Iterator for FilterIter<'_> { let prev_hash = next_header .previous_block_hash .ok_or(Error::ReorgDepthExceeded)?; - let prev_header = self.client.get_block_header_info(&prev_hash)?; + let prev_header = self.client.get_block_header_verbose(&prev_hash)?; next_header = prev_header; } next_hash = next_header.hash; - let next_height: u32 = next_header.height.try_into()?; + let next_height = next_header.height; cp = cp.insert(next_height, next_hash); @@ -153,13 +154,11 @@ impl Iterator for FilterIter<'_> { #[derive(Debug)] pub enum Error { /// RPC error - Rpc(bitcoincore_rpc::Error), + Rpc(bitcoind_client::Error), /// `bitcoin::bip158` error Bip158(bitcoin::bip158::Error), /// Max reorg depth exceeded. ReorgDepthExceeded, - /// Error converting an integer - TryFromInt(core::num::TryFromIntError), } impl core::fmt::Display for Error { @@ -168,30 +167,14 @@ impl core::fmt::Display for Error { Self::Rpc(e) => write!(f, "{e}"), Self::Bip158(e) => write!(f, "{e}"), Self::ReorgDepthExceeded => write!(f, "maximum reorg depth exceeded"), - Self::TryFromInt(e) => write!(f, "{e}"), } } } impl core::error::Error for Error {} -impl From for Error { - fn from(e: bitcoincore_rpc::Error) -> Self { +impl From for Error { + fn from(e: bitcoind_client::Error) -> Self { Self::Rpc(e) } } - -impl From for Error { - fn from(e: core::num::TryFromIntError) -> Self { - Self::TryFromInt(e) - } -} - -/// Whether the RPC error is a "not found" error (code: `-5`). -fn is_not_found(e: &bitcoincore_rpc::Error) -> bool { - matches!( - e, - bitcoincore_rpc::Error::JsonRpc(bitcoincore_rpc::jsonrpc::Error::Rpc(e)) - if e.code == -5 - ) -} diff --git a/crates/bitcoind_rpc/src/lib.rs b/crates/bitcoind_rpc/src/lib.rs index 9ab4bd1924..4b44f7e18c 100644 --- a/crates/bitcoind_rpc/src/lib.rs +++ b/crates/bitcoind_rpc/src/lib.rs @@ -2,7 +2,7 @@ //! use the wallet RPC API, so this crate can be used with wallet-disabled Bitcoin Core nodes. //! //! [`Emitter`] is the main structure which sources blockchain data from -//! [`bitcoincore_rpc::Client`]. +//! [`bitcoind_client::bitreq::Client`]. //! //! To only get block updates (exclude mempool transactions), the caller can use //! [`Emitter::next_block`] until it returns `Ok(None)` (which means the chain tip is reached). A @@ -18,20 +18,22 @@ use alloc::sync::Arc; use bdk_core::collections::{HashMap, HashSet}; use bdk_core::{BlockId, CheckPoint}; use bitcoin::{Block, BlockHash, Transaction, Txid}; -use bitcoincore_rpc::{bitcoincore_rpc_json, RpcApi}; -use core::ops::Deref; +use bitcoind_client::bitreq::Client; +use bitcoind_client::corepc_types::model::GetBlockVerboseOne; + +pub extern crate bitcoind_client; pub mod bip158; -pub use bitcoincore_rpc; +pub(crate) use bitcoind_client::corepc_types; -/// The [`Emitter`] is used to emit data sourced from [`bitcoincore_rpc::Client`]. +/// The [`Emitter`] is used to emit data sourced from [`bitcoind_client::bitreq::Client`]. /// /// Refer to [module-level documentation] for more. /// /// [module-level documentation]: crate -pub struct Emitter { - client: C, +pub struct Emitter<'a> { + client: &'a Client, start_height: u32, /// The checkpoint of the last-emitted block that is in the best chain. If it is later found @@ -42,7 +44,7 @@ pub struct Emitter { /// next block's block hash (which we use to fetch the next block), we set this to `None` /// whenever there are no more blocks, or the next block is no longer in the best chain. This /// gives us an opportunity to re-fetch this result. - last_block: Option, + last_block: Option, /// The last snapshot of mempool transactions. /// @@ -62,11 +64,7 @@ pub struct Emitter { /// to start empty (i.e. with no unconfirmed transactions). pub const NO_EXPECTED_MEMPOOL_TXS: core::iter::Empty> = core::iter::empty(); -impl Emitter -where - C: Deref, - C::Target: RpcApi, -{ +impl<'a> Emitter<'a> { /// Construct a new [`Emitter`]. /// /// `last_cp` informs the emitter of the chain we are starting off with. This way, the emitter @@ -79,7 +77,7 @@ where /// 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, + client: &'a Client, last_cp: CheckPoint, start_height: u32, expected_mempool_txs: impl IntoIterator>>, @@ -138,7 +136,7 @@ where /// # Ok::<_, bdk_bitcoind_rpc::bitcoincore_rpc::Error>(()) /// ``` #[cfg(feature = "std")] - pub fn mempool(&mut self) -> Result { + pub fn mempool(&mut self) -> Result { let sync_time = std::time::UNIX_EPOCH .elapsed() .expect("must get current time") @@ -151,8 +149,8 @@ where /// `sync_time` is in unix seconds. /// /// This is the no-std version of [`mempool`](Self::mempool). - pub fn mempool_at(&mut self, sync_time: u64) -> Result { - let client = &*self.client; + pub fn mempool_at(&mut self, sync_time: u64) -> Result { + let client = self.client; let mut rpc_tip_height; let mut rpc_tip_hash; @@ -175,10 +173,10 @@ where let mut mempool_event = MempoolEvent { update: rpc_mempool .into_iter() - .filter_map(|txid| -> Option> { + .filter_map(|txid| -> Option> { let tx = match self.mempool_snapshot.get(&txid) { Some(tx) => tx.clone(), - None => match client.get_raw_transaction(&txid, None) { + None => match client.get_raw_transaction(&txid) { Ok(tx) => { let tx = Arc::new(tx); self.mempool_snapshot.insert(txid, tx.clone()); @@ -194,8 +192,7 @@ where ..Default::default() }; - let at_tip = - rpc_tip_height == self.last_cp.height() as u64 && rpc_tip_hash == self.last_cp.hash(); + let at_tip = rpc_tip_height == self.last_cp.height() && rpc_tip_hash == self.last_cp.hash(); if at_tip { // We only emit evicted transactions when we have already emitted the RPC tip. This is @@ -229,35 +226,7 @@ where } /// Emit the next block height and block (if any). - /// - /// Call this repeatedly until it returns `Ok(None)`, which means the chain tip has been - /// reached. Each [`BlockEvent`] carries the block alongside the [`CheckPoint`] it connects to, - /// so it can be applied to BDK structures that require block connectivity. - /// - /// # Example - /// - /// ```no_run - /// use bdk_bitcoind_rpc::{ - /// bitcoincore_rpc::{Auth, Client}, - /// Emitter, NO_EXPECTED_MEMPOOL_TXS, - /// }; - /// # use bdk_core::CheckPoint; - /// # use bitcoin::{constants::genesis_block, Network}; - /// - /// let client = Client::new("127.0.0.1:8332", Auth::None)?; - /// let start_height = 0; - /// # let last_cp = CheckPoint::new(0, genesis_block(Network::Bitcoin).block_hash()); - /// // Use the checkpoint from your receiving structure (e.g. `LocalChain::tip()`). - /// let mut emitter = Emitter::new(&client, last_cp, start_height, NO_EXPECTED_MEMPOOL_TXS); - /// - /// while let Some(event) = emitter.next_block()? { - /// let height = event.block_height(); - /// let connected_to = event.connected_to(); - /// // apply `event.block` to your chain and tx graph here - /// } - /// # Ok::<_, bdk_bitcoind_rpc::bitcoincore_rpc::Error>(()) - /// ``` - pub fn next_block(&mut self) -> Result>, bitcoincore_rpc::Error> { + pub fn next_block(&mut self) -> Result>, bitcoind_client::Error> { if let Some((checkpoint, block)) = poll(self, move |hash, client| client.get_block(hash))? { // Stop tracking unconfirmed transactions that have been confirmed in this block. for tx in &block.txdata { @@ -323,39 +292,35 @@ impl BlockEvent { } enum PollResponse { - Block(bitcoincore_rpc_json::GetBlockResult), + Block(GetBlockVerboseOne), NoMoreBlocks, /// Fetched block is not in the best chain. BlockNotInBestChain, - AgreementFound(bitcoincore_rpc_json::GetBlockResult, CheckPoint), + AgreementFound(GetBlockVerboseOne, CheckPoint), /// Force the genesis checkpoint down the receiver's throat. AgreementPointNotFound(BlockHash), } -fn poll_once(emitter: &Emitter) -> Result -where - C: Deref, - C::Target: RpcApi, -{ - let client = &*emitter.client; +fn poll_once(emitter: &Emitter<'_>) -> Result { + let client = emitter.client; if let Some(last_res) = &emitter.last_block { - let next_hash = if last_res.height + 1 < emitter.start_height as _ { + let next_hash = if last_res.height + 1 < emitter.start_height { // enforce start height - let next_hash = client.get_block_hash(emitter.start_height as _)?; + let next_hash = client.get_block_hash(emitter.start_height)?; // make sure last emission is still in best chain - if client.get_block_hash(last_res.height as _)? != last_res.hash { + if client.get_block_hash(last_res.height)? != last_res.hash { return Ok(PollResponse::BlockNotInBestChain); } next_hash } else { - match last_res.nextblockhash { + match last_res.next_block_hash { None => return Ok(PollResponse::NoMoreBlocks), Some(next_hash) => next_hash, } }; - let res = client.get_block_info(&next_hash)?; + let res = client.get_block_verbose(&next_hash)?; if res.confirmations < 0 { return Ok(PollResponse::BlockNotInBestChain); } @@ -364,7 +329,7 @@ where } for cp in emitter.last_cp.iter() { - let res = match client.get_block_info(&cp.hash()) { + let res = match client.get_block_verbose(&cp.hash()) { // block not in best chain Ok(res) if res.confirmations < 0 => continue, Ok(res) => res, @@ -386,21 +351,19 @@ where Ok(PollResponse::AgreementPointNotFound(genesis_hash)) } -fn poll( - emitter: &mut Emitter, +fn poll( + emitter: &mut Emitter<'_>, get_item: F, -) -> Result, V)>, bitcoincore_rpc::Error> +) -> Result, V)>, bitcoind_client::Error> where - C: Deref, - C::Target: RpcApi, - F: Fn(&BlockHash, &C::Target) -> Result, + F: Fn(&BlockHash, &Client) -> Result, { loop { match poll_once(emitter)? { PollResponse::Block(res) => { - let height = res.height as u32; + let height = res.height; let hash = res.hash; - let item = get_item(&hash, &emitter.client)?; + let item = get_item(&hash, emitter.client)?; let new_cp = emitter .last_cp @@ -422,10 +385,8 @@ where PollResponse::AgreementFound(res, cp) => { // When a reorg happens, the agreement point drops below `last_cp`. We // override `start_height` so the emitter revisits the invalidated heights. - if (res.height as u32) < emitter.start_height - && (res.height as u32) < emitter.last_cp.height() - { - emitter.start_height = res.height as _; + if res.height < emitter.start_height && res.height < emitter.last_cp.height() { + emitter.start_height = res.height; } // get rid of evicted blocks emitter.last_cp = cp; @@ -441,26 +402,6 @@ where } } -/// Extends [`bitcoincore_rpc::Error`]. -pub trait BitcoindRpcErrorExt { - /// Returns whether the error is a "not found" error. - /// - /// This is useful since [`Emitter`] emits [`Result<_, bitcoincore_rpc::Error>`]s as - /// [`Iterator::Item`]. - fn is_not_found_error(&self) -> bool; -} - -impl BitcoindRpcErrorExt for bitcoincore_rpc::Error { - fn is_not_found_error(&self) -> bool { - if let bitcoincore_rpc::Error::JsonRpc(bitcoincore_rpc::jsonrpc::Error::Rpc(rpc_err)) = self - { - rpc_err.code == -5 - } else { - false - } - } -} - #[cfg(test)] #[cfg_attr(coverage_nightly, coverage(off))] mod test { @@ -476,9 +417,9 @@ mod test { let (chain, _) = LocalChain::from_genesis(env.genesis_hash()?); let chain_tip = chain.tip(); - let rpc_client = bitcoincore_rpc::Client::new( + let rpc_client = bitcoind_client::bitreq::Client::with_auth( &env.bitcoind.rpc_url(), - bitcoincore_rpc::Auth::CookieFile(env.bitcoind.params.cookie_file.clone()), + bitcoind_client::bitreq::Auth::CookieFile(env.bitcoind.params.cookie_file.clone()), )?; let mut emitter = Emitter::new(&rpc_client, chain_tip.clone(), 1, NO_EXPECTED_MEMPOOL_TXS); diff --git a/crates/bitcoind_rpc/tests/common/mod.rs b/crates/bitcoind_rpc/tests/common/mod.rs index bbd914cca7..2d75f5bf30 100644 --- a/crates/bitcoind_rpc/tests/common/mod.rs +++ b/crates/bitcoind_rpc/tests/common/mod.rs @@ -1,20 +1,20 @@ use bdk_testenv::anyhow; use bdk_testenv::TestEnv; +use bitcoind_client::bitreq::{Auth, Client}; -/// This trait is used for testing. It allows creating a new [`bitcoincore_rpc::Client`] connected +/// This trait is used for testing. It allows creating a new [`Client`] connected /// to the instance of bitcoind running in the test environment. This way the `TestEnv` and the -/// `Emitter` aren't required to share the same client. In the future when we no longer depend on -/// `bitcoincore-rpc`, this can be updated to return the production client that is used by BDK. +/// `Emitter` aren't required to share the same client. pub trait ClientExt { - /// Creates a new [`bitcoincore_rpc::Client`] connected to the current node instance. - fn get_rpc_client(&self) -> anyhow::Result; + /// Creates a new [`Client`] connected to the current node instance. + fn get_rpc_client(&self) -> anyhow::Result; } impl ClientExt for TestEnv { - fn get_rpc_client(&self) -> anyhow::Result { - Ok(bitcoincore_rpc::Client::new( + fn get_rpc_client(&self) -> anyhow::Result { + Ok(Client::with_auth( &self.bitcoind.rpc_url(), - bitcoincore_rpc::Auth::CookieFile(self.bitcoind.params.cookie_file.clone()), + Auth::CookieFile(self.bitcoind.params.cookie_file.clone()), )?) } } diff --git a/crates/bitcoind_rpc/tests/test_emitter.rs b/crates/bitcoind_rpc/tests/test_emitter.rs index 2aeede6283..61e1c09398 100644 --- a/crates/bitcoind_rpc/tests/test_emitter.rs +++ b/crates/bitcoind_rpc/tests/test_emitter.rs @@ -1,4 +1,4 @@ -use std::{collections::BTreeSet, ops::Deref}; +use std::collections::BTreeSet; use bdk_bitcoind_rpc::{Emitter, NO_EXPECTED_MEMPOOL_TXS}; use bdk_chain::{ @@ -298,15 +298,11 @@ fn process_block( Ok(()) } -fn sync_from_emitter( +fn sync_from_emitter( recv_chain: &mut LocalChain, recv_graph: &mut IndexedTxGraph>, - emitter: &mut Emitter, -) -> anyhow::Result<()> -where - C: Deref, - C::Target: bitcoincore_rpc::RpcApi, -{ + emitter: &mut Emitter<'_>, +) -> anyhow::Result<()> { while let Some(emission) = emitter.next_block()? { let height = emission.block_height(); process_block(recv_chain, recv_graph, emission.block, height)?; @@ -560,7 +556,6 @@ fn no_agreement_point() -> anyhow::Result<()> { /// 3. Insert the eviction into the graph and assert tx1 is no longer canonical. #[test] fn test_expect_tx_evicted() -> anyhow::Result<()> { - use bdk_bitcoind_rpc::bitcoincore_rpc::bitcoin; use bdk_chain::miniscript; use bdk_chain::spk_txout::SpkTxOutIndex; use bitcoin::constants::genesis_block; diff --git a/crates/bitcoind_rpc/tests/test_filter_iter.rs b/crates/bitcoind_rpc/tests/test_filter_iter.rs index d6136ecfbc..f6aeea62b4 100644 --- a/crates/bitcoind_rpc/tests/test_filter_iter.rs +++ b/crates/bitcoind_rpc/tests/test_filter_iter.rs @@ -2,7 +2,6 @@ use bdk_bitcoind_rpc::bip158::{Error, FilterIter}; use bdk_core::CheckPoint; use bdk_testenv::{anyhow, bitcoind, TestEnv}; use bitcoin::{Address, Amount, Network, ScriptBuf}; -use bitcoincore_rpc::RpcApi; use crate::common::ClientExt; @@ -21,9 +20,7 @@ fn testenv() -> anyhow::Result { #[test] fn filter_iter_matches_blocks() -> anyhow::Result<()> { let env = testenv()?; - let addr = ClientExt::get_rpc_client(&env)? - .get_new_address(None, None)? - .assume_checked(); + let addr = env.bitcoind.client.new_address()?; let _ = env.mine_blocks(100, Some(addr.clone()))?; assert_eq!(ClientExt::get_rpc_client(&env)?.get_block_count()?, 101); @@ -78,7 +75,7 @@ fn filter_iter_detects_reorgs() -> anyhow::Result<()> { let env = testenv()?; let rpc = ClientExt::get_rpc_client(&env)?; - while rpc.get_block_count()? < MINE_TO as u64 { + while rpc.get_block_count()? < MINE_TO { let _ = env.mine_blocks(1, None)?; } From 1806f7d7ac9b2564c5d9f108643b914a270c090e Mon Sep 17 00:00:00 2001 From: valued mammal Date: Sat, 11 Jul 2026 14:02:58 -0400 Subject: [PATCH 2/7] refactor(rpc): Move Emitter to src/emitter.rs The Emitter struct, its impl, MempoolEvent, BlockEvent, PollResponse, the poll/poll_once helpers, NO_EXPECTED_MEMPOOL_TXS, and the inline test module are relocated from lib.rs into a new src/emitter.rs. lib.rs now declares `mod emitter` and re-exports its public items, keeping the crate's public API unchanged. --- crates/bitcoind_rpc/src/emitter.rs | 428 ++++++++++++++++++++++++++ crates/bitcoind_rpc/src/lib.rs | 462 +---------------------------- 2 files changed, 431 insertions(+), 459 deletions(-) create mode 100644 crates/bitcoind_rpc/src/emitter.rs diff --git a/crates/bitcoind_rpc/src/emitter.rs b/crates/bitcoind_rpc/src/emitter.rs new file mode 100644 index 0000000000..3f8e9f3b29 --- /dev/null +++ b/crates/bitcoind_rpc/src/emitter.rs @@ -0,0 +1,428 @@ +use alloc::sync::Arc; +use bdk_core::collections::{HashMap, HashSet}; +use bdk_core::{BlockId, CheckPoint}; +use bitcoin::{Block, BlockHash, Transaction, Txid}; +use bitcoind_client::bitreq::Client; +use bitcoind_client::corepc_types::model::GetBlockVerboseOne; + +/// The [`Emitter`] is used to emit data sourced from [`bitcoind_client::bitreq::Client`]. +/// +/// Refer to [module-level documentation] for more. +/// +/// [module-level documentation]: crate +pub struct Emitter<'a> { + client: &'a Client, + start_height: u32, + + /// The checkpoint of the last-emitted block that is in the best chain. If it is later found + /// that the block is no longer in the best chain, it will be popped off from here. + last_cp: CheckPoint, + + /// The block result returned from rpc of the last-emitted block. As this result contains the + /// next block's block hash (which we use to fetch the next block), we set this to `None` + /// whenever there are no more blocks, or the next block is no longer in the best chain. This + /// gives us an opportunity to re-fetch this result. + last_block: Option, + + /// The last snapshot of mempool transactions. + /// + /// This is used to detect mempool evictions and as a cache for transactions to emit. + /// + /// For mempool evictions, the latest call to `getrawmempool` is compared against this field. + /// Any transaction that is missing from this field is considered evicted. The exception is if + /// the transaction is confirmed into a block - therefore, we only emit evictions when we are + /// 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>, +} + +/// Indicates that there are no initially-expected mempool transactions. +/// +/// Use this as the `expected_mempool_txs` field of [`Emitter::new`] when the wallet is known +/// to start empty (i.e. with no unconfirmed transactions). +pub const NO_EXPECTED_MEMPOOL_TXS: core::iter::Empty> = core::iter::empty(); + +impl<'a> Emitter<'a> { + /// 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: &'a Client, + last_cp: CheckPoint, + start_height: u32, + expected_mempool_txs: impl IntoIterator>>, + ) -> Self { + Self { + client, + start_height, + last_cp, + last_block: None, + mempool_snapshot: expected_mempool_txs + .into_iter() + .map(|tx| { + let tx: Arc = tx.into(); + (tx.compute_txid(), tx) + }) + .collect(), + } + } + + /// 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 + /// and hash. Until `next_block()` advances the checkpoint to tip, `mempool()` will always + /// return an empty `evicted` set. + #[cfg(feature = "std")] + pub fn mempool(&mut self) -> Result { + let sync_time = std::time::UNIX_EPOCH + .elapsed() + .expect("must get current time") + .as_secs(); + self.mempool_at(sync_time) + } + + /// Emit mempool transactions and any evicted [`Txid`]s at the given `sync_time`. + /// + /// `sync_time` is in unix seconds. + /// + /// This is the no-std version of [`mempool`](Self::mempool). + pub fn mempool_at(&mut self, sync_time: u64) -> Result { + let client = self.client; + + let mut rpc_tip_height; + let mut rpc_tip_hash; + let mut rpc_mempool; + let mut rpc_mempool_txids; + + // Ensure we get a mempool snapshot consistent with `rpc_tip_hash` as the tip. + loop { + rpc_tip_height = client.get_block_count()?; + rpc_tip_hash = client.get_block_hash(rpc_tip_height)?; + rpc_mempool = client.get_raw_mempool()?; + rpc_mempool_txids = rpc_mempool.iter().copied().collect::>(); + let is_still_at_tip = rpc_tip_hash == client.get_block_hash(rpc_tip_height)? + && rpc_tip_height == client.get_block_count()?; + if is_still_at_tip { + break; + } + } + + let mut mempool_event = MempoolEvent { + update: rpc_mempool + .into_iter() + .filter_map(|txid| -> Option> { + let tx = match self.mempool_snapshot.get(&txid) { + Some(tx) => tx.clone(), + None => match client.get_raw_transaction(&txid) { + Ok(tx) => { + let tx = Arc::new(tx); + self.mempool_snapshot.insert(txid, tx.clone()); + tx + } + Err(err) if err.is_not_found_error() => return None, + Err(err) => return Some(Err(err)), + }, + }; + Some(Ok((tx, sync_time))) + }) + .collect::, _>>()?, + ..Default::default() + }; + + let at_tip = rpc_tip_height == self.last_cp.height() && rpc_tip_hash == self.last_cp.hash(); + + if at_tip { + // We only emit evicted transactions when we have already emitted the RPC tip. This is + // because we cannot differentiate between transactions that are confirmed and + // transactions that are evicted, so we rely on emitted blocks to remove + // transactions from the `mempool_snapshot`. + mempool_event.evicted = self + .mempool_snapshot + .keys() + .filter(|&txid| !rpc_mempool_txids.contains(txid)) + .map(|&txid| (txid, sync_time)) + .collect(); + self.mempool_snapshot = mempool_event + .update + .iter() + .map(|(tx, _)| (tx.compute_txid(), tx.clone())) + .collect(); + } else { + // Since we are still catching up to the tip (a.k.a tip has not been emitted), we + // accumulate more transactions in `mempool_snapshot` so that we can emit evictions in + // a batch once we catch up. + self.mempool_snapshot.extend( + mempool_event + .update + .iter() + .map(|(tx, _)| (tx.compute_txid(), tx.clone())), + ); + }; + + Ok(mempool_event) + } + + /// Emit the next block height and block (if any). + pub fn next_block(&mut self) -> Result>, bitcoind_client::Error> { + if let Some((checkpoint, block)) = poll(self, move |hash, client| client.get_block(hash))? { + // Stop tracking unconfirmed transactions that have been confirmed in this block. + for tx in &block.txdata { + self.mempool_snapshot.remove(&tx.compute_txid()); + } + return Ok(Some(BlockEvent { block, checkpoint })); + } + Ok(None) + } +} + +/// A new emission from mempool. +#[derive(Debug, Default)] +pub struct MempoolEvent { + /// Transactions currently in the mempool alongside their seen-at timestamp. + pub update: Vec<(Arc, u64)>, + + /// Transactions evicted from the mempool alongside their evicted-at timestamp. + pub evicted: Vec<(Txid, u64)>, +} + +/// A newly emitted block from [`Emitter`]. +#[derive(Debug)] +pub struct BlockEvent { + /// The block. + pub block: B, + + /// The checkpoint of the new block. + /// + /// A [`CheckPoint`] is a node of a linked list of [`BlockId`]s. This checkpoint is linked to + /// all [`BlockId`]s originally passed in [`Emitter::new`] as well as emitted blocks since + /// then. These blocks are guaranteed to be of the same chain. + /// + /// This is important as BDK structures require block-to-apply to be connected with another + /// block in the original chain. + pub checkpoint: CheckPoint, +} + +impl BlockEvent { + /// The block height of this new block. + pub fn block_height(&self) -> u32 { + self.checkpoint.height() + } + + /// The block hash of this new block. + pub fn block_hash(&self) -> BlockHash { + self.checkpoint.hash() + } + + /// The [`BlockId`] of a previous block that this block connects to. + /// + /// This either returns a [`BlockId`] of a previously emitted block or from the chain we started + /// with (passed in as `last_cp` in [`Emitter::new`]). + /// + /// This value is derived from [`BlockEvent::checkpoint`]. + pub fn connected_to(&self) -> BlockId { + match self.checkpoint.prev() { + Some(prev_cp) => prev_cp.block_id(), + // there is no previous checkpoint, so just connect with itself + None => self.checkpoint.block_id(), + } + } +} + +enum PollResponse { + Block(GetBlockVerboseOne), + NoMoreBlocks, + /// Fetched block is not in the best chain. + BlockNotInBestChain, + AgreementFound(GetBlockVerboseOne, CheckPoint), + /// Force the genesis checkpoint down the receiver's throat. + AgreementPointNotFound(BlockHash), +} + +fn poll_once(emitter: &Emitter<'_>) -> Result { + let client = emitter.client; + + if let Some(last_res) = &emitter.last_block { + let next_hash = if last_res.height + 1 < emitter.start_height { + // enforce start height + let next_hash = client.get_block_hash(emitter.start_height)?; + // make sure last emission is still in best chain + if client.get_block_hash(last_res.height)? != last_res.hash { + return Ok(PollResponse::BlockNotInBestChain); + } + next_hash + } else { + match last_res.next_block_hash { + None => return Ok(PollResponse::NoMoreBlocks), + Some(next_hash) => next_hash, + } + }; + + let res = client.get_block_verbose(&next_hash)?; + if res.confirmations < 0 { + return Ok(PollResponse::BlockNotInBestChain); + } + + return Ok(PollResponse::Block(res)); + } + + for cp in emitter.last_cp.iter() { + let res = match client.get_block_verbose(&cp.hash()) { + // block not in best chain + Ok(res) if res.confirmations < 0 => continue, + Ok(res) => res, + Err(e) if e.is_not_found_error() => { + if cp.height() > 0 { + continue; + } + // if we can't find genesis block, we can't create an update that connects + break; + } + Err(e) => return Err(e), + }; + + // agreement point found + return Ok(PollResponse::AgreementFound(res, cp)); + } + + let genesis_hash = client.get_block_hash(0)?; + Ok(PollResponse::AgreementPointNotFound(genesis_hash)) +} + +fn poll( + emitter: &mut Emitter<'_>, + get_item: F, +) -> Result, V)>, bitcoind_client::Error> +where + F: Fn(&BlockHash, &Client) -> Result, +{ + loop { + match poll_once(emitter)? { + PollResponse::Block(res) => { + let height = res.height; + let hash = res.hash; + let item = get_item(&hash, emitter.client)?; + + let new_cp = emitter + .last_cp + .clone() + .push(height, hash) + .expect("must push"); + emitter.last_cp = new_cp.clone(); + emitter.last_block = Some(res); + return Ok(Some((new_cp, item))); + } + PollResponse::NoMoreBlocks => { + emitter.last_block = None; + return Ok(None); + } + PollResponse::BlockNotInBestChain => { + emitter.last_block = None; + continue; + } + PollResponse::AgreementFound(res, cp) => { + // When a reorg happens, the agreement point drops below `last_cp`. We + // override `start_height` so the emitter revisits the invalidated heights. + if res.height < emitter.start_height && res.height < emitter.last_cp.height() { + emitter.start_height = res.height; + } + // get rid of evicted blocks + emitter.last_cp = cp; + emitter.last_block = Some(res); + continue; + } + PollResponse::AgreementPointNotFound(genesis_hash) => { + emitter.last_cp = CheckPoint::new(0, genesis_hash); + emitter.last_block = None; + continue; + } + } + } +} + +#[cfg(test)] +#[cfg_attr(coverage_nightly, coverage(off))] +mod test { + use crate::{Emitter, NO_EXPECTED_MEMPOOL_TXS}; + use bdk_chain::local_chain::LocalChain; + use bdk_testenv::{anyhow, TestEnv}; + use bitcoin::{hashes::Hash, Address, Amount, ScriptBuf, Txid, WScriptHash}; + use std::collections::HashSet; + + #[test] + fn test_expected_mempool_txids_accumulate_and_remove() -> anyhow::Result<()> { + let env = TestEnv::new()?; + let (chain, _) = LocalChain::from_genesis(env.genesis_hash()?); + let chain_tip = chain.tip(); + + let rpc_client = bitcoind_client::bitreq::Client::with_auth( + &env.bitcoind.rpc_url(), + bitcoind_client::bitreq::Auth::CookieFile(env.bitcoind.params.cookie_file.clone()), + )?; + + let mut emitter = Emitter::new(&rpc_client, chain_tip.clone(), 1, NO_EXPECTED_MEMPOOL_TXS); + + env.mine_blocks(100, None)?; + while emitter.next_block()?.is_some() {} + + let spk_to_track = ScriptBuf::new_p2wsh(&WScriptHash::all_zeros()); + let addr_to_track = Address::from_script(&spk_to_track, bitcoin::Network::Regtest)?; + let mut mempool_txids = HashSet::new(); + + // Send a tx at different heights and ensure txs are accumulating in expected_mempool_txids. + for _ in 0..10 { + let sent_txid = env.send(&addr_to_track, Amount::from_sat(1_000))?; + mempool_txids.insert(sent_txid); + emitter.mempool()?; + env.mine_blocks(1, None)?; + + for txid in &mempool_txids { + assert!( + emitter.mempool_snapshot.contains_key(txid), + "Expected txid {txid:?} missing" + ); + } + } + + // Process each block and check that confirmed txids are removed from from + // expected_mempool_txids. + while let Some(block_event) = emitter.next_block()? { + let confirmed_txids: HashSet = block_event + .block + .txdata + .iter() + .map(|tx| tx.compute_txid()) + .collect(); + mempool_txids = mempool_txids + .difference(&confirmed_txids) + .copied() + .collect::>(); + for txid in confirmed_txids { + assert!( + !emitter.mempool_snapshot.contains_key(&txid), + "Expected txid {txid:?} should have been removed" + ); + } + for txid in &mempool_txids { + assert!( + emitter.mempool_snapshot.contains_key(txid), + "Expected txid {txid:?} missing" + ); + } + } + + assert!(emitter.mempool_snapshot.is_empty()); + + Ok(()) + } +} diff --git a/crates/bitcoind_rpc/src/lib.rs b/crates/bitcoind_rpc/src/lib.rs index 4b44f7e18c..4043e16c03 100644 --- a/crates/bitcoind_rpc/src/lib.rs +++ b/crates/bitcoind_rpc/src/lib.rs @@ -14,467 +14,11 @@ #[macro_use] extern crate alloc; -use alloc::sync::Arc; -use bdk_core::collections::{HashMap, HashSet}; -use bdk_core::{BlockId, CheckPoint}; -use bitcoin::{Block, BlockHash, Transaction, Txid}; -use bitcoind_client::bitreq::Client; -use bitcoind_client::corepc_types::model::GetBlockVerboseOne; - pub extern crate bitcoind_client; +mod emitter; +pub use emitter::*; + pub mod bip158; pub(crate) use bitcoind_client::corepc_types; - -/// The [`Emitter`] is used to emit data sourced from [`bitcoind_client::bitreq::Client`]. -/// -/// Refer to [module-level documentation] for more. -/// -/// [module-level documentation]: crate -pub struct Emitter<'a> { - client: &'a Client, - start_height: u32, - - /// The checkpoint of the last-emitted block that is in the best chain. If it is later found - /// that the block is no longer in the best chain, it will be popped off from here. - last_cp: CheckPoint, - - /// The block result returned from rpc of the last-emitted block. As this result contains the - /// next block's block hash (which we use to fetch the next block), we set this to `None` - /// whenever there are no more blocks, or the next block is no longer in the best chain. This - /// gives us an opportunity to re-fetch this result. - last_block: Option, - - /// The last snapshot of mempool transactions. - /// - /// This is used to detect mempool evictions and as a cache for transactions to emit. - /// - /// For mempool evictions, the latest call to `getrawmempool` is compared against this field. - /// Any transaction that is missing from this field is considered evicted. The exception is if - /// the transaction is confirmed into a block - therefore, we only emit evictions when we are - /// 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>, -} - -/// Indicates that there are no initially-expected mempool transactions. -/// -/// Use this as the `expected_mempool_txs` field of [`Emitter::new`] when the wallet is known -/// to start empty (i.e. with no unconfirmed transactions). -pub const NO_EXPECTED_MEMPOOL_TXS: core::iter::Empty> = core::iter::empty(); - -impl<'a> Emitter<'a> { - /// 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: &'a Client, - last_cp: CheckPoint, - start_height: u32, - expected_mempool_txs: impl IntoIterator>>, - ) -> Self { - Self { - client, - start_height, - last_cp, - last_block: None, - mempool_snapshot: expected_mempool_txs - .into_iter() - .map(|tx| { - let tx: Arc = tx.into(); - (tx.compute_txid(), tx) - }) - .collect(), - } - } - - /// Emit mempool transactions and any evicted [`Txid`]s. - /// - /// This method returns a [`MempoolEvent`] containing the full transactions that were emitted, - /// each stamped with the call's `sync_time` (unix seconds captured at the start of the call, - /// not when the node first saw the transaction), and [`MempoolEvent::evicted`] which are any - /// [`Txid`]s which were previously seen in the mempool and are now missing. The timestamp - /// advances on each poll, so callers should treat it as a "last seen" value. Evicted txids - /// 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. - /// - /// # Example - /// - /// ```no_run - /// use bdk_bitcoind_rpc::{ - /// bitcoincore_rpc::{Auth, Client}, - /// Emitter, NO_EXPECTED_MEMPOOL_TXS, - /// }; - /// # use bdk_core::CheckPoint; - /// # use bitcoin::{constants::genesis_block, Network}; - /// - /// let client = Client::new("127.0.0.1:8332", Auth::None)?; - /// # let last_cp = CheckPoint::new(0, genesis_block(Network::Bitcoin).block_hash()); - /// // Use the checkpoint from your receiving structure (e.g. `LocalChain::tip()`). - /// let mut emitter = Emitter::new(&client, last_cp, 0, NO_EXPECTED_MEMPOOL_TXS); - /// - /// // Drain blocks first so evictions can be reported once the checkpoint reaches tip. - /// while emitter.next_block()?.is_some() {} - /// - /// let event = emitter.mempool()?; - /// for (tx, seen_at) in event.update { - /// // index unconfirmed `tx` (last seen at `seen_at`) - /// } - /// for (txid, evicted_at) in event.evicted { - /// // mark `txid` as evicted from the mempool at `evicted_at` - /// } - /// # Ok::<_, bdk_bitcoind_rpc::bitcoincore_rpc::Error>(()) - /// ``` - #[cfg(feature = "std")] - pub fn mempool(&mut self) -> Result { - let sync_time = std::time::UNIX_EPOCH - .elapsed() - .expect("must get current time") - .as_secs(); - self.mempool_at(sync_time) - } - - /// Emit mempool transactions and any evicted [`Txid`]s at the given `sync_time`. - /// - /// `sync_time` is in unix seconds. - /// - /// This is the no-std version of [`mempool`](Self::mempool). - pub fn mempool_at(&mut self, sync_time: u64) -> Result { - let client = self.client; - - let mut rpc_tip_height; - let mut rpc_tip_hash; - let mut rpc_mempool; - let mut rpc_mempool_txids; - - // Ensure we get a mempool snapshot consistent with `rpc_tip_hash` as the tip. - loop { - rpc_tip_height = client.get_block_count()?; - rpc_tip_hash = client.get_block_hash(rpc_tip_height)?; - rpc_mempool = client.get_raw_mempool()?; - rpc_mempool_txids = rpc_mempool.iter().copied().collect::>(); - let is_still_at_tip = rpc_tip_hash == client.get_block_hash(rpc_tip_height)? - && rpc_tip_height == client.get_block_count()?; - if is_still_at_tip { - break; - } - } - - let mut mempool_event = MempoolEvent { - update: rpc_mempool - .into_iter() - .filter_map(|txid| -> Option> { - let tx = match self.mempool_snapshot.get(&txid) { - Some(tx) => tx.clone(), - None => match client.get_raw_transaction(&txid) { - Ok(tx) => { - let tx = Arc::new(tx); - self.mempool_snapshot.insert(txid, tx.clone()); - tx - } - Err(err) if err.is_not_found_error() => return None, - Err(err) => return Some(Err(err)), - }, - }; - Some(Ok((tx, sync_time))) - }) - .collect::, _>>()?, - ..Default::default() - }; - - let at_tip = rpc_tip_height == self.last_cp.height() && rpc_tip_hash == self.last_cp.hash(); - - if at_tip { - // We only emit evicted transactions when we have already emitted the RPC tip. This is - // because we cannot differentiate between transactions that are confirmed and - // transactions that are evicted, so we rely on emitted blocks to remove - // transactions from the `mempool_snapshot`. - mempool_event.evicted = self - .mempool_snapshot - .keys() - .filter(|&txid| !rpc_mempool_txids.contains(txid)) - .map(|&txid| (txid, sync_time)) - .collect(); - self.mempool_snapshot = mempool_event - .update - .iter() - .map(|(tx, _)| (tx.compute_txid(), tx.clone())) - .collect(); - } else { - // Since we are still catching up to the tip (a.k.a tip has not been emitted), we - // accumulate more transactions in `mempool_snapshot` so that we can emit evictions in - // a batch once we catch up. - self.mempool_snapshot.extend( - mempool_event - .update - .iter() - .map(|(tx, _)| (tx.compute_txid(), tx.clone())), - ); - }; - - Ok(mempool_event) - } - - /// Emit the next block height and block (if any). - pub fn next_block(&mut self) -> Result>, bitcoind_client::Error> { - if let Some((checkpoint, block)) = poll(self, move |hash, client| client.get_block(hash))? { - // Stop tracking unconfirmed transactions that have been confirmed in this block. - for tx in &block.txdata { - self.mempool_snapshot.remove(&tx.compute_txid()); - } - return Ok(Some(BlockEvent { block, checkpoint })); - } - Ok(None) - } -} - -/// A new emission from mempool. -#[derive(Debug, Default)] -pub struct MempoolEvent { - /// Transactions currently in the mempool alongside their seen-at timestamp. - pub update: Vec<(Arc, u64)>, - - /// Transactions evicted from the mempool alongside their evicted-at timestamp. - pub evicted: Vec<(Txid, u64)>, -} - -/// A newly emitted block from [`Emitter`]. -#[derive(Debug)] -pub struct BlockEvent { - /// The block. - pub block: B, - - /// The checkpoint of the new block. - /// - /// A [`CheckPoint`] is a node of a linked list of [`BlockId`]s. This checkpoint is linked to - /// all [`BlockId`]s originally passed in [`Emitter::new`] as well as emitted blocks since - /// then. These blocks are guaranteed to be of the same chain. - /// - /// This is important as BDK structures require block-to-apply to be connected with another - /// block in the original chain. - pub checkpoint: CheckPoint, -} - -impl BlockEvent { - /// The block height of this new block. - pub fn block_height(&self) -> u32 { - self.checkpoint.height() - } - - /// The block hash of this new block. - pub fn block_hash(&self) -> BlockHash { - self.checkpoint.hash() - } - - /// The [`BlockId`] of a previous block that this block connects to. - /// - /// This either returns a [`BlockId`] of a previously emitted block or from the chain we started - /// with (passed in as `last_cp` in [`Emitter::new`]). - /// - /// This value is derived from [`BlockEvent::checkpoint`]. - pub fn connected_to(&self) -> BlockId { - match self.checkpoint.prev() { - Some(prev_cp) => prev_cp.block_id(), - // there is no previous checkpoint, so just connect with itself - None => self.checkpoint.block_id(), - } - } -} - -enum PollResponse { - Block(GetBlockVerboseOne), - NoMoreBlocks, - /// Fetched block is not in the best chain. - BlockNotInBestChain, - AgreementFound(GetBlockVerboseOne, CheckPoint), - /// Force the genesis checkpoint down the receiver's throat. - AgreementPointNotFound(BlockHash), -} - -fn poll_once(emitter: &Emitter<'_>) -> Result { - let client = emitter.client; - - if let Some(last_res) = &emitter.last_block { - let next_hash = if last_res.height + 1 < emitter.start_height { - // enforce start height - let next_hash = client.get_block_hash(emitter.start_height)?; - // make sure last emission is still in best chain - if client.get_block_hash(last_res.height)? != last_res.hash { - return Ok(PollResponse::BlockNotInBestChain); - } - next_hash - } else { - match last_res.next_block_hash { - None => return Ok(PollResponse::NoMoreBlocks), - Some(next_hash) => next_hash, - } - }; - - let res = client.get_block_verbose(&next_hash)?; - if res.confirmations < 0 { - return Ok(PollResponse::BlockNotInBestChain); - } - - return Ok(PollResponse::Block(res)); - } - - for cp in emitter.last_cp.iter() { - let res = match client.get_block_verbose(&cp.hash()) { - // block not in best chain - Ok(res) if res.confirmations < 0 => continue, - Ok(res) => res, - Err(e) if e.is_not_found_error() => { - if cp.height() > 0 { - continue; - } - // if we can't find genesis block, we can't create an update that connects - break; - } - Err(e) => return Err(e), - }; - - // agreement point found - return Ok(PollResponse::AgreementFound(res, cp)); - } - - let genesis_hash = client.get_block_hash(0)?; - Ok(PollResponse::AgreementPointNotFound(genesis_hash)) -} - -fn poll( - emitter: &mut Emitter<'_>, - get_item: F, -) -> Result, V)>, bitcoind_client::Error> -where - F: Fn(&BlockHash, &Client) -> Result, -{ - loop { - match poll_once(emitter)? { - PollResponse::Block(res) => { - let height = res.height; - let hash = res.hash; - let item = get_item(&hash, emitter.client)?; - - let new_cp = emitter - .last_cp - .clone() - .push(height, hash) - .expect("must push"); - emitter.last_cp = new_cp.clone(); - emitter.last_block = Some(res); - return Ok(Some((new_cp, item))); - } - PollResponse::NoMoreBlocks => { - emitter.last_block = None; - return Ok(None); - } - PollResponse::BlockNotInBestChain => { - emitter.last_block = None; - continue; - } - PollResponse::AgreementFound(res, cp) => { - // When a reorg happens, the agreement point drops below `last_cp`. We - // override `start_height` so the emitter revisits the invalidated heights. - if res.height < emitter.start_height && res.height < emitter.last_cp.height() { - emitter.start_height = res.height; - } - // get rid of evicted blocks - emitter.last_cp = cp; - emitter.last_block = Some(res); - continue; - } - PollResponse::AgreementPointNotFound(genesis_hash) => { - emitter.last_cp = CheckPoint::new(0, genesis_hash); - emitter.last_block = None; - continue; - } - } - } -} - -#[cfg(test)] -#[cfg_attr(coverage_nightly, coverage(off))] -mod test { - use crate::{Emitter, NO_EXPECTED_MEMPOOL_TXS}; - use bdk_chain::local_chain::LocalChain; - use bdk_testenv::{anyhow, TestEnv}; - use bitcoin::{hashes::Hash, Address, Amount, ScriptBuf, Txid, WScriptHash}; - use std::collections::HashSet; - - #[test] - fn test_expected_mempool_txids_accumulate_and_remove() -> anyhow::Result<()> { - let env = TestEnv::new()?; - let (chain, _) = LocalChain::from_genesis(env.genesis_hash()?); - let chain_tip = chain.tip(); - - let rpc_client = bitcoind_client::bitreq::Client::with_auth( - &env.bitcoind.rpc_url(), - bitcoind_client::bitreq::Auth::CookieFile(env.bitcoind.params.cookie_file.clone()), - )?; - - let mut emitter = Emitter::new(&rpc_client, chain_tip.clone(), 1, NO_EXPECTED_MEMPOOL_TXS); - - env.mine_blocks(100, None)?; - while emitter.next_block()?.is_some() {} - - let spk_to_track = ScriptBuf::new_p2wsh(&WScriptHash::all_zeros()); - let addr_to_track = Address::from_script(&spk_to_track, bitcoin::Network::Regtest)?; - let mut mempool_txids = HashSet::new(); - - // Send a tx at different heights and ensure txs are accumulating in expected_mempool_txids. - for _ in 0..10 { - let sent_txid = env.send(&addr_to_track, Amount::from_sat(1_000))?; - mempool_txids.insert(sent_txid); - emitter.mempool()?; - env.mine_blocks(1, None)?; - - for txid in &mempool_txids { - assert!( - emitter.mempool_snapshot.contains_key(txid), - "Expected txid {txid:?} missing" - ); - } - } - - // Process each block and check that confirmed txids are removed from from - // expected_mempool_txids. - while let Some(block_event) = emitter.next_block()? { - let confirmed_txids: HashSet = block_event - .block - .txdata - .iter() - .map(|tx| tx.compute_txid()) - .collect(); - mempool_txids = mempool_txids - .difference(&confirmed_txids) - .copied() - .collect::>(); - for txid in confirmed_txids { - assert!( - !emitter.mempool_snapshot.contains_key(&txid), - "Expected txid {txid:?} should have been removed" - ); - } - for txid in &mempool_txids { - assert!( - emitter.mempool_snapshot.contains_key(txid), - "Expected txid {txid:?} missing" - ); - } - } - - assert!(emitter.mempool_snapshot.is_empty()); - - Ok(()) - } -} From a9af72669661b44715361de9ffb35d5249fe984d Mon Sep 17 00:00:00 2001 From: valued mammal Date: Sat, 11 Jul 2026 15:12:51 -0400 Subject: [PATCH 3/7] refactor(rpc)!: Tidy poll loop, mempool snapshot, start_height Return a bounded `EmitterError::AgreementNotFound` instead of silently resetting to the genesis checkpoint when no agreement block is found. This surfaces the wrong-network case (e.g. a testnet checkpoint polled against a mainnet node) to the caller rather than resetting emission to genesis. Agreement failures are bounded by `MAX_AGREEMENT_FAILURES` (3). Other changes: - Extract private `Emitter::raw_mempool()` for the tip-consistent mempool snapshot, replacing the mut-before-loop pattern. - Fix `BlockEvent::connected_to()` to derive the connected block from `(height - 1, header.prev_blockhash)` instead of returning itself when there is no previous checkpoint. - Convert `start_height` from a constructor parameter into a builder method. `Emitter::new` is simplified by using last_cp.height() as the default start height. - Introde `EmitterError` and use it in the definition of `next_block` and mempool methods. - Remove NO_EXPECTED_MEMPOOL_TXS, which restricted type inference at the call site. --- crates/bitcoind_rpc/src/emitter.rs | 373 ++++++++++++---------- crates/bitcoind_rpc/tests/test_emitter.rs | 44 +-- 2 files changed, 236 insertions(+), 181 deletions(-) diff --git a/crates/bitcoind_rpc/src/emitter.rs b/crates/bitcoind_rpc/src/emitter.rs index 3f8e9f3b29..a6bcd5ffca 100644 --- a/crates/bitcoind_rpc/src/emitter.rs +++ b/crates/bitcoind_rpc/src/emitter.rs @@ -1,10 +1,16 @@ use alloc::sync::Arc; +use core::fmt; + use bdk_core::collections::{HashMap, HashSet}; use bdk_core::{BlockId, CheckPoint}; use bitcoin::{Block, BlockHash, Transaction, Txid}; use bitcoind_client::bitreq::Client; use bitcoind_client::corepc_types::model::GetBlockVerboseOne; +/// Maximum number of consecutive failed attempts to find an agreement point before +/// [`Emitter::next_block`] returns [`EmitterError::AgreementNotFound`]. +const MAX_AGREEMENT_FAILURES: u8 = 3; + /// The [`Emitter`] is used to emit data sourced from [`bitcoind_client::bitreq::Client`]. /// /// Refer to [module-level documentation] for more. @@ -24,6 +30,11 @@ pub struct Emitter<'a> { /// gives us an opportunity to re-fetch this result. last_block: Option, + /// Number of consecutive times polling has failed to find an agreement point. Used to bound + /// the retry loop and surface [`EmitterError::AgreementNotFound`] rather than spinning + /// forever. + agreement_failure_count: u8, + /// The last snapshot of mempool transactions. /// /// This is used to detect mempool evictions and as a cache for transactions to emit. @@ -36,20 +47,14 @@ pub struct Emitter<'a> { mempool_snapshot: HashMap>, } -/// Indicates that there are no initially-expected mempool transactions. -/// -/// Use this as the `expected_mempool_txs` field of [`Emitter::new`] when the wallet is known -/// to start empty (i.e. with no unconfirmed transactions). -pub const NO_EXPECTED_MEMPOOL_TXS: core::iter::Empty> = core::iter::empty(); - impl<'a> Emitter<'a> { /// 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). + /// By default emission starts from `last_cp.height()`. Use [`Emitter::start_height`] to start + /// from a later height (for example, a wallet's birthday). /// /// `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. @@ -57,14 +62,15 @@ impl<'a> Emitter<'a> { pub fn new( client: &'a Client, last_cp: CheckPoint, - start_height: u32, expected_mempool_txs: impl IntoIterator>>, ) -> Self { + let start_height = last_cp.height(); Self { client, start_height, last_cp, last_block: None, + agreement_failure_count: 0, mempool_snapshot: expected_mempool_txs .into_iter() .map(|tx| { @@ -75,6 +81,29 @@ impl<'a> Emitter<'a> { } } + /// Set the block height from which to start emitting blocks. + /// + /// By default the emitter starts from `last_cp.height()`. Use this when `last_cp` is at a low + /// height (for example, genesis) but the wallet only needs blocks from a later point — its + /// "birthday". During a reorg, if the agreement point is found below the local checkpoint, the + /// emitter automatically lowers `start_height` so that no invalidated heights are skipped. + pub fn start_height(mut self, start_height: u32) -> Self { + self.start_height = start_height; + self + } + + /// Retrieve all txids currently in the mempool, ensuring the snapshot is consistent with the + /// remote tip (the best block hash is unchanged across the call). + fn raw_mempool(&self) -> Result<(BlockHash, Vec), EmitterError> { + loop { + let block_hash = self.client.get_best_block_hash()?; + let raw_mempool = self.client.get_raw_mempool()?; + if self.client.get_best_block_hash()? == block_hash { + return Ok((block_hash, raw_mempool)); + } + } + } + /// Emit mempool transactions and any evicted [`Txid`]s. /// /// This method returns a [`MempoolEvent`] containing the full transactions (with their @@ -84,7 +113,7 @@ impl<'a> Emitter<'a> { /// and hash. Until `next_block()` advances the checkpoint to tip, `mempool()` will always /// return an empty `evicted` set. #[cfg(feature = "std")] - pub fn mempool(&mut self) -> Result { + pub fn mempool(&mut self) -> Result { let sync_time = std::time::UNIX_EPOCH .elapsed() .expect("must get current time") @@ -97,50 +126,39 @@ impl<'a> Emitter<'a> { /// `sync_time` is in unix seconds. /// /// This is the no-std version of [`mempool`](Self::mempool). - pub fn mempool_at(&mut self, sync_time: u64) -> Result { - let client = self.client; - - let mut rpc_tip_height; - let mut rpc_tip_hash; - let mut rpc_mempool; - let mut rpc_mempool_txids; - - // Ensure we get a mempool snapshot consistent with `rpc_tip_hash` as the tip. - loop { - rpc_tip_height = client.get_block_count()?; - rpc_tip_hash = client.get_block_hash(rpc_tip_height)?; - rpc_mempool = client.get_raw_mempool()?; - rpc_mempool_txids = rpc_mempool.iter().copied().collect::>(); - let is_still_at_tip = rpc_tip_hash == client.get_block_hash(rpc_tip_height)? - && rpc_tip_height == client.get_block_count()?; - if is_still_at_tip { - break; - } - } + pub fn mempool_at(&mut self, sync_time: u64) -> Result { + let (tip_hash, raw_mempool) = self.raw_mempool()?; + + let mempool_txids: HashSet = raw_mempool.iter().copied().collect(); + + let mempool_txs: Vec<(Txid, Arc, u64)> = raw_mempool + .into_iter() + .filter_map(|txid| -> Option> { + let tx = match self.mempool_snapshot.get(&txid) { + Some(tx) => tx.clone(), + None => match self.client.get_raw_transaction(&txid) { + Ok(tx) => { + let tx = Arc::new(tx); + self.mempool_snapshot.insert(txid, tx.clone()); + tx + } + Err(err) if err.is_not_found_error() => return None, + Err(err) => return Some(Err(err)), + }, + }; + Some(Ok((txid, tx, sync_time))) + }) + .collect::>()?; let mut mempool_event = MempoolEvent { - update: rpc_mempool - .into_iter() - .filter_map(|txid| -> Option> { - let tx = match self.mempool_snapshot.get(&txid) { - Some(tx) => tx.clone(), - None => match client.get_raw_transaction(&txid) { - Ok(tx) => { - let tx = Arc::new(tx); - self.mempool_snapshot.insert(txid, tx.clone()); - tx - } - Err(err) if err.is_not_found_error() => return None, - Err(err) => return Some(Err(err)), - }, - }; - Some(Ok((tx, sync_time))) - }) - .collect::, _>>()?, + update: mempool_txs + .iter() + .map(|(_, tx, sync_time)| (tx.clone(), *sync_time)) + .collect(), ..Default::default() }; - let at_tip = rpc_tip_height == self.last_cp.height() && rpc_tip_hash == self.last_cp.hash(); + let at_tip = self.last_cp.hash() == tip_hash; if at_tip { // We only emit evicted transactions when we have already emitted the RPC tip. This is @@ -150,32 +168,27 @@ impl<'a> Emitter<'a> { mempool_event.evicted = self .mempool_snapshot .keys() - .filter(|&txid| !rpc_mempool_txids.contains(txid)) + .filter(|&txid| !mempool_txids.contains(txid)) .map(|&txid| (txid, sync_time)) .collect(); - self.mempool_snapshot = mempool_event - .update + self.mempool_snapshot = mempool_txs .iter() - .map(|(tx, _)| (tx.compute_txid(), tx.clone())) + .map(|(txid, tx, _)| (*txid, tx.clone())) .collect(); } else { // Since we are still catching up to the tip (a.k.a tip has not been emitted), we // accumulate more transactions in `mempool_snapshot` so that we can emit evictions in // a batch once we catch up. - self.mempool_snapshot.extend( - mempool_event - .update - .iter() - .map(|(tx, _)| (tx.compute_txid(), tx.clone())), - ); + self.mempool_snapshot + .extend(mempool_txs.iter().map(|(txid, tx, _)| (*txid, tx.clone()))); }; Ok(mempool_event) } /// Emit the next block height and block (if any). - pub fn next_block(&mut self) -> Result>, bitcoind_client::Error> { - if let Some((checkpoint, block)) = poll(self, move |hash, client| client.get_block(hash))? { + pub fn next_block(&mut self) -> Result, EmitterError> { + if let Some((checkpoint, block)) = self.poll()? { // Stop tracking unconfirmed transactions that have been confirmed in this block. for tx in &block.txdata { self.mempool_snapshot.remove(&tx.compute_txid()); @@ -198,9 +211,9 @@ pub struct MempoolEvent { /// A newly emitted block from [`Emitter`]. #[derive(Debug)] -pub struct BlockEvent { +pub struct BlockEvent { /// The block. - pub block: B, + pub block: Block, /// The checkpoint of the new block. /// @@ -213,7 +226,7 @@ pub struct BlockEvent { pub checkpoint: CheckPoint, } -impl BlockEvent { +impl BlockEvent { /// The block height of this new block. pub fn block_height(&self) -> u32 { self.checkpoint.height() @@ -233,130 +246,164 @@ impl BlockEvent { pub fn connected_to(&self) -> BlockId { match self.checkpoint.prev() { Some(prev_cp) => prev_cp.block_id(), - // there is no previous checkpoint, so just connect with itself - None => self.checkpoint.block_id(), + // No previous checkpoint (e.g. the first emission after a `start_height` skip); derive + // the parent from this block's header instead of connecting the block to itself. + None => BlockId { + height: self.checkpoint.height().saturating_sub(1), + hash: self.block.header.prev_blockhash, + }, } } } +/// Internal state machine responses from [`Emitter::poll_once`]. enum PollResponse { - Block(GetBlockVerboseOne), - NoMoreBlocks, - /// Fetched block is not in the best chain. - BlockNotInBestChain, - AgreementFound(GetBlockVerboseOne, CheckPoint), - /// Force the genesis checkpoint down the receiver's throat. - AgreementPointNotFound(BlockHash), + /// The next consecutive block is ready to be emitted. + NextBlock(GetBlockVerboseOne), + /// The emitter is current with the node's tip; no next block exists. + Tip, + /// The last emitted block is no longer in the best chain. + Reorged, + /// A checkpoint still in the best chain was found; resume scanning from here. + AgreementAt(GetBlockVerboseOne, CheckPoint), + /// No checkpoint matches the node's best chain. + AgreementNotFound, } -fn poll_once(emitter: &Emitter<'_>) -> Result { - let client = emitter.client; +impl<'a> Emitter<'a> { + /// Probe the node once and return the appropriate [`PollResponse`]. + fn poll_once(&self) -> Result { + if let Some(last_block_info) = &self.last_block { + let next_hash = if last_block_info.height + 1 < self.start_height { + // enforce start height + self.client.get_block_hash(self.start_height)? + } else { + match last_block_info.next_block_hash { + None => return Ok(PollResponse::Tip), + Some(next_hash) => next_hash, + } + }; + + let block_info = self.client.get_block_verbose(&next_hash)?; + return if block_info.confirmations < 0 { + Ok(PollResponse::Reorged) + } else { + Ok(PollResponse::NextBlock(block_info)) + }; + } - if let Some(last_res) = &emitter.last_block { - let next_hash = if last_res.height + 1 < emitter.start_height { - // enforce start height - let next_hash = client.get_block_hash(emitter.start_height)?; - // make sure last emission is still in best chain - if client.get_block_hash(last_res.height)? != last_res.hash { - return Ok(PollResponse::BlockNotInBestChain); - } - next_hash - } else { - match last_res.next_block_hash { - None => return Ok(PollResponse::NoMoreBlocks), - Some(next_hash) => next_hash, - } - }; + for cp in self.last_cp.iter() { + let block_info = match self.client.get_block_verbose(&cp.hash()) { + // block not in best chain + Ok(block_info) if block_info.confirmations < 0 => continue, + Ok(block_info) => block_info, + Err(e) if e.is_not_found_error() => { + if cp.height() > 0 { + continue; + } + // if we can't find genesis block, we can't create an update that connects + break; + } + Err(e) => return Err(e), + }; - let res = client.get_block_verbose(&next_hash)?; - if res.confirmations < 0 { - return Ok(PollResponse::BlockNotInBestChain); + // agreement point found + return Ok(PollResponse::AgreementAt(block_info, cp)); } - return Ok(PollResponse::Block(res)); + Ok(PollResponse::AgreementNotFound) } - for cp in emitter.last_cp.iter() { - let res = match client.get_block_verbose(&cp.hash()) { - // block not in best chain - Ok(res) if res.confirmations < 0 => continue, - Ok(res) => res, - Err(e) if e.is_not_found_error() => { - if cp.height() > 0 { - continue; + /// Drive the emitter state machine forward until a block is ready to emit or the tip is + /// reached. + fn poll(&mut self) -> Result, Block)>, EmitterError> { + loop { + match self.poll_once()? { + PollResponse::NextBlock(block_info) => { + let height = block_info.height; + let hash = block_info.hash; + let block = self.client.get_block(&hash)?; + + let new_cp = self + .last_cp + .clone() + .push(height, hash) + .expect("NextBlock height must only increase"); + self.last_cp = new_cp.clone(); + self.last_block = Some(block_info); + self.agreement_failure_count = 0; + return Ok(Some((new_cp, block))); + } + PollResponse::Tip => { + self.last_block = None; + return Ok(None); + } + PollResponse::Reorged => { + self.last_block = None; + } + PollResponse::AgreementAt(block_info, cp) => { + // When a reorg happens, the agreement point drops below `last_cp`. We lower + // `start_height` so the emitter revisits the invalidated heights. + if block_info.height < self.last_cp.height() { + self.start_height = block_info.height; + } + self.last_cp = cp; + self.last_block = Some(block_info); + self.agreement_failure_count = 0; + } + PollResponse::AgreementNotFound => { + self.agreement_failure_count += 1; + if self.agreement_failure_count >= MAX_AGREEMENT_FAILURES { + return Err(EmitterError::AgreementNotFound); + } + self.last_block = None; } - // if we can't find genesis block, we can't create an update that connects - break; } - Err(e) => return Err(e), - }; - - // agreement point found - return Ok(PollResponse::AgreementFound(res, cp)); + } } +} - let genesis_hash = client.get_block_hash(0)?; - Ok(PollResponse::AgreementPointNotFound(genesis_hash)) +/// Errors returned by [`Emitter`] methods. +#[derive(Debug)] +#[non_exhaustive] +pub enum EmitterError { + /// An RPC call to bitcoind failed. + Rpc(bitcoind_client::Error), + /// The emitter exhausted all checkpoints without finding a block that is part of the node's + /// best chain. This indicates either a catastrophic reorg that rolled back further than any + /// known checkpoint, or an inconsistent node (e.g. a checkpoint from the wrong network, whose + /// genesis differs from the connected node). The caller should reinitialise the [`Emitter`] + /// with a fresh checkpoint against the correct node. + AgreementNotFound, } -fn poll( - emitter: &mut Emitter<'_>, - get_item: F, -) -> Result, V)>, bitcoind_client::Error> -where - F: Fn(&BlockHash, &Client) -> Result, -{ - loop { - match poll_once(emitter)? { - PollResponse::Block(res) => { - let height = res.height; - let hash = res.hash; - let item = get_item(&hash, emitter.client)?; - - let new_cp = emitter - .last_cp - .clone() - .push(height, hash) - .expect("must push"); - emitter.last_cp = new_cp.clone(); - emitter.last_block = Some(res); - return Ok(Some((new_cp, item))); - } - PollResponse::NoMoreBlocks => { - emitter.last_block = None; - return Ok(None); - } - PollResponse::BlockNotInBestChain => { - emitter.last_block = None; - continue; - } - PollResponse::AgreementFound(res, cp) => { - // When a reorg happens, the agreement point drops below `last_cp`. We - // override `start_height` so the emitter revisits the invalidated heights. - if res.height < emitter.start_height && res.height < emitter.last_cp.height() { - emitter.start_height = res.height; - } - // get rid of evicted blocks - emitter.last_cp = cp; - emitter.last_block = Some(res); - continue; - } - PollResponse::AgreementPointNotFound(genesis_hash) => { - emitter.last_cp = CheckPoint::new(0, genesis_hash); - emitter.last_block = None; - continue; - } +impl fmt::Display for EmitterError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + EmitterError::Rpc(e) => write!(f, "bitcoind RPC error: {e}"), + EmitterError::AgreementNotFound => write!( + f, + "no agreement point found between local checkpoints and the node's best chain", + ), } } } +impl core::error::Error for EmitterError {} + +impl From for EmitterError { + fn from(e: bitcoind_client::Error) -> Self { + EmitterError::Rpc(e) + } +} + #[cfg(test)] #[cfg_attr(coverage_nightly, coverage(off))] mod test { - use crate::{Emitter, NO_EXPECTED_MEMPOOL_TXS}; + use crate::Emitter; use bdk_chain::local_chain::LocalChain; use bdk_testenv::{anyhow, TestEnv}; - use bitcoin::{hashes::Hash, Address, Amount, ScriptBuf, Txid, WScriptHash}; + use bitcoin::{hashes::Hash, Address, Amount, ScriptBuf, Transaction, Txid, WScriptHash}; use std::collections::HashSet; #[test] @@ -370,7 +417,11 @@ mod test { bitcoind_client::bitreq::Auth::CookieFile(env.bitcoind.params.cookie_file.clone()), )?; - let mut emitter = Emitter::new(&rpc_client, chain_tip.clone(), 1, NO_EXPECTED_MEMPOOL_TXS); + let mut emitter = Emitter::new( + &rpc_client, + chain_tip.clone(), + core::iter::empty::(), + ); env.mine_blocks(100, None)?; while emitter.next_block()?.is_some() {} diff --git a/crates/bitcoind_rpc/tests/test_emitter.rs b/crates/bitcoind_rpc/tests/test_emitter.rs index 61e1c09398..d8667d4edb 100644 --- a/crates/bitcoind_rpc/tests/test_emitter.rs +++ b/crates/bitcoind_rpc/tests/test_emitter.rs @@ -1,6 +1,6 @@ use std::collections::BTreeSet; -use bdk_bitcoind_rpc::{Emitter, NO_EXPECTED_MEMPOOL_TXS}; +use bdk_bitcoind_rpc::Emitter; use bdk_chain::{ bitcoin::{Address, Amount, Txid}, local_chain::{CheckPoint, LocalChain}, @@ -12,7 +12,7 @@ use bdk_testenv::{ bitcoind::{Input, Output}, TestEnv, }; -use bitcoin::{hashes::Hash, Block, Network, ScriptBuf, WScriptHash}; +use bitcoin::{hashes::Hash, Block, Network, ScriptBuf, Transaction, WScriptHash}; use crate::common::ClientExt; @@ -31,7 +31,11 @@ pub fn test_sync_local_chain() -> anyhow::Result<()> { let (mut local_chain, _) = LocalChain::from_genesis(env.genesis_hash()?); let client = ClientExt::get_rpc_client(&env)?; - let mut emitter = Emitter::new(&client, local_chain.tip(), 0, NO_EXPECTED_MEMPOOL_TXS); + let mut emitter = Emitter::new( + &client, + local_chain.tip(), + core::iter::empty::(), + ); // Mine some blocks and return the actual block hashes. // Because initializing `ElectrsD` already mines some blocks, we must include those too when @@ -171,7 +175,7 @@ fn test_into_tx_graph() -> anyhow::Result<()> { }); let client = ClientExt::get_rpc_client(&env)?; - let emitter = &mut Emitter::new(&client, chain.tip(), 0, NO_EXPECTED_MEMPOOL_TXS); + let emitter = &mut Emitter::new(&client, chain.tip(), core::iter::empty::()); while let Some(emission) = emitter.next_block()? { let height = emission.block_height(); @@ -263,9 +267,9 @@ fn ensure_block_emitted_after_reorg_is_at_reorg_height() -> anyhow::Result<()> { let mut emitter = Emitter::new( &client, CheckPoint::new(0, env.genesis_hash()?), - EMITTER_START_HEIGHT as _, - NO_EXPECTED_MEMPOOL_TXS, - ); + core::iter::empty::(), + ) + .start_height(EMITTER_START_HEIGHT as _); env.mine_blocks(CHAIN_TIP_HEIGHT, None)?; while emitter.next_block()?.is_some() {} @@ -339,8 +343,7 @@ fn tx_can_become_unconfirmed_after_reorg() -> anyhow::Result<()> { let mut emitter = Emitter::new( &client, CheckPoint::new(0, env.genesis_hash()?), - 0, - NO_EXPECTED_MEMPOOL_TXS, + core::iter::empty::(), ); // setup addresses @@ -432,8 +435,7 @@ fn mempool_avoids_re_emission() -> anyhow::Result<()> { let mut emitter = Emitter::new( &client, CheckPoint::new(0, env.genesis_hash()?), - 0, - NO_EXPECTED_MEMPOOL_TXS, + core::iter::empty::(), ); // mine blocks and sync up emitter @@ -504,9 +506,9 @@ fn no_agreement_point() -> anyhow::Result<()> { let mut emitter = Emitter::new( &client, CheckPoint::new(0, env.genesis_hash()?), - (PREMINE_COUNT - 3) as u32, - NO_EXPECTED_MEMPOOL_TXS, - ); + core::iter::empty::(), + ) + .start_height((PREMINE_COUNT - 3) as u32); // mine 101 blocks env.mine_blocks(PREMINE_COUNT, None)?; @@ -585,7 +587,7 @@ fn test_expect_tx_evicted() -> anyhow::Result<()> { let tx_1 = env.rpc_client().get_transaction(txid_1)?.into_model()?.tx; let client = ClientExt::get_rpc_client(&env)?; - let mut emitter = Emitter::new(&client, chain.tip(), 1, core::iter::once(tx_1)); + let mut emitter = Emitter::new(&client, chain.tip(), core::iter::once(tx_1)); while let Some(emission) = emitter.next_block()? { let height = emission.block_height(); chain.apply_header(&emission.block.header, height)?; @@ -667,7 +669,11 @@ fn test_sync_with_new_emitter_after_reorg() -> anyhow::Result<()> { env.mine_blocks(110, None)?; - let mut emitter = Emitter::new(&client, local_chain.tip(), 0, NO_EXPECTED_MEMPOOL_TXS); + let mut emitter = Emitter::new( + &client, + local_chain.tip(), + core::iter::empty::(), + ); while let Some(emission) = emitter.next_block()? { let _ = local_chain.apply_update(emission.checkpoint)?; } @@ -681,8 +687,7 @@ fn test_sync_with_new_emitter_after_reorg() -> anyhow::Result<()> { let mut emitter = Emitter::new( &client, local_chain.tip(), - tip_height, - NO_EXPECTED_MEMPOOL_TXS, + core::iter::empty::(), ); while let Some(emission) = emitter.next_block()? { @@ -712,8 +717,7 @@ fn detect_new_mempool_txs() -> anyhow::Result<()> { let mut emitter = Emitter::new( &client, CheckPoint::new(0, env.genesis_hash()?), - 0, - NO_EXPECTED_MEMPOOL_TXS, + core::iter::empty::(), ); while emitter.next_block()?.is_some() {} From 1ba4a7e275a79ea94824929f7821a01210c3398e Mon Sep 17 00:00:00 2001 From: valued mammal Date: Sat, 11 Jul 2026 15:46:45 -0400 Subject: [PATCH 4/7] docs(rpc): Rewrite emitter/poll docs, mempool timestamp semantics, README Rewrite the crate README with a usage overview and a runnable example, and wire it in as the crate-level docs via `#![doc = include_str!("../README.md")]`. Correct the mempool timestamp semantics: `MempoolEvent::update`/`evicted` are timestamped with the `sync_time` of the call, not the time a transaction was first seen or evicted. Document that this is a full snapshot, not a delta, and that evictions are only reported once the emitter is at tip. Tidy the poll state-machine docs (`PollResponse`, `poll_once`, `poll`), `next_block`, and `connected_to`, and trim redundant field comments so the code stays self-describing. Documentation only; no behavior change. --- crates/bitcoind_rpc/README.md | 61 +++++++++++++- crates/bitcoind_rpc/src/emitter.rs | 129 +++++++++++++++-------------- crates/bitcoind_rpc/src/lib.rs | 10 +-- 3 files changed, 126 insertions(+), 74 deletions(-) diff --git a/crates/bitcoind_rpc/README.md b/crates/bitcoind_rpc/README.md index 12de87020d..24b1715032 100644 --- a/crates/bitcoind_rpc/README.md +++ b/crates/bitcoind_rpc/README.md @@ -1,3 +1,62 @@ # BDK Bitcoind RPC -This crate is used for emitting blockchain data from the `bitcoind` RPC interface. +This crate emits blockchain data from the `bitcoind` RPC interface. It does not use the wallet +RPC API, so it works against wallet-disabled Bitcoin Core nodes. + +[`Emitter`] is the main entry point. It sources blocks and mempool transactions from a +[`bitcoind_client::bitreq::Client`] and produces updates that connect to a local chain you already +hold, handling reorgs along the way. + +## Usage + +Give the emitter a checkpoint describing the chain you already know about (for a fresh wallet this +is just the genesis block). Then: + +- Call [`Emitter::next_block`] in a loop until it returns `Ok(None)` to walk the chain forward, + block by block, up to the node's tip. Each [`BlockEvent`] carries a [`CheckPoint`] that connects + to your existing chain, so it can be applied directly to a `LocalChain`. +- Call [`Emitter::mempool`] to emit the current mempool: the full set of unconfirmed transactions + plus any that have been evicted since the last call. + +If your wallet has a known creation height ("birthday"), use [`Emitter::start_height`] to skip +directly to it instead of scanning from the checkpoint height. + +```rust,no_run +use bdk_bitcoind_rpc::{Emitter, EmitterError}; +use bdk_chain::local_chain::LocalChain; +use bitcoin::{constants::genesis_block, BlockHash, Network, Transaction}; +use bitcoind_client::bitreq::{Auth, Client}; + +// The chain we already know about. A fresh wallet starts at the network's genesis block. +let genesis_hash: BlockHash = genesis_block(Network::Bitcoin).block_hash(); +let (local_chain, _) = LocalChain::from_genesis(genesis_hash); + +let rpc_client = Client::with_auth( + "127.0.0.1:8332", + Auth::CookieFile("/home/user/.bitcoin/.cookie".into()), +)?; + +let mut emitter = Emitter::new( + &rpc_client, + local_chain.tip(), + core::iter::empty::(), +); + +// Walk the chain forward until the node's tip is reached. +while let Some(event) = emitter.next_block()? { + println!("block {}: {}", event.block_height(), event.block_hash()); +} + +// Emit the current mempool state. +let mempool = emitter.mempool()?; +println!("{} mempool txs, {} evicted", mempool.update.len(), mempool.evicted.len()); +# >::Ok(()) +``` + +[`Emitter`]: crate::Emitter +[`Emitter::next_block`]: crate::Emitter::next_block +[`Emitter::mempool`]: crate::Emitter::mempool +[`Emitter::start_height`]: crate::Emitter::start_height +[`BlockEvent`]: crate::BlockEvent +[`CheckPoint`]: bdk_core::CheckPoint +[`bitcoind_client::bitreq::Client`]: bitcoind_client::bitreq::Client diff --git a/crates/bitcoind_rpc/src/emitter.rs b/crates/bitcoind_rpc/src/emitter.rs index a6bcd5ffca..a25a8fdd91 100644 --- a/crates/bitcoind_rpc/src/emitter.rs +++ b/crates/bitcoind_rpc/src/emitter.rs @@ -18,47 +18,40 @@ const MAX_AGREEMENT_FAILURES: u8 = 3; /// [module-level documentation]: crate pub struct Emitter<'a> { client: &'a Client, + + /// Height from which to start emitting blocks. Defaults to `last_cp.height()`. start_height: u32, - /// The checkpoint of the last-emitted block that is in the best chain. If it is later found - /// that the block is no longer in the best chain, it will be popped off from here. + /// Checkpoint of the last-emitted block known to be in the best chain, and the tail of the + /// linked checkpoint list threaded through each [`BlockEvent`]. Blocks reorged out are popped + /// off during agreement scanning. last_cp: CheckPoint, - /// The block result returned from rpc of the last-emitted block. As this result contains the - /// next block's block hash (which we use to fetch the next block), we set this to `None` - /// whenever there are no more blocks, or the next block is no longer in the best chain. This - /// gives us an opportunity to re-fetch this result. + /// RPC result of the last-emitted block, kept for its `nextblockhash` so the next block can be + /// fetched without a height-to-hash lookup. `None` before the first emission, at tip, or after + /// the last block was reorged out. last_block: Option, - /// Number of consecutive times polling has failed to find an agreement point. Used to bound - /// the retry loop and surface [`EmitterError::AgreementNotFound`] rather than spinning - /// forever. + /// Consecutive polls that failed to find an agreement point. Bounds the retry loop so the + /// emitter surfaces [`EmitterError::AgreementNotFound`] instead of spinning forever. agreement_failure_count: u8, - /// The last snapshot of mempool transactions. - /// - /// This is used to detect mempool evictions and as a cache for transactions to emit. - /// - /// For mempool evictions, the latest call to `getrawmempool` is compared against this field. - /// Any transaction that is missing from this field is considered evicted. The exception is if - /// the transaction is confirmed into a block - therefore, we only emit evictions when we are - /// sure the tip block is already emitted. When a block is emitted, the transactions in the - /// block are removed from this field. + /// Unconfirmed transactions seen so far. Doubles as a fetch cache (avoids re-fetching txs) and + /// as the reference set for eviction detection: at tip, any txid here but absent from the + /// latest `getrawmempool` is reported evicted. Entries are removed once confirmed in a block. mempool_snapshot: HashMap>, } impl<'a> Emitter<'a> { /// 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. - /// - /// By default emission starts from `last_cp.height()`. Use [`Emitter::start_height`] to start - /// from a later height (for example, a wallet's birthday). + /// `last_cp` is the chain the emitter starts from: it scans this checkpoint chain to find the + /// deepest block still in the node's best chain, then emits forward from there. Emission + /// starts at `last_cp.height()` by default; use [`Emitter::start_height`] to skip ahead (for + /// example, to a wallet's birthday). /// - /// `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. + /// `expected_mempool_txs` is the wallet's set of already-known unconfirmed transactions, which + /// lets the emitter report evictions for them. Pass `core::iter::empty()` when empty. pub fn new( client: &'a Client, last_cp: CheckPoint, @@ -104,14 +97,10 @@ impl<'a> Emitter<'a> { } } - /// Emit mempool transactions and any evicted [`Txid`]s. + /// Emit a full snapshot of the mempool along with 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 - /// and hash. Until `next_block()` advances the checkpoint to tip, `mempool()` will always - /// return an empty `evicted` set. + /// The returned [`MempoolEvent`] timestamps every entry with the current time (see + /// [`mempool_at`](Self::mempool_at) for details on eviction reporting). #[cfg(feature = "std")] pub fn mempool(&mut self) -> Result { let sync_time = std::time::UNIX_EPOCH @@ -121,11 +110,13 @@ impl<'a> Emitter<'a> { self.mempool_at(sync_time) } - /// Emit mempool transactions and any evicted [`Txid`]s at the given `sync_time`. + /// Emit a full snapshot of the mempool along with any evicted [`Txid`]s, timestamped with the + /// given `sync_time` (unix seconds). This is the no-std version of [`mempool`](Self::mempool). /// - /// `sync_time` is in unix seconds. - /// - /// This is the no-std version of [`mempool`](Self::mempool). + /// Evictions are only reported once the emitter has caught up to the node's tip (its + /// checkpoint hash matches the best block). Until then we cannot tell an evicted transaction + /// apart from one confirmed in a not-yet-emitted block, so [`MempoolEvent::evicted`] stays + /// empty and transactions accumulate in the snapshot to be reconciled once at tip. pub fn mempool_at(&mut self, sync_time: u64) -> Result { let (tip_hash, raw_mempool) = self.raw_mempool()?; @@ -161,10 +152,8 @@ impl<'a> Emitter<'a> { let at_tip = self.last_cp.hash() == tip_hash; if at_tip { - // We only emit evicted transactions when we have already emitted the RPC tip. This is - // because we cannot differentiate between transactions that are confirmed and - // transactions that are evicted, so we rely on emitted blocks to remove - // transactions from the `mempool_snapshot`. + // At tip we can trust that a missing txid was evicted rather than confirmed, so report + // evictions and replace the snapshot with the current mempool. mempool_event.evicted = self .mempool_snapshot .keys() @@ -176,9 +165,7 @@ impl<'a> Emitter<'a> { .map(|(txid, tx, _)| (*txid, tx.clone())) .collect(); } else { - // Since we are still catching up to the tip (a.k.a tip has not been emitted), we - // accumulate more transactions in `mempool_snapshot` so that we can emit evictions in - // a batch once we catch up. + // Still catching up: accumulate so evictions can be reconciled in one batch at tip. self.mempool_snapshot .extend(mempool_txs.iter().map(|(txid, tx, _)| (*txid, tx.clone()))); }; @@ -186,10 +173,16 @@ impl<'a> Emitter<'a> { Ok(mempool_event) } - /// Emit the next block height and block (if any). + /// Emit the next block in chain order, or `Ok(None)` once the tip is reached. + /// + /// Blocks are emitted consecutively from the agreement point (the deepest checkpoint still in + /// the node's best chain). On a reorg the emitter rescans for a new agreement point and + /// re-emits from there. Returns [`EmitterError::AgreementNotFound`] if no agreement point can + /// be found after multiple consecutive attempts. pub fn next_block(&mut self) -> Result, EmitterError> { if let Some((checkpoint, block)) = self.poll()? { - // Stop tracking unconfirmed transactions that have been confirmed in this block. + // Confirmed transactions leave the mempool snapshot so they aren't misreported as + // evictions on the next mempool() call. for tx in &block.txdata { self.mempool_snapshot.remove(&tx.compute_txid()); } @@ -202,10 +195,12 @@ impl<'a> Emitter<'a> { /// A new emission from mempool. #[derive(Debug, Default)] pub struct MempoolEvent { - /// Transactions currently in the mempool alongside their seen-at timestamp. + /// A full snapshot of the current mempool, each transaction paired with the + /// `sync_time` of the call that produced it. pub update: Vec<(Arc, u64)>, - /// Transactions evicted from the mempool alongside their evicted-at timestamp. + /// Transactions evicted from the mempool since the last call, paired with the call's + /// `sync_time`. Only populated once the emitter is at the node's tip; empty while catching up. pub evicted: Vec<(Txid, u64)>, } @@ -237,17 +232,18 @@ impl BlockEvent { self.checkpoint.hash() } - /// The [`BlockId`] of a previous block that this block connects to. + /// The [`BlockId`] this block's checkpoint chain connects to. /// - /// This either returns a [`BlockId`] of a previously emitted block or from the chain we started - /// with (passed in as `last_cp` in [`Emitter::new`]). - /// - /// This value is derived from [`BlockEvent::checkpoint`]. + /// This is the previous entry in the emitter's checkpoint chain. For consecutive emissions + /// it is the Bitcoin parent block; when [`Emitter::start_height`] skips ahead, the first + /// emission connects to whatever checkpoint was at the tail of `last_cp` (possibly an + /// ancestor further back than the direct parent). pub fn connected_to(&self) -> BlockId { match self.checkpoint.prev() { Some(prev_cp) => prev_cp.block_id(), - // No previous checkpoint (e.g. the first emission after a `start_height` skip); derive - // the parent from this block's header instead of connecting the block to itself. + // No previous checkpoint; derive the parent from this block's header. + // This should be unreachable in practice, since every emitted block connects + // to `last_cp`. None => BlockId { height: self.checkpoint.height().saturating_sub(1), hash: self.block.header.prev_blockhash, @@ -256,13 +252,14 @@ impl BlockEvent { } } -/// Internal state machine responses from [`Emitter::poll_once`]. +/// Outcome of a single node poll, driving the [`Emitter`] state machine (see +/// [`Emitter::poll_once`]). enum PollResponse { - /// The next consecutive block is ready to be emitted. + /// The next consecutive block is ready to emit. NextBlock(GetBlockVerboseOne), - /// The emitter is current with the node's tip; no next block exists. + /// The emitter is current with the node's tip; there is no next block. Tip, - /// The last emitted block is no longer in the best chain. + /// The last emitted block is no longer in the best chain; fall through to agreement scanning. Reorged, /// A checkpoint still in the best chain was found; resume scanning from here. AgreementAt(GetBlockVerboseOne, CheckPoint), @@ -271,7 +268,11 @@ enum PollResponse { } impl<'a> Emitter<'a> { - /// Probe the node once and return the appropriate [`PollResponse`]. + /// Probe the node once and return the appropriate `PollResponse`. + /// + /// `last_block` determines the next phase: when `Some`, follow its `next_block_hash` to the + /// next block (reporting `PollResponse::Reorged` if it has been reorged out); when + /// `None`, walk `last_cp` backwards to find the nearest checkpoint still in the best chain. fn poll_once(&self) -> Result { if let Some(last_block_info) = &self.last_block { let next_hash = if last_block_info.height + 1 < self.start_height { @@ -301,21 +302,21 @@ impl<'a> Emitter<'a> { if cp.height() > 0 { continue; } - // if we can't find genesis block, we can't create an update that connects + // genesis not found; cannot form a connected update break; } Err(e) => return Err(e), }; - // agreement point found return Ok(PollResponse::AgreementAt(block_info, cp)); } Ok(PollResponse::AgreementNotFound) } - /// Drive the emitter state machine forward until a block is ready to emit or the tip is - /// reached. + /// Drive the state machine until a block is ready to emit (`Some`) or the tip is reached + /// (`None`). Returns [`EmitterError::AgreementNotFound`] after [`MAX_AGREEMENT_FAILURES`] + /// consecutive failures to find an agreement point. fn poll(&mut self) -> Result, Block)>, EmitterError> { loop { match self.poll_once()? { diff --git a/crates/bitcoind_rpc/src/lib.rs b/crates/bitcoind_rpc/src/lib.rs index 4043e16c03..4691a60a83 100644 --- a/crates/bitcoind_rpc/src/lib.rs +++ b/crates/bitcoind_rpc/src/lib.rs @@ -1,12 +1,4 @@ -//! This crate is used for emitting blockchain data from the `bitcoind` RPC interface. It does not -//! use the wallet RPC API, so this crate can be used with wallet-disabled Bitcoin Core nodes. -//! -//! [`Emitter`] is the main structure which sources blockchain data from -//! [`bitcoind_client::bitreq::Client`]. -//! -//! To only get block updates (exclude mempool transactions), the caller can use -//! [`Emitter::next_block`] until it returns `Ok(None)` (which means the chain tip is reached). A -//! separate method, [`Emitter::mempool`] can be used to emit the whole mempool. +#![doc = include_str!("../README.md")] #![cfg_attr(coverage_nightly, feature(coverage_attribute))] #![warn(missing_docs)] From 98d163933286363ffd9225172df2f81a8fbf1f51 Mon Sep 17 00:00:00 2001 From: valued mammal Date: Sat, 11 Jul 2026 17:14:17 -0400 Subject: [PATCH 5/7] test(rpc): Rewrite test_emitter.rs Rewrite the emitter integration tests with behavior-focused names and expanded coverage. Each test now documents the scenario it exercises: - blocks_emitted_in_order_and_after_reorg - unconfirmed_txs_anchored_on_confirmation - ensure_block_emitted_after_reorg_is_at_reorg_height - tx_can_become_unconfirmed_after_reorg - mempool_update_is_complete_snapshot - reorg_past_start_checkpoint - double_spent_tx_evicted_and_removed_from_canonical_set - detect_new_mempool_txs - birthday_checkpoint_skips_earlier_blocks - mempool_at_uses_provided_timestamp - start_height_reset_on_reorg_prevents_height_gaps - evictions_withheld_until_at_tip - start_height_beyond_tip_returns_rpc_error - wrong_genesis_returns_agreement_not_found New coverage includes behind-tip eviction withholding (live and wallet-restart paths), the start_height birthday guarantee, custom mempool_at sync_time semantics, and the recoverable EmitterError cases (Rpc for an out-of-range start_height, AgreementNotFound for a wrong genesis hash). --- crates/bitcoind_rpc/tests/test_emitter.rs | 560 ++++++++++++++++------ 1 file changed, 419 insertions(+), 141 deletions(-) diff --git a/crates/bitcoind_rpc/tests/test_emitter.rs b/crates/bitcoind_rpc/tests/test_emitter.rs index d8667d4edb..f05c9b5b37 100644 --- a/crates/bitcoind_rpc/tests/test_emitter.rs +++ b/crates/bitcoind_rpc/tests/test_emitter.rs @@ -1,8 +1,9 @@ use std::collections::BTreeSet; +use std::sync::Arc; -use bdk_bitcoind_rpc::Emitter; +use bdk_bitcoind_rpc::{Emitter, EmitterError}; use bdk_chain::{ - bitcoin::{Address, Amount, Txid}, + bitcoin::{Address, Amount, Transaction, Txid}, local_chain::{CheckPoint, LocalChain}, spk_txout::SpkTxOutIndex, Balance, BlockId, IndexedTxGraph, Merge, @@ -12,20 +13,21 @@ use bdk_testenv::{ bitcoind::{Input, Output}, TestEnv, }; -use bitcoin::{hashes::Hash, Block, Network, ScriptBuf, Transaction, WScriptHash}; +use bitcoin::{hashes::Hash, Block, Network, ScriptBuf, WScriptHash}; use crate::common::ClientExt; mod common; -/// Ensure that blocks are emitted in order even after reorg. +/// Ensures blocks are emitted consecutively with correct hashes, and that after a reorg the +/// emitter re-emits the replacement blocks at the same heights with updated hashes. /// /// 1. Mine 101 blocks. /// 2. Emit blocks from [`Emitter`] and update the [`LocalChain`]. /// 3. Reorg highest 6 blocks. /// 4. Emit blocks from [`Emitter`] and re-update the [`LocalChain`]. #[test] -pub fn test_sync_local_chain() -> anyhow::Result<()> { +pub fn blocks_emitted_in_order_and_after_reorg() -> anyhow::Result<()> { let env = TestEnv::new()?; let network_tip = env.rpc_client().get_block_count()?.into_model().0; let (mut local_chain, _) = LocalChain::from_genesis(env.genesis_hash()?); @@ -34,11 +36,11 @@ pub fn test_sync_local_chain() -> anyhow::Result<()> { let mut emitter = Emitter::new( &client, local_chain.tip(), - core::iter::empty::(), + core::iter::empty::(), ); // Mine some blocks and return the actual block hashes. - // Because initializing `ElectrsD` already mines some blocks, we must include those too when + // Because initializing `TestEnv` already mines some blocks, we must include those too when // returning block hashes. let exp_hashes = { let mut hashes = (0..=network_tip) @@ -137,12 +139,15 @@ pub fn test_sync_local_chain() -> anyhow::Result<()> { Ok(()) } -/// Ensure that [`EmittedUpdate::into_tx_graph_update`] behaves appropriately for both mempool and -/// block updates. +/// Verifies the mempool → confirmation pipeline: unconfirmed transactions appear in +/// [`Emitter::mempool`] and receive block anchors once mined. /// -/// [`EmittedUpdate::into_tx_graph_update`]: bdk_bitcoind_rpc::EmittedUpdate::into_tx_graph_update +/// 1. Mine 101 blocks and sync emitter to tip. +/// 2. Send 3 transactions to a tracked address — they will be in the mempool. +/// 3. Assert `next_block` returns `None` (at tip) and `mempool` returns all 3 txs. +/// 4. Mine a block confirming those txs and assert the emitter produces anchors for them. #[test] -fn test_into_tx_graph() -> anyhow::Result<()> { +fn unconfirmed_txs_anchored_on_confirmation() -> anyhow::Result<()> { let env = TestEnv::new()?; let addr_0 = env @@ -151,37 +156,27 @@ fn test_into_tx_graph() -> anyhow::Result<()> { .address()? .assume_checked(); - let addr_1 = env - .rpc_client() - .get_new_address(None, None)? - .address()? - .assume_checked(); - - let addr_2 = env - .rpc_client() - .get_new_address(None, None)? - .address()? - .assume_checked(); - env.mine_blocks(101, None)?; let (mut chain, _) = LocalChain::from_genesis(env.genesis_hash()?); let mut indexed_tx_graph = IndexedTxGraph::::new({ let mut index = SpkTxOutIndex::::default(); index.insert_spk(0, addr_0.script_pubkey()); - index.insert_spk(1, addr_1.script_pubkey()); - index.insert_spk(2, addr_2.script_pubkey()); index }); let client = ClientExt::get_rpc_client(&env)?; - let emitter = &mut Emitter::new(&client, chain.tip(), core::iter::empty::()); + let emitter = &mut Emitter::new( + &client, + chain.tip(), + core::iter::empty::(), + ); while let Some(emission) = emitter.next_block()? { let height = emission.block_height(); let _ = chain.apply_update(emission.checkpoint)?; - let indexed_additions = indexed_tx_graph.apply_block_relevant(&emission.block, height); - assert!(indexed_additions.is_empty()); + let changeset = indexed_tx_graph.apply_block_relevant(&emission.block, height); + assert!(changeset.is_empty()); } // send 3 txs to a tracked address, these txs will be in the mempool @@ -203,9 +198,9 @@ fn test_into_tx_graph() -> anyhow::Result<()> { assert!(emitter.next_block()?.is_none()); let mempool_txs = emitter.mempool()?; - let indexed_additions = indexed_tx_graph.batch_insert_unconfirmed(mempool_txs.update); + let changeset = indexed_tx_graph.batch_insert_unconfirmed(mempool_txs.update); assert_eq!( - indexed_additions + changeset .tx_graph .txs .iter() @@ -214,7 +209,7 @@ fn test_into_tx_graph() -> anyhow::Result<()> { exp_txids, "changeset should have the 3 mempool transactions", ); - assert!(indexed_additions.tx_graph.anchors.is_empty()); + assert!(changeset.tx_graph.anchors.is_empty()); } // mine a block that confirms the 3 txs @@ -239,10 +234,10 @@ fn test_into_tx_graph() -> anyhow::Result<()> { let emission = emitter.next_block()?.expect("must get mined block"); let height = emission.block_height(); let _ = chain.apply_update(emission.checkpoint)?; - let indexed_additions = indexed_tx_graph.apply_block_relevant(&emission.block, height); - assert!(indexed_additions.tx_graph.txs.is_empty()); - assert!(indexed_additions.tx_graph.txouts.is_empty()); - assert_eq!(indexed_additions.tx_graph.anchors, exp_anchors); + let changeset = indexed_tx_graph.apply_block_relevant(&emission.block, height); + assert!(changeset.tx_graph.txs.is_empty()); + assert!(changeset.tx_graph.txouts.is_empty()); + assert_eq!(changeset.tx_graph.anchors, exp_anchors); } Ok(()) @@ -250,28 +245,31 @@ fn test_into_tx_graph() -> anyhow::Result<()> { /// Ensure next block emitted after reorg is at reorg height. /// -/// After a reorg, if the last-emitted block height is equal or greater than the reorg height, and -/// the fallback height is equal to or lower than the reorg height, the next block/header emission -/// should be at the reorg height. -/// -/// TODO: If the reorg height is lower than the fallback height, how do we find a block height to -/// emit that can connect with our receiver chain? +/// After a reorg, if the last-emitted block height is equal or greater than the reorg height, +/// the next emission should be at the reorg height. This is guaranteed by the agreement-scanning +/// algorithm: the emitter walks back through its checkpoint list to find the deepest block still +/// in the best chain and resumes consecutive emission from there. Because `last_cp` is built from +/// the actual birthday hash (not just a height integer), the agreement point is always well-defined +/// regardless of how deep the reorg goes. #[test] fn ensure_block_emitted_after_reorg_is_at_reorg_height() -> anyhow::Result<()> { - const EMITTER_START_HEIGHT: usize = 100; + const EMITTER_START_HEIGHT: u64 = 100; const CHAIN_TIP_HEIGHT: usize = 110; let env = TestEnv::new()?; - let client = ClientExt::get_rpc_client(&env)?; + + env.mine_blocks(CHAIN_TIP_HEIGHT, None)?; + + // Encode the birthday directly in last_cp rather than using a bare start_height integer. + // This ensures agreement-scanning works correctly even when the birthday block is reorged out. + let start_hash = env.get_block_hash(EMITTER_START_HEIGHT)?; let mut emitter = Emitter::new( &client, - CheckPoint::new(0, env.genesis_hash()?), - core::iter::empty::(), - ) - .start_height(EMITTER_START_HEIGHT as _); + CheckPoint::new(EMITTER_START_HEIGHT as u32, start_hash), + core::iter::empty::(), + ); - env.mine_blocks(CHAIN_TIP_HEIGHT, None)?; while emitter.next_block()?.is_some() {} for reorg_count in 1..=10 { @@ -305,7 +303,7 @@ fn process_block( fn sync_from_emitter( recv_chain: &mut LocalChain, recv_graph: &mut IndexedTxGraph>, - emitter: &mut Emitter<'_>, + emitter: &mut Emitter, ) -> anyhow::Result<()> { while let Some(emission) = emitter.next_block()? { let height = emission.block_height(); @@ -343,7 +341,7 @@ fn tx_can_become_unconfirmed_after_reorg() -> anyhow::Result<()> { let mut emitter = Emitter::new( &client, CheckPoint::new(0, env.genesis_hash()?), - core::iter::empty::(), + core::iter::empty::(), ); // setup addresses @@ -419,13 +417,10 @@ fn tx_can_become_unconfirmed_after_reorg() -> anyhow::Result<()> { Ok(()) } -/// Ensure avoid-re-emission-logic is sound when [`Emitter`] is synced to tip. -/// -/// The receiver (bdk_chain structures) is synced to the chain tip, and there is txs in the mempool. -/// When we call Emitter::mempool multiple times, mempool txs should not be re-emitted, even if the -/// chain tip is extended. +/// Every call to mempool should return all currently-known unconfirmed transactions, +/// including ones returned on previous calls. #[test] -fn mempool_avoids_re_emission() -> anyhow::Result<()> { +fn mempool_update_is_complete_snapshot() -> anyhow::Result<()> { const BLOCKS_TO_MINE: usize = 101; const MEMPOOL_TX_COUNT: usize = 2; @@ -435,7 +430,7 @@ fn mempool_avoids_re_emission() -> anyhow::Result<()> { let mut emitter = Emitter::new( &client, CheckPoint::new(0, env.genesis_hash()?), - core::iter::empty::(), + core::iter::empty::(), ); // mine blocks and sync up emitter @@ -485,79 +480,71 @@ fn mempool_avoids_re_emission() -> anyhow::Result<()> { Ok(()) } -/// If blockchain re-org includes the start height, emit new start height block +/// If a reorg invalidates the emitter's starting checkpoint, the emitter must find a lower +/// agreement point and resume consecutive emission from there. /// /// 1. mine 101 blocks -/// 2. emit blocks 98a, 99a, 100a -/// 3. invalidate blocks 99a, 100a, 101a -/// 4. mine new blocks 99b, 100b, 101b -/// 5. emit block 99b +/// 2. create emitter with last_cp at block 98 (one below the reorg point) +/// 3. emit blocks 99a, 100a +/// 4. reorg 3 blocks deep (replaces 99a, 100a, 101a with 99b, 100b, 101b) +/// 5. emit block 99b — agreement found at 98, next consecutive block is 99b /// /// The block hash of 99b should be different than 99a, but their previous block hashes should -/// be the same. +/// be the same (both build on block 98). #[test] -fn no_agreement_point() -> anyhow::Result<()> { - const PREMINE_COUNT: usize = 101; - +fn reorg_past_start_checkpoint() -> anyhow::Result<()> { let env = TestEnv::new()?; - let client = ClientExt::get_rpc_client(&env)?; - // start height is 99 - let mut emitter = Emitter::new( - &client, - CheckPoint::new(0, env.genesis_hash()?), - core::iter::empty::(), - ) - .start_height((PREMINE_COUNT - 3) as u32); - // mine 101 blocks - env.mine_blocks(PREMINE_COUNT, None)?; + // mine 101 blocks first so block 98 exists for the checkpoint + env.mine_blocks(100, None)?; - // emit blocks: 98a, 99a, 100a - let block_98a = emitter.next_block()?.expect("block 98a"); - let block_99a = emitter.next_block()?.expect("block 99a"); - let block_100a = emitter.next_block()?.expect("block 100a"); - assert_eq!(block_98a.block_height(), 98); - assert_eq!(block_99a.block_height(), 99); - assert_eq!(block_100a.block_height(), 100); + assert_eq!(env.bitcoind.client.get_block_count()?.0, 101); - // get hash for block 101a - let blockhash_101a = env.rpc_client().get_block_hash(101)?.block_hash()?; + // Encode last_cp at block 98 — the last block before the reorg zone. + let cp_height: u64 = 98; + let cp_hash = env.get_block_hash(cp_height)?; + let mut emitter = Emitter::new( + &client, + CheckPoint::new(cp_height as u32, cp_hash), + core::iter::empty::(), + ); - // invalidate blocks 99a, 100a, 101a - env.rpc_client().invalidate_block(blockhash_101a)?; - env.rpc_client().invalidate_block(block_100a.block_hash())?; - env.rpc_client().invalidate_block(block_99a.block_hash())?; + // emit block 99a + let event_99a = emitter.next_block()?.expect("block 99a header"); + assert_eq!(event_99a.block_height(), 99); + let block_header_99a = event_99a.block.header; + let block_hash_99a = block_header_99a.block_hash(); + let block_hash_98a = block_header_99a.prev_blockhash; - // mine new blocks 99b, 100b, 101b - env.mine_blocks(3, None)?; + // emit block 100a (advance the emitter past 99a so the reorg spans 3 blocks) + let _block_100a = emitter.next_block()?.expect("block 100a header"); - // emit block header 99b - let block_99b = emitter.next_block()?.expect("block 99b"); - assert_eq!(block_99b.block_height(), 99); + // Reorg depth 3: invalidates 99a, 100a, 101a and mines new 99b, 100b, 101b. + env.reorg(3)?; - assert_ne!(block_99a.block_hash(), block_99b.block_hash()); - assert_eq!( - block_98a.block_hash(), - block_99a.block.header.prev_blockhash - ); - assert_eq!( - block_98a.block_hash(), - block_99b.block.header.prev_blockhash - ); + // emit block 99b: agreement found at block 98, which is unchanged, so next block is 99b + let event_99b = emitter.next_block()?.expect("block 99b header"); + assert_eq!(event_99b.block_height(), 99); + let block_header_99b = event_99b.block.header; + let block_hash_99b = block_header_99b.block_hash(); + + assert_ne!(block_hash_99a, block_hash_99b); + assert_eq!(block_hash_98a, block_header_99a.prev_blockhash); + assert_eq!(block_hash_98a, block_header_99b.prev_blockhash); Ok(()) } /// Validates that when an unconfirmed transaction is double-spent (and thus evicted from the -/// mempool), the emitter reports it in `evicted_txids`, and after inserting that eviction into the +/// mempool), the emitter reports it in `evicted`, and after inserting that eviction into the /// graph it no longer appears in the set of canonical transactions. /// /// 1. Broadcast a first tx (tx1) and confirm it arrives in unconfirmed set. /// 2. Double-spend tx1 with tx1b and verify `mempool()` reports tx1 as evicted. /// 3. Insert the eviction into the graph and assert tx1 is no longer canonical. #[test] -fn test_expect_tx_evicted() -> anyhow::Result<()> { +fn double_spent_tx_evicted_and_removed_from_canonical_set() -> anyhow::Result<()> { use bdk_chain::miniscript; use bdk_chain::spk_txout::SpkTxOutIndex; use bitcoin::constants::genesis_block; @@ -565,11 +552,11 @@ fn test_expect_tx_evicted() -> anyhow::Result<()> { use bitcoin::Network; let env = TestEnv::new()?; - let s = bdk_testenv::utils::DESCRIPTORS[0]; - let desc = miniscript::Descriptor::parse_descriptor(&Secp256k1::new(), s) + let desc_str = bdk_testenv::utils::DESCRIPTORS[0]; + let descriptor = miniscript::Descriptor::parse_descriptor(&Secp256k1::new(), desc_str) .unwrap() .0; - let spk = desc.at_derivation_index(0)?.script_pubkey(); + let spk = descriptor.at_derivation_index(0)?.script_pubkey(); let mut chain = LocalChain::from_genesis(genesis_block(Network::Regtest).block_hash()).0; let chain_tip = chain.tip().block_id(); @@ -602,33 +589,36 @@ fn test_expect_tx_evicted() -> anyhow::Result<()> { // Double spend tx1. - // Get `prevout` from core. - let core = env.rpc_client(); - let tx1 = core.get_transaction(txid_1)?.into_model()?.tx; + // Get `prevout` from bitcoin core. + let rpc_client = env.rpc_client(); + let tx1 = rpc_client.get_transaction(txid_1)?.into_model()?.tx; let txin = &tx1.input[0]; let op = txin.previous_output; // Create `tx1b` using the previous output from tx1. - let utxo = Input { + let input = Input { txid: op.txid, vout: op.vout as u64, sequence: None, }; - let addr = core + let addr = rpc_client .get_new_address(None, None)? .address()? .assume_checked(); let outputs = [Output::new(addr, Amount::from_btc(49.99)?)]; - let tx = core - .create_raw_transaction(&[utxo], &outputs)? + let tx = rpc_client + .create_raw_transaction(&[input], &outputs)? .into_model()? .0; - let tx1b = core.sign_raw_transaction_with_wallet(&tx)?.into_model()?.tx; + let tx1b = rpc_client + .sign_raw_transaction_with_wallet(&tx)? + .into_model()? + .tx; // Send the tx. - let _txid_2 = core.send_raw_transaction(&tx1b)?; + let _txid_2 = rpc_client.send_raw_transaction(&tx1b)?; // Retrieve the expected unconfirmed txids and spks from the graph. let exp_spk_txids = chain @@ -658,21 +648,144 @@ fn test_expect_tx_evicted() -> anyhow::Result<()> { Ok(()) } -/// Creating a new [`Emitter`] after a reorg with `start_height` at the tip should still -/// produce a connectable checkpoint. When blocks are invalidated, the emitted checkpoint must -/// include the invalidation height so the update can connect with the original chain. #[test] -fn test_sync_with_new_emitter_after_reorg() -> anyhow::Result<()> { +fn detect_new_mempool_txs() -> anyhow::Result<()> { + let env = TestEnv::new()?; + env.mine_blocks(101, None)?; + + let addr = env + .rpc_client() + .get_new_address(None, None)? + .address()? + .require_network(Network::Regtest)?; + + let client = ClientExt::get_rpc_client(&env)?; + let mut emitter = Emitter::new( + &client, + CheckPoint::new(0, env.genesis_hash()?), + core::iter::empty::(), + ); + + while emitter.next_block()?.is_some() {} + + for n in 0..5 { + let txid = env.send(&addr, Amount::ONE_BTC)?; + let new_txs = emitter.mempool()?.update; + assert!( + new_txs.iter().any(|(tx, _)| tx.compute_txid() == txid), + "must detect new tx {n}" + ); + } + + Ok(()) +} + +/// Verifies that encoding a birthday block directly in `last_cp` causes the emitter to skip all +/// blocks at or below the checkpoint height, starting emission from the next block. +#[test] +fn birthday_checkpoint_skips_earlier_blocks() -> anyhow::Result<()> { + const BIRTHDAY_HEIGHT: u64 = 50; + const CHAIN_TIP: usize = 101; + + let env = TestEnv::new()?; + let client = ClientExt::get_rpc_client(&env)?; + + env.mine_blocks(CHAIN_TIP, None)?; + + let birthday_hash = env.get_block_hash(BIRTHDAY_HEIGHT)?; + let mut emitter = Emitter::new( + &client, + CheckPoint::new(BIRTHDAY_HEIGHT as u32, birthday_hash), + core::iter::empty::(), + ); + + let mut emitted_heights = Vec::new(); + while let Some(event) = emitter.next_block()? { + emitted_heights.push(event.block_height()); + } + + assert!( + !emitted_heights.is_empty(), + "should emit blocks above birthday" + ); + assert_eq!( + emitted_heights.first().copied(), + Some(BIRTHDAY_HEIGHT as u32 + 1), + "first emitted block must be immediately after birthday" + ); + assert!( + emitted_heights.iter().all(|&h| h > BIRTHDAY_HEIGHT as u32), + "no block at or below birthday height should be emitted" + ); + + Ok(()) +} + +/// Verifies that all `(tx, ts)` pairs in a [`MempoolEvent`] carry the exact `sync_time` passed +/// to [`Emitter::mempool_at`], ensuring callers control the timestamp semantics. +#[test] +fn mempool_at_uses_provided_timestamp() -> anyhow::Result<()> { + const SYNC_TIME: u64 = 42; + + let env = TestEnv::new()?; + env.mine_blocks(101, None)?; + + let addr = env + .rpc_client() + .get_new_address(None, None)? + .address()? + .require_network(Network::Regtest)?; + + let client = ClientExt::get_rpc_client(&env)?; + let mut emitter = Emitter::new( + &client, + CheckPoint::new(0, env.genesis_hash()?), + core::iter::empty::(), + ); + + // Advance to tip so the emitter tracks the current chain position. + while emitter.next_block()?.is_some() {} + + // Place a few transactions in the mempool. + for _ in 0..3 { + env.send(&addr, Amount::ONE_BTC)?; + } + + let event = emitter.mempool_at(SYNC_TIME)?; + + assert!( + !event.update.is_empty(), + "should have received mempool transactions" + ); + for (_, sync_time) in &event.update { + assert_eq!( + *sync_time, SYNC_TIME, + "all update timestamps must equal sync_time" + ); + } + + Ok(()) +} + +/// Verifies that when a reorg's agreement point falls below `start_height`, the emitter resets +/// `start_height` to the agreement height so that no reorged heights are skipped. +/// +/// Concretely, with a 6-block reorg the emitter must re-emit all 6 reorged heights in order +/// without skipping any, even though `start_height` was set to the pre-reorg tip. +#[test] +fn start_height_reset_on_reorg_prevents_height_gaps() -> anyhow::Result<()> { let env = TestEnv::new()?; let (mut local_chain, _) = LocalChain::from_genesis(env.genesis_hash()?); let client = ClientExt::get_rpc_client(&env)?; + const REORG_DEPTH: u32 = 6; + env.mine_blocks(110, None)?; let mut emitter = Emitter::new( &client, local_chain.tip(), - core::iter::empty::(), + core::iter::empty::>(), ); while let Some(emission) = emitter.next_block()? { let _ = local_chain.apply_update(emission.checkpoint)?; @@ -681,55 +794,220 @@ fn test_sync_with_new_emitter_after_reorg() -> anyhow::Result<()> { let pre_reorg_tip = local_chain.tip(); let tip_height = pre_reorg_tip.height(); - env.reorg(6)?; + env.reorg(REORG_DEPTH as usize)?; - // New emitter with start_height = tip height (common caller pattern). + // New emitter with start_height = tip height. The emitter must detect the reorg, walk back + // to the agreement point, and reset start_height so no invalidated heights are skipped. let mut emitter = Emitter::new( &client, local_chain.tip(), - core::iter::empty::(), - ); + core::iter::empty::>(), + ) + .start_height(tip_height); + let mut emitted_heights = Vec::new(); while let Some(emission) = emitter.next_block()? { + emitted_heights.push(emission.block_height()); let _ = local_chain .apply_update(emission.checkpoint) .expect("emission checkpoint must connect with local chain"); } + // All reorged heights must be re-emitted consecutively — no gaps. + let reorg_start = tip_height - REORG_DEPTH + 1; + let exp_heights: Vec = (reorg_start..=tip_height).collect(); + assert_eq!( + emitted_heights, exp_heights, + "emitter must re-emit all reorged heights without skipping; \ + got {:?}, expected {:?}", + emitted_heights, exp_heights, + ); + assert_eq!(local_chain.tip().height(), tip_height); assert_ne!(local_chain.tip().hash(), pre_reorg_tip.hash()); Ok(()) } +/// Evictions are withheld while the emitter is behind the node's best-block tip and are only +/// surfaced once [`Emitter::next_block`] has drained the chain to tip. This applies both to +/// live evictions and to transactions seeded via `expected_mempool_txs` at construction time +/// (the wallet-restart scenario). +/// +/// **Phase 1 — live eviction while catching up:** +/// 1. Mine 110 blocks; emitter starts at genesis — behind tip. +/// 2. Broadcast tx1; `mempool()` shows tx1 in `update`, `evicted` is empty (behind tip). +/// 3. Double-spend tx1 (tx1b) to evict it from the mempool. +/// 4. Call `mempool()` again — still behind tip, `evicted` must still be empty. +/// 5. Drain all blocks to tip; `mempool()` — tx1 must appear in `evicted`. +/// +/// **Phase 2 — wallet-restart: seeded tx already absent from mempool:** +/// 6. Create a fresh emitter from genesis, seeding tx1 (already evicted) as a known-unconfirmed tx. +/// 7. Drain all blocks to tip; `mempool()` — tx1 must again appear in `evicted`. #[test] -fn detect_new_mempool_txs() -> anyhow::Result<()> { +fn evictions_withheld_until_at_tip() -> anyhow::Result<()> { let env = TestEnv::new()?; - env.mine_blocks(101, None)?; + let rpc_client = env.rpc_client(); + let client = ClientExt::get_rpc_client(&env)?; - let addr = env - .rpc_client() + env.mine_blocks(110, None)?; + + let spk = ScriptBuf::new_p2wsh(&WScriptHash::all_zeros()); + let recipient = Address::from_script(&spk, Network::Regtest)?; + let txid_1 = env.send(&recipient, Amount::from_sat(10_000))?; + // Fetch the full tx now so we can seed it in phase 2. + let tx_1 = rpc_client.get_transaction(txid_1)?.into_model()?.tx; + + // --- Phase 1: live eviction while catching up --- + + // Emitter starts at genesis — deliberately behind the node's tip. + let mut emitter = Emitter::new( + &client, + CheckPoint::new(0, env.genesis_hash()?), + core::iter::empty::(), + ); + + // Behind tip: tx1 appears in update, but no evictions are reported yet. + let event = emitter.mempool()?; + assert!( + event + .update + .iter() + .any(|(tx, _)| tx.compute_txid() == txid_1), + "phase 1: tx1 should appear in mempool update", + ); + assert!( + event.evicted.is_empty(), + "phase 1: evicted must be empty while emitter is behind tip", + ); + + // Double-spend tx1 to evict it from the mempool. + let outpoint = tx_1.input[0].previous_output; + let new_addr = rpc_client .get_new_address(None, None)? .address()? - .require_network(Network::Regtest)?; + .assume_checked(); + let input = Input { + txid: outpoint.txid, + vout: outpoint.vout as u64, + sequence: None, + }; + let outputs = [Output::new(new_addr, Amount::from_btc(49.99)?)]; + let raw = rpc_client + .create_raw_transaction(&[input], &outputs)? + .into_model()? + .0; + let tx1b = rpc_client + .sign_raw_transaction_with_wallet(&raw)? + .into_model()? + .tx; + rpc_client.send_raw_transaction(&tx1b)?; + + // Still behind tip: evicted must remain empty even though tx1 is gone from the mempool. + let event = emitter.mempool()?; + assert!( + event.evicted.is_empty(), + "phase 1: evicted must remain empty while emitter is still catching up", + ); + + // Drain all blocks to tip. + while emitter.next_block()?.is_some() {} + // Now at tip: tx1 was evicted and must be reported. + let event = emitter.mempool()?; + assert!( + event.evicted.iter().any(|(txid, _)| txid == &txid_1), + "phase 1: tx1 must appear in evicted once emitter is at tip", + ); + + // --- Phase 2: wallet restart — seeded tx already absent from mempool --- + + // tx1 is still evicted. Simulate a wallet restart by creating a fresh emitter from + // genesis and seeding tx1 as a known-unconfirmed transaction. + let mut emitter2 = Emitter::new( + &client, + CheckPoint::new(0, env.genesis_hash()?), + core::iter::once(tx_1), + ); + + // Drain all blocks to tip. tx1 is not confirmed in any of them (it was evicted, not mined). + while emitter2.next_block()?.is_some() {} + + // At tip: the snapshot has tx1 (from seeding) but the node's mempool does not. Must be evicted. + let event = emitter2.mempool()?; + assert!( + event.evicted.iter().any(|(txid, _)| txid == &txid_1), + "phase 2: seeded-but-absent tx1 must be reported as evicted once at tip", + ); + + Ok(()) +} + +/// Setting `start_height` to a height that does not yet exist on the remote node results in +/// [`EmitterError::Rpc`] on the first [`Emitter::next_block`] call that tries to fetch that block. +/// +/// This is the documented contract: callers should not set `start_height` beyond the node's +/// current tip. The error is recoverable — the caller can reinitialise with a valid height. +#[test] +fn start_height_beyond_tip_returns_rpc_error() -> anyhow::Result<()> { + let env = TestEnv::new()?; let client = ClientExt::get_rpc_client(&env)?; + + // Mine enough to be able to coinbase-spend, but well short of 999_999. + env.mine_blocks(10, None)?; + let mut emitter = Emitter::new( &client, CheckPoint::new(0, env.genesis_hash()?), - core::iter::empty::(), + core::iter::empty::(), + ) + .start_height(999_999); + + // The emitter finds agreement at genesis, then tries get_block_hash(999_999) which the + // node rejects because that height does not exist. + let err = emitter + .next_block() + .expect_err("must fail with Rpc error for out-of-range start_height"); + assert!( + matches!(err, EmitterError::Rpc(_)), + "expected EmitterError::Rpc, got {err:?}", ); - while emitter.next_block()?.is_some() {} + Ok(()) +} - for n in 0..5 { - let txid = env.send(&addr, Amount::ONE_BTC)?; - let new_txs = emitter.mempool()?.update; - assert!( - new_txs.iter().any(|(tx, _)| tx.compute_txid() == txid), - "must detect new tx {n}" - ); - } +/// An emitter whose `last_cp` has a genesis hash that does not exist on the connected node +/// (e.g. a mainnet genesis hash used against a regtest node) exhausts all agreement candidates +/// and returns [`EmitterError::AgreementNotFound`]. +/// +/// This is the documented "catastrophic mismatch" safety net: the caller should reinitialise +/// the emitter against the correct node. +#[test] +fn wrong_genesis_returns_agreement_not_found() -> anyhow::Result<()> { + use bitcoin::constants::genesis_block; + + let env = TestEnv::new()?; + let client = ClientExt::get_rpc_client(&env)?; + + env.mine_blocks(10, None)?; + + // Use the mainnet genesis hash — it will never exist on the regtest node. + let mainnet_genesis = genesis_block(bitcoin::params::Params::MAINNET).block_hash(); + let mut emitter = Emitter::new( + &client, + CheckPoint::new(0, mainnet_genesis), + core::iter::empty::(), + ); + + // The agreement scanner finds no matching block and, after MAX_AGREEMENT_FAILURES retries + // within this single call, surfaces the error. + let err = emitter + .next_block() + .expect_err("must fail with AgreementNotFound for wrong genesis hash"); + assert!( + matches!(err, EmitterError::AgreementNotFound), + "expected EmitterError::AgreementNotFound, got {err:?}", + ); Ok(()) } From 8c5f55b5754a6251a0d79c6869fda35181547a1a Mon Sep 17 00:00:00 2001 From: valued mammal Date: Sat, 11 Jul 2026 17:28:26 -0400 Subject: [PATCH 6/7] feat(rpc)!: Support generic CheckPoint in Emitter and FilterIter Introduce a generic block-data type parameter B on Emitter and FilterIter so callers can thread full Headers through emissions, not just BlockHash. - emitter: Emitter<'a> is now Emitter<'a, B> and BlockEvent is BlockEvent. Added B: From
to existing generic bounds. - bip158: FilterIter<'a> and Event are now FilterIter<'a, B> and Event with the same bound on its Iterator impl. A private `as_header` helper reconstructs a bitcoin Header from GetBlockHeaderVerbose so the emitted checkpoint carries the header. - bip158: Error is now #[non_exhaustive] - test: Add `emitter_collects_header_checkpoints` to exercise Emitter with CheckPoint
--- crates/bitcoind_rpc/src/bip158.rs | 55 +++++++++++++----- crates/bitcoind_rpc/src/emitter.rs | 38 +++++++----- crates/bitcoind_rpc/tests/test_emitter.rs | 58 ++++++++++++++++++- crates/bitcoind_rpc/tests/test_filter_iter.rs | 4 +- 4 files changed, 121 insertions(+), 34 deletions(-) diff --git a/crates/bitcoind_rpc/src/bip158.rs b/crates/bitcoind_rpc/src/bip158.rs index 82a3cfa2ec..bf1591bacc 100644 --- a/crates/bitcoind_rpc/src/bip158.rs +++ b/crates/bitcoind_rpc/src/bip158.rs @@ -6,10 +6,12 @@ //! [0]: https://github.com/bitcoin/bips/blob/master/bip-0157.mediawiki //! [1]: https://github.com/bitcoin/bips/blob/master/bip-0158.mediawiki +use core::fmt::{self, Debug, Display}; + use bdk_core::bitcoin; -use bdk_core::CheckPoint; -use bitcoin::BlockHash; +use bdk_core::{CheckPoint, ToBlockHash}; use bitcoin::{bip158::BlockFilter, Block, ScriptBuf}; +use bitcoin::{block::Header, hashes::Hash, BlockHash}; use bitcoind_client::bitreq::Client; use crate::corepc_types::model::GetBlockHeaderVerbose; @@ -30,22 +32,22 @@ use crate::corepc_types::model::GetBlockHeaderVerbose; /// 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> { +pub struct FilterIter<'a, B> { /// RPC client client: &'a Client, /// SPK inventory spks: Vec, /// checkpoint - cp: CheckPoint, + cp: CheckPoint, /// Header info, contains the prev and next hashes for each header. header: Option, } -impl<'a> FilterIter<'a> { +impl<'a, B> FilterIter<'a, B> { /// Construct [`FilterIter`] with checkpoint, RPC client and SPKs. pub fn new( client: &'a Client, - cp: CheckPoint, + cp: CheckPoint, spks: impl IntoIterator, ) -> Self { Self { @@ -74,14 +76,14 @@ impl<'a> FilterIter<'a> { /// Event returned by [`FilterIter`]. #[derive(Debug, Clone)] -pub struct Event { +pub struct Event { /// Checkpoint - pub cp: CheckPoint, + pub cp: CheckPoint, /// Block, will be `Some(..)` for matching blocks pub block: Option, } -impl Event { +impl Event { /// Whether this event contains a matching block. pub fn is_match(&self) -> bool { self.block.is_some() @@ -93,8 +95,11 @@ impl Event { } } -impl Iterator for FilterIter<'_> { - type Item = Result; +impl Iterator for FilterIter<'_, B> +where + B: ToBlockHash + Debug + Clone + From
, +{ + type Item = Result, Error>; fn next(&mut self) -> Option { (|| -> Result, Error> { @@ -127,7 +132,7 @@ impl Iterator for FilterIter<'_> { next_hash = next_header.hash; let next_height = next_header.height; - cp = cp.insert(next_height, next_hash); + cp = cp.insert(next_height, next_header.as_header().into()); let mut block = None; let filter = @@ -152,6 +157,7 @@ impl Iterator for FilterIter<'_> { /// Error that may be thrown by [`FilterIter`]. #[derive(Debug)] +#[non_exhaustive] pub enum Error { /// RPC error Rpc(bitcoind_client::Error), @@ -161,8 +167,8 @@ pub enum Error { ReorgDepthExceeded, } -impl core::fmt::Display for Error { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::Rpc(e) => write!(f, "{e}"), Self::Bip158(e) => write!(f, "{e}"), @@ -178,3 +184,24 @@ impl From for Error { Self::Rpc(e) } } + +/// Trait used internally to derive a Bitcoin block [`Header`] from an instance of +/// [`GetBlockHeaderVerbose`]. +trait AsHeader { + fn as_header(&self) -> Header; +} + +impl AsHeader for GetBlockHeaderVerbose { + fn as_header(&self) -> Header { + Header { + version: self.version, + prev_blockhash: self + .previous_block_hash + .unwrap_or(BlockHash::from_byte_array([0x00; 32])), + merkle_root: self.merkle_root, + time: self.time, + bits: self.bits, + nonce: self.nonce, + } + } +} diff --git a/crates/bitcoind_rpc/src/emitter.rs b/crates/bitcoind_rpc/src/emitter.rs index a25a8fdd91..98f01799bc 100644 --- a/crates/bitcoind_rpc/src/emitter.rs +++ b/crates/bitcoind_rpc/src/emitter.rs @@ -2,8 +2,8 @@ use alloc::sync::Arc; use core::fmt; use bdk_core::collections::{HashMap, HashSet}; -use bdk_core::{BlockId, CheckPoint}; -use bitcoin::{Block, BlockHash, Transaction, Txid}; +use bdk_core::{BlockId, CheckPoint, ToBlockHash}; +use bitcoin::{block::Header, Block, BlockHash, Transaction, Txid}; use bitcoind_client::bitreq::Client; use bitcoind_client::corepc_types::model::GetBlockVerboseOne; @@ -16,7 +16,7 @@ const MAX_AGREEMENT_FAILURES: u8 = 3; /// Refer to [module-level documentation] for more. /// /// [module-level documentation]: crate -pub struct Emitter<'a> { +pub struct Emitter<'a, B> { client: &'a Client, /// Height from which to start emitting blocks. Defaults to `last_cp.height()`. @@ -25,7 +25,7 @@ pub struct Emitter<'a> { /// Checkpoint of the last-emitted block known to be in the best chain, and the tail of the /// linked checkpoint list threaded through each [`BlockEvent`]. Blocks reorged out are popped /// off during agreement scanning. - last_cp: CheckPoint, + last_cp: CheckPoint, /// RPC result of the last-emitted block, kept for its `nextblockhash` so the next block can be /// fetched without a height-to-hash lookup. `None` before the first emission, at tip, or after @@ -42,7 +42,10 @@ pub struct Emitter<'a> { mempool_snapshot: HashMap>, } -impl<'a> Emitter<'a> { +impl<'a, B> Emitter<'a, B> +where + B: ToBlockHash + Clone + fmt::Debug + From
, +{ /// Construct a new [`Emitter`]. /// /// `last_cp` is the chain the emitter starts from: it scans this checkpoint chain to find the @@ -54,7 +57,7 @@ impl<'a> Emitter<'a> { /// lets the emitter report evictions for them. Pass `core::iter::empty()` when empty. pub fn new( client: &'a Client, - last_cp: CheckPoint, + last_cp: CheckPoint, expected_mempool_txs: impl IntoIterator>>, ) -> Self { let start_height = last_cp.height(); @@ -179,7 +182,7 @@ impl<'a> Emitter<'a> { /// the node's best chain). On a reorg the emitter rescans for a new agreement point and /// re-emits from there. Returns [`EmitterError::AgreementNotFound`] if no agreement point can /// be found after multiple consecutive attempts. - pub fn next_block(&mut self) -> Result, EmitterError> { + pub fn next_block(&mut self) -> Result>, EmitterError> { if let Some((checkpoint, block)) = self.poll()? { // Confirmed transactions leave the mempool snapshot so they aren't misreported as // evictions on the next mempool() call. @@ -206,7 +209,7 @@ pub struct MempoolEvent { /// A newly emitted block from [`Emitter`]. #[derive(Debug)] -pub struct BlockEvent { +pub struct BlockEvent { /// The block. pub block: Block, @@ -218,10 +221,10 @@ pub struct BlockEvent { /// /// This is important as BDK structures require block-to-apply to be connected with another /// block in the original chain. - pub checkpoint: CheckPoint, + pub checkpoint: CheckPoint, } -impl BlockEvent { +impl BlockEvent { /// The block height of this new block. pub fn block_height(&self) -> u32 { self.checkpoint.height() @@ -254,7 +257,7 @@ impl BlockEvent { /// Outcome of a single node poll, driving the [`Emitter`] state machine (see /// [`Emitter::poll_once`]). -enum PollResponse { +enum PollResponse { /// The next consecutive block is ready to emit. NextBlock(GetBlockVerboseOne), /// The emitter is current with the node's tip; there is no next block. @@ -262,18 +265,21 @@ enum PollResponse { /// The last emitted block is no longer in the best chain; fall through to agreement scanning. Reorged, /// A checkpoint still in the best chain was found; resume scanning from here. - AgreementAt(GetBlockVerboseOne, CheckPoint), + AgreementAt(GetBlockVerboseOne, CheckPoint), /// No checkpoint matches the node's best chain. AgreementNotFound, } -impl<'a> Emitter<'a> { +impl<'a, B> Emitter<'a, B> +where + B: ToBlockHash + Clone + fmt::Debug + From
, +{ /// Probe the node once and return the appropriate `PollResponse`. /// /// `last_block` determines the next phase: when `Some`, follow its `next_block_hash` to the /// next block (reporting `PollResponse::Reorged` if it has been reorged out); when /// `None`, walk `last_cp` backwards to find the nearest checkpoint still in the best chain. - fn poll_once(&self) -> Result { + fn poll_once(&self) -> Result, bitcoind_client::Error> { if let Some(last_block_info) = &self.last_block { let next_hash = if last_block_info.height + 1 < self.start_height { // enforce start height @@ -317,7 +323,7 @@ impl<'a> Emitter<'a> { /// Drive the state machine until a block is ready to emit (`Some`) or the tip is reached /// (`None`). Returns [`EmitterError::AgreementNotFound`] after [`MAX_AGREEMENT_FAILURES`] /// consecutive failures to find an agreement point. - fn poll(&mut self) -> Result, Block)>, EmitterError> { + fn poll(&mut self) -> Result, Block)>, EmitterError> { loop { match self.poll_once()? { PollResponse::NextBlock(block_info) => { @@ -328,7 +334,7 @@ impl<'a> Emitter<'a> { let new_cp = self .last_cp .clone() - .push(height, hash) + .push(height, block.header.into()) .expect("NextBlock height must only increase"); self.last_cp = new_cp.clone(); self.last_block = Some(block_info); diff --git a/crates/bitcoind_rpc/tests/test_emitter.rs b/crates/bitcoind_rpc/tests/test_emitter.rs index f05c9b5b37..b24e7ca3ff 100644 --- a/crates/bitcoind_rpc/tests/test_emitter.rs +++ b/crates/bitcoind_rpc/tests/test_emitter.rs @@ -1,9 +1,9 @@ use std::collections::BTreeSet; use std::sync::Arc; -use bdk_bitcoind_rpc::{Emitter, EmitterError}; +use bdk_bitcoind_rpc::EmitterError; use bdk_chain::{ - bitcoin::{Address, Amount, Transaction, Txid}, + bitcoin::{Address, Amount, BlockHash, Transaction, Txid}, local_chain::{CheckPoint, LocalChain}, spk_txout::SpkTxOutIndex, Balance, BlockId, IndexedTxGraph, Merge, @@ -19,6 +19,8 @@ use crate::common::ClientExt; mod common; +type Emitter<'a> = bdk_bitcoind_rpc::Emitter<'a, BlockHash>; + /// Ensures blocks are emitted consecutively with correct hashes, and that after a reorg the /// emitter re-emits the replacement blocks at the same heights with updated hashes. /// @@ -1011,3 +1013,55 @@ fn wrong_genesis_returns_agreement_not_found() -> anyhow::Result<()> { Ok(()) } + +/// Exercises the emitter with a generic `CheckPoint
` rather than the default +/// `CheckPoint`. The generic `B` parameter must thread full block [`Header`]s through +/// every emitted checkpoint, and the checkpoint's derived block hash must match the emitted block. +#[test] +fn emitter_collects_header_checkpoints() -> anyhow::Result<()> { + use bitcoin::block::Header; + use bitcoin::constants::genesis_block; + + const CHAIN_TIP: usize = 20; + + let env = TestEnv::new()?; + let client = ClientExt::get_rpc_client(&env)?; + + env.mine_blocks(CHAIN_TIP, None)?; + let network_tip = env.rpc_client().get_block_count()?.into_model().0; + + // Start from a genesis checkpoint whose `data` is a full block `Header`. + let genesis_header = genesis_block(Network::Regtest).header; + let cp = CheckPoint::
::new(0, genesis_header); + + let mut emitter = + bdk_bitcoind_rpc::Emitter::new(&client, cp, core::iter::empty::()); + + let mut last_height = 0; + while let Some(block_event) = emitter.next_block()? { + let height = block_event.block_height(); + assert_eq!(height, last_height + 1, "heights must be consecutive"); + + // The emitted checkpoint carries the block's `Header`; its hash must match the block. + let header: Header = block_event.checkpoint.data(); + assert_eq!( + header, block_event.block.header, + "checkpoint header must match block header" + ); + assert_eq!( + header.block_hash(), + block_event.block_hash(), + "checkpoint hash must derive from the header", + ); + + last_height = height; + } + + assert_eq!( + u64::from(last_height), + network_tip, + "emitter must advance to the node's tip", + ); + + Ok(()) +} diff --git a/crates/bitcoind_rpc/tests/test_filter_iter.rs b/crates/bitcoind_rpc/tests/test_filter_iter.rs index f6aeea62b4..8e2683850c 100644 --- a/crates/bitcoind_rpc/tests/test_filter_iter.rs +++ b/crates/bitcoind_rpc/tests/test_filter_iter.rs @@ -1,7 +1,7 @@ use bdk_bitcoind_rpc::bip158::{Error, FilterIter}; use bdk_core::CheckPoint; use bdk_testenv::{anyhow, bitcoind, TestEnv}; -use bitcoin::{Address, Amount, Network, ScriptBuf}; +use bitcoin::{Address, Amount, BlockHash, Network, ScriptBuf}; use crate::common::ClientExt; @@ -60,7 +60,7 @@ fn filter_iter_error_wrong_network() -> anyhow::Result<()> { let _ = env.mine_blocks(10, None)?; // Try to initialize FilterIter with a CP on the wrong network - let cp = CheckPoint::new(0, bitcoin::hashes::Hash::hash(b"wrong-hash")); + let cp = CheckPoint::::new(0, bitcoin::hashes::Hash::hash(b"wrong-hash")); let client = ClientExt::get_rpc_client(&env)?; let mut iter = FilterIter::new(&client, cp, [ScriptBuf::new()]); assert!(matches!(iter.next(), Some(Err(Error::ReorgDepthExceeded)))); From 5a3298e4459a485adee2eb1a0c70aae68298768c Mon Sep 17 00:00:00 2001 From: valued mammal Date: Mon, 13 Jul 2026 11:37:20 -0400 Subject: [PATCH 7/7] deps(core,testenv)!: Bump electrsd to 0.41.0 The `download` feature now enables electrsd/bitcoind_30_2. This brings the downloaded bitcoind version in line with the latest version feature in bdk_bitcoind_rpc. electrsd 0.41.0 throws a compile error if no electrs version is set, causing compilation of bdk_core to fail when no testenv default features are set. We can't specify an electrs version directly as a feature of electrsd because that breaks docs.rs which has no network access when building the docs. Instead enable bdk_testenv default features in crates/core/Cargo.toml to ensure a version of electrs is set. However it's valid to note that since bdk_core only uses the test macros (not TestEnv), another option is to make electrsd an optional dependency in bdk_testenv, explicitly enabled by the download feature. That way crates like bdk_core can use bdk_testenv without pulling in the extra download dependencies. --- .github/workflows/cont_integration.yml | 6 ++---- crates/core/Cargo.toml | 2 +- crates/testenv/Cargo.toml | 4 ++-- crates/testenv/src/lib.rs | 1 + 4 files changed, 6 insertions(+), 7 deletions(-) diff --git a/.github/workflows/cont_integration.yml b/.github/workflows/cont_integration.yml index d97fccbeff..9e71e3fb98 100644 --- a/.github/workflows/cont_integration.yml +++ b/.github/workflows/cont_integration.yml @@ -226,12 +226,10 @@ jobs: with: toolchain: ${{ needs.prepare.outputs.rust_version }} cache: true + - name: Build (no default features + download) + run: cargo build --no-default-features --features download - name: Build (default features) run: cargo build - - name: Build (no default features) - run: cargo build --no-default-features - - name: Build (std, no download) - run: cargo build --no-default-features --features std - name: Build (all features) run: cargo build --all-features - name: Test (all features) diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml index ed313f3ae3..05aa89a1d7 100644 --- a/crates/core/Cargo.toml +++ b/crates/core/Cargo.toml @@ -22,7 +22,7 @@ serde = ["dep:serde", "bitcoin/serde", "hashbrown?/serde"] [dev-dependencies] bdk_chain = { path = "../chain" } -bdk_testenv = { path = "../testenv", default-features = false } +bdk_testenv = { path = "../testenv" } criterion = { version = "0.7" } proptest = "1.2.0" diff --git a/crates/testenv/Cargo.toml b/crates/testenv/Cargo.toml index 6cedcf4d30..8d3a198175 100644 --- a/crates/testenv/Cargo.toml +++ b/crates/testenv/Cargo.toml @@ -17,7 +17,7 @@ workspace = true [dependencies] bdk_chain = { path = "../chain", version = "0.23.1", default-features = false } -electrsd = { version = "0.38.0", features = [ "legacy" ], default-features = false } +electrsd = { version = "0.41.0", features = [ "legacy" ], default-features = false } bitcoin = { version = "0.32.0", default-features = false } [dev-dependencies] @@ -25,7 +25,7 @@ bdk_testenv = { path = "." } [features] default = ["std", "download"] -download = ["electrsd/bitcoind_download", "electrsd/bitcoind_28_2", "electrsd/esplora_a33e97e1"] +download = ["electrsd/bitcoind_download", "electrsd/bitcoind_30_2", "electrsd/esplora_a33e97e1"] std = ["bdk_chain/std", "bitcoin/rand-std"] serde = ["bdk_chain/serde"] diff --git a/crates/testenv/src/lib.rs b/crates/testenv/src/lib.rs index 3c3f8a6f48..1b365becfd 100644 --- a/crates/testenv/src/lib.rs +++ b/crates/testenv/src/lib.rs @@ -163,6 +163,7 @@ impl TestEnv { TemplateRules::Taproot, TemplateRules::Csv, ], + ..Default::default() })? .into_model()?) }