feat(bip158): FilterIter detects reorgs

Author: ValuedMammal

crates/bitcoind_rpc/src/bip158.rs

@@ -6,7 +6,7 @@
 //! [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 bdk_core::collections::{BTreeMap, BTreeSet};
 use core::fmt;
 
 use bdk_core::bitcoin;
@@ -33,21 +33,30 @@ pub struct FilterIter<'c, C> {
     cp: Option<CheckPoint>,
     // blocks map
     blocks: BTreeMap<Height, BlockHash>,
+    // heights of matching blocks
+    matched: BTreeSet<Height>,
     // best height counter
     height: Height,
+    // initial height
+    start: Height,
     // stop height
     stop: Height,
 }
 
 impl<'c, C: RpcApi> FilterIter<'c, C> {
+    /// Hard cap on how far to walk back when a reorg is detected.
+    const MAX_REORG_DEPTH: u32 = 100;
+
     /// 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(),
+            matched: BTreeSet::new(),
             height,
+            start: height,
             stop: 0,
         }
     }
@@ -69,57 +78,21 @@ impl<'c, C: RpcApi> FilterIter<'c, C> {
         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 the remote height is not strictly greater than the height of this
-    /// [`FilterIter`].
+    /// Returns `None` if the remote height is less than the height of this [`FilterIter`].
     pub fn get_tip(&mut self) -> Result<Option<BlockId>, Error> {
         let tip_hash = self.client.get_best_block_hash()?;
-        let mut header = self.client.get_block_header_info(&tip_hash)?;
+        let header = self.client.get_block_header_info(&tip_hash)?;
         let tip_height = header.height as u32;
-        if self.height >= tip_height {
-            // nothing to do
+        if self.height > tip_height {
             return Ok(None);
         }
-        self.blocks.insert(tip_height, tip_hash);
 
-        // if we have a checkpoint we use a lookback of ten blocks
-        // to ensure consistency of the local chain
+        // start scanning from point of agreement + 1
         if let Some(cp) = self.cp.as_ref() {
-            // adjust start height to point of agreement + 1
             let base = self.find_base_with(cp.clone())?;
-            self.height = base.height + 1;
-
-            for _ in 0..9 {
-                let hash = match header.previous_block_hash {
-                    Some(hash) => hash,
-                    None => break,
-                };
-                header = self.client.get_block_header_info(&hash)?;
-                let height = header.height as u32;
-                if height < self.height {
-                    break;
-                }
-                self.blocks.insert(height, hash);
-            }
+            self.height = base.height.saturating_add(1);
         }
 
         self.stop = tip_height;
@@ -131,9 +104,6 @@ impl<'c, C: RpcApi> FilterIter<'c, C> {
     }
 }
 
-/// Alias for a compact filter and associated block id.
-type NextFilter = (BlockId, BlockFilter);
-
 /// Event inner type
 #[derive(Debug, Clone)]
 pub struct EventInner {
@@ -171,27 +141,71 @@ impl<C: RpcApi> Iterator for FilterIter<'_, C> {
     type Item = Result<Event, Error>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        (|| -> 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
-            self.next_filter()?.map_or(Ok(None), |(block, filter)| {
-                let height = block.height;
-                let hash = block.hash;
-
-                if self.spks.is_empty() {
-                    Err(Error::NoScripts)
-                } else 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 };
-                    Ok(Some(Event::Block(inner)))
-                } else {
-                    Ok(Some(Event::NoMatch(height)))
+        (|| -> Result<Option<_>, Error> {
+            if self.height > self.stop {
+                return Ok(None);
+            }
+            // Fetch next header.
+            let mut height = self.height;
+            let mut hash = self.client.get_block_hash(height as u64)?;
+
+            let mut reorg_depth = 0;
+
+            loop {
+                if reorg_depth >= Self::MAX_REORG_DEPTH {
+                    return Err(Error::ReorgDepthExceeded);
                 }
-            })
+
+                let header = self.client.get_block_header(&hash)?;
+
+                let prev_height = height.saturating_sub(1);
+                match self.blocks.get(&prev_height).copied() {
+                    // Not enough data.
+                    None => break,
+                    // Ok, the chain is consistent.
+                    Some(prev_hash) if prev_hash == header.prev_blockhash => break,
+                    _ => {
+                        // Reorg detected, keep backtracking.
+                        height = height.saturating_sub(1);
+                        hash = self.client.get_block_hash(height as u64)?;
+                        reorg_depth += 1;
+                    }
+                }
+            }
+
+            let filter_bytes = self.client.get_block_filter(&hash)?.filter;
+            let filter = BlockFilter::new(&filter_bytes);
+
+            // If the filter matches any of our watched SPKs, fetch the full
+            // block and prepare the next event.
+            let next_event = if self.spks.is_empty() {
+                Err(Error::NoScripts)
+            } else if filter
+                .match_any(&hash, self.spks.iter().map(|s| s.as_bytes()))
+                .map_err(Error::Bip158)?
+            {
+                let block = self.client.get_block(&hash)?;
+                let inner = EventInner { height, block };
+                Ok(Some(Event::Block(inner)))
+            } else {
+                Ok(Some(Event::NoMatch(height)))
+            };
+
+            // In case of a reorg, throw out any stale entries.
+            if reorg_depth > 0 {
+                self.blocks.split_off(&height);
+                self.matched.split_off(&height);
+            }
+            // Record the scanned block
+            self.blocks.insert(height, hash);
+            // Record the matching block
+            if let Ok(Some(Event::Block(..))) = next_event {
+                self.matched.insert(height);
+            }
+            // Increment next height
+            self.height = height.saturating_add(1);
+
+            next_event
         })()
         .transpose()
     }
@@ -220,17 +234,27 @@ impl<C: RpcApi> FilterIter<'_, C> {
     /// Returns a chain update from the newly scanned blocks.
     ///
     /// Returns `None` if this [`FilterIter`] was not constructed using a [`CheckPoint`], or
-    /// if no blocks have been fetched for example by using [`get_tip`](Self::get_tip).
-    pub fn chain_update(&mut self) -> Option<CheckPoint> {
-        if self.cp.is_none() || self.blocks.is_empty() {
+    /// if not all events have been emitted (by calling `next`).
+    pub fn chain_update(&self) -> Option<CheckPoint> {
+        if self.cp.is_none() || self.blocks.is_empty() || self.height <= self.stop {
             return None;
         }
 
-        // note: to connect with the local chain we must guarantee that `self.blocks.first()`
-        // is also the point of agreement with `self.cp`.
+        // We return blocks up to and including the initial height, all of the matching blocks,
+        // and blocks in the terminal range.
+        let tail_range = self.stop.saturating_sub(9)..=self.stop;
         Some(
-            CheckPoint::from_block_ids(self.blocks.iter().map(BlockId::from))
-                .expect("blocks must be in order"),
+            CheckPoint::from_block_ids(self.blocks.iter().filter_map(|(&height, &hash)| {
+                if height <= self.start
+                    || self.matched.contains(&height)
+                    || tail_range.contains(&height)
+                {
+                    Some(BlockId { height, hash })
+                } else {
+                    None
+                }
+            }))
+            .expect("blocks must be in order"),
         )
     }
 }

crates/bitcoind_rpc/tests/test_filter_iter.rs

@@ -99,6 +99,7 @@ fn get_tip_and_chain_update() -> anyhow::Result<()> {
         let cp = CheckPoint::from_block_ids(test.chain).unwrap();
         let mut iter = FilterIter::new_with_checkpoint(env.rpc_client(), cp);
         assert_eq!(iter.get_tip().unwrap(), Some(new_tip));
+        for _res in iter.by_ref() {}
         let update_cp = iter.chain_update().unwrap();
         let mut update_blocks: Vec<_> = update_cp.iter().map(|cp| cp.block_id()).collect();
         update_blocks.reverse();
@@ -161,6 +162,45 @@ fn filter_iter_error_no_scripts() -> anyhow::Result<()> {
     Ok(())
 }
 
+// Test that while a reorg is detected we delay incrementing the best height
+#[test]
+fn repeat_reorgs() -> anyhow::Result<()> {
+    const MINE_TO: u32 = 11;
+
+    let env = testenv()?;
+    let rpc = env.rpc_client();
+    while rpc.get_block_count()? < MINE_TO as u64 {
+        let _ = env.mine_blocks(1, None)?;
+    }
+
+    let spk = ScriptBuf::from_hex("0014446906a6560d8ad760db3156706e72e171f3a2aa")?;
+
+    let mut iter = FilterIter::new_with_height(env.rpc_client(), 1);
+    iter.add_spk(spk);
+    assert_eq!(iter.get_tip()?.unwrap().height, MINE_TO);
+
+    // Process events to height (MINE_TO - 1)
+    loop {
+        if iter.next().unwrap()?.height() == MINE_TO - 1 {
+            break;
+        }
+    }
+
+    for _ in 0..3 {
+        // Invalidate 2 blocks and remine to height = MINE_TO
+        let _ = env.reorg(2)?;
+
+        // Call next. If we detect a reorg, we'll see no change in the event height
+        assert_eq!(iter.next().unwrap()?.height(), MINE_TO - 1);
+    }
+
+    // If no reorg, then height should increment normally from here on
+    assert_eq!(iter.next().unwrap()?.height(), MINE_TO);
+    assert!(iter.next().is_none());
+
+    Ok(())
+}
+
 #[test]
 fn filter_iter_error_wrong_network() -> anyhow::Result<()> {
     let env = testenv()?;
@@ -181,3 +221,37 @@ fn filter_iter_error_wrong_network() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn filter_iter_max_reorg_depth() -> anyhow::Result<()> {
+    use bdk_bitcoind_rpc::bip158::Error;
+
+    let env = testenv()?;
+    let client = env.rpc_client();
+
+    const BASE_HEIGHT: u32 = 10;
+    const REORG_LEN: u32 = 101;
+    const STOP_HEIGHT: u32 = BASE_HEIGHT + REORG_LEN;
+
+    while client.get_block_count()? < STOP_HEIGHT as u64 {
+        env.mine_blocks(1, None)?;
+    }
+
+    let mut iter = FilterIter::new_with_height(client, BASE_HEIGHT);
+    let spk = ScriptBuf::from_hex("0014446906a6560d8ad760db3156706e72e171f3a2aa")?;
+    iter.add_spk(spk.clone());
+    assert_eq!(iter.get_tip()?.unwrap().height, STOP_HEIGHT);
+
+    // Consume events up to final height - 1.
+    loop {
+        if iter.next().unwrap()?.height() == STOP_HEIGHT - 1 {
+            break;
+        }
+    }
+
+    let _ = env.reorg(REORG_LEN as usize)?;
+
+    assert!(matches!(iter.next(), Some(Err(Error::ReorgDepthExceeded))));
+
+    Ok(())
+}