feat(rpc): Introduce FilterIter

crates/bitcoind_rpc/Cargo.toml

@@ -28,3 +28,6 @@ bdk_chain = { path = "../chain" }
 default = ["std"]
 std = ["bitcoin/std", "bdk_core/std"]
 serde = ["bitcoin/serde", "bdk_core/serde"]
+
+[[example]]
+name = "bip158"

crates/bitcoind_rpc/examples/bip158.rs

@@ -0,0 +1,75 @@
+#![allow(clippy::print_stdout)]
+use bdk_bitcoind_rpc::bip158::{Event, EventInner, FilterIter};
+use bdk_chain::bitcoin::{constants::genesis_block, secp256k1::Secp256k1, Network};
+use bdk_chain::indexer::keychain_txout::KeychainTxOutIndex;
+use bdk_chain::local_chain::LocalChain;
+use bdk_chain::miniscript::Descriptor;
+use bdk_chain::{BlockId, ConfirmationBlockTime, IndexedTxGraph, SpkIterator};
+use bdk_testenv::anyhow;
+
+// This example shows how BDK chain and tx-graph structures are updated using compact filters syncing.
+// assumes a local Signet node, and "RPC_COOKIE" set in environment.
+
+// Usage: `cargo run -p bdk_bitcoind_rpc --example bip158`
+
+const EXTERNAL: &str = "tr([83737d5e/86h/1h/0h]tpubDDR5GgtoxS8fJyjjvdahN4VzV5DV6jtbcyvVXhEKq2XtpxjxBXmxH3r8QrNbQqHg4bJM1EGkxi7Pjfkgnui9jQWqS7kxHvX6rhUeriLDKxz/0/*)";
+const INTERNAL: &str = "tr([83737d5e/86h/1h/0h]tpubDDR5GgtoxS8fJyjjvdahN4VzV5DV6jtbcyvVXhEKq2XtpxjxBXmxH3r8QrNbQqHg4bJM1EGkxi7Pjfkgnui9jQWqS7kxHvX6rhUeriLDKxz/1/*)";
+const SPK_COUNT: u32 = 10;
+const NETWORK: Network = Network::Signet;
+
+fn main() -> anyhow::Result<()> {
+    // Setup receiving chain and graph structures.
+    let secp = Secp256k1::new();
+    let (descriptor, _) = Descriptor::parse_descriptor(&secp, EXTERNAL)?;
+    let (change_descriptor, _) = Descriptor::parse_descriptor(&secp, INTERNAL)?;
+    let (mut chain, _) = LocalChain::from_genesis_hash(genesis_block(NETWORK).block_hash());
+    let mut graph = IndexedTxGraph::<ConfirmationBlockTime, KeychainTxOutIndex<usize>>::new({
+        let mut index = KeychainTxOutIndex::default();
+        index.insert_descriptor(0, descriptor.clone())?;
+        index.insert_descriptor(1, change_descriptor.clone())?;
+        index
+    });
+
+    // Assume a minimum birthday height
+    let block = BlockId {
+        height: 205_000,
+        hash: "0000002bd0f82f8c0c0f1e19128f84c938763641dba85c44bdb6aed1678d16cb".parse()?,
+    };
+    let _ = chain.insert_block(block)?;
+
+    // Configure RPC client
+    let rpc_client = bitcoincore_rpc::Client::new(
+        "127.0.0.1:38332",
+        bitcoincore_rpc::Auth::CookieFile(std::env::var("RPC_COOKIE")?.into()),
+    )?;
+
+    // Initialize block emitter
+    let mut emitter = FilterIter::new_with_checkpoint(&rpc_client, chain.tip());
+    for (_, desc) in graph.index.keychains() {
+        let spks = SpkIterator::new_with_range(desc, 0..SPK_COUNT).map(|(_, spk)| spk);
+        emitter.add_spks(spks);
+    }
+
+    // Sync
+    if emitter.get_tip()?.is_some() {
+        // apply relevant blocks
+        for event in emitter.by_ref() {
+            if let Event::Block(EventInner { height, block }) = event? {
+                let _ = graph.apply_block_relevant(&block, height);
+            }
+        }
+        // update chain
+        if let Some(tip) = emitter.chain_update()? {
+            let _ = chain.apply_update(tip)?;
+        }
+    }
+
+    println!("Local tip: {}", chain.tip().height());
+
+    println!("Unspent");
+    for (_, outpoint) in graph.index.outpoints() {
+        println!("{outpoint}");
+    }
+
+    Ok(())
+}

