test: add multi-block reorg and watermark min() coverage

Address review feedback: strengthen reorg integration tests with
multi-block reverts and multiple-reorgs-between-polls scenarios.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: prestwich

crates/node-tests/tests/reorg.rs

@@ -109,7 +109,6 @@ async fn test_block_filter_reorg() {
         // Poll: should have 1 block hash.
         let hashes: Vec<B256> = ctx.alloy_provider.get_filter_changes(filter_id).await.unwrap();
         assert_eq!(hashes.len(), 1);
-        let block2_hash = hashes[0];
 
         // Process block 3 (increment), keep clone for revert.
         let block3 = process_increment(&ctx, *contract.address()).await;
@@ -128,8 +127,9 @@ async fn test_block_filter_reorg() {
         // Poll: should return the new block 3 hash.
         let hashes: Vec<B256> = ctx.alloy_provider.get_filter_changes(filter_id).await.unwrap();
         assert_eq!(hashes.len(), 1);
-        // Verify it is NOT the old block 2 hash (it should be the new block 3).
-        assert_ne!(hashes[0], block2_hash);
+        let new_block3_hash =
+            ctx.alloy_provider.get_block_by_number(3.into()).await.unwrap().unwrap().hash();
+        assert_eq!(hashes[0], new_block3_hash);
 
         ctx
     })
@@ -383,3 +383,167 @@ async fn test_no_regression_filters_and_subscriptions() {
     })
     .await;
 }
