Event Scanner

⚠️ WARNING: ACTIVE DEVELOPMENT ⚠️

This project is under active development and likely contains bugs. APIs and behaviour may change without notice. Use at your own risk.

About

Event Scanner is a Rust library for streaming EVM-based smart contract events. It is built on top of the 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

• Features
• Architecture Overview
• Quick Start
• Usage
  • Building a Scanner
  • Defining Event Filters
  • Scanning Modes
• Examples
• Testing

---

Features

• 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.

---

Architecture Overview

The library exposes two primary layers:

• EventScanner – the main scanner type the application will interact with.
• BlockRangeScanner – lower-level component that streams block ranges, handles reorg, batching, and provider subscriptions.

---

Quick Start

Add event-scanner to your Cargo.toml:

[dependencies]
event-scanner = "0.9.0-alpha"

Create an event stream for the given event filters registered with the EventScanner:

use alloy::{network::Ethereum, providers::ProviderBuilder, sol_types::SolEvent};
use event_scanner::{EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder};
use tokio_stream::StreamExt;
use tracing::{error, info};

use crate::MyContract;

async fn run_scanner(
    ws_url: &str,
    contract: alloy::primitives::Address,
) -> Result<(), Box<dyn std::error::Error>> {
    // Connect to provider
    let provider = ProviderBuilder::new().connect(ws_url).await?;
    let robust_provider = RobustProviderBuilder::new(provider).build().await?;
    
    // Configure scanner with custom batch size (optional)
    let mut scanner = EventScannerBuilder::live()
        .max_block_range(500)  // Process up to 500 blocks per batch
        .connect(robust_provider)
        .await?;

    // Register an event listener
    let filter = EventFilter::new()
        .contract_address(contract)
        .event(MyContract::SomeEvent::SIGNATURE);

    let mut stream = scanner.subscribe(filter);

    // Start the scanner
    scanner.start().await?;

    // Process messages from the stream
    while let Some(message) = stream.next().await {
        match message {
            Ok(Message::Data(logs)) => {
                for log in logs {
                    info!("Callback successfully executed with event {:?}", log.inner.data);
                }
            }
            Ok(Message::Notification(notification)) => {
                info!("Received notification: {:?}", notification);
            }
            Err(e) => {
                error!("Received error: {}", e);
            }
        }
    }

    Ok(())
}

---

Usage

Building a Scanner

EventScannerBuilder provides mode-specific constructors and functions to configure settings before connecting. Once configured, connect using:

• connect(provider) - Connect using a RobustProvider wrapping your alloy provider or using an alloy provider directly

This will connect the EventScanner and allow you to create event streams and start scanning in various modes.

use alloy::providers::ProviderBuilder;
use event_scanner::{EventScannerBuilder, robust_provider::RobustProviderBuilder};

// Connect to provider (example with WebSocket)
let provider = ProviderBuilder::new().connect("ws://localhost:8545").await?;

// Live streaming mode
let scanner = EventScannerBuilder::live()
    .max_block_range(500)  // Optional: set max blocks per read (default: 1000)
    .block_confirmations(12)  // Optional: set block confirmations (default: 12)
    .connect(provider.clone())
    .await?;

// Historical block range mode
let scanner = EventScannerBuilder::historic()
    .from_block(1_000_000)
    .to_block(2_000_000)
    .max_block_range(500)
    .connect(provider.clone())
    .await?;

// we can also wrap the provider in a RobustProvider
// for more advanced configurations like retries and fallbacks
let robust_provider = RobustProviderBuilder::new(provider).build().await?;

// Latest events mode
let scanner = EventScannerBuilder::latest(100)
    // .from_block(1_000_000)  // Optional: set start of search range
    // .to_block(2_000_000)    // Optional: set end of search range
    .max_block_range(500)
    .connect(robust_provider.clone())
    .await?;

// Sync from block then switch to live mode
let scanner = EventScannerBuilder::sync()
    .from_block(100)
    .max_block_range(500)
    .block_confirmations(12)
    .connect(robust_provider.clone())
    .await?;

// Sync the latest 60 events then switch to live mode
let scanner = EventScannerBuilder::sync()
    .from_latest(60)
    .block_confirmations(12)
    .connect(robust_provider)
    .await?;

Invoking scanner.start() starts the scanner in the specified mode.

Defining Event Filters

Create an EventFilter for each event stream you wish to process. The filter specifies the contract address where events originated, and event signatures (tip: you can use the value stored in SolEvent::SIGNATURE).

use alloy::sol_types::SolEvent;
use event_scanner::EventFilter;

// Track a SPECIFIC event from a SPECIFIC contract
let specific_filter = EventFilter::new()
    .contract_address(*my_contract.address())
    .event(MyContract::SomeEvent::SIGNATURE);

// Track multiple events from a SPECIFIC contract
let specific_filter = EventFilter::new()
    .contract_address(*my_contract.address())
    .event(MyContract::SomeEvent::SIGNATURE)
    .event(MyContract::OtherEvent::SIGNATURE);