crates/bitcoind_rpc/src/bip158.rs

@@ -0,0 +1,261 @@
+//! Compact block filters sync over RPC, see also [BIP157][0].
+//!
+//! This module is home to [`FilterIter`], a structure that returns bitcoin blocks by matching
+//! a list of script pubkeys against a [BIP158][1] [`BlockFilter`].
+//!
+//! [0]: https://github.com/bitcoin/bips/blob/master/bip-0157.mediawiki
+//! [1]: https://github.com/bitcoin/bips/blob/master/bip-0158.mediawiki
+
+use bdk_core::collections::BTreeMap;
+use core::fmt;
+
+use bdk_core::bitcoin;
+use bdk_core::{BlockId, CheckPoint};
+use bitcoin::{
+    bip158::{self, BlockFilter},
+    Block, BlockHash, ScriptBuf,
+};
+use bitcoincore_rpc;
+use bitcoincore_rpc::RpcApi;
+
+/// Block height
+type Height = u32;
+
+/// Type that generates block [`Event`]s by matching a list of script pubkeys against a
+/// [`BlockFilter`].
+#[derive(Debug)]
+pub struct FilterIter<'c, C> {
+    // RPC client
+    client: &'c C,
+    // SPK inventory
+    spks: Vec<ScriptBuf>,
+    // local cp
+    cp: Option<CheckPoint>,
+    // blocks map
+    blocks: BTreeMap<Height, BlockHash>,
+    // next filter
+    next_filter: Option<NextFilter>,
+    // best height counter
+    height: Height,
+    // stop height
+    stop: Height,
+}
+
+impl<'c, C: RpcApi> FilterIter<'c, C> {
+    /// Construct [`FilterIter`] from a given `client` and start `height`.
+    pub fn new_with_height(client: &'c C, height: u32) -> Self {
+        Self {
+            client,
+            spks: vec![],
+            cp: None,
+            blocks: BTreeMap::new(),
+            next_filter: None,
+            height,
+            stop: 0,
+        }
+    }
+
+    /// Construct [`FilterIter`] from a given `client` and [`CheckPoint`].
+    pub fn new_with_checkpoint(client: &'c C, cp: CheckPoint) -> Self {
+        let mut filter_iter = Self::new_with_height(client, cp.height());
+        filter_iter.cp = Some(cp);
+        filter_iter
+    }
+
+    /// Extends `self` with an iterator of spks.
+    pub fn add_spks(&mut self, spks: impl IntoIterator<Item = ScriptBuf>) {
+        self.spks.extend(spks)
+    }
+
+    /// Add spk to the list of spks to scan with.
+    pub fn add_spk(&mut self, spk: ScriptBuf) {
+        self.spks.push(spk);
+    }
+
+    /// Get the next filter and increment the current best height.
+    ///
+    /// Returns `Ok(None)` when the stop height is exceeded.
+    fn next_filter(&mut self) -> Result<Option<NextFilter>, Error> {
+        if self.height > self.stop {
+            return Ok(None);
+        }
+        let height = self.height;
+        let hash = match self.blocks.get(&height) {
+            Some(h) => *h,
+            None => self.client.get_block_hash(height as u64)?,
+        };
+        let filter_bytes = self.client.get_block_filter(&hash)?.filter;
+        let filter = BlockFilter::new(&filter_bytes);
+        self.height += 1;
+        Ok(Some((BlockId { height, hash }, filter)))
+    }
+
+    /// Get the remote tip.
+    ///
+    /// Returns `None` if there's no difference between the height of this [`FilterIter`] and the
+    /// remote height.
+    pub fn get_tip(&mut self) -> Result<Option<BlockId>, Error> {
+        let tip_hash = self.client.get_best_block_hash()?;
+        let mut header_info = self.client.get_block_header_info(&tip_hash)?;
+        let tip_height = header_info.height as u32;
+        if self.height == tip_height {
+            // nothing to do
+            return Ok(None);
+        }
+        self.blocks.insert(tip_height, tip_hash);
+        // if we have a checkpoint we mandate a lookback of ten blocks
+        // to ensure consistency of the local chain
+        if let Some(cp) = self.cp.as_ref() {
+            for _ in 0..9 {
+                match header_info.previous_block_hash {
+                    None => break,
+                    Some(hash) => {
+                        header_info = self.client.get_block_header_info(&hash)?;
+                        let height = header_info.height as u32;
+                        self.blocks.insert(height, hash);
+                    }
+                }
+            }
+            let start_height = self.blocks.keys().next().expect("blocks not empty");
+            self.height = cp.height().min(*start_height);
+        }
+
+        self.stop = tip_height;
+
+        // get the first filter
+        self.next_filter = self.next_filter()?;
+
+        Ok(Some(BlockId {
+            height: tip_height,
+            hash: self.blocks[&tip_height],
+        }))
+    }
+}
+
+/// Alias for a compact filter and associated block id.
+type NextFilter = (BlockId, BlockFilter);
+
+/// Event inner type
+#[derive(Debug, Clone)]
+pub struct EventInner {
+    /// Height
+    pub height: Height,
+    /// Block
+    pub block: Block,
+}
+
+/// Kind of event produced by [`FilterIter`].
+#[derive(Debug, Clone)]
+pub enum Event {
+    /// Block
+    Block(EventInner),
+    /// No match
+    NoMatch,
+}
+
+impl<'c, C: RpcApi> Iterator for FilterIter<'c, C> {
+    type Item = Result<Event, Error>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let (block, filter) = self.next_filter.clone()?;
+
+        (|| -> Result<_, Error> {
+            // if the next filter matches any of our watched spks, get the block
+            // and return it, inserting relevant block ids along the way
+            let height = block.height;
+            let hash = block.hash;
+            let event = if filter
+                .match_any(&hash, self.spks.iter().map(|script| script.as_bytes()))
+                .map_err(Error::Bip158)?
+            {
+                let block = self.client.get_block(&hash)?;
+                self.blocks.insert(height, hash);
+                let inner = EventInner { height, block };
+                Event::Block(inner)
+            } else {
+                Event::NoMatch
+            };
+
+            self.next_filter = self.next_filter()?;
+            Ok(Some(event))
+        })()
+        .transpose()
+    }
+}
+
+impl<'c, C: RpcApi> FilterIter<'c, C> {
+    /// Get block from `self` if it exists or else get it from the RPC client
+    /// and return the block hash.
+    fn fetch_block(&mut self, height: Height) -> Result<BlockHash, Error> {
+        match self.blocks.get(&height).cloned() {
+            Some(hash) => Ok(hash),
+            None => {
+                let hash = self.client.get_block_hash(height as _)?;
+                self.blocks.insert(height, hash);
+                Ok(hash)
+            }
+        }
+    }
+
+    /// Returns a chain update from the newly scanned blocks, which will be `None` if this
+    /// [`FilterIter`] was not constructed using a [`CheckPoint`].
+    ///
+    /// # Errors
+    ///
+    /// This will error if the RPC client decides to fail.
+    pub fn chain_update(&mut self) -> Result<Option<CheckPoint>, Error> {
+        if self.cp.is_none() {
+            return Ok(None);
+        }
+        let mut cp = self.cp.clone().expect("must have cp");
+        let base = loop {
+            let height = cp.height();
+            let remote_hash = if height > 0 {
+                self.fetch_block(height)?
+            } else {
+                // fallback to genesis hash
+                cp.hash()
+            };
+            if cp.hash() == remote_hash {
+                break cp.block_id();
+            }
+            cp = cp.prev().expect("cp height must be > 0");
+        };
+
+        #[rustfmt::skip]
+        let blocks = self.blocks.iter().skip_while(|(&height, _)| height <= base.height)
+            .map(BlockId::from);
+
+        Ok(Some(
+            CheckPoint::from_block_ids(core::iter::once(base).chain(blocks))
+                .expect("blocks must be in order"),
+        ))
+    }
+}
+
+/// Errors that may occur during a compact filters sync.
+#[derive(Debug)]
+pub enum Error {
+    /// bitcoin bip158 error
+    Bip158(bip158::Error),
+    /// bitcoincore_rpc error
+    Rpc(bitcoincore_rpc::Error),
+}
+
+impl From<bitcoincore_rpc::Error> for Error {
+    fn from(e: bitcoincore_rpc::Error) -> Self {
+        Self::Rpc(e)
+    }
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Bip158(e) => e.fmt(f),
+            Self::Rpc(e) => e.fmt(f),
+        }
+    }
+}
+
+#[cfg(feature = "std")]
+impl std::error::Error for Error {}

crates/bitcoind_rpc/src/lib.rs

@@ -11,9 +11,12 @@
 
 use bdk_core::{BlockId, CheckPoint};
 use bitcoin::{block::Header, Block, BlockHash, Transaction};
-pub use bitcoincore_rpc;
 use bitcoincore_rpc::bitcoincore_rpc_json;
 
+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.