fix(rpc)!: improve checkpoint update

Change `Event` to a simple struct containing `cp` and optional
`block`. The checkpoint is updated on each iteration whether
or not it corresponds to a matching block. We use
`CheckPoint::insert` which will also purge evicted blocks
if needed.

Change implementation of `find_base` to use `get_block_header_info`
which helps to reduce the number of RPC calls.

Add test `event_checkpoint_connects_to_local_chain` to check
the expected events after a reorg, and check that intermediate
updates can be applied to the local chain.

Author: ValuedMammal

crates/bitcoind_rpc/examples/filter_iter.rs

@@ -61,20 +61,12 @@ fn main() -> anyhow::Result<()> {
     let start = Instant::now();
 
     for res in iter {
-        let event = res?;
-        match event {
-            Event::NoMatch { .. } => {}
-            Event::Block { cp, ref block } => {
-                // Apply relevant tx data
-                let height = cp.height();
-                let _ = graph.apply_block_relevant(block, height);
-                // Update chain tip
-                let _ = chain.apply_update(cp)?;
-                println!("Matched block {height}");
-            }
-            Event::Tip { cp } => {
-                let _ = chain.apply_update(cp)?;
-            }
+        let Event { cp, block } = res?;
+        let height = cp.height();
+        let _ = chain.apply_update(cp)?;
+        if let Some(block) = block {
+            let _ = graph.apply_block_relevant(&block, height);
+            println!("Matched block {height}");
         }
     }
 

crates/bitcoind_rpc/src/bip158.rs

@@ -41,59 +41,40 @@ impl<'a> FilterIter<'a> {
         }
     }
 
-    /// Find the agreement height with the remote node and return the corresponding
-    /// header info.
+    /// Return the agreement header with the remote node.
     ///
-    /// Error if no agreement height is found.
+    /// Error if no agreement header is found.
     fn find_base(&self) -> Result<GetBlockHeaderResult, Error> {
         for cp in self.cp.iter() {
-            let height = cp.height();
-
-            let fetched_hash = self.client.get_block_hash(height as u64)?;
-
-            if fetched_hash == cp.hash() {
-                return Ok(self.client.get_block_header_info(&fetched_hash)?);
+            match self.client.get_block_header_info(&cp.hash()) {
+                Err(e) if is_not_found(&e) => continue,
+                Ok(header) if header.confirmations <= 0 => continue,
+                Ok(header) => return Ok(header),
+                Err(e) => return Err(Error::Rpc(e)),
             }
         }
-
         Err(Error::ReorgDepthExceeded)
     }
 }
 
-/// Kind of event produced by `FilterIter`.
+/// Event returned by [`FilterIter`].
 #[derive(Debug, Clone)]
-pub enum Event {
-    /// Block
-    Block {
-        /// checkpoint
-        cp: CheckPoint,
-        /// block
-        block: Block,
-    },
-    /// No match
-    NoMatch {
-        /// block id
-        id: BlockId,
-    },
-    /// Tip
-    Tip {
-        /// checkpoint
-        cp: CheckPoint,
-    },
+pub struct Event {
+    /// Checkpoint
+    pub cp: CheckPoint,
+    /// Block, will be `Some(..)` for matching blocks
+    pub block: Option<Block>,
 }
 
 impl Event {
     /// Whether this event contains a matching block.
     pub fn is_match(&self) -> bool {
-        matches!(self, Event::Block { .. })
+        self.block.is_some()
     }
 
     /// Return the height of the event.
     pub fn height(&self) -> u32 {
-        match self {
-            Self::Block { cp, .. } | Self::Tip { cp } => cp.height(),
-            Self::NoMatch { id } => id.height,
-        }
+        self.cp.height()
     }
 }
 
@@ -106,15 +87,9 @@ impl Iterator for FilterIter<'_> {
 
             let header = match self.header.take() {
                 Some(header) => header,
-                None => {
-                    // If no header is cached we need to locate a base of the local
-                    // checkpoint from which the scan may proceed.
-                    let header = self.find_base()?;
-                    let height: u32 = header.height.try_into()?;
-                    cp = cp.range(..=height).next().expect("we found a base");
-
-                    header
-                }
+                // If no header is cached we need to locate a base of the local
+                // checkpoint from which the scan may proceed.
+                None => self.find_base()?,
             };
 
             let mut next_hash = match header.next_block_hash {
@@ -126,58 +101,38 @@ impl Iterator for FilterIter<'_> {
 
             // In case of a reorg, rewind by fetching headers of previous hashes until we find
             // one with enough confirmations.
-            let mut reorg_ct: i32 = 0;
             while next_header.confirmations < 0 {
                 let prev_hash = next_header
                     .previous_block_hash
                     .ok_or(Error::ReorgDepthExceeded)?;
                 let prev_header = self.client.get_block_header_info(&prev_hash)?;
                 next_header = prev_header;
-                reorg_ct += 1;
             }
 
             next_hash = next_header.hash;
             let next_height: u32 = next_header.height.try_into()?;
 
-            // Purge any no longer valid checkpoints.
-            if reorg_ct.is_positive() {
-                cp = cp
-                    .range(..=next_height)
-                    .next()
-                    .ok_or(Error::ReorgDepthExceeded)?;
-            }
-            let block_id = BlockId {
+            cp = cp.insert(BlockId {
                 height: next_height,
                 hash: next_hash,
-            };
-            let filter_bytes = self.client.get_block_filter(&next_hash)?.filter;
-            let filter = BlockFilter::new(&filter_bytes);
+            });
 
-            let next_event = if filter
+            let mut block = None;
+            let filter =
+                BlockFilter::new(self.client.get_block_filter(&next_hash)?.filter.as_slice());
+            if filter
                 .match_any(&next_hash, self.spks.iter().map(ScriptBuf::as_ref))
                 .map_err(Error::Bip158)?
             {
-                let block = self.client.get_block(&next_hash)?;
-                cp = cp.insert(block_id);
-
-                Ok(Some(Event::Block {
-                    cp: cp.clone(),
-                    block,
-                }))
-            } else if next_header.next_block_hash.is_none() {
-                cp = cp.insert(block_id);
-
-                Ok(Some(Event::Tip { cp: cp.clone() }))
-            } else {
-                Ok(Some(Event::NoMatch { id: block_id }))
-            };
+                block = Some(self.client.get_block(&next_hash)?);
+            }
 
             // Store the next header
             self.header = Some(next_header);
             // Update self.cp
-            self.cp = cp;
+            self.cp = cp.clone();
 
-            next_event
+            Ok(Some(Event { cp, block }))
         })()
         .transpose()
     }
