diff --git a/.github/workflows/cont_integration.yml b/.github/workflows/cont_integration.yml index d97fccbef..9e71e3fb9 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/bitcoind_rpc/Cargo.toml b/crates/bitcoind_rpc/Cargo.toml index 572d1f523..0cf58e8e4 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/README.md b/crates/bitcoind_rpc/README.md index 12de87020..24b171503 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/examples/filter_iter.rs b/crates/bitcoind_rpc/examples/filter_iter.rs index a346a2db0..03ceb1761 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 4caf59fcd..bf1591bac 100644 --- a/crates/bitcoind_rpc/src/bip158.rs +++ b/crates/bitcoind_rpc/src/bip158.rs @@ -6,18 +6,21 @@ //! [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 bitcoincore_rpc; -use bitcoincore_rpc::{json::GetBlockHeaderResult, RpcApi}; +use bitcoin::{block::Header, hashes::Hash, BlockHash}; +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 @@ -29,22 +32,22 @@ use bitcoincore_rpc::{json::GetBlockHeaderResult, RpcApi}; /// 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 bitcoincore_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, + 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 bitcoincore_rpc::Client, - cp: CheckPoint, + client: &'a Client, + cp: CheckPoint, spks: impl IntoIterator, ) -> Self { Self { @@ -58,10 +61,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)), @@ -73,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() @@ -92,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> { @@ -111,7 +117,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,14 +125,14 @@ 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); + cp = cp.insert(next_height, next_header.as_header().into()); let mut block = None; let filter = @@ -151,47 +157,51 @@ impl Iterator for FilterIter<'_> { /// Error that may be thrown by [`FilterIter`]. #[derive(Debug)] +#[non_exhaustive] 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 { - 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}"), 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) - } +/// Trait used internally to derive a Bitcoin block [`Header`] from an instance of +/// [`GetBlockHeaderVerbose`]. +trait AsHeader { + fn as_header(&self) -> Header; } -/// 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 - ) +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 new file mode 100644 index 000000000..98f01799b --- /dev/null +++ b/crates/bitcoind_rpc/src/emitter.rs @@ -0,0 +1,486 @@ +use alloc::sync::Arc; +use core::fmt; + +use bdk_core::collections::{HashMap, HashSet}; +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; + +/// 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. +/// +/// [module-level documentation]: crate +pub struct Emitter<'a, B> { + client: &'a Client, + + /// Height from which to start emitting blocks. Defaults to `last_cp.height()`. + start_height: u32, + + /// 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, + + /// 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, + + /// 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, + + /// 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, 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 + /// 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 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, + 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| { + let tx: Arc = tx.into(); + (tx.compute_txid(), tx) + }) + .collect(), + } + } + + /// 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 a full snapshot of the mempool along with any evicted [`Txid`]s. + /// + /// 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 + .elapsed() + .expect("must get current time") + .as_secs(); + self.mempool_at(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). + /// + /// 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()?; + + 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: mempool_txs + .iter() + .map(|(_, tx, sync_time)| (tx.clone(), *sync_time)) + .collect(), + ..Default::default() + }; + + let at_tip = self.last_cp.hash() == tip_hash; + + if at_tip { + // 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() + .filter(|&txid| !mempool_txids.contains(txid)) + .map(|&txid| (txid, sync_time)) + .collect(); + self.mempool_snapshot = mempool_txs + .iter() + .map(|(txid, tx, _)| (*txid, tx.clone())) + .collect(); + } else { + // 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()))); + }; + + Ok(mempool_event) + } + + /// 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()? { + // 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()); + } + return Ok(Some(BlockEvent { block, checkpoint })); + } + Ok(None) + } +} + +/// A new emission from mempool. +#[derive(Debug, Default)] +pub struct MempoolEvent { + /// 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 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)>, +} + +/// A newly emitted block from [`Emitter`]. +#[derive(Debug)] +pub struct BlockEvent { + /// The block. + pub block: Block, + + /// 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`] this block's checkpoint chain connects to. + /// + /// 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; 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, + }, + } + } +} + +/// Outcome of a single node poll, driving the [`Emitter`] state machine (see +/// [`Emitter::poll_once`]). +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. + Tip, + /// 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), + /// No checkpoint matches the node's best chain. + AgreementNotFound, +} + +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, 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 + 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)) + }; + } + + 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; + } + // genesis not found; cannot form a connected update + break; + } + Err(e) => return Err(e), + }; + + return Ok(PollResponse::AgreementAt(block_info, cp)); + } + + Ok(PollResponse::AgreementNotFound) + } + + /// 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()? { + 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, block.header.into()) + .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; + } + } + } + } +} + +/// 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, +} + +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; + use bdk_chain::local_chain::LocalChain; + use bdk_testenv::{anyhow, TestEnv}; + use bitcoin::{hashes::Hash, Address, Amount, ScriptBuf, Transaction, 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(), + core::iter::empty::(), + ); + + 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 9ab4bd192..4691a60a8 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 -//! [`bitcoincore_rpc::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)] @@ -14,526 +6,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 bitcoincore_rpc::{bitcoincore_rpc_json, RpcApi}; -use core::ops::Deref; +pub extern crate bitcoind_client; -pub mod bip158; - -pub use bitcoincore_rpc; - -/// The [`Emitter`] is used to emit data sourced from [`bitcoincore_rpc::Client`]. -/// -/// Refer to [module-level documentation] for more. -/// -/// [module-level documentation]: crate -pub struct Emitter { - client: C, - 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 Emitter -where - C: Deref, - C::Target: RpcApi, -{ - /// Construct a new [`Emitter`]. - /// - /// `last_cp` informs the emitter of the chain we are starting off with. This way, the emitter - /// can start emission from a block that connects to the original chain. - /// - /// `start_height` starts emission from a given height (if there are no conflicts with the - /// original chain). - /// - /// `expected_mempool_txs` is the initial set of unconfirmed transactions provided by the - /// wallet. This allows the [`Emitter`] to inform the wallet about relevant mempool evictions. - /// If it is known that the wallet is empty, [`NO_EXPECTED_MEMPOOL_TXS`] can be used. - pub fn new( - client: C, - last_cp: CheckPoint, - 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, None) { - 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() as u64 && 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). - /// - /// 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> { - 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() - } +mod emitter; +pub use emitter::*; - /// 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(bitcoincore_rpc_json::GetBlockResult), - NoMoreBlocks, - /// Fetched block is not in the best chain. - BlockNotInBestChain, - AgreementFound(bitcoincore_rpc_json::GetBlockResult, 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; - - if let Some(last_res) = &emitter.last_block { - let next_hash = if last_res.height + 1 < emitter.start_height as _ { - // enforce start height - let next_hash = client.get_block_hash(emitter.start_height as _)?; - // make sure last emission is still in best chain - if client.get_block_hash(last_res.height as _)? != last_res.hash { - return Ok(PollResponse::BlockNotInBestChain); - } - next_hash - } else { - match last_res.nextblockhash { - None => return Ok(PollResponse::NoMoreBlocks), - Some(next_hash) => next_hash, - } - }; - - let res = client.get_block_info(&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_info(&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)>, bitcoincore_rpc::Error> -where - C: Deref, - C::Target: RpcApi, - F: Fn(&BlockHash, &C::Target) -> Result, -{ - loop { - match poll_once(emitter)? { - PollResponse::Block(res) => { - let height = res.height as u32; - 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 as u32) < emitter.start_height - && (res.height as u32) < emitter.last_cp.height() - { - emitter.start_height = res.height as _; - } - // 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; - } - } - } -} - -/// 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 { - 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 = bitcoincore_rpc::Client::new( - &env.bitcoind.rpc_url(), - bitcoincore_rpc::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()); +pub mod bip158; - Ok(()) - } -} +pub(crate) use bitcoind_client::corepc_types; diff --git a/crates/bitcoind_rpc/tests/common/mod.rs b/crates/bitcoind_rpc/tests/common/mod.rs index bbd914cca..2d75f5bf3 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 2aeede628..b24e7ca3f 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, ops::Deref}; +use std::collections::BTreeSet; +use std::sync::Arc; -use bdk_bitcoind_rpc::{Emitter, NO_EXPECTED_MEMPOOL_TXS}; +use bdk_bitcoind_rpc::EmitterError; use bdk_chain::{ - bitcoin::{Address, Amount, Txid}, + bitcoin::{Address, Amount, BlockHash, Transaction, Txid}, local_chain::{CheckPoint, LocalChain}, spk_txout::SpkTxOutIndex, Balance, BlockId, IndexedTxGraph, Merge, @@ -18,23 +19,30 @@ use crate::common::ClientExt; mod common; -/// Ensure that blocks are emitted in order even after reorg. +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. /// /// 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()?); 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 + // 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) @@ -133,12 +141,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 @@ -147,37 +158,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(), 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(); 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 @@ -199,9 +200,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() @@ -210,7 +211,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 @@ -235,10 +236,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(()) @@ -246,28 +247,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()?), - EMITTER_START_HEIGHT as _, - NO_EXPECTED_MEMPOOL_TXS, + 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 { @@ -298,15 +302,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)?; @@ -343,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 @@ -420,13 +419,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; @@ -436,8 +432,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 @@ -487,80 +482,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 + + // mine 101 blocks first so block 98 exists for the checkpoint + env.mine_blocks(100, None)?; + + assert_eq!(env.bitcoind.client.get_block_count()?.0, 101); + + // 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(0, env.genesis_hash()?), - (PREMINE_COUNT - 3) as u32, - NO_EXPECTED_MEMPOOL_TXS, + CheckPoint::new(cp_height as u32, cp_hash), + core::iter::empty::(), ); - // mine 101 blocks - env.mine_blocks(PREMINE_COUNT, None)?; + // 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; - // 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); + // emit block 100a (advance the emitter past 99a so the reorg spans 3 blocks) + let _block_100a = emitter.next_block()?.expect("block 100a header"); - // get hash for block 101a - let blockhash_101a = env.rpc_client().get_block_hash(101)?.block_hash()?; + // Reorg depth 3: invalidates 99a, 100a, 101a and mines new 99b, 100b, 101b. + env.reorg(3)?; - // 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 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(); - // mine new blocks 99b, 100b, 101b - env.mine_blocks(3, None)?; - - // emit block header 99b - let block_99b = emitter.next_block()?.expect("block 99b"); - assert_eq!(block_99b.block_height(), 99); - - 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 - ); + 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<()> { - use bdk_bitcoind_rpc::bitcoincore_rpc::bitcoin; +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; @@ -568,11 +554,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(); @@ -590,7 +576,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)?; @@ -605,33 +591,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 @@ -661,18 +650,145 @@ 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(), 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)?; } @@ -680,57 +796,272 @@ 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(), - tip_height, - NO_EXPECTED_MEMPOOL_TXS, - ); + 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()?), - 0, - NO_EXPECTED_MEMPOOL_TXS, + 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(()) +} + +/// 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 d6136ecfb..8e2683850 100644 --- a/crates/bitcoind_rpc/tests/test_filter_iter.rs +++ b/crates/bitcoind_rpc/tests/test_filter_iter.rs @@ -1,8 +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 bitcoincore_rpc::RpcApi; +use bitcoin::{Address, Amount, BlockHash, Network, ScriptBuf}; 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); @@ -63,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)))); @@ -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)?; } diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml index ed313f3ae..05aa89a1d 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 6cedcf4d3..8d3a19817 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 3c3f8a6f4..1b365becf 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()?) }