+
+// ---------------------------------------------------------------------------
+// 8. Multi-block reorg with log filter
+// ---------------------------------------------------------------------------
+
+#[serial]
+#[tokio::test]
+async fn test_multi_block_reorg_log_filter() {
+    rpc_test(|ctx, contract| async move {
+        let addr = *contract.address();
+
+        // Install a log filter on the Counter address.
+        let filter_id = ctx.alloy_provider.new_filter(&Filter::new().address(addr)).await.unwrap();
+
+        // Process blocks 2, 3, 4 (increment each → count 1, 2, 3).
+        let block2 = process_increment(&ctx, addr).await;
+        let block3 = process_increment(&ctx, addr).await;
+        let block4 = process_increment(&ctx, addr).await;
+
+        // Poll: expect 3 logs.
+        let logs: Vec<Log<LogData>> =
+            ctx.alloy_provider.get_filter_changes(filter_id).await.unwrap();
+        assert_eq!(logs.len(), 3);
+        for (i, log) in logs.iter().enumerate() {
+            assert_eq!(log.inner.topics()[1], B256::with_last_byte(i as u8 + 1));
+        }
+
+        // Revert blocks 4, 3, 2 (back to block 1).
+        ctx.revert_block(block4).await.unwrap();
+        ctx.revert_block(block3).await.unwrap();
+        ctx.revert_block(block2).await.unwrap();
+
+        // Poll: empty (watermark rewinds start to 2, but latest=1).
+        let logs: Vec<Log<LogData>> =
+            ctx.alloy_provider.get_filter_changes(filter_id).await.unwrap();
+        assert!(logs.is_empty());
+
+        // Rebuild blocks 2, 3 (increment each → count 1, 2).
+        let _new_b2 = process_increment(&ctx, addr).await;
+        let _new_b3 = process_increment(&ctx, addr).await;
+
+        // Poll: expect 2 logs with count 1, 2.
+        let logs: Vec<Log<LogData>> =
+            ctx.alloy_provider.get_filter_changes(filter_id).await.unwrap();
+        assert_eq!(logs.len(), 2);
+        assert_eq!(logs[0].inner.topics()[1], B256::with_last_byte(1));
+        assert_eq!(logs[1].inner.topics()[1], B256::with_last_byte(2));
+
+        ctx
+    })
+    .await;
+}
+
+// ---------------------------------------------------------------------------
+// 9. Multi-block reorg with log subscription
+// ---------------------------------------------------------------------------
+
+#[serial]
+#[tokio::test]
+async fn test_multi_block_reorg_log_subscription() {
+    rpc_test(|ctx, contract| async move {
+        let addr = *contract.address();
+        let mut sub =
+            ctx.alloy_provider.subscribe_logs(&Filter::new().address(addr)).await.unwrap();
+
+        // Process blocks 2, 3 (increment each → count 1, 2).
+        let block2 = process_increment(&ctx, addr).await;
+        let block3 = process_increment(&ctx, addr).await;
+
+        // Receive 2 normal logs.
+        for expected in [1u8, 2] {
+            let log =
+                tokio::time::timeout(Duration::from_secs(5), sub.recv()).await.unwrap().unwrap();
+            assert!(!log.removed);
+            assert_eq!(log.inner.topics()[1], B256::with_last_byte(expected));
+        }
+
+        // Revert blocks 3, 2.
+        ctx.revert_block(block3).await.unwrap();
+        ctx.revert_block(block2).await.unwrap();
+
+        // Receive 2 removed logs (one per reverted block).
+        for _ in 0..2 {
+            let log =
+                tokio::time::timeout(Duration::from_secs(5), sub.recv()).await.unwrap().unwrap();
+            assert!(log.removed);
+            assert_eq!(log.inner.address, addr);
+        }
+
+        // Rebuild block 2 (increment → count 1).
+        let _new_b2 = process_increment(&ctx, addr).await;
+
+        // Receive the new normal log.
+        let log = tokio::time::timeout(Duration::from_secs(5), sub.recv()).await.unwrap().unwrap();
+        assert!(!log.removed);
+        assert_eq!(log.inner.topics()[1], B256::with_last_byte(1));
+
+        ctx
+    })
+    .await;
+}
+
+// ---------------------------------------------------------------------------
+// 10. Multiple reorgs between polls (watermark min path)
+// ---------------------------------------------------------------------------
+
+#[serial]
+#[tokio::test]
+async fn test_multiple_reorgs_between_polls() {
+    rpc_test(|ctx, contract| async move {
+        let addr = *contract.address();
+
+        let filter_id = ctx.alloy_provider.new_filter(&Filter::new().address(addr)).await.unwrap();
+
+        // Process blocks 2, 3, 4, 5 (count 1, 2, 3, 4).
+        let _b2 = process_increment(&ctx, addr).await;
+        let _b3 = process_increment(&ctx, addr).await;
+        let b4 = process_increment(&ctx, addr).await;
+        let b5 = process_increment(&ctx, addr).await;
+
+        // Poll to advance the filter cursor past block 5.
+        let logs: Vec<Log<LogData>> =
+            ctx.alloy_provider.get_filter_changes(filter_id).await.unwrap();
+        assert_eq!(logs.len(), 4);
+
+        // --- Reorg 1: revert block 5 only (common_ancestor=4, watermark=4) ---
+        ctx.revert_block(b5).await.unwrap();
+
+        // Rebuild block 5 (count=5).
+        let new_b5 = process_increment(&ctx, addr).await;
+
+        // --- Reorg 2 (deeper): revert blocks 5 AND 4 ---
+        // common_ancestor for block 5 revert = 4, for block 4 revert = 3
+        // watermark = min(4, min(4, 3)) = 3
+        ctx.revert_block(new_b5).await.unwrap();
+        ctx.revert_block(b4).await.unwrap();
+
+        // DO NOT POLL between reorg 1 and reorg 2 — this is the key.
+        // The filter now has watermark=3 from the min() of both reorgs.
+
+        // Poll: empty (watermark+1=4, but latest=3 → start > latest).
+        let logs: Vec<Log<LogData>> =
+            ctx.alloy_provider.get_filter_changes(filter_id).await.unwrap();
+        assert!(logs.is_empty());
+
+        // Rebuild blocks 4, 5 (count 3, 4).
+        let _new_b4 = process_increment(&ctx, addr).await;
+        let _new_b5 = process_increment(&ctx, addr).await;
+
+        // Poll: 2 logs from blocks 4 and 5 (count 3, 4).
+        // After reverting blocks 4+5, count was back to 2. Rebuilding
+        // increments to 3 then 4. This proves the deeper watermark (3)
+        // was kept — if only the shallow watermark (4) had been kept,
+        // we'd see 1 log from block 5 only.
+        let logs: Vec<Log<LogData>> =
+            ctx.alloy_provider.get_filter_changes(filter_id).await.unwrap();
+        assert_eq!(logs.len(), 2);
+        assert_eq!(logs[0].inner.topics()[1], B256::with_last_byte(3));
+        assert_eq!(logs[1].inner.topics()[1], B256::with_last_byte(4));
+
+        ctx
+    })
+    .await;
+}