@@ -221,3 +176,12 @@ impl From<core::num::TryFromIntError> for Error {
         Self::TryFromInt(e)
     }
 }
+
+/// Whether the RPC error is a "not found" error (code: `-5`).
+fn is_not_found(e: &bitcoincore_rpc::Error) -> bool {
+    matches!(
+        e,
+        bitcoincore_rpc::Error::JsonRpc(bitcoincore_rpc::jsonrpc::Error::Rpc(e))
+        if e.code == -5
+    )
+}

crates/bitcoind_rpc/tests/test_filter_iter.rs

@@ -112,3 +112,44 @@ fn filter_iter_detects_reorgs() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn event_checkpoint_connects_to_local_chain() -> anyhow::Result<()> {
+    use bitcoin::BlockHash;
+    use std::collections::BTreeMap;
+    let env = testenv()?;
+    let _ = env.mine_blocks(15, None)?;
+
+    let cp = env.make_checkpoint_tip();
+    let mut chain = bdk_chain::local_chain::LocalChain::from_tip(cp.clone())?;
+    assert_eq!(chain.tip().height(), 16);
+    let old_hashes: Vec<BlockHash> = [14, 15, 16]
+        .into_iter()
+        .map(|height| chain.get(height).unwrap().hash())
+        .collect();
+
+    // Construct iter
+    let mut iter = FilterIter::new(&env.bitcoind.client, cp, vec![ScriptBuf::new()]);
+
+    // Now reorg 3 blocks (14, 15, 16)
+    let new_hashes: BTreeMap<u32, BlockHash> = (14..=16).zip(env.reorg(3)?).collect();
+
+    // Expect events from height 14 on...
+    while let Some(event) = iter.next().transpose()? {
+        let _ = chain
+            .apply_update(event.cp)
+            .expect("chain update should connect");
+    }
+
+    for height in 14..=16 {
+        let hash = chain.get(height).unwrap().hash();
+        assert!(!old_hashes.contains(&hash), "Old hashes were reorged out");
+        assert_eq!(
+            new_hashes.get(&height),
+            Some(&hash),
+            "Chain must include new hashes"
+        );
+    }
+
+    Ok(())
+}