feat: Add "fetch latest X events" stream function (#121)

Author: 0xNeshi

CONTRIBUTING.md

@@ -60,9 +60,9 @@ cargo build --locked --all-targets --all-features
 Run the test suite (we recommend nextest):
 
 ```bash
-cargo nextest run
+cargo nextest run --features test-utils
 # or
-cargo test
+cargo test --features test-utils
 ```
 
 Run examples:
@@ -123,7 +123,7 @@ cargo clippy --all-targets --all-features -- -D warnings -D clippy::pedantic
 typos
 
 # Tests (prefer nextest)
-cargo nextest run
+cargo nextest run --features test-utils
 ```
 
 ---
@@ -160,7 +160,7 @@ Key implementation details, trade-offs, and alternatives considered.
 - [ ] `cargo +nightly fmt --all --check`
 - [ ] `cargo clippy --all-targets --all-features -- -D warnings -D clippy::pedantic`
 - [ ] `typos`
-- [ ] Tests pass (`cargo nextest run`)
+- [ ] Tests pass (`cargo nextest run --features test-utils`)
 - [ ] Docs/README/examples updated (if applicable)
 ```
 

Cargo.lock

@@ -1,5 +1,10 @@
 [workspace]
-members = [".", "examples/historical_scanning", "examples/simple_counter"]
+members = [
+    ".",
+    "examples/historical_scanning",
+    "examples/simple_counter",
+    "examples/latest_events_scanning",
+]
 resolver = "2"
 
 [workspace.package]
@@ -63,3 +68,6 @@ tracing-subscriber.workspace = true
 
 [lints]
 workspace = true
+
+[features]
+test-utils = []

Cargo.toml

@@ -11,6 +11,7 @@
 Event Scanner is a Rust library for streaming EVM-based smart contract events. It is built on top of the [`alloy`](https://github.com/alloy-rs/alloy) ecosystem and focuses on in-memory scanning without a backing database. Applications provide event filters; the scanner takes care of fetching historical ranges, bridging into live streaming mode, all whilst delivering the events as streams of data.
 
 ---
+ 
 
 ## Table of Contents
 
@@ -21,7 +22,7 @@ Event Scanner is a Rust library for streaming EVM-based smart contract events. I
   - [Building a Scanner](#building-a-scanner)
   - [Defining Event Filters](#defining-event-filters)
   - [Scanning Modes](#scanning-modes)
-  - [Working with Callbacks](#working-with-callbacks)
+  - [Scanning Latest Events](#scanning-latest-events)
 - [Examples](#examples)
 - [Testing](#testing)
 
@@ -32,6 +33,7 @@ Event Scanner is a Rust library for streaming EVM-based smart contract events. I
 - **Historical replay** – stream events from past block ranges.
 - **Live subscriptions** – stay up to date with latest events via WebSocket or IPC transports.
 - **Hybrid flow** – automatically transition from historical catch-up into streaming mode.
+- **Latest events fetch** – one-shot rewind to collect the most recent matching logs.
 - **Composable filters** – register one or many contract + event signature pairs.
 - **No database** – processing happens in-memory; persistence is left to the host application.
 
@@ -137,20 +139,71 @@ The flexibility provided by `EventFilter` allows you to build sophisticated even
 
 ### Scanning Modes
 
-- **Live mode** – `start_scanner(BlockNumberOrTag::Latest, None)` subscribes to new blocks only.
-- **Historical mode** – `start_scanner(BlockNumberOrTag::Number(start), Some(BlockNumberOrTag::Number(end)))`, scanner fetches events from a historical block range.
-- **Historical → Live** – `start_scanner(BlockNumberOrTag::Number(start), None)` replays from `start` to current head, then streams future blocks.
+- **Live mode** - `start_scanner(BlockNumberOrTag::Latest, None)` subscribes to new blocks only. On detecting a reorg, the scanner emits `ScannerStatus::ReorgDetected` and recalculates the confirmed window, streaming logs from the corrected confirmed block range.
+- **Historical mode** - `start_scanner(BlockNumberOrTag::Number(start), Some(BlockNumberOrTag::Number(end)))`, scanner fetches events from a historical block range. While syncing ranges, the scanner verifies continuity. If a reorg is detected, it rewinds by `with_reorg_rewind_depth` blocks and resumes forward syncing.
+- **Historical → Live** - `start_scanner(BlockNumberOrTag::Number(start), None)` replays from `start` to current head, then streams future blocks. Reorgs are handled as per the particular mode phase the scanner is in (historical or live).
 
 For now modes are deduced from the `start` and `end` parameters. In the future, we might add explicit commands to select the mode.
 
 See the integration tests under `tests/live_mode`, `tests/historic_mode`, and `tests/historic_to_live` for concrete examples.
 
+### Scanning Latest Events
+
+`scan_latest` collects the most recent matching events for each registered stream.
+
+- It does not enter live mode; it scans a block range and then returns.
+- Each registered stream receives at most `count` logs in a single message, chronologically ordered.
+
+Basic usage:
+
+```rust
+use alloy::{eips::BlockNumberOrTag, network::Ethereum};
+use event_scanner::{EventFilter, event_scanner::{EventScanner, EventScannerMessage}};
+use tokio_stream::StreamExt;
+
+async fn latest_example(ws_url: alloy::transports::http::reqwest::Url, addr: alloy::primitives::Address) -> eyre::Result<()> {
+    let mut client = EventScanner::new().connect_ws::<Ethereum>(ws_url).await?;
+
+    let filter = EventFilter::new().with_contract_address(addr);
+    let mut stream = client.create_event_stream(filter);
+
+    // Collect the latest 10 events across Earliest..=Latest
+    client.scan_latest(10).await?;
+
+    // Expect a single message with up to 10 logs, then the stream ends
+    while let Some(msg) = stream.next().await {
+        if let EventScannerMessage::Data(logs) = msg {
+            println!("Latest logs: {}", logs.len());
+        }
+    }
+
+    Ok(())
+}
+```
+
+Restricting to a specific block range:
+
+```rust
+// Collect the latest 5 events between blocks [1_000_000, 1_100_000]
+client
+    .scan_latest_in_range(5, BlockNumberOrTag::Number(1_000_000), BlockNumberOrTag::Number(1_100_000))
+    .await?;
+```
+
+The scanner periodically checks the tip to detect reorgs. On reorg, the scanner emits `ScannerStatus::ReorgDetected`, resets to the updated tip, and restarts the scan. Final delivery to log listeners is in chronological order.
+
+Notes:
+
+- Ensure you create streams via `create_event_stream()` before calling `scan_latest*` so listeners are registered.
+<!-- TODO: uncomment once implemented - The function returns after delivering the messages; to continuously stream new blocks, use `scan_latest_then_live`. -->
+
 ---
 
 ## Examples
 
 - `examples/simple_counter` – minimal live-mode scanner
 - `examples/historical_scanning` – demonstrates replaying from genesis (block 0) before continuing streaming latest blocks
+- `examples/latest_events_scanning` – demonstrates scanning the latest events
 
 Run an example with:
 
@@ -166,10 +219,11 @@ Both examples spin up a local `anvil` instance, deploy a demo counter contract,
 
 ## Testing
 
-Integration tests cover live, historical, and hybrid flows:
 (We recommend using [nextest](https://crates.io/crates/cargo-nextest) to run the tests)
 
+Integration tests cover all modes:
+
 ```bash
-cargo nextest run
+cargo nextest run --features test-utils
 ```
 

README.md

@@ -0,0 +1,21 @@
+[package]
+name = "latest_events_scanning"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+publish = false
+
+[[bin]]
+name = "latest_events_scanning"
+path = "main.rs"
+
+[dependencies]
+alloy.workspace = true
+alloy-node-bindings.workspace = true
+tokio.workspace = true
+tokio-stream.workspace = true
+anyhow.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+event-scanner = { path = "../.." }

examples/latest_events_scanning/Cargo.toml

@@ -0,0 +1,84 @@
+use alloy::{network::Ethereum, providers::ProviderBuilder, sol, sol_types::SolEvent};
+use alloy_node_bindings::Anvil;
+use event_scanner::{
+    EventFilter,
+    event_scanner::{EventScanner, EventScannerMessage},
+};
+
+use tokio_stream::StreamExt;
+use tracing::{error, info};
+use tracing_subscriber::EnvFilter;
+
+sol! {
+    #[allow(missing_docs)]
+    #[sol(rpc, bytecode="608080604052346015576101b0908161001a8239f35b5f80fdfe6080806040526004361015610012575f80fd5b5f3560e01c90816306661abd1461016157508063a87d942c14610145578063d732d955146100ad5763e8927fbc14610048575f80fd5b346100a9575f3660031901126100a9575f5460018101809111610095576020817f7ca2ca9527391044455246730762df008a6b47bbdb5d37a890ef78394535c040925f55604051908152a1005b634e487b7160e01b5f52601160045260245ffd5b5f80fd5b346100a9575f3660031901126100a9575f548015610100575f198101908111610095576020817f53a71f16f53e57416424d0d18ccbd98504d42a6f98fe47b09772d8f357c620ce925f55604051908152a1005b60405162461bcd60e51b815260206004820152601860248201527f436f756e742063616e6e6f74206265206e6567617469766500000000000000006044820152606490fd5b346100a9575f3660031901126100a95760205f54604051908152f35b346100a9575f3660031901126100a9576020905f548152f3fea2646970667358221220b846b706f79f5ae1fc4a4238319e723a092f47ce4051404186424739164ab02264736f6c634300081e0033")]
+    contract Counter {
+        uint256 public count;
+
+        event CountIncreased(uint256 newCount);
+        event CountDecreased(uint256 newCount);
+
+        function increase() public {
+            count += 1;
+            emit CountIncreased(count);
+        }
+
+        function decrease() public {
+            require(count > 0, "Count cannot be negative");
+            count -= 1;
+            emit CountDecreased(count);
+        }
+
+        function getCount() public view returns (uint256) {
+            return count;
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let _ = tracing_subscriber::fmt().with_env_filter(EnvFilter::from_default_env()).try_init();
+
+    let anvil = Anvil::new().block_time_f64(0.5).try_spawn()?;
+    let wallet = anvil.wallet();
+    let provider =
+        ProviderBuilder::new().wallet(wallet.unwrap()).connect(anvil.endpoint().as_str()).await?;
+    let counter_contract = Counter::deploy(provider).await?;
+
+    let contract_address = counter_contract.address();
+
+    let increase_filter = EventFilter::new()
+        .with_contract_address(*contract_address)
+        .with_event(Counter::CountIncreased::SIGNATURE);
+
+    let mut client = EventScanner::new().connect_ws::<Ethereum>(anvil.ws_endpoint_url()).await?;
+
+    let mut stream = client.create_event_stream(increase_filter);
+
+    for _ in 0..10 {
+        _ = counter_contract.increase().send().await?;
+    }
+
+    tokio::spawn(async move {
+        client.scan_latest(5).await.expect("failed to start scanner");
+    });
+
+    // only the last 5 events will be streamed
+    while let Some(message) = stream.next().await {
+        match message {
+            EventScannerMessage::Data(logs) => {
+                for log in logs {
+                    info!("Callback successfully executed with event {:?}", log.inner.data);
+                }
+            }
+            EventScannerMessage::Error(e) => {
+                error!("Received error: {}", e);
+            }
+            EventScannerMessage::Status(info) => {
+                info!("Received info: {:?}", info);
+            }
+        }
+    }
+
+    Ok(())
+}