Abstract block looping to seperate crate (#247)

Co-authored-by: Nenad <[email protected]>

Author: LeoPatOZ

src/block_range_scanner.rs

@@ -79,18 +79,21 @@ use alloy::{
     network::{BlockResponse, Network, primitives::HeaderResponse},
     primitives::BlockNumber,
 };
-use tracing::{debug, error, info, warn};
+use tracing::{error, info, warn};
 
 mod common;
+mod range_iterator;
 mod reorg_handler;
 mod ring_buffer;
 mod sync_handler;
 
+pub(crate) use range_iterator::RangeIterator;
+
 use reorg_handler::ReorgHandler;
 pub use ring_buffer::RingBufferCapacity;
 
 pub const DEFAULT_MAX_BLOCK_RANGE: u64 = 1000;
-// copied form https://github.com/taikoxyz/taiko-mono/blob/f4b3a0e830e42e2fee54829326389709dd422098/packages/taiko-client/pkg/chain_iterator/block_batch_iterator.go#L19
+
 pub const DEFAULT_BLOCK_CONFIRMATIONS: u64 = 0;
 
 pub const DEFAULT_STREAM_BUFFER_CAPACITY: usize = 50000;
@@ -472,8 +475,6 @@ impl<N: Network> Service<N> {
         provider: &RobustProvider<N>,
         reorg_handler: &mut ReorgHandler<N>,
     ) {
-        let mut batch_count = 0;
-
         // for checking whether reorg occurred
         let mut tip = from;
 
@@ -490,29 +491,15 @@ impl<N: Network> Service<N> {
             }
         };
 
-        // we're iterating in reverse
-        let mut batch_from = from;
         let finalized_number = finalized_block.header().number();
 
         // only check reorg if our tip is after the finalized block
         let check_reorg = tip.header().number() > finalized_number;
 
-        while batch_from >= to {
-            let batch_to = batch_from.saturating_sub(max_block_range - 1).max(to);
-
+        let mut iter = RangeIterator::reverse(from, to, max_block_range);
+        for range in &mut iter {
             // stream the range regularly, i.e. from smaller block number to greater
-            if !sender.try_stream(batch_to..=batch_from).await {
-                break;
-            }
-
-            batch_count += 1;
-            if batch_count % 10 == 0 {
-                debug!(batch_count = batch_count, "Processed rewind batches");
-            }
-
-            // check early if end of stream achieved to avoid subtraction overflow when `to
-            // == 0`
-            if batch_to == to {
+            if !sender.try_stream(range).await {
                 break;
             }
 
@@ -539,11 +526,9 @@ impl<N: Network> Service<N> {
                     return;
                 }
             }
-
-            batch_from = batch_to - 1;
         }
 
-        info!(batch_count = batch_count, "Rewind completed");
+        info!(batch_count = iter.batch_count(), "Rewind completed");
     }
 
     /// Handles re-scanning of reorged blocks.
@@ -586,18 +571,10 @@ impl<N: Network> Service<N> {
         // Re-scan only the affected range (from common_ancestor + 1 up to tip)
         let rescan_from = common_ancestor + 1;
 
-        let mut rescan_batch_start = rescan_from;
-        while rescan_batch_start <= tip_number {
-            let rescan_batch_end = (rescan_batch_start + max_block_range - 1).min(tip_number);
-
-            if !sender.try_stream(rescan_batch_start..=rescan_batch_end).await {
+        for batch in RangeIterator::forward(rescan_from, tip_number, max_block_range) {
+            if !sender.try_stream(batch).await {
                 return false;
             }
-
-            if rescan_batch_end == tip_number {
-                break;
-            }
-            rescan_batch_start = rescan_batch_end + 1;
         }
 
         true

src/block_range_scanner/common.rs

@@ -3,7 +3,7 @@ use tokio_stream::StreamExt;
 
 use crate::{
     ScannerError,
-    block_range_scanner::{BlockScannerResult, reorg_handler::ReorgHandler},
+    block_range_scanner::{BlockScannerResult, RangeIterator, reorg_handler::ReorgHandler},
     robust_provider::{RobustProvider, RobustSubscription, subscription},
     types::{Notification, TryStream},
 };
@@ -13,7 +13,7 @@ use alloy::{
     network::{BlockResponse, Network},
     primitives::BlockNumber,
 };
-use tracing::{debug, error, info, warn};
+use tracing::{error, info, warn};
 
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn stream_live_blocks<N: Network>(
@@ -347,18 +347,17 @@ pub(crate) async fn stream_historical_range<N: Network>(
     };
 
     // no reorg check for finalized blocks
-    let mut batch_start = start;
     let finalized_batch_end = finalized.min(end);
-    while batch_start <= finalized_batch_end {
-        let batch_end = batch_start.saturating_add(max_block_range - 1).min(finalized_batch_end);
-
-        if !sender.try_stream(batch_start..=batch_end).await {
+    for range in RangeIterator::forward(start, finalized_batch_end, max_block_range) {
+        if !sender.try_stream(range).await {
             return None; // channel closed
         }
-
-        batch_start = batch_end + 1;
     }
 
+    // If start > finalized_batch_end, the loop above was empty and we should
+    // continue from start. Otherwise, continue from after finalized_batch_end.
+    let batch_start = start.max(finalized_batch_end + 1);
+
     // covers case when `end <= finalized`
     if batch_start > end {
         return Some(()); // we're done
@@ -392,35 +391,31 @@ pub(crate) async fn stream_historical_range<N: Network>(
 /// Assumes that `min_common_ancestor <= next_start_block <= end`, performs no internal checks.
 pub(crate) async fn stream_range_with_reorg_handling<N: Network>(
     min_common_ancestor: BlockNumber,
-    mut next_start_block: BlockNumber,
+    next_start_block: BlockNumber,
     end: BlockNumber,
     max_block_range: u64,
     sender: &mpsc::Sender<BlockScannerResult>,
     provider: &RobustProvider<N>,
     reorg_handler: &mut ReorgHandler<N>,
 ) -> Option<N::BlockResponse> {
-    let mut batch_count = 0;
+    let mut last_batch_end: Option<N::BlockResponse> = None;
+    let mut iter = RangeIterator::forward(next_start_block, end, max_block_range);
 
-    loop {
-        let batch_end_num = next_start_block.saturating_add(max_block_range - 1).min(end);
+    while let Some(batch) = iter.next() {
+        let batch_end_num = *batch.end();
         let batch_end = match provider.get_block_by_number(batch_end_num.into()).await {
             Ok(block) => block,
             Err(e) => {
-                error!(batch_start = next_start_block, batch_end = batch_end_num, error = %e, "Failed to get ending block of the current batch");
+                error!(batch_start = batch.start(), batch_end = batch_end_num, error = %e, "Failed to get ending block of the current batch");
                 _ = sender.try_stream(e).await;
                 return None;
             }
         };
 
-        if !sender.try_stream(next_start_block..=batch_end_num).await {
+        if !sender.try_stream(batch).await {
             return None; // channel closed
         }
 
-        batch_count += 1;
-        if batch_count % 10 == 0 {
-            debug!(batch_count = batch_count, "Processed historical batches");
-        }
-
         let reorged_opt = match reorg_handler.check(&batch_end).await {
             Ok(opt) => opt,
             Err(e) => {
@@ -430,19 +425,17 @@ pub(crate) async fn stream_range_with_reorg_handling<N: Network>(
             }
         };
 
-        next_start_block = if let Some(common_ancestor) = reorged_opt {
+        if let Some(common_ancestor) = reorged_opt {
             let common_ancestor = common_ancestor.header().number();
             if !sender.try_stream(Notification::ReorgDetected { common_ancestor }).await {
                 return None;
             }
-            (common_ancestor + 1).max(min_common_ancestor)
-        } else {
-            batch_end_num + 1
-        };
-
-        if next_start_block > end {
-            info!(batch_count = batch_count, "Historical sync completed");
-            return Some(batch_end);
+            iter.reset_to((common_ancestor + 1).max(min_common_ancestor));
         }
+
+        last_batch_end = Some(batch_end);
     }
+
+    info!(batch_count = iter.batch_count(), "Historical sync completed");
+    last_batch_end
 }