test: Create Macro That Addresses Unpredictability in Streamed Block Ranges (#189)

Author: 0xNeshi

.github/workflows/test.yml

@@ -54,4 +54,4 @@ jobs:
 
         # https://github.com/rust-lang/cargo/issues/6669
       - name: Run doc tests
-        run: cargo test --locked --doc
+        run: cargo test --locked --doc --all-features --no-fail-fast

src/test_utils/macros.rs

@@ -42,3 +42,115 @@ macro_rules! assert_empty {
         tokio_stream::wrappers::ReceiverStream::new(inner)
     }};
 }
+
+/// Asserts that a stream of block ranges completely covers an expected block range.
+///
+/// This macro consumes messages from a stream and verifies that the block ranges received
+/// sequentially cover the entire `expected_range` without gaps or overlaps. Each streamed
+/// range must start exactly where the previous one ended, and all ranges must fit within
+/// the expected bounds.
+///
+/// The macro expects the stream to yield `Message::Data(range)` variants containing
+/// `RangeInclusive<u64>` values representing block ranges. It tracks coverage by ensuring
+/// each new range starts at the next expected block number and doesn't exceed the end of
+/// the expected range. Once the entire range is covered, the assertion succeeds.
+///
+/// # Example
+///
+/// ```rust
+/// use event_scanner::{assert_range_coverage, block_range_scanner::Message};
+/// use tokio::sync::mpsc;
+/// use tokio_stream::wrappers::ReceiverStream;
+///
+/// #[tokio::test]
+/// async fn test_scanner_covers_range() {
+///     let (tx, rx) = mpsc::channel(10);
+///     let mut stream = ReceiverStream::new(rx);
+///
+///     // Simulate a scanner that splits blocks 100-199 into chunks
+///     tokio::spawn(async move {
+///         tx.send(Message::Data(100..=149)).await.unwrap();
+///         tx.send(Message::Data(150..=199)).await.unwrap();
+///     });
+///
+///     // Assert that the stream covers blocks 100-199
+///     assert_range_coverage!(stream, 100..=199);
+/// }
+/// ```
+///
+/// # Panics
+///
+/// * **Timeout**: The stream doesn't produce the next expected range within the timeout period
+///   (default 5 seconds, configurable via `timeout = N` parameter).
+/// * **Gap or overlap**: A streamed range doesn't start exactly at the next expected block number,
+///   indicating a gap or overlap in coverage.
+/// * **Out of bounds**: A streamed range extends beyond the end of the expected range.
+/// * **Wrong message type**: The stream yields a non-`Data` message (e.g., `Error` or `Status`)
+///   when a block range is expected.
+/// * **Stream closed early**: The stream ends before the entire expected range is covered.
+///
+/// On panic, the error message includes the expected remaining range and all previously
+/// streamed ranges for debugging.
+#[macro_export]
+macro_rules! assert_range_coverage {
+    ($stream: expr, $expected_range: expr) => {
+        assert_range_coverage!($stream, $expected_range, timeout = 5)
+    };
+    ($stream: expr, $expected_range: expr, timeout = $secs: expr) => {{
+        fn bounds<R: ::std::ops::RangeBounds<u64>>(range: &R) -> (u64, u64) {
+            let start_bound = match range.start_bound() {
+                ::std::ops::Bound::Unbounded => 0,
+                ::std::ops::Bound::Excluded(&x) => x + 1,
+                ::std::ops::Bound::Included(&x) => x,
+            };
+            let end_bound = match range.end_bound() {
+                ::std::ops::Bound::Unbounded => u64::MAX,
+                ::std::ops::Bound::Excluded(&x) => x - 1,
+                ::std::ops::Bound::Included(&x) => x,
+            };
+            (start_bound, end_bound)
+        }
+
+        let original_bounds = bounds(&$expected_range);
+        let (mut start, end) = original_bounds;
+
+        let start_time = ::std::time::Instant::now();
+        let timeout_duration = ::std::time::Duration::from_secs($secs);
+
+        // log all streamed ranges on failures
+        let mut streamed_ranges = vec![];
+
+        while start <= end {
+            let elapsed = start_time.elapsed();
+
+            assert!(elapsed < timeout_duration, "Timed out. Still expecting: {:#?}", start..=end,);
+
+            let time_left = timeout_duration - elapsed;
+            let message =
+                tokio::time::timeout(time_left, tokio_stream::StreamExt::next(&mut $stream))
+                    .await
+                    .expect("Timed out waiting for the next block range");
+
+            match message {
+                Some( $crate::block_range_scanner::Message::Data(range)) => {
+                    let (streamed_start, streamed_end) = bounds(&range);
+                    streamed_ranges.push(range.clone());
+                    assert!(
+                        start == streamed_start && streamed_end <= end,
+                        "Unexpected range bounds, expected max. range: {:#?}, got: {:#?}\nPrevious streams:\n{:#?}",
+                        start..=end,
+                        range,
+                        streamed_ranges,
+                    );
+                    start = streamed_end + 1;
+                }
+                Some(other) => {
+                    panic!("Expected a block range, got: {other:#?}");
+                }
+                None => {
+                    panic!("Stream closed without covering range: {:#?}", start..=end);
+                }
+            }
+        }
+    }};
+}

