Optimization: Fetching Historic Block Ranges Before the Finalized Block Should Omit Reorg Check (#221)

Author: 0xNeshi

src/block_range_scanner.rs

@@ -370,14 +370,17 @@ impl<N: Network> Service<N> {
             _ => (start_block_num, end_block_num),
         };
 
-        info!(start_block = start_block_num, end_block = end_block_num, "Syncing historical data");
+        info!(
+            start_block = start_block_num,
+            end_block = end_block_num,
+            "Normalized the block range"
+        );
 
         tokio::spawn(async move {
             let mut reorg_handler =
                 ReorgHandler::new(provider.clone(), past_blocks_storage_capacity);
 
-            common::stream_block_range(
-                start_block_num,
+            _ = common::stream_historical_range(
                 start_block_num,
                 end_block_num,
                 max_block_range,

src/block_range_scanner/common.rs

@@ -9,6 +9,7 @@ use crate::{
 };
 use alloy::{
     consensus::BlockHeader,
+    eips::BlockNumberOrTag,
     network::{BlockResponse, Network},
     primitives::BlockNumber,
 };
@@ -135,9 +136,12 @@ async fn initialize_live_streaming_state<N: Network>(
 
     let confirmed = incoming_block_num.saturating_sub(block_confirmations);
 
+    // The minimum common ancestor is the block before the stream start
+    let min_common_ancestor = stream_start.saturating_sub(1);
+
     // Catch up on any confirmed blocks between stream_start and the confirmed tip
-    let previous_batch_end = stream_block_range(
-        stream_start,
+    let previous_batch_end = stream_range_with_reorg_handling(
+        min_common_ancestor,
         stream_start,
         confirmed,
         max_block_range,
@@ -289,8 +293,11 @@ async fn stream_next_batch<N: Network>(
         return true;
     }
 
-    state.previous_batch_end = stream_block_range(
-        stream_start,
+    // The minimum common ancestor is the block before the stream start
+    let min_common_ancestor = stream_start.saturating_sub(1);
+
+    state.previous_batch_end = stream_range_with_reorg_handling(
+        min_common_ancestor,
         state.batch_start,
         batch_end_num,
         max_block_range,
@@ -319,9 +326,72 @@ struct LiveStreamingState<N: Network> {
     previous_batch_end: Option<N::BlockResponse>,
 }
 
-/// Assumes that `min_block <= next_start_block <= end`.
-pub(crate) async fn stream_block_range<N: Network>(
-    min_block: BlockNumber,
+#[must_use]
+pub(crate) async fn stream_historical_range<N: Network>(
+    start: BlockNumber,
+    end: BlockNumber,
+    max_block_range: u64,
+    sender: &mpsc::Sender<BlockScannerResult>,
+    provider: &RobustProvider<N>,
+    reorg_handler: &mut ReorgHandler<N>,
+) -> Option<()> {
+    info!("Getting finalized block number");
+    let finalized = match provider.get_block_number_by_id(BlockNumberOrTag::Finalized.into()).await
+    {
+        Ok(block) => block,
+        Err(e) => {
+            error!(error = %e, "Failed to get finalized block");
+            _ = sender.try_stream(e).await;
+            return None;
+        }
+    };
+
+    // 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(end);
+
+        if !sender.try_stream(batch_start..=batch_end).await {
+            return None; // channel closed
+        }
+
+        batch_start = batch_end + 1;
+    }
+
+    // covers case when `end <= finalized`
+    if batch_start > end {
+        return Some(()); // we're done
+    }
+
+    // we have non-finalized block numbers to stream, a reorg can occur
+
+    // Possible minimal common ancestors when a reorg occurs:
+    // * start > finalized -> the common ancestor we care about is the block before `start`, that's
+    //   where the stream should restart -> this is why we used `start - 1`
+    // * start == finalized -> `start` should never be re-streamed on reorgs; stream should restart
+    //   on `start + 1`
+    // * start < finalized -> if we got here, then `end > finalized`; on reorg, we should only
+    //   re-stream non-finalized blocks
+    let min_common_ancestor = (start.saturating_sub(1)).max(finalized);
+
+    stream_range_with_reorg_handling(
+        min_common_ancestor,
+        batch_start,
+        end,
+        max_block_range,
+        sender,
+        provider,
+        reorg_handler,
+    )
+    .await?;
+
+    Some(())
+}
+
+/// 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,
     end: BlockNumber,
     max_block_range: u64,
@@ -364,13 +434,10 @@ pub(crate) async fn stream_block_range<N: Network>(
             if !sender.try_stream(Notification::ReorgDetected).await {
                 return None;
             }
-            if common_ancestor.header().number() < min_block {
-                min_block
-            } else {
-                common_ancestor.header().number() + 1
-            }
+
+            min_common_ancestor.max(common_ancestor.header().number()) + 1
         } else {
-            batch_end_num.saturating_add(1)
+            batch_end_num + 1
         };
 
         if next_start_block > end {

src/block_range_scanner/sync_handler.rs

@@ -127,7 +127,10 @@ impl<N: Network> SyncHandler<N> {
             )
             .await
             {
-                Ok(start_block) => start_block,
+                Ok(Some(start_block)) => start_block,
+                Ok(None) => {
+                    return; // channel closed
+                }
                 Err(e) => {
                     error!(error = %e, "Error during historical catchup, shutting down");
                     _ = sender.try_stream(e).await;
@@ -158,18 +161,21 @@ impl<N: Network> SyncHandler<N> {
         sender: &mpsc::Sender<BlockScannerResult>,
         provider: &RobustProvider<N>,
         reorg_handler: &mut ReorgHandler<N>,
-    ) -> Result<BlockNumber, ScannerError> {
+    ) -> Result<Option<BlockNumber>, ScannerError> {
         while start_block < confirmed_tip {
-            common::stream_block_range(
-                start_block,
+            if common::stream_historical_range(
                 start_block,
                 confirmed_tip,
                 max_block_range,
                 sender,
                 provider,
                 reorg_handler,
             )
-            .await;
+            .await
+            .is_none()
+            {
+                return Ok(None);
+            }
 
             let latest = provider.get_block_number().await?;
 
@@ -179,7 +185,7 @@ impl<N: Network> SyncHandler<N> {
 
         info!("Historical catchup complete, ready to transition to live");
 
-        Ok(start_block)
+        Ok(Some(start_block))
     }
 
     /// Subscribes to live blocks and begins streaming