Reorg Rewind optimisation

src/block_range_scanner.rs

@@ -75,7 +75,7 @@ use crate::{
 
 use alloy::{
     consensus::BlockHeader,
-    eips::BlockId,
+    eips::{BlockId, BlockNumberOrTag},
     network::{BlockResponse, Network, primitives::HeaderResponse},
     primitives::BlockNumber,
 };
@@ -445,6 +445,11 @@ impl<N: Network> Service<N> {
     ///
     /// The `from` block is assumed to be greater than or equal to the `to` block.
     ///
+    /// # Reorg Handling
+    ///
+    /// Reorg checks are only performed when the tip is above the current finalized
+    /// block height.
+    ///
     /// # Errors
     ///
     /// Returns an error if the stream fails
@@ -464,8 +469,22 @@ impl<N: Network> Service<N> {
         let from = tip.header().number();
         let to = to.header().number();
 
+        let finalized_block = match provider.get_block_by_number(BlockNumberOrTag::Finalized).await
+        {
+            Ok(block) => block,
+            Err(e) => {
+                error!(error = %e, "Failed to get finalized block");
+                _ = sender.try_stream(e).await;
+                return;
+            }
+        };
+
         // 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);
@@ -486,51 +505,92 @@ impl<N: Network> Service<N> {
                 break;
             }
 
-            let reorged_opt = match reorg_handler.check(&tip).await {
-                Ok(opt) => {
-                    info!(block_number = %from, hash = %tip.header().hash(), "Reorg detected");
-                    opt
-                }
-                Err(e) => {
-                    error!(error = %e, "Terminal RPC call error, shutting down");
-                    _ = sender.try_stream(e).await;
-                    return;
-                }
-            };
-
-            // For now we only care if a reorg occurred, not which block it was.
-            // Once we optimize 'latest' mode to update only the reorged logs, we will need the
-            // exact common ancestor.
-            if reorged_opt.is_some() {
-                info!(block_number = %from, hash = %tip.header().hash(), "Reorg detected");
-
-                if !sender.try_stream(Notification::ReorgDetected).await {
-                    break;
-                }
-
-                // restart rewind
-                batch_from = from;
-                // store the updated end block hash
-                tip = match provider.get_block_by_number(from.into()).await {
-                    Ok(block) => block,
-                    Err(RobustProviderError::BlockNotFound(_)) => {
-                        panic!("Block with number '{from}' should exist post-reorg");
-                    }
+            if check_reorg {
+                let reorg = match reorg_handler.check(&tip).await {
+                    Ok(opt) => opt,
                     Err(e) => {
                         error!(error = %e, "Terminal RPC call error, shutting down");
                         _ = sender.try_stream(e).await;
                         return;
                     }
                 };
-            } else {
-                // `batch_to` is always greater than `to`, so `batch_to - 1` is always a valid
-                // unsigned integer
-                batch_from = batch_to - 1;
+
+                if let Some(common_ancestor) = reorg &&
+                    !Self::handle_reorg_rescan(
+                        &mut tip,
+                        common_ancestor,
+                        max_block_range,
+                        sender,
+                        provider,
+                    )
+                    .await
+                {
+                    return;
+                }
             }
+
+            batch_from = batch_to - 1;
         }
 
         info!(batch_count = batch_count, "Rewind completed");
     }
+
+    /// Handles re-scanning of reorged blocks.
+    ///
+    /// Returns `true` on success, `false` if stream closed or terminal error occurred.
+    async fn handle_reorg_rescan(
+        tip: &mut N::BlockResponse,
+        common_ancestor: N::BlockResponse,
+        max_block_range: u64,
+        sender: &mpsc::Sender<BlockScannerResult>,
+        provider: &RobustProvider<N>,
+    ) -> bool {
+        let tip_number = tip.header().number();
+        let common_ancestor_block = common_ancestor.header().number();
+        info!(
+            block_number = %tip_number,
+            hash = %tip.header().hash(),
+            common_ancestor_block = %common_ancestor_block,
+            "Reorg detected"
+        );
+
+        if !sender.try_stream(Notification::ReorgDetected { common_ancestor_block }).await {
+            return false;
+        }
+
+        // Get the new tip block (same height as original tip, but new hash)
+        *tip = match provider.get_block_by_number(tip_number.into()).await {
+            Ok(block) => block,
+            Err(RobustProviderError::BlockNotFound(_)) => {
+                panic!("Block with number '{tip_number}' should exist post-reorg");
+            }
+            Err(e) => {
+                error!(error = %e, "Terminal RPC call error, shutting down");
+                _ = sender.try_stream(e).await;
+                return false;
+            }
+        };
+
+        // Re-scan only the affected range (from tip down to common_ancestor + 1)
+        let rescan_to = common_ancestor_block + 1;
+
+        let mut rescan_batch_from = tip_number;
+        while rescan_batch_from >= rescan_to {
+            let rescan_batch_to =
+                rescan_batch_from.saturating_sub(max_block_range - 1).max(rescan_to);
+
+            if !sender.try_stream(rescan_batch_to..=rescan_batch_from).await {
+                return false;
+            }
+
+            if rescan_batch_to == rescan_to {
+                break;
+            }
+            rescan_batch_from = rescan_batch_to - 1;
+        }
+
+        true
+    }
 }
 
 pub struct BlockRangeScannerClient {

src/block_range_scanner/common.rs

@@ -251,12 +251,13 @@ async fn handle_reorg_detected<N: Network>(
     state: &mut LiveStreamingState<N>,
     sender: &mpsc::Sender<BlockScannerResult>,
 ) -> bool {
-    if !sender.try_stream(Notification::ReorgDetected).await {
+    let ancestor_num = common_ancestor.header().number();
+
+    if !sender.try_stream(Notification::ReorgDetected { common_ancestor_block: ancestor_num }).await
+    {
         return false;
     }
 
-    let ancestor_num = common_ancestor.header().number();
-
     // Reset streaming position based on common ancestor
     if ancestor_num < stream_start {
         // Reorg went before our starting point - restart from stream_start
@@ -431,11 +432,11 @@ pub(crate) async fn stream_range_with_reorg_handling<N: Network>(
         };
 
         next_start_block = if let Some(common_ancestor) = reorged_opt {
-            if !sender.try_stream(Notification::ReorgDetected).await {
+            let common_ancestor_block = common_ancestor.header().number();
+            if !sender.try_stream(Notification::ReorgDetected { common_ancestor_block }).await {
                 return None;
             }
-
-            min_common_ancestor.max(common_ancestor.header().number()) + 1
+            (common_ancestor_block + 1).max(min_common_ancestor)
         } else {
             batch_end_num + 1
         };

src/event_scanner/scanner/common.rs

@@ -185,6 +185,35 @@ pub fn spawn_log_consumers_in_collection_mode<N: Network>(
                                 }
                             }
                         }
+                        Ok(ScannerMessage::Notification(Notification::ReorgDetected {
+                            common_ancestor_block,
+                        })) => {
+                            info!(
+                                common_ancestor_block = common_ancestor_block,
+                                "Received ReorgDetected notification"
+                            );
+
+                            // Invalidate logs from reorged blocks
+                            // Logs are ordered newest -> oldest, so skip logs with
+                            // block_number > common_ancestor at the front
+                            let before_count = collected.len();
+                            collected = collected
+                                .into_iter()
+                                .skip_while(|log| {
+                                    log.block_number.is_some_and(|n| n > common_ancestor_block)
+                                })
+                                .collect();
+                            let removed_count = before_count - collected.len();
+                            if removed_count > 0 {
+                                info!(
+                                    removed_count = removed_count,
+                                    remaining_count = collected.len(),
+                                    "Invalidated logs from reorged blocks"
+                                );
+                            }
+                            // Don't forward the notification to the user in CollectLatest mode
+                            // since logs haven't been sent yet
+                        }
                         Ok(ScannerMessage::Notification(notification)) => {
                             info!(notification = ?notification, "Received notification");
                             if !sender.try_stream(notification).await {

src/types.rs

@@ -24,8 +24,6 @@ pub enum Notification {
     /// in sync scanners.
     SwitchingToLive,
 
-    /// Emitted when a blockchain reorganization is detected during scanning.
-    ///
     /// When a reorg occurs, the scanner adjusts its position to re-stream events from the
     /// canonical chain state. The specific behavior depends on the scanning mode (see individual
     /// scanner mode documentation for details).
@@ -49,7 +47,10 @@ pub enum Notification {
     /// handle duplicate logs idempotently (e.g., using transaction hashes or log indices as
     /// deduplication keys). The scanner prioritizes correctness by ensuring all logs from the
     /// canonical chain are delivered, even if it means occasional duplicates during reorgs.
-    ReorgDetected,
+    ///
+    /// The `common_ancestor_block` field contains the block number of the last block
+    /// that is still valid on the canonical chain.
+    ReorgDetected { common_ancestor_block: u64 },
 
     /// Emitted during the latest events phase when no matching logs are found in the
     /// scanned range.