Merge branch 'main' into start-typestate

Author: LeoPatOZ

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(finalized_batch_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

src/types.rs

@@ -25,6 +25,30 @@ pub enum Notification {
     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).
+    ///
+    /// # Redundant Notifications
+    ///
+    /// Due to the asynchronous nature of block scanning and log fetching, you may occasionally
+    /// receive this notification even after the reorg has already been accounted for. This happens
+    /// when:
+    ///
+    /// 1. `BlockRangeScanner` validates and emits a block range
+    /// 2. A reorg occurs on the chain
+    /// 3. `EventScanner` fetches logs for that range, but the RPC provider returns logs from the
+    ///    post-reorg chain state (the provider's view has already updated)
+    /// 4. `BlockRangeScanner` detects the reorg on its next check and emits
+    ///    `Notification::ReorgDetected` with a new range starting from the first reorged block
+    /// 5. `EventScanner` re-fetches logs for this range, which may return duplicate logs already
+    ///    delivered in step 3 (the new range might also extend beyond the original range)
+    ///
+    /// **How to handle**: This is a benign race condition. Your application should be designed to
+    /// 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,
 
     /// Emitted during the latest events phase when no matching logs are found in the

tests/block_range_scanner.rs

@@ -188,16 +188,16 @@ async fn historical_emits_correction_range_when_reorg_below_end() -> anyhow::Res
         .stream_historical(BlockNumberOrTag::Number(0), BlockNumberOrTag::Number(end_num))
         .await?;
 
-    let depth = 15;
-    _ = provider.anvil_reorg(ReorgOptions { depth, tx_block_pairs: vec![] }).await;
-    _ = provider.anvil_mine(Some(20), None).await;
-
     assert_next!(stream, 0..=29);
-    assert_next!(stream, 30..=59);
-    assert_next!(stream, 60..=89);
-    assert_next!(stream, 90..=110);
+    assert_next!(stream, 30..=56);
+    assert_next!(stream, 57..=86);
+    assert_next!(stream, 87..=110);
+    let mut stream = assert_empty!(stream);
+
+    _ = provider.anvil_reorg(ReorgOptions { depth: 15, tx_block_pairs: vec![] }).await;
+
     assert_next!(stream, Notification::ReorgDetected);
-    assert_next!(stream, 105..=110);
+    assert_next!(stream, 106..=110);
     assert_closed!(stream);
 
     Ok(())
@@ -211,32 +211,55 @@ async fn historical_emits_correction_range_when_end_num_reorgs() -> anyhow::Resu
 
     provider.anvil_mine(Some(120), None).await?;
 
-    let end_num = 120;
-
     let client =
         BlockRangeScanner::new().max_block_range(30).connect(provider.clone()).await?.run()?;
 
-    let mut stream = client
-        .stream_historical(BlockNumberOrTag::Number(0), BlockNumberOrTag::Number(end_num))
-        .await?;
-
-    let pre_reorg_mine = 20;
-    _ = provider.anvil_mine(Some(pre_reorg_mine), None).await;
-    let depth = pre_reorg_mine + 1;
-    _ = provider.anvil_reorg(ReorgOptions { depth, tx_block_pairs: vec![] }).await;
-    _ = provider.anvil_mine(Some(20), None).await;
+    let mut stream =
+        client.stream_historical(BlockNumberOrTag::Number(0), BlockNumberOrTag::Latest).await?;
 
     assert_next!(stream, 0..=29);
-    assert_next!(stream, 30..=59);
-    assert_next!(stream, 60..=89);
-    assert_next!(stream, 90..=120);
+    assert_next!(stream, 30..=56);
+    assert_next!(stream, 57..=86);
+    assert_next!(stream, 87..=116);
+    assert_next!(stream, 117..=120);
+    let mut stream = assert_empty!(stream);
+
+    _ = provider.anvil_reorg(ReorgOptions { depth: 1, tx_block_pairs: vec![] }).await;
+
     assert_next!(stream, Notification::ReorgDetected);
     assert_next!(stream, 120..=120);
     assert_closed!(stream);
 
     Ok(())
 }
 
+#[tokio::test]
+#[ignore = "enable when ack-channels are introduced: https://github.com/OpenZeppelin/Event-Scanner/issues/218"]
+async fn historical_reorg_occurring_immediately_after_finalized_processing_is_ignored()
+-> anyhow::Result<()> {
+    let anvil = Anvil::new().try_spawn()?;
+    let provider = ProviderBuilder::new().connect(anvil.ws_endpoint_url().as_str()).await?;
+
+    provider.anvil_mine(Some(11), None).await?;
+
+    let client =
+        BlockRangeScanner::new().max_block_range(10).connect(provider.clone()).await?.run()?;
+
+    let mut stream =
+        client.stream_historical(BlockNumberOrTag::Earliest, BlockNumberOrTag::Latest).await?;
+
+    assert_next!(stream, 0..=0);
+    let mut stream = assert_empty!(stream);
+
+    _ = provider.anvil_reorg(ReorgOptions { depth: 3, tx_block_pairs: vec![] }).await;
+
+    // no Notification::ReorgDetected is streamed
+    assert_next!(stream, 1..=11);
+    assert_closed!(stream);
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn historic_mode_respects_blocks_read_per_epoch() -> anyhow::Result<()> {
     let anvil = Anvil::new().try_spawn()?;
@@ -279,7 +302,8 @@ async fn historic_mode_respects_blocks_read_per_epoch() -> anyhow::Result<()> {
     assert_closed!(stream);
 
     let mut stream = client.stream_historical(0, 99).await?;
-    assert_next!(stream, 0..=99);
+    assert_next!(stream, 0..=36);
+    assert_next!(stream, 37..=99);
     assert_closed!(stream);
 
     Ok(())
@@ -295,15 +319,15 @@ async fn historic_mode_normalises_start_and_end_block() -> anyhow::Result<()> {
     let client = BlockRangeScanner::new().max_block_range(5).connect(provider).await?.run()?;
 
     let mut stream = client.stream_historical(10, 0).await?;
-    assert_next!(stream, 0..=4);
-    assert_next!(stream, 5..=9);
-    assert_next!(stream, 10..=10);
+    assert_next!(stream, 0..=0);
+    assert_next!(stream, 1..=5);
+    assert_next!(stream, 6..=10);
     assert_closed!(stream);
 
     let mut stream = client.stream_historical(0, 10).await?;
-    assert_next!(stream, 0..=4);
-    assert_next!(stream, 5..=9);
-    assert_next!(stream, 10..=10);
+    assert_next!(stream, 0..=0);
+    assert_next!(stream, 1..=5);
+    assert_next!(stream, 6..=10);
     assert_closed!(stream);
 
     Ok(())