fix: validate predicate start_block and end_block (#489)

### Description

Previously, the scanning threads had some validation for the
`start_block` and `end_block` that was incorrect. This PR introduces
validation that does the following:
- We now allow `start_block` to be omitted by the user and we default to
0
- If there are no blocks in the database, we abort the scan and go to
streaming mode rather than erroring (fixes #477)
- If the user provides an `end_block`, we validate that it is greater
than the `start_block`
- If the `start_block` is greater than chain tip, we abort the scan and
go to streaming mode rather than erroring (fixes #464)

This PR also adds some validation to the `BlockHeights` class.
Previously, it was possible to overload the
`BlockHeights::BlockRange(start_block, end_block)` function to allocate
a lot of memory into an empty array. We now have limits on this class.
However, due to the above validation, it should not be possible to pass
through parameters that reach theses limits (with the current usage of
the class) until a chain height is up to `1_000_000`.

---

### Checklist

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

components/chainhook-cli/src/scan/bitcoin.rs

@@ -1,4 +1,5 @@
 use crate::config::{Config, PredicatesApi};
+use crate::scan::common::get_block_heights_to_scan;
 use crate::service::{
     open_readwrite_predicates_db_conn_or_panic, set_confirmed_expiration_status,
     set_predicate_scanning_status, set_unconfirmed_expiration_status, ScanningData,
@@ -19,7 +20,7 @@ use chainhook_sdk::observer::{gather_proofs, EventObserverConfig};
 use chainhook_sdk::types::{
     BitcoinBlockData, BitcoinChainEvent, BitcoinChainUpdatedWithBlocksData, BlockIdentifier, Chain,
 };
-use chainhook_sdk::utils::{file_append, send_request, BlockHeights, Context};
+use chainhook_sdk::utils::{file_append, send_request, Context};
 use std::collections::HashMap;
 
 pub async fn scan_bitcoin_chainstate_via_rpc_using_predicate(
@@ -39,46 +40,28 @@ pub async fn scan_bitcoin_chainstate_via_rpc_using_predicate(
             return Err(format!("Bitcoin RPC error: {}", message.to_string()));
         }
     };
-    let mut floating_end_block = false;
 
-    let mut block_heights_to_scan = if let Some(ref blocks) = predicate_spec.blocks {
-        // todo: if a user provides a number of blocks where start_block + blocks > chain tip,
-        // the predicate will fail to scan all blocks. we should calculate a valid end_block and
-        // switch to streaming mode at some point
-        BlockHeights::Blocks(blocks.clone()).get_sorted_entries()
-    } else {
-        let start_block = match predicate_spec.start_block {
-            Some(start_block) => match &unfinished_scan_data {
-                Some(scan_data) => scan_data.last_evaluated_block_height,
-                None => start_block,
-            },
-            None => {
-                return Err(
-                    "Bitcoin chainhook specification must include a field start_block in replay mode"
-                        .into(),
-                );
-            }
-        };
-        let (end_block, update_end_block) = match bitcoin_rpc.get_blockchain_info() {
-            Ok(result) => match predicate_spec.end_block {
-                Some(end_block) => {
-                    if end_block > result.blocks {
-                        (result.blocks, true)
-                    } else {
-                        (end_block, false)
-                    }
-                }
-                None => (result.blocks, true),
-            },
-            Err(e) => {
-                return Err(format!(
-                    "unable to retrieve Bitcoin chain tip ({})",
-                    e.to_string()
-                ));
-            }
-        };
-        floating_end_block = update_end_block;
-        BlockHeights::BlockRange(start_block, end_block).get_sorted_entries()
+    let mut chain_tip = match bitcoin_rpc.get_blockchain_info() {
+        Ok(result) => result.blocks,
+        Err(e) => {
+            return Err(format!(
+                "unable to retrieve Bitcoin chain tip ({})",
+                e.to_string()
+            ));
+        }
+    };
+
+    let block_heights_to_scan = get_block_heights_to_scan(
+        &predicate_spec.blocks,
+        &predicate_spec.start_block,
+        &predicate_spec.end_block,
+        &chain_tip,
+        &unfinished_scan_data,
+    )?;
+    let mut block_heights_to_scan = match block_heights_to_scan {
+        Some(h) => h,
+        // no blocks to scan, go straight to streaming
+        None => return Ok(false),
     };
 
     let mut predicates_db_conn = match config.http_api {
@@ -115,6 +98,30 @@ pub async fn scan_bitcoin_chainstate_via_rpc_using_predicate(
     let http_client = build_http_client();
 
     while let Some(current_block_height) = block_heights_to_scan.pop_front() {
+        if current_block_height > chain_tip {
+            let prev_chain_tip = chain_tip;
+            // we've scanned up to the chain tip as of the start of this scan
+            // so see if the chain has progressed since then
+            chain_tip = match bitcoin_rpc.get_blockchain_info() {
+                Ok(result) => result.blocks,
+                Err(e) => {
+                    return Err(format!(
+                        "unable to retrieve Bitcoin chain tip ({})",
+                        e.to_string()
+                    ));
+                }
+            };
+            // if the chain hasn't progressed, break out so we can enter streaming mode
+            // and put back the block we weren't able to scan
+            if current_block_height > chain_tip {
+                block_heights_to_scan.push_front(current_block_height);
+                break;
+            } else {
+                // if the chain has progressed, update our total number of blocks to scan and keep scanning
+                number_of_blocks_to_scan += chain_tip - prev_chain_tip;
+            }
+        }
+
         number_of_blocks_scanned += 1;
 
         let block_hash = retrieve_block_hash_with_retry(
@@ -189,30 +196,8 @@ pub async fn scan_bitcoin_chainstate_via_rpc_using_predicate(
                 );
             }
         }
-
-        if block_heights_to_scan.is_empty() && floating_end_block {
-            let new_tip = match bitcoin_rpc.get_blockchain_info() {
-                Ok(result) => match predicate_spec.end_block {
-                    Some(end_block) => {
-                        if end_block > result.blocks {
-                            result.blocks
-                        } else {
-                            end_block
-                        }
-                    }
-                    None => result.blocks,
-                },
-                Err(_e) => {
-                    continue;
-                }
-            };
-
-            for entry in (current_block_height + 1)..new_tip {
-                block_heights_to_scan.push_back(entry);
-            }
-            number_of_blocks_to_scan += block_heights_to_scan.len() as u64;
-        }
     }
+
     info!(
         ctx.expect_logger(),
         "{number_of_blocks_scanned} blocks scanned, {actions_triggered} actions triggered"
@@ -228,24 +213,28 @@ pub async fn scan_bitcoin_chainstate_via_rpc_using_predicate(
             predicates_db_conn,
             ctx,
         );
-        if let Some(predicate_end_block) = predicate_spec.end_block {
-            if predicate_end_block == last_block_scanned.index {
-                // todo: we need to find a way to check if this block is confirmed
-                // and if so, set the status to confirmed expiration
-                set_unconfirmed_expiration_status(
-                    &Chain::Bitcoin,
-                    number_of_blocks_scanned,
-                    predicate_end_block,
-                    &predicate_spec.key(),
-                    predicates_db_conn,
-                    ctx,
-                );
-                if last_scanned_block_confirmations >= CONFIRMED_SEGMENT_MINIMUM_LENGTH {
-                    set_confirmed_expiration_status(&predicate_spec.key(), predicates_db_conn, ctx);
-                }
-                return Ok(true);
+    }
+    // if an end block was provided, or a fixed number of blocks were set to be scanned,
+    // check to see if we've processed all of the blocks and can expire the predicate.
+    if (predicate_spec.blocks.is_some()
+        || (predicate_spec.end_block.is_some()
+            && predicate_spec.end_block.unwrap() == last_block_scanned.index))
+        && block_heights_to_scan.is_empty()
+    {
+        if let Some(ref mut predicates_db_conn) = predicates_db_conn {
+            set_unconfirmed_expiration_status(
+                &Chain::Bitcoin,
+                number_of_blocks_scanned,
+                last_block_scanned.index,
+                &predicate_spec.key(),
+                predicates_db_conn,
+                ctx,
+            );
+            if last_scanned_block_confirmations >= CONFIRMED_SEGMENT_MINIMUM_LENGTH {
+                set_confirmed_expiration_status(&predicate_spec.key(), predicates_db_conn, ctx);
             }
         }
+        return Ok(true);
     }
 
     return Ok(false);

components/chainhook-cli/src/scan/common.rs

@@ -0,0 +1,67 @@
+use crate::service::ScanningData;
+use chainhook_sdk::utils::{BlockHeights, BlockHeightsError};
+use std::collections::VecDeque;
+
+pub fn get_block_heights_to_scan(
+    blocks: &Option<Vec<u64>>,
+    start_block: &Option<u64>,
+    end_block: &Option<u64>,
+    chain_tip: &u64,
+    unfinished_scan_data: &Option<ScanningData>,
+) -> Result<Option<VecDeque<u64>>, String> {
+    let block_heights_to_scan = if let Some(ref blocks) = blocks {
+        match BlockHeights::Blocks(blocks.clone()).get_sorted_entries() {
+            Ok(heights) => Some(heights),
+            Err(e) => match e {
+                BlockHeightsError::ExceedsMaxEntries(max, specified) => {
+                    return Err(format!("Chainhook specification exceeds max number of blocks to scan. Maximum: {}, Attempted: {}", max, specified));
+                }
+                BlockHeightsError::StartLargerThanEnd => {
+                    // this code path should not be reachable
+                    return Err(
+                        "Chainhook specification field `end_block` should be greater than `start_block`."
+                            .into(),
+                    );
+                }
+            },
+        }
+    } else {
+        let start_block = match &unfinished_scan_data {
+            Some(scan_data) => scan_data.last_evaluated_block_height,
+            None => start_block.unwrap_or(0),
+        };
+
+        let end_block = if let Some(end_block) = end_block {
+            if &start_block > end_block {
+                return Err(
+                    "Chainhook specification field `end_block` should be greater than `start_block`."
+                        .into(),
+                );
+            }
+            end_block
+        } else {
+            chain_tip
+        };
+        if &start_block > end_block {
+            return Ok(None);
+        }
+        let block_heights_to_scan = match BlockHeights::BlockRange(start_block, *end_block)
+            .get_sorted_entries()
+        {
+            Ok(heights) => heights,
+            Err(e) => match e {
+                BlockHeightsError::ExceedsMaxEntries(max, specified) => {
+                    return Err(format!("Chainhook specification exceeds max number of blocks to scan. Maximum: {}, Attempted: {}", max, specified));
+                }
+                BlockHeightsError::StartLargerThanEnd => {
+                    return Err(
+                        "Chainhook specification field `end_block` should be greater than `start_block`."
+                            .into(),
+                    );
+                }
+            },
+        };
+        Some(block_heights_to_scan)
+    };
+    Ok(block_heights_to_scan)
+}

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

@@ -1,2 +1,6 @@
 pub mod bitcoin;
+pub mod common;
 pub mod stacks;
+
+#[cfg(test)]
+pub mod tests;