Accessing event stream now depends on scanner.start() being Invoked (#220)

Co-authored-by: 0xNeshi <[email protected]>

Author: LeoPatOZ

README.md

@@ -85,10 +85,13 @@ async fn run_scanner(
         .contract_address(contract)
         .event(MyContract::SomeEvent::SIGNATURE);
 
-    let mut stream = scanner.subscribe(filter);
+    let subscription = scanner.subscribe(filter);
 
-    // Start the scanner
-    scanner.start().await?;
+    // Start the scanner and get the proof
+    let proof = scanner.start().await?;
+
+    // Access the stream using the proof
+    let mut stream = subscription.stream(&proof);
 
     // Process messages from the stream
     while let Some(message) = stream.next().await {
@@ -174,7 +177,7 @@ let scanner = EventScannerBuilder::sync()
     .await?;
 ```
 
-Invoking `scanner.start()` starts the scanner in the specified mode.
+Invoking `scanner.start()` starts the scanner in the specified mode and returns a `StartProof` that must be passed to `subscription.stream()` to access the event stream. This compile-time guarantee ensures the scanner is started before attempting to read events.
 
 ### Defining Event Filters
 

examples/historical_scanning/main.rs

@@ -63,9 +63,12 @@ async fn main() -> anyhow::Result<()> {
 
     let mut scanner = EventScannerBuilder::historic().connect(robust_provider).await?;
 
-    let mut stream = scanner.subscribe(increase_filter);
+    let subscription = scanner.subscribe(increase_filter);
 
-    scanner.start().await.expect("failed to start scanner");
+    let proof = scanner.start().await.expect("failed to start scanner");
+
+    // Access the stream using the proof (proves scanner is started)
+    let mut stream = subscription.stream(&proof);
 
     while let Some(message) = stream.next().await {
         match message {

examples/latest_events_scanning/main.rs

@@ -60,13 +60,16 @@ async fn main() -> anyhow::Result<()> {
 
     let mut scanner = EventScannerBuilder::latest(5).connect(robust_provider).await?;
 
-    let mut stream = scanner.subscribe(increase_filter);
+    let subscription = scanner.subscribe(increase_filter);
 
     for _ in 0..8 {
         _ = counter_contract.increase().send().await?;
     }
 
-    scanner.start().await?;
+    let proof = scanner.start().await?;
+
+    // Access the stream using the proof (proves scanner is started)
+    let mut stream = subscription.stream(&proof);
 
     while let Some(message) = stream.next().await {
         match message {

examples/live_scanning/main.rs

@@ -61,9 +61,12 @@ async fn main() -> anyhow::Result<()> {
 
     let mut scanner = EventScannerBuilder::live().connect(robust_provider).await?;
 
-    let mut stream = scanner.subscribe(increase_filter);
+    let subscription = scanner.subscribe(increase_filter);
 
-    scanner.start().await.expect("failed to start scanner");
+    let proof = scanner.start().await.expect("failed to start scanner");
+
+    // Access the stream using the proof (proves scanner is started)
+    let mut stream = subscription.stream(&proof);
 
     _ = counter_contract.increase().send().await?;
 

examples/sync_from_block_scanning/main.rs

@@ -69,10 +69,13 @@ async fn main() -> anyhow::Result<()> {
 
     let mut scanner = EventScannerBuilder::sync().from_block(0).connect(robust_provider).await?;
 
-    let mut stream = scanner.subscribe(increase_filter);
+    let subscription = scanner.subscribe(increase_filter);
 
     info!("Starting sync scanner...");
-    scanner.start().await.expect("failed to start scanner");
+    let proof = scanner.start().await.expect("failed to start scanner");
+
+    // Access the stream using the proof (proves scanner is started)
+    let mut stream = subscription.stream(&proof);
 
     info!("Creating live events...");
     for i in 0..2 {

examples/sync_from_latest_scanning/main.rs

@@ -61,13 +61,16 @@ async fn main() -> anyhow::Result<()> {
 
     let mut client = EventScannerBuilder::sync().from_latest(5).connect(robust_provider).await?;
 
-    let mut stream = client.subscribe(increase_filter);
+    let subscription = client.subscribe(increase_filter);
 
     for _ in 0..10 {
         _ = counter_contract.increase().send().await?;
     }
 
-    client.start().await.expect("failed to start scanner");
+    let proof = client.start().await.expect("failed to start scanner");
+
+    // Access the stream using the proof (proves scanner is started)
+    let mut stream = subscription.stream(&proof);
 
     // emit some events for live mode to pick up
     _ = counter_contract.increase().send().await?;

src/event_scanner/mod.rs

@@ -16,6 +16,6 @@ mod scanner;
 pub use filter::EventFilter;
 pub use message::{EventScannerResult, Message};
 pub use scanner::{
-    DEFAULT_MAX_CONCURRENT_FETCHES, EventScanner, EventScannerBuilder, Historic, LatestEvents,
-    Live, SyncFromBlock, SyncFromLatestEvents, block_range_handler,
+    DEFAULT_MAX_CONCURRENT_FETCHES, EventScanner, EventScannerBuilder, EventSubscription, Historic,
+    LatestEvents, Live, StartProof, SyncFromBlock, SyncFromLatestEvents, block_range_handler,
 };

src/event_scanner/scanner/historic.rs

@@ -7,7 +7,10 @@ use alloy::{
 use super::block_range_handler::StreamHandler;
 use crate::{
     EventScannerBuilder, ScannerError,
-    event_scanner::scanner::{EventScanner, Historic, block_range_handler::BlockRangeHandler},
+    event_scanner::{
+        StartProof,
+        scanner::{EventScanner, Historic, block_range_handler::BlockRangeHandler},
+    },
     robust_provider::IntoRobustProvider,
 };
 
@@ -133,7 +136,7 @@ impl<N: Network> EventScanner<Historic, N> {
     /// * [`ScannerError::Timeout`] - if an RPC call required for startup times out.
     /// * [`ScannerError::RpcError`] - if an RPC call required for startup fails.
     /// * [`ScannerError::BlockNotFound`] - if `from_block` or `to_block` cannot be resolved.
-    pub async fn start(self) -> Result<(), ScannerError> {
+    pub async fn start(self) -> Result<StartProof, ScannerError> {
         info!(
             from_block = ?self.config.from_block,
             to_block = ?self.config.to_block,
@@ -159,7 +162,7 @@ impl<N: Network> EventScanner<Historic, N> {
             handler.handle(stream).await;
         });
 
-        Ok(())
+        Ok(StartProof::new())
     }
 }
 

src/event_scanner/scanner/latest.rs

@@ -7,7 +7,7 @@ use alloy::{
 use crate::{
     EventScannerBuilder, ScannerError,
     event_scanner::{
-        EventScanner, LatestEvents,
+        EventScanner, LatestEvents, StartProof,
         scanner::block_range_handler::{BlockRangeHandler, LatestEventsHandler},
     },
     robust_provider::IntoRobustProvider,
@@ -146,7 +146,15 @@ impl<N: Network> EventScanner<LatestEvents, N> {
     /// * [`ScannerError::Timeout`] - if an RPC call required for startup times out.
     /// * [`ScannerError::RpcError`] - if an RPC call required for startup fails.
     /// * [`ScannerError::BlockNotFound`] - if `from_block` or `to_block` cannot be resolved.
-    pub async fn start(self) -> Result<(), ScannerError> {
+    pub async fn start(self) -> Result<StartProof, ScannerError> {
+        info!(
+            from_block = ?self.config.from_block,
+            to_block = ?self.config.to_block,
+            count = ?self.config.count,
+            listener_count = self.listeners.len(),
+            "Starting EventScanner in LatestEvents mode"
+        );
+
         let stream = self
             .block_range_scanner
             .stream_rewind(self.config.from_block, self.config.to_block)
@@ -166,7 +174,7 @@ impl<N: Network> EventScanner<LatestEvents, N> {
             handler.handle(stream).await;
         });
 
-        Ok(())
+        Ok(StartProof::new())
     }
 }
 

src/event_scanner/scanner/live.rs

@@ -3,7 +3,7 @@ use alloy::network::Network;
 use crate::{
     EventScannerBuilder, ScannerError,
     event_scanner::{
-        EventScanner,
+        EventScanner, StartProof,
         scanner::{
             Live,
             block_range_handler::{BlockRangeHandler, StreamHandler},
@@ -72,7 +72,7 @@ impl<N: Network> EventScanner<Live, N> {
     ///
     /// * [`ScannerError::Timeout`] - if an RPC call required for startup times out.
     /// * [`ScannerError::RpcError`] - if an RPC call required for startup fails.
-    pub async fn start(self) -> Result<(), ScannerError> {
+    pub async fn start(self) -> Result<StartProof, ScannerError> {
         info!(
             block_confirmations = self.config.block_confirmations,
             listener_count = self.listeners.len(),
@@ -93,7 +93,7 @@ impl<N: Network> EventScanner<Live, N> {
             handler.handle(stream).await;
         });
 
-        Ok(())
+        Ok(StartProof::new())
     }
 }