// Track a SPECIFIC event from ALL contracts
let specific_filter = EventFilter::new()
    .event(MyContract::SomeEvent::SIGNATURE);

// Track ALL events from SPECIFIC contracts
let all_contract_events_filter = EventFilter::new()
    .contract_address(*my_contract.address())
    .contract_address(*other_contract.address());

// Track ALL events from ALL contracts
let all_events_filter = EventFilter::new();

Register multiple filters by invoking subscribe repeatedly.

The flexibility provided by EventFilter allows you to build sophisticated event monitoring systems that can track events at different granularities depending on your application's needs.

Event Filter Batch Builders

Batch builder examples:

// Multiple contract addresses at once
let multi_addr = EventFilter::new()
    .contract_addresses([*my_contract.address(), *other_contract.address()]);

// Multiple event names at once
let multi_events = EventFilter::new()
    .events([MyContract::SomeEvent::SIGNATURE, MyContract::OtherEvent::SIGNATURE]);

// Multiple event signature hashes at once
let multi_sigs = EventFilter::new()
    .event_signatures([
        MyContract::SomeEvent::SIGNATURE_HASH,
        MyContract::OtherEvent::SIGNATURE_HASH,
    ]);

Message Types

The scanner delivers three types of messages through the event stream:

• Message::Data(Vec<Log>) – Contains a batch of matching event logs. Each log includes the raw event data, transaction hash, block number, and other metadata.
• Message::Notification(Notification) – Notifications from the scanner.
• ScannerError – Errors indicating that the scanner has encountered issues (e.g., RPC failures, connection problems, or a lagging consumer).

Always handle all message types in your stream processing loop to ensure robust error handling and proper reorg detection.

Notes:

• Ordering is guaranteed only within a single subscription stream. There is no global ordering guarantee across multiple subscriptions.
• When the scanner detects a reorg, it emits Notification::ReorgDetected. Consumers should assume the same events might be delivered more than once around reorgs (i.e. benign duplicates are possible). Depending on the application's needs, this could be handled via idempotency/deduplication or by rolling back application state on reorg notifications.

Scanning Modes

• Live – scanner that streams new blocks as they arrive.
• Historic – scanner for streaming events from a past block range (default: genesis..=latest).
• Latest Events – scanner that collects up to count most recent events per listener. Final delivery is in chronological order (oldest to newest).
• Sync from Block – scanner that streams events from a given start block up to the current confirmed tip, then automatically transitions to live streaming.
• Sync from Latest Events - scanner that collects the most recent count events, then automatically transitions to live streaming.

Important Notes

• Set max_block_range based on your RPC provider's limits (e.g., Alchemy, Infura may limit queries to 2000 blocks). Default is 1000 blocks.
• The modes come with sensible defaults; for example, not specifying a start block for historic mode automatically sets it to the genesis block.
• In live mode, if the block subscription lags and the scanner needs to catch up by querying past blocks, catch-up queries are performed in ranges bounded by max_block_range to respect provider limits.

---

Examples

• examples/live_scanning – minimal live-mode scanner using EventScannerBuilder::live()
• examples/historical_scanning – demonstrates replaying historical data using EventScannerBuilder::historic()
• examples/sync_from_block_scanning – demonstrates replaying from genesis (block 0) before continuing to stream the latest blocks using EventScannerBuilder::sync().from_block(block_id)
• examples/latest_events_scanning – demonstrates scanning the latest events using EventScannerBuilder::latest()
• examples/sync_from_latest_scanning – demonstrates scanning the latest events before switching to live mode using EventScannerBuilder::sync().from_latest(count).

Run an example with:

RUST_LOG=info cargo run -p live_scanning

All examples spin up a local anvil instance, deploy a demo counter contract, and demonstrate using event streams to process events.

---

Robust Provider

event-scanner ships with a robust_provider module that wraps Alloy providers with:

• bounded per-call timeouts and exponential backoff retries
• automatic failover from a primary provider to one or more fallbacks
• resilient WebSocket block subscriptions with timeout handling and reconnection.

The main entry point is robust_provider::RobustProviderBuilder, which accepts a wide range of provider types (URLs, RootProvider, layered providers, etc.) through the IntoRobustProvider and IntoRobustProvider traits.

A typical setup looks like:

use alloy::providers::ProviderBuilder;
use event_scanner::robust_provider::RobustProviderBuilder;
use std::time::Duration;

async fn example() -> anyhow::Result<()> {
    let ws = ProviderBuilder::new().connect("ws://localhost:8545").await?;
    let http = ProviderBuilder::new().connect_http("http://localhost:8545".parse()?);

    let provider = RobustProviderBuilder::new(ws)
        .fallback(http)
        .call_timeout(Duration::from_secs(30))
        .subscription_timeout(Duration::from_secs(120))
        .build()
        .await?;

    // ...

    Ok(())
}

You can then pass this robust provider into EventScannerBuilder::connect just like any other provider.

---

Testing

(We recommend using nextest to run the tests)

Integration tests cover all modes:

cargo nextest run --features test-utils