Struct EventScannerBuilder 

pub struct EventScannerBuilder<Mode> { /* private fields */ }

Builder for constructing an EventScanner in a particular mode.

Implementations

impl EventScannerBuilder<Unspecified>

pub fn historic() -> EventScannerBuilder<Historic>

Streams events from a historical block range.

Example

// Stream all events from genesis to latest block
let mut scanner = EventScannerBuilder::historic().connect(provider).await?;

let filter = EventFilter::new().contract_address(contract_address);
let subscription = scanner.subscribe(filter);
let proof = scanner.start().await?;
let mut stream = subscription.stream(&proof);

while let Some(Ok(Message::Data(logs))) = stream.next().await {
    println!("Received {} logs", logs.len());
}

Specifying a custom block range:

// Stream events between blocks [1_000_000, 2_000_000]
let mut scanner = EventScannerBuilder::historic()
    .from_block(1_000_000)
    .to_block(2_000_000)
    .connect(provider)
    .await?;

How it works

The scanner streams events in chronological order (oldest to newest) within the specified block range. Events are delivered in batches as they are fetched from the provider, with batch sizes controlled by the max_block_range configuration.

Key behaviors

• Continuous streaming: Events are delivered in multiple messages as they are fetched

• Chronological order: Events are always delivered oldest to newest

• Concurrent log fetching: Logs are fetched concurrently to reduce the execution time. The maximum number of concurrent RPC calls is controlled by max_concurrent_fetches

• Default range: By default, scans from Earliest to Latest block

• Batch control: Use max_block_range to control how many blocks are queried per RPC call

• Reorg handling:

  • Blocks up to the chain’s finalized height are streamed without reorg checks.
  • For the non-finalized portion of the range, the scanner takes a snapshot of the requested end block, streams the non-finalized portion once, and then verifies that the end block is still the same. If a reorg is detected, it emits Notification::ReorgDetected and re-streams the non-finalized portion of the range from the reported common ancestor.

  Consumers should be prepared to observe benign duplicate events around reorg boundaries (e.g. by applying idempotency/deduplication).

• Completion: The scanner completes when the entire range has been processed.

pub fn live() -> EventScannerBuilder<Live>

Streams new events as blocks are produced on-chain.

Example

// Stream new events as they arrive
let mut scanner = EventScannerBuilder::live()
    .block_confirmations(20)
    .connect(provider)
    .await?;

let filter = EventFilter::new().contract_address(contract_address);
let subscription = scanner.subscribe(filter);
let proof = scanner.start().await?;
let mut stream = subscription.stream(&proof);

while let Some(msg) = stream.next().await {
    match msg {
        Ok(Message::Data(logs)) => {
            println!("Received {} new events", logs.len());
        }
        Ok(Message::Notification(notification)) => {
            println!("Notification received: {:?}", notification);
        }
        Err(e) => {
            eprintln!("Error: {}", e);
        }
    }
}

How it works

The scanner subscribes to new blocks via WebSocket and streams events from confirmed blocks. The block_confirmations setting determines how many blocks to wait before considering a block confirmed, providing protection against chain reorganizations.

Key behaviors

• Real-time streaming: Events are delivered as new blocks are confirmed
• Reorg protection: Waits for configured confirmations before emitting events
• Continuous operation: Runs indefinitely until the scanner is dropped or encounters an error
• Default confirmations: By default, waits for 12 block confirmations

Reorg behavior

When a reorg is detected:

1. Emits Notification::ReorgDetected to all listeners
2. Adjusts the next confirmed range using block_confirmations
3. Re-emits events from the corrected confirmed block range
4. Continues streaming from the new chain state

Important: If a reorg occurs, the scanner will only restream blocks from the new canonical chain that have block numbers greater than or equal to the block number that was the “latest block” at the time when the live stream was first started. Blocks with lower block numbers will not be restreamed, even if they are part of the new canonical chain.

pub fn sync() -> EventScannerBuilder<Synchronize>

Creates a builder for sync mode scanners that combine historical catch-up with live streaming.

This method returns a builder that must be further narrowed down:

// Sync from block mode
EventScannerBuilder::sync().from_block(1_000_000);
// Sync from latest events mode
EventScannerBuilder::sync().from_latest(10);

See from_block and from_latest for details on each mode.

