Merge branch 'OpenZeppelin:main' into feat/benchmarks

Author: yug49

README.md

@@ -59,7 +59,7 @@ event-scanner = "0.9.0-alpha"
 Create an event stream for the given event filters registered with the `EventScanner`:
 
 ```rust
-use alloy::{network::Ethereum, providers::{Provider, ProviderBuilder}, sol_types::SolEvent};
+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};
@@ -125,8 +125,8 @@ Once configured, connect using:
 This will connect the `EventScanner` and allow you to create event streams and start scanning in various [modes](#scanning-modes).
 
 ```rust
-use alloy::providers::{Provider, ProviderBuilder};
-use event_scanner::robust_provider::RobustProviderBuilder;
+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?;
@@ -135,14 +135,16 @@ let provider = ProviderBuilder::new().connect("ws://localhost:8545").await?;
 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());
+    .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());
+    .connect(provider.clone())
+    .await?;
 
 // we can also wrap the provider in a RobustProvider
 // for more advanced configurations like retries and fallbacks
@@ -153,20 +155,23 @@ 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());
+    .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());
+    .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);
+    .connect(robust_provider)
+    .await?;
 ```
 
 Invoking `scanner.start()` starts the scanner in the specified mode.
@@ -176,25 +181,28 @@ Invoking `scanner.start()` starts the scanner in the specified mode.
 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`).
 
 ```rust
+use alloy::sol_types::SolEvent;
+use event_scanner::EventFilter;
+
 // Track a SPECIFIC event from a SPECIFIC contract
 let specific_filter = EventFilter::new()
-    .contract_address(*counter_contract.address())
-    .event(Counter::CountIncreased::SIGNATURE);
+    .contract_address(*my_contract.address())
+    .event(MyContract::SomeEvent::SIGNATURE);
 
 // Track multiple events from a SPECIFIC contract
 let specific_filter = EventFilter::new()
-    .contract_address(*counter_contract.address())
-    .event(Counter::CountIncreased::SIGNATURE)
-    .event(Counter::CountDecreased::SIGNATURE);
+    .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(Counter::CountIncreased::SIGNATURE);
+    .event(MyContract::SomeEvent::SIGNATURE);
 
 // Track ALL events from SPECIFIC contracts
 let all_contract_events_filter = EventFilter::new()
-    .contract_address(*counter_contract.address())
-    .contract_address(*other_counter_contract.address());
+    .contract_address(*my_contract.address())
+    .contract_address(*other_contract.address());
 
 // Track ALL events from ALL contracts
 let all_events_filter = EventFilter::new();
@@ -211,17 +219,17 @@ Batch builder examples:
 ```rust
 // Multiple contract addresses at once
 let multi_addr = EventFilter::new()
-    .contract_addresses([*counter_contract.address(), *other_counter_contract.address()]);
+    .contract_addresses([*my_contract.address(), *other_contract.address()]);
 
 // Multiple event names at once
 let multi_events = EventFilter::new()
-    .events([Counter::CountIncreased::SIGNATURE, Counter::CountDecreased::SIGNATURE]);
+    .events([MyContract::SomeEvent::SIGNATURE, MyContract::OtherEvent::SIGNATURE]);
 
 // Multiple event signature hashes at once
 let multi_sigs = EventFilter::new()
     .event_signatures([
-        Counter::CountIncreased::SIGNATURE_HASH,
-        Counter::CountDecreased::SIGNATURE_HASH,
+        MyContract::SomeEvent::SIGNATURE_HASH,
+        MyContract::OtherEvent::SIGNATURE_HASH,
     ]);
 ```
 
@@ -303,6 +311,9 @@ async fn example() -> anyhow::Result<()> {
         .subscription_timeout(Duration::from_secs(120))
         .build()
         .await?;
+
+    // ...
+
     Ok(())
 }
 ```

src/block_range_scanner/builder.rs