tests/block_range_scanner.rs

@@ -5,7 +5,7 @@ use alloy::{
 };
 use alloy_node_bindings::Anvil;
 use event_scanner::{
-    ScannerError, ScannerStatus, assert_closed, assert_empty, assert_next,
+    ScannerError, ScannerStatus, assert_closed, assert_empty, assert_next, assert_range_coverage,
     block_range_scanner::BlockRangeScanner,
 };
 
@@ -22,11 +22,7 @@ async fn live_mode_processes_all_blocks_respecting_block_confirmations() -> anyh
 
     provider.anvil_mine(Some(5), None).await?;
 
-    assert_next!(stream, 1..=1);
-    assert_next!(stream, 2..=2);
-    assert_next!(stream, 3..=3);
-    assert_next!(stream, 4..=4);
-    assert_next!(stream, 5..=5);
+    assert_range_coverage!(stream, 1..=5);
     let mut stream = assert_empty!(stream);
 
     provider.anvil_mine(Some(1), None).await?;
@@ -40,11 +36,7 @@ async fn live_mode_processes_all_blocks_respecting_block_confirmations() -> anyh
 
     provider.anvil_mine(Some(5), None).await?;
 
-    assert_next!(stream, 6..=6);
-    assert_next!(stream, 7..=7);
-    assert_next!(stream, 8..=8);
-    assert_next!(stream, 9..=9);
-    assert_next!(stream, 10..=10);
+    assert_range_coverage!(stream, 6..=10);
     let mut stream = assert_empty!(stream);
 
     provider.anvil_mine(Some(1), None).await?;
@@ -97,26 +89,18 @@ async fn continuous_blocks_if_reorg_less_than_block_confirmation() -> anyhow::Re
     // assert initial block ranges immediately to avoid Anvil race condition:
     //
     // when a reorg happens after anvil_mine, Anvil occasionally first streams a non-zero block
-    // number, which makes it impossible to deterministically assert the next expected block range
+    // number, which makes it impossible to deterministically assert the first expected block range
     // streamed by the scanner
-    assert_next!(stream, 0..=0);
-    assert_next!(stream, 1..=1);
-    assert_next!(stream, 2..=2);
-    assert_next!(stream, 3..=3);
-    assert_next!(stream, 4..=4);
-    assert_next!(stream, 5..=5);
+    assert_range_coverage!(stream, 0..=5);
+    let mut stream = assert_empty!(stream);
 
     // reorg less blocks than the block_confirmation config
     provider.anvil_reorg(ReorgOptions { depth: 4, tx_block_pairs: vec![] }).await?;
     // mint additional blocks so the scanner processes reorged blocks
     provider.anvil_mine(Some(5), None).await?;
 
     // no ReorgDetected should be emitted
-    assert_next!(stream, 6..=6);
-    assert_next!(stream, 7..=7);
-    assert_next!(stream, 8..=8);
-    assert_next!(stream, 9..=9);
-    assert_next!(stream, 10..=10);
+    assert_range_coverage!(stream, 6..=10);
     assert_empty!(stream);
 
     Ok(())
@@ -137,34 +121,18 @@ async fn shallow_block_confirmation_does_not_mitigate_reorg() -> anyhow::Result<
     // assert initial block ranges immediately to avoid Anvil race condition:
     //
     // when a reorg happens after anvil_mine, Anvil occasionally first streams a non-zero block
-    // number, which makes it impossible to deterministically assert the next expected block range
+    // number, which makes it impossible to deterministically assert the first expected block range
     // streamed by the scanner
-    assert_next!(stream, 0..=0);
-    assert_next!(stream, 1..=1);
-    assert_next!(stream, 2..=2);
-    assert_next!(stream, 3..=3);
-    assert_next!(stream, 4..=4);
-    assert_next!(stream, 5..=5);
-    assert_next!(stream, 6..=6);
-    assert_next!(stream, 7..=7);
+    assert_range_coverage!(stream, 0..=7);
+    let mut stream = assert_empty!(stream);
 
     // reorg more blocks than the block_confirmation config
     provider.anvil_reorg(ReorgOptions { depth: 8, tx_block_pairs: vec![] }).await?;
     // mint additional blocks
     provider.anvil_mine(Some(3), None).await?;
 
     assert_next!(stream, ScannerStatus::ReorgDetected);
-    assert_next!(stream, 0..=0);
-    assert_next!(stream, 1..=1);
-    assert_next!(stream, 2..=2);
-    assert_next!(stream, 3..=3);
-    assert_next!(stream, 4..=4);
-    assert_next!(stream, 5..=5);
-    assert_next!(stream, 6..=6);
-    assert_next!(stream, 7..=7);
-    assert_next!(stream, 8..=8);
-    assert_next!(stream, 9..=9);
-    assert_next!(stream, 10..=10);
+    assert_range_coverage!(stream, 0..=10);
     assert_empty!(stream);
 
     Ok(())