fix: order and filter blocks used to seed forking block pool (#534)

### Description

<!-- Describe the bug this PR fixes or the feature it adds. Link to any
related issues and PRs -->

#### Breaking change?

<!-- If applicable, list the APIs/functionality which this PR breaks -->

### Example

<!-- If applicable, add an example on how this improves the application
-->

---

### Checklist

- [ ] All tests pass
- [ ] Tests added in this PR (if applicable)

components/chainhook-cli/src/service/mod.rs

@@ -7,7 +7,8 @@ use crate::service::http_api::{load_predicates_from_redis, start_predicate_api_s
 use crate::service::runloops::{start_bitcoin_scan_runloop, start_stacks_scan_runloop};
 use crate::storage::{
     confirm_entries_in_stacks_blocks, draft_entries_in_stacks_blocks, get_all_unconfirmed_blocks,
-    open_readonly_stacks_db_conn_with_retry, open_readwrite_stacks_db_conn,
+    get_last_block_height_inserted, open_readonly_stacks_db_conn_with_retry,
+    open_readwrite_stacks_db_conn,
 };
 
 use chainhook_sdk::chainhooks::types::{ChainhookConfig, ChainhookFullSpecification};
@@ -218,7 +219,17 @@ impl Service {
         let stacks_db =
             open_readonly_stacks_db_conn_with_retry(&config.expected_cache_path(), 3, &ctx)?;
         let unconfirmed_blocks = match get_all_unconfirmed_blocks(&stacks_db, &ctx) {
-            Ok(blocks) => Some(blocks),
+            Ok(blocks) => {
+                let confirmed_tip = get_last_block_height_inserted(&stacks_db, &ctx).unwrap_or(0);
+                // any unconfirmed blocks that are earlier than confirmed blocks are invalid
+                Some(
+                    blocks
+                        .iter()
+                        .filter(|&b| b.block_identifier.index > confirmed_tip)
+                        .cloned()
+                        .collect(),
+                )
+            }
             Err(e) => {
                 info!(
                     self.ctx.expect_logger(),

components/chainhook-cli/src/storage/mod.rs

@@ -1,8 +1,9 @@
+use std::collections::VecDeque;
 use std::path::PathBuf;
 
 use chainhook_sdk::types::{BlockIdentifier, StacksBlockData, StacksBlockUpdate};
 use chainhook_sdk::utils::Context;
-use rocksdb::{Direction, IteratorMode, Options, DB};
+use rocksdb::{Options, DB};
 
 const UNCONFIRMED_KEY_PREFIX: &[u8; 2] = b"~:";
 const CONFIRMED_KEY_PREFIX: &[u8; 2] = b"b:";
@@ -172,28 +173,22 @@ pub fn get_last_unconfirmed_block_height_inserted(stacks_db: &DB, _ctx: &Context
 
 pub fn get_all_unconfirmed_blocks(
     stacks_db: &DB,
-    _ctx: &Context,
-) -> Result<Vec<StacksBlockData>, String> {
-    let unconfirmed_key_prefix = UNCONFIRMED_KEY_PREFIX;
-    let mut blocks = vec![];
-    let iter = stacks_db.iterator(IteratorMode::From(
-        unconfirmed_key_prefix,
-        Direction::Forward,
-    ));
-    for item in iter {
-        match item {
-            Ok((k, v)) => {
-                if k.starts_with(unconfirmed_key_prefix) {
-                    let spec: StacksBlockData = serde_json::from_slice(&v[..]).map_err(|e| {
-                        format!("unable to deserialize Stacks block {}", e.to_string())
-                    })?;
-                    blocks.push(spec);
-                } else {
-                    // we're past the set of keys we're looking for, so we've found all unconfirmed
-                    return Ok(blocks);
+    ctx: &Context,
+) -> Result<VecDeque<StacksBlockData>, String> {
+    let mut blocks = VecDeque::new();
+    let Some(mut cursor) = get_last_unconfirmed_block_height_inserted(stacks_db, ctx) else {
+        return Ok(blocks);
+    };
+    loop {
+        match get_stacks_block_at_block_height(cursor, false, 3, stacks_db) {
+            Ok(block) => match block {
+                Some(block) => {
+                    blocks.push_front(block.clone());
+                    cursor = block.parent_block_identifier.index;
                 }
-            }
-            Err(e) => return Err(format!("failed to get all unconfirmed blocks: {e}")),
+                None => break,
+            },
+            Err(e) => return Err(e),
         };
     }
     Ok(blocks)

components/chainhook-sdk/src/indexer/stacks/blocks_pool.rs

@@ -47,7 +47,7 @@ impl StacksBlockPool {
         ctx.try_log(|logger| {
             slog::info!(logger, "Seeding block pool with {} blocks", blocks.len())
         });
-        for block in blocks {
+        for block in blocks.into_iter() {
             let existing_entry = self.block_store.get(&block.block_identifier.clone());
             if existing_entry.is_some() {
                 ctx.try_log(|logger| {