@@ -0,0 +1,101 @@
+use alloy::network::Network;
+
+use crate::{
+    ScannerError,
+    block_range_scanner::{
+        DEFAULT_MAX_BLOCK_RANGE, DEFAULT_STREAM_BUFFER_CAPACITY, RingBufferCapacity,
+        scanner::BlockRangeScanner,
+    },
+    robust_provider::IntoRobustProvider,
+};
+
+/// Builder/configuration for the block-range streaming service.
+#[derive(Clone, Debug)]
+pub struct BlockRangeScannerBuilder {
+    /// Maximum number of blocks per streamed range.
+    pub max_block_range: u64,
+    /// How many past block hashes to keep in memory for reorg detection.
+    ///
+    /// If set to `RingBufferCapacity::Limited(0)`, reorg detection is disabled.
+    pub past_blocks_storage_capacity: RingBufferCapacity,
+    pub buffer_capacity: usize,
+}
+
+impl Default for BlockRangeScannerBuilder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl BlockRangeScannerBuilder {
+    /// Creates a scanner with default configuration.
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            max_block_range: DEFAULT_MAX_BLOCK_RANGE,
+            past_blocks_storage_capacity: RingBufferCapacity::Limited(10),
+            buffer_capacity: DEFAULT_STREAM_BUFFER_CAPACITY,
+        }
+    }
+
+    /// Sets the maximum number of blocks per streamed range.
+    ///
+    /// This controls batching for historical scans and for catch-up in live/sync scanners.
+    ///
+    /// Must be greater than 0.
+    #[must_use]
+    pub fn max_block_range(mut self, max_block_range: u64) -> Self {
+        self.max_block_range = max_block_range;
+        self
+    }
+
+    /// Sets how many past block hashes to keep in memory for reorg detection.
+    ///
+    /// If set to `RingBufferCapacity::Limited(0)`, reorg detection is disabled.
+    #[must_use]
+    pub fn past_blocks_storage_capacity(
+        mut self,
+        past_blocks_storage_capacity: RingBufferCapacity,
+    ) -> Self {
+        self.past_blocks_storage_capacity = past_blocks_storage_capacity;
+        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)
+    #[must_use]
+    pub fn buffer_capacity(mut self, buffer_capacity: usize) -> Self {
+        self.buffer_capacity = buffer_capacity;
+        self
+    }
+
+    /// Connects to an existing provider
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the provider connection fails.
+    pub async fn connect<N: Network>(
+        self,
+        provider: impl IntoRobustProvider<N>,
+    ) -> Result<BlockRangeScanner<N>, ScannerError> {
+        if self.max_block_range == 0 {
+            return Err(ScannerError::InvalidMaxBlockRange);
+        }
+        if self.buffer_capacity == 0 {
+            return Err(ScannerError::InvalidBufferCapacity);
+        }
+        let provider = provider.into_robust_provider().await?;
+        Ok(BlockRangeScanner {
+            provider,
+            max_block_range: self.max_block_range,
+            past_blocks_storage_capacity: self.past_blocks_storage_capacity,
+            buffer_capacity: self.buffer_capacity,
+        })
+    }
+}

src/block_range_scanner/common.rs

@@ -1,11 +1,13 @@
+use std::ops::RangeInclusive;
+
 use tokio::sync::mpsc;
 use tokio_stream::StreamExt;
 
 use crate::{
-    ScannerError,
-    block_range_scanner::{BlockScannerResult, RangeIterator, reorg_handler::ReorgHandler},
+    ScannerError, ScannerMessage,
+    block_range_scanner::{range_iterator::RangeIterator, reorg_handler::ReorgHandler},
     robust_provider::{RobustProvider, RobustSubscription, subscription},
-    types::{Notification, TryStream},
+    types::{IntoScannerResult, Notification, ScannerResult, TryStream},
 };
 use alloy::{
     consensus::BlockHeader,
@@ -14,6 +16,39 @@ use alloy::{
     primitives::BlockNumber,
 };
 
+/// Default maximum number of blocks per streamed range.
+pub const DEFAULT_MAX_BLOCK_RANGE: u64 = 1000;
+
+/// Default confirmation depth used by scanners that accept a `block_confirmations` setting.
+pub const DEFAULT_BLOCK_CONFIRMATIONS: u64 = 0;
+
+/// Default per-stream buffer size used by scanners.
+pub const DEFAULT_STREAM_BUFFER_CAPACITY: usize = 50000;
+
+/// The result type yielded by block-range streams.
+pub type BlockScannerResult = ScannerResult<RangeInclusive<BlockNumber>>;
+
+/// Convenience alias for a streamed block-range message.
+pub type Message = ScannerMessage<RangeInclusive<BlockNumber>>;
+
+impl From<RangeInclusive<BlockNumber>> for Message {
+    fn from(range: RangeInclusive<BlockNumber>) -> Self {
+        Message::Data(range)
+    }
+}
+
+impl PartialEq<RangeInclusive<BlockNumber>> for Message {
+    fn eq(&self, other: &RangeInclusive<BlockNumber>) -> bool {
+        if let Message::Data(range) = self { range.eq(other) } else { false }
+    }
+}
+
+impl IntoScannerResult<RangeInclusive<BlockNumber>> for RangeInclusive<BlockNumber> {
+    fn into_scanner_message_result(self) -> BlockScannerResult {
+        Ok(Message::Data(self))
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn stream_live_blocks<N: Network>(
     stream_start: BlockNumber,

src/block_range_scanner/mod.rs

@@ -0,0 +1,17 @@
+mod builder;
+mod common;
+mod range_iterator;
+mod reorg_handler;
+mod rewind_handler;
+mod ring_buffer;
+mod scanner;
+mod sync_handler;
+
+pub use builder::BlockRangeScannerBuilder;
+pub use common::BlockScannerResult;
+pub use ring_buffer::RingBufferCapacity;
+pub use scanner::BlockRangeScanner;
+
+pub use common::{
+    DEFAULT_BLOCK_CONFIRMATIONS, DEFAULT_MAX_BLOCK_RANGE, DEFAULT_STREAM_BUFFER_CAPACITY,
+};