pub fn latest(count: usize) -> EventScannerBuilder<LatestEvents>

Streams the latest count matching events per registered listener.

Example

// Collect the latest 10 events across Earliest..=Latest
let mut scanner = EventScannerBuilder::latest(10).connect(provider).await?;

let filter = EventFilter::new().contract_address(contract_address);
let subscription = scanner.subscribe(filter);
let proof = scanner.start().await?;
let mut stream = subscription.stream(&proof);

// Expect a single message with up to 10 logs, then the stream ends
while let Some(Ok(Message::Data(logs))) = stream.next().await {
    println!("Latest logs: {}", logs.len());
}

Restricting to a specific block range:

// Collect the latest 5 events between blocks [1_000_000, 1_100_000]
let mut scanner = EventScannerBuilder::latest(5)
    .from_block(1_000_000)
    .to_block(1_100_000)
    .connect(provider)
    .await?;

How it works

The scanner performs a reverse-ordered scan (newest to oldest) within the specified block range, collecting up to count events per registered listener. Once the target count is reached or the range is exhausted, it delivers the events in chronological order (oldest to newest) and completes.

When using a custom block range, the scanner automatically normalizes the range boundaries. This means you can specify from_block and to_block in any order - the scanner will always scan from the higher block number down to the lower one, regardless of which parameter holds which value.

Key behaviors

• Single delivery: Each registered stream receives at most count logs in a single message, chronologically ordered
• One-shot operation: The scanner completes after delivering messages; it does not continue streaming
• Concurrent log fetching: Logs are fetched concurrently to reduce the execution time. The maximum number of concurrent RPC calls is controlled by max_concurrent_fetches
• Flexible count: If fewer than count events exist in the range, returns all available events
• Default range: By default, scans from Earliest to Latest block
• Reorg handling: Periodically checks the tip to detect reorgs during the scan

Notifications

The scanner can emit the following notifications:

• Notification::NoPastLogsFound: Emitted when no matching logs are found in the scanned range.
• Notification::ReorgDetected: Emitted when a reorg is detected during the scan.

Arguments

• count - Maximum number of recent events to collect per listener (must be greater than 0)

Reorg behavior

The scanner can detect reorgs during the scan by periodically checking that the range tip has not changed. This is done only when the specified range tip is not a finalized block.

On reorg detection:

1. Emits Notification::ReorgDetected to all listeners
2. Resets to the updated tip
3. Reloads logs from the block range affected by the reorg
4. Continues until count events are collected

Final delivery to log listeners preserves chronological order regardless of reorgs.

Notes

For continuous streaming after collecting latest events, use EventScannerBuilder::sync().from_latest(count) instead

impl EventScannerBuilder<LatestEvents>

pub fn new(count: usize) -> Self

impl EventScannerBuilder<SyncFromLatestEvents>

pub fn new(count: usize) -> Self

impl EventScannerBuilder<SyncFromBlock>

pub fn new(from_block: impl Into<BlockId>) -> Self

impl<Mode> EventScannerBuilder<Mode>

pub fn max_block_range(self, max_block_range: u64) -> Self

Sets the maximum block range per event batch.

Controls how the scanner splits a large block range into smaller batches for processing. Each batch corresponds to a single RPC call to fetch logs. This prevents timeouts and respects rate limits imposed by node providers.

Arguments

• max_block_range - Maximum number of blocks to process per batch (must be greater than 0)

Example

If scanning events from blocks 1000–1099 (100 blocks total) with max_block_range(30):

• Batch 1: blocks 1000–1029 (30 blocks)
• Batch 2: blocks 1030–1059 (30 blocks)
• Batch 3: blocks 1060–1089 (30 blocks)
• Batch 4: blocks 1090–1099 (10 blocks)

pub fn past_blocks_storage_capacity( self, past_blocks_storage_capacity: RingBufferCapacity, ) -> Self

Sets how many of past blocks to keep in memory for reorg detection.

IMPORTANT: If zero, reorg detection is disabled.

Arguments

• past_blocks_storage_capacity - Maximum number of blocks to keep in memory.

pub fn buffer_capacity(self, buffer_capacity: usize) -> Self

Sets the stream buffer capacity.

Controls the maximum number of messages that can be buffered in the stream before backpressure is applied.

Arguments

• buffer_capacity - Maximum number of messages to buffer (must be greater than 0)

impl EventScannerBuilder<Historic>

pub fn from_block(self, block_id: impl Into<BlockId>) -> Self

Sets the starting block for the historic scan.

Note

Although passing BlockNumberOrTag::Pending will compile, the subsequent call to connect will fail at runtime. See issue https://github.com/OpenZeppelin/Event-Scanner/issues/244

pub fn to_block(self, block_id: impl Into<BlockId>) -> Self

Sets the starting block for the historic scan.

Note

Although passing BlockNumberOrTag::Pending will compile, the subsequent call to connect will fail at runtime. See issue https://github.com/OpenZeppelin/Event-Scanner/issues/244

pub fn max_concurrent_fetches(self, max_concurrent_fetches: usize) -> Self

Sets the maximum number of block-range fetches to process concurrently when scanning a historical block range.

Increasing this value can improve throughput by issuing multiple RPC requests concurrently, at the cost of higher load on the provider.

Note: This limit applies per listener. With N listeners and a limit of M, up to N × M concurrent RPC requests may be in-flight simultaneously.

Must be greater than 0.

Defaults to DEFAULT_MAX_CONCURRENT_FETCHES.

pub async fn connect<N: Network>( self, provider: impl IntoRobustProvider<N>, ) -> Result<EventScanner<Historic, N>, ScannerError>

Connects to an existing provider with block range validation.

Validates that the maximum of from_block and to_block does not exceed the latest block on the chain.

Errors

Returns an error if:

• The provider connection fails
• The specified block range exceeds the latest block on the chain
• The max block range is zero

impl EventScannerBuilder<LatestEvents>

pub fn block_confirmations(self, confirmations: u64) -> Self

Sets the number of confirmations required before a block is considered stable enough to include when collecting the latest events.

Higher values reduce the likelihood of emitting logs from blocks that are later reorged, at the cost of potentially excluding very recent events.

pub fn from_block(self, block_id: impl Into<BlockId>) -> Self

Sets the starting block for the historic scan.

Note

Although passing BlockNumberOrTag::Pending will compile, the subsequent call to connect will fail at runtime. See issue https://github.com/OpenZeppelin/Event-Scanner/issues/244

pub fn to_block(self, block_id: impl Into<BlockId>) -> Self

Sets the starting block for the historic scan.

Note

Although passing BlockNumberOrTag::Pending will compile, the subsequent call to connect will fail at runtime. See issue https://github.com/OpenZeppelin/Event-Scanner/issues/244

pub fn max_concurrent_fetches(self, max_concurrent_fetches: usize) -> Self

Sets the maximum number of block-range fetches to process concurrently when collecting the latest events.

Higher values can increase throughput by issuing multiple RPC requests concurrently, at the expense of more load on the provider.

Note: This limit applies per listener. With N listeners and a limit of M, up to N × M concurrent RPC requests may be in-flight simultaneously.

Must be greater than 0.

Defaults to DEFAULT_MAX_CONCURRENT_FETCHES.

pub async fn connect<N: Network>( self, provider: impl IntoRobustProvider<N>, ) -> Result<EventScanner<LatestEvents, N>, ScannerError>

Connects to an existing provider.

Errors

Returns an error if:

• The provider connection fails
• The event count is zero
• The max block range is zero

impl EventScannerBuilder<Live>

pub fn block_confirmations(self, confirmations: u64) -> Self

Sets the number of confirmations required before a block is considered stable enough to scan in live mode.

Higher values reduce the likelihood of emitting logs from blocks that are later reorged, at the cost of increased event delivery latency.

pub fn max_concurrent_fetches(self, max_concurrent_fetches: usize) -> Self

Sets the maximum number of block-range fetches to process concurrently for live streaming.

This knob primarily exists to handle an edge case: when the scanner falls significantly behind the head of the chain (for example due to a temporary outage or slow consumer), it can catch up faster by processing multiple block ranges in parallel. Increasing the value improves throughput at the expense of higher load on the provider.

Note: This limit applies per listener. With N listeners and a limit of M, up to N × M concurrent RPC requests may be in-flight simultaneously.

Must be greater than 0.

Defaults to DEFAULT_MAX_CONCURRENT_FETCHES.

pub async fn connect<N: Network>( self, provider: impl IntoRobustProvider<N>, ) -> Result<EventScanner<Live, N>, ScannerError>

Connects to an existing provider.

Errors

Returns an error if:

• The provider connection fails
• The max block range is zero

impl EventScannerBuilder<SyncFromBlock>

pub fn block_confirmations(self, confirmations: u64) -> Self

Sets the number of confirmations required before a block is considered stable enough to scan in the live phase.

This affects the post-sync live streaming phase; higher values reduce reorg risk at the cost of increased event delivery latency.

pub fn max_concurrent_fetches(self, max_concurrent_fetches: usize) -> Self

Sets the maximum number of block-range fetches to process concurrently when synchronizing from a specific block.

Higher values can improve throughput by issuing multiple RPC requests concurrently, at the cost of additional load on the provider.

Note: This limit applies per listener. With N listeners and a limit of M, up to N × M concurrent RPC requests may be in-flight simultaneously.

Must be greater than 0.

Defaults to DEFAULT_MAX_CONCURRENT_FETCHES.

pub async fn connect<N: Network>( self, provider: impl IntoRobustProvider<N>, ) -> Result<EventScanner<SyncFromBlock, N>, ScannerError>

Connects to an existing provider.

Errors

Returns an error if:

• The provider connection fails
• The max block range is zero

impl EventScannerBuilder<SyncFromLatestEvents>

pub fn block_confirmations(self, confirmations: u64) -> Self

Sets the number of confirmations required before a block is considered stable enough to scan in the live phase.

This affects the post-sync live streaming phase; higher values reduce reorg risk at the cost of increased event delivery latency.

pub fn max_concurrent_fetches(self, max_concurrent_fetches: usize) -> Self

Sets the maximum number of block-range fetches to process concurrently when fetching the latest events before switching to live streaming.

Increasing this value can improve catch-up throughput by issuing multiple RPC requests concurrently, at the cost of additional load on the provider.

Note: This limit applies per listener. With N listeners and a limit of M, up to N × M concurrent RPC requests may be in-flight simultaneously.

Must be greater than 0.

Defaults to DEFAULT_MAX_CONCURRENT_FETCHES.

pub async fn connect<N: Network>( self, provider: impl IntoRobustProvider<N>, ) -> Result<EventScanner<SyncFromLatestEvents, N>, ScannerError>

Connects to an existing provider.

Errors

Returns an error if:

• The provider connection fails
• The event count is zero
• The max block range is zero

impl EventScannerBuilder<Synchronize>

pub fn from_latest( self, count: usize, ) -> EventScannerBuilder<SyncFromLatestEvents>

Scans the latest count matching events per registered listener, then automatically transitions to live streaming mode.

This method combines two scanning phases into a single operation:

1. Latest events phase: Collects up to count most recent events by scanning backwards from the current chain tip. Events are delivered in chronological order.
2. Automatic transition: Emits Notification::SwitchingToLive to signal the mode change
3. Live streaming phase: Continuously monitors and streams new events as they arrive on-chain

Example

// Fetch the latest 10 events, then stream new events continuously
let provider = ProviderBuilder::new().connect("ws://localhost:8545").await?;
let robust_provider = RobustProviderBuilder::new(provider).build().await?;
let mut scanner = EventScannerBuilder::sync()
    .from_latest(10)
    .connect(robust_provider)
    .await?;

let filter = EventFilter::new().contract_address(contract_address);
let subscription = scanner.subscribe(filter);

let proof = scanner.start().await?;
let mut stream = subscription.stream(&proof);

while let Some(msg) = stream.next().await {
    match msg {
        Ok(Message::Data(logs)) => {
            println!("Received {} events", logs.len());
        }
        Ok(Message::Notification(notification)) => {
            println!("Notification received: {:?}", notification);
            // You'll see Notification::SwitchingToLive when transitioning
        }
        Err(e) => {
            eprintln!("Error: {}", e);
        }
    }
}

How it works

The scanner captures the latest block number before starting to establish a clear boundary between phases. The “latest events” phase scans from the current latest block to the genesis block, while the live phase starts from the block after the latest block. This design prevents duplicate events and handles race conditions where new blocks arrive during setup.

Key behaviors

• No duplicates: Events are not delivered twice across the phase transition
• Flexible count: If fewer than count events exist, returns all available events
• Concurrent log fetching: Logs are fetched concurrently to reduce the sync phase execution time. The maximum number of concurrent RPC calls is controlled by max_concurrent_fetches
• Reorg handling: Both phases handle reorgs appropriately:
  • Latest events phase: resets and rescans on reorg detection
  • Live phase: resets stream to the first post-reorg block that satisfies the configured block confirmations
• Continuous operation: Live phase continues indefinitely until the scanner is dropped

Notifications

During the latest events phase, the scanner can emit the following notification before transitioning to live mode:

• Notification::NoPastLogsFound: Emitted when no matching logs are found in the scanned range

After the latest events phase completes, Notification::SwitchingToLive is emitted before transitioning to the live streaming phase.

Arguments

• count - Maximum number of recent events to collect per listener before switching to live streaming (must be greater than 0)

Important notes

• The live phase continues indefinitely until the scanner is dropped or encounters an error

Detailed reorg behavior

• Latest events phase: Restart the scanner. On detecting a reorg, emits Notification::ReorgDetected, resets the rewind start to the new tip, and continues until collectors accumulate count logs. Final delivery to listeners preserves chronological order.
• Live streaming phase: Starts from latest_block + 1 and respects the configured block confirmations. On reorg, emits Notification::ReorgDetected, adjusts the next confirmed window (possibly re-emitting confirmed portions), and continues streaming.

pub fn from_block( self, block_id: impl Into<BlockId>, ) -> EventScannerBuilder<SyncFromBlock>

Streams events from a specific starting block to the present, then automatically transitions to live streaming mode.

This method combines two scanning phases into a single operation:

1. Historical sync phase: Streams events from from_block up to the current confirmed tip
2. Automatic transition: Emits Notification::SwitchingToLive to signal the mode change
3. Live streaming phase: Continuously monitors and streams new events as they arrive on-chain

Example

// Sync from block 1_000_000 to present, then stream new events
let provider = ProviderBuilder::new().connect("ws://localhost:8545").await?;
let robust_provider = RobustProviderBuilder::new(provider).build().await?;
let mut scanner = EventScannerBuilder::sync()
    .from_block(1_000_000)
    .connect(robust_provider)
    .await?;

let filter = EventFilter::new().contract_address(contract_address);
let subscription = scanner.subscribe(filter);

let proof = scanner.start().await?;
let mut stream = subscription.stream(&proof);

while let Some(msg) = stream.next().await {
    match msg {
        Ok(Message::Data(logs)) => {
            println!("Received {} events", logs.len());
        }
        Ok(Message::Notification(notification)) => {
            println!("Notification received: {:?}", notification);
            // You'll see Notification::SwitchingToLive when transitioning
        }
        Err(e) => {
            eprintln!("Error: {}", e);
        }
    }
}

Using block tags:

// Sync from genesis block
let provider = ProviderBuilder::new().connect("ws://localhost:8545").await?;
let robust_provider = RobustProviderBuilder::new(provider).build().await?;
let mut scanner = EventScannerBuilder::sync()
    .from_block(BlockNumberOrTag::Earliest)
    .connect(robust_provider)
    .await?;

How it works

The scanner first streams all events from the specified starting block up to the current confirmed tip (respecting block_confirmations). Once caught up, it seamlessly transitions to live mode and continues streaming new events as blocks are produced.

Key behaviors

• No duplicates: Events are not delivered twice across the phase transition
• Chronological order: Historical events are delivered oldest to newest
• Concurrent log fetching: Logs are fetched concurrently to reduce the sync phase execution time. The maximum number of concurrent RPC calls is controlled by max_concurrent_fetches
• Seamless transition: Automatically switches to live mode when caught up
• Continuous operation: Live phase continues indefinitely until the scanner is dropped
• Reorg detection: When a reorg is detected, Notification::ReorgDetected is emitted, the next confirmed window is adjusted to stream the reorged blocks, and continues streaming. While syncing, reorg checks are only performed for non-finalized blocks.

Arguments

• block_id - Starting block id

Important notes

The live phase continues indefinitely until the scanner is dropped or encounters an error.

Trait Implementations

impl<Mode: Debug> Debug for EventScannerBuilder<Mode>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<Mode: Default> Default for EventScannerBuilder<Mode>

fn default() -> EventScannerBuilder<Mode>

Returns the “default value” for a type.