feat(bitcoind_rpc): Various fixes for FilterIter

* Remove `MAX_REORG_DEPTH` check and error variant. For the scenario
  where the `FilterIter` is of a different network than the receiver, the
  genesis block will be emitted, thus making the receiver error out.
* Fix `Iterator::next` impl so that we only mutate internal state after
  we are certain that no more errors will be returned.
* `Error::NoScripts` is no longer returned since I could not justify a
  rationale for it.
* Refactor `Iterator::next` impl be cleaner.

Author: evanlinjin

crates/bitcoind_rpc/src/bip158.rs

@@ -15,8 +15,8 @@ use bitcoin::{
     bip158::{self, BlockFilter},
     Block, BlockHash, ScriptBuf,
 };
-use bitcoincore_rpc;
 use bitcoincore_rpc::RpcApi;
+use bitcoincore_rpc::{self, jsonrpc};
 
 /// Block height
 type Height = u32;
@@ -44,9 +44,6 @@ pub struct FilterIter<'c, C> {
 }
 
 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 {
@@ -148,82 +145,7 @@ impl<C: RpcApi> Iterator for FilterIter<'_, C> {
     type Item = Result<Event, Error>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        (|| -> 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 _)?;
-            let mut header = self.client.get_block_header(&hash)?;
-
-            // Detect and resolve reorgs: either block at height changed, or its parent changed.
-            let stored_hash = self.blocks.get(&height).copied();
-            let prev_hash = height
-                .checked_sub(1)
-                .and_then(|height| self.blocks.get(&height).copied());
-
-            // If we've seen this height before but the hash has changed, or parent changed, trigger
-            // reorg.
-            let reorg_detected = if let Some(old_hash) = stored_hash {
-                old_hash != hash
-            } else if let Some(expected_prev) = prev_hash {
-                header.prev_blockhash != expected_prev
-            } else {
-                false
-            };
-
-            // Reorg detected, rewind to last known-good ancestor.
-            if reorg_detected {
-                let mut reorg_depth = 0;
-                loop {
-                    if reorg_depth >= Self::MAX_REORG_DEPTH || height == 0 {
-                        return Err(Error::ReorgDepthExceeded);
-                    }
-
-                    height = height.saturating_sub(1);
-                    hash = self.client.get_block_hash(height as _)?;
-                    header = self.client.get_block_header(&hash)?;
-
-                    let prev_height = height.saturating_sub(1);
-                    if let Some(prev_hash) = self.blocks.get(&prev_height) {
-                        if header.prev_blockhash == *prev_hash {
-                            break;
-                        }
-                    }
-
-                    reorg_depth += 1;
-                }
-
-                self.blocks.split_off(&height);
-                self.matched.split_off(&height);
-            }
-
-            let filter_bytes = self.client.get_block_filter(&hash)?.filter;
-            let filter = BlockFilter::new(&filter_bytes);
-
-            // record the scanned block
-            self.blocks.insert(height, hash);
-            // increment best height
-            self.height = height.saturating_add(1);
-
-            // If the filter matches any of our watched SPKs, fetch the full
-            // block, and record the matching block entry.
-            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)?;
-                self.matched.insert(height);
-                let inner = EventInner { height, block };
-                Ok(Some(Event::Block(inner)))
-            } else {
-                Ok(Some(Event::NoMatch(height)))
-            }
-        })()
-        .transpose()
+        self.next_event().transpose()
     }
 }
 
@@ -274,6 +196,99 @@ impl<C: RpcApi> FilterIter<'_, C> {
             .expect("blocks must be in order"),
         )
     }
+
+    fn next_event(&mut self) -> Result<Option<Event>, Error> {
+        let (height, hash) = match self.find_next_block()? {
+            None => return Ok(None),
+            Some((height, _)) if height > self.stop => return Ok(None),
+            Some(block) => block,
+        };
+
+        // Emit and increment `height` (which should really be `next_height`).
+        let is_match = BlockFilter::new(&self.client.get_block_filter(&hash)?.filter)
+            .match_any(&hash, self.spks.iter().map(ScriptBuf::as_ref))
+            .map_err(Error::Bip158)?;
+
+        let event = if is_match {
+            Event::Block(EventInner {
+                height,
+                block: self.client.get_block(&hash)?,
+            })
+        } else {
+            Event::NoMatch(height)
+        };
+
+        // Mutate internal state at the end, once we are sure there are no more errors.
+        if is_match {
+            self.matched.insert(height);
+        }
+        self.matched.split_off(&height);
+        self.blocks.split_off(&height);
+        self.blocks.insert(height, hash);
+        self.height = height.saturating_add(1);
+        self.cp = self
+            .cp
+            .as_ref()
+            .and_then(|cp| cp.range(..=cp.height()).next());
+
+        Ok(Some(event))
+    }
+
+    /// Non-mutating method that finds the next block which connects with our previously-emitted
+    /// history.
+    fn find_next_block(&self) -> Result<Option<(Height, BlockHash)>, bitcoincore_rpc::Error> {
+        let mut height = self.height;
+
+        // Search blocks backwards until we find a block which connects with something the consumer
+        // has already seen.
+        let hash = loop {
+            let hash = match self.client.get_block_hash(height as _) {
+                Ok(hash) => hash,
+                Err(bitcoincore_rpc::Error::JsonRpc(jsonrpc::Error::Rpc(rpc_err)))
+                    // -8: Out of bounds, -5: Not found
+                    if rpc_err.code == -8 || rpc_err.code == -5 =>
+                {
+                    return Ok(None)
+                }
+                Err(err) => return Err(err),
+            };
+            let header = self.client.get_block_header(&hash)?;
+
+            let prev_height = match height.checked_sub(1) {
+                Some(prev_height) => prev_height,
+                // Always emit the genesis block as it cannot change.
+                None => break hash,
+            };
+
+            let prev_hash_remote = header.prev_blockhash;
+            if let Some(&prev_hash) = self.blocks.get(&prev_height) {
+                if prev_hash == prev_hash_remote {
+                    break hash;
+                }
+                height = prev_height;
+                continue;
+            }
+
+            let maybe_prev_cp = self
+                .cp
+                .as_ref()
+                .and_then(|cp| cp.range(..=prev_height).next());
+            if let Some(prev_cp) = maybe_prev_cp {
+                if prev_cp.height() != prev_height {
+                    // Try again at a height that the consumer can compare against.
+                    height = prev_cp.height();
+                    continue;
+                }
+                if prev_cp.hash() != prev_hash_remote {
+                    height = prev_height;
+                    continue;
+                }
+            }
+            break hash;
+        };
+
+        Ok(Some((height, hash)))
+    }
 }
 
 /// Errors that may occur during a compact filters sync.
@@ -285,8 +300,6 @@ pub enum Error {
     NoScripts,
     /// `bitcoincore_rpc` error
     Rpc(bitcoincore_rpc::Error),
-    /// `MAX_REORG_DEPTH` exceeded
-    ReorgDepthExceeded,
 }
 
 impl From<bitcoincore_rpc::Error> for Error {
@@ -301,7 +314,6 @@ impl fmt::Display for Error {
             Self::Bip158(e) => e.fmt(f),
             Self::NoScripts => write!(f, "no script pubkeys were provided to match with"),
             Self::Rpc(e) => e.fmt(f),
-            Self::ReorgDepthExceeded => write!(f, "maximum reorg depth exceeded"),
         }
     }
 }

crates/bitcoind_rpc/tests/test_filter_iter.rs

@@ -146,24 +146,6 @@ fn filter_iter_returns_matched_blocks() -> anyhow::Result<()> {
     Ok(())
 }
 
-#[test]
-fn filter_iter_error_no_scripts() -> anyhow::Result<()> {
-    use bdk_bitcoind_rpc::bip158::Error;
-    let env = testenv()?;
-    let _ = env.mine_blocks(2, None)?;
-
-    let mut iter = FilterIter::new_with_height(env.rpc_client(), 1);
-    assert_eq!(iter.get_tip()?.unwrap().height, 3);
-
-    // iterator should return three errors
-    for _ in 0..3 {
-        assert!(matches!(iter.next().unwrap(), Err(Error::NoScripts)));
-    }
-    assert!(iter.next().is_none());
-
-    Ok(())
-}
-
 #[test]
 #[allow(clippy::print_stdout)]
 fn filter_iter_handles_reorg() -> anyhow::Result<()> {
@@ -593,70 +575,3 @@ fn repeat_reorgs() -> anyhow::Result<()> {
 
     Ok(())
 }
-
-#[test]
-#[allow(clippy::print_stdout)]
-fn filter_iter_max_reorg_depth() -> anyhow::Result<()> {
-    use bdk_bitcoind_rpc::bip158::{Error, FilterIter};
-    use bdk_chain::{
-        bitcoin::{Address, Amount, Network, ScriptBuf},
-        BlockId, CheckPoint,
-    };
-
-    let env = testenv()?;
-    let client = env.rpc_client();
-
-    const BASE_HEIGHT: u32 = 110;
-    const REORG_LENGTH: u32 = 101;
-    const START_HEIGHT: u32 = BASE_HEIGHT + 1;
-    const FINAL_HEIGHT: u32 = START_HEIGHT + REORG_LENGTH - 1;
-
-    while client.get_block_count()? < BASE_HEIGHT as u64 {
-        env.mine_blocks(1, None)?;
-    }
-
-    let spk = ScriptBuf::from_hex("0014446906a6560d8ad760db3156706e72e171f3a2aa")?;
-    let addr = Address::from_script(&spk, Network::Regtest)?;
-
-    // Mine blocks 111-211, each with a tx.
-    for _ in 0..REORG_LENGTH {
-        env.send(&addr, Amount::from_sat(1000))?;
-        env.mine_blocks(1, None)?;
-    }
-
-    // Create `CheckPoint` and build `FilterIter`.
-    let cp = CheckPoint::from_block_ids([BlockId {
-        height: BASE_HEIGHT,
-        hash: client.get_block_hash(BASE_HEIGHT as u64)?,
-    }])
-    .unwrap();
-    let mut iter = FilterIter::new_with_checkpoint(client, cp);
-    iter.add_spk(spk.clone());
-    iter.get_tip()?;
-
-    // Scan blocks 111–211 so their hashes reside in `self.blocks`.
-    for res in iter.by_ref() {
-        res?;
-    }
-
-    // Invalidate blocks 111-211 and mine replacement blocks.
-    for h in (START_HEIGHT..=FINAL_HEIGHT).rev() {
-        let hash = client.get_block_hash(h as u64)?;
-        client.invalidate_block(&hash)?;
-    }
-
-    // Mine one extra block so the new tip (212) isn’t below `iter.height`.
-    env.mine_blocks((REORG_LENGTH + 1) as usize, None)?;
-
-    iter.get_tip()?;
-    match iter.next() {
-        Some(Err(Error::ReorgDepthExceeded)) => {
-            println!("SUCCESS: Detected ReorgDepthExceeded");
-        }
-        Some(Err(e)) => panic!("Expected ReorgDepthExceeded, got {:?}", e),
-        Some(Ok(ev)) => panic!("Expected error, got event {:?}", ev),
-        None => panic!("Expected error, got None"),
-    }
-
-    Ok(())
-}