Fixes for finalized block processing in historic mode (#225)

Author: 0xNeshi

src/block_range_scanner/common.rs

@@ -349,8 +349,8 @@ 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(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

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