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

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())
     }
 }
 

src/event_scanner/scanner/mod.rs

@@ -39,9 +39,12 @@ use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
 
 use crate::{
-    BlockRangeScannerBuilder, EventFilter, ScannerError,
-    block_range_scanner::{BlockRangeScanner, DEFAULT_BLOCK_CONFIRMATIONS, RingBufferCapacity},
-    event_scanner::{EventScannerResult, listener::EventListener},
+    block_range_scanner::{
+        BlockRangeScanner, BlockRangeScannerBuilder, DEFAULT_BLOCK_CONFIRMATIONS,
+        RingBufferCapacity,
+    },
+    error::ScannerError,
+    event_scanner::{EventScannerResult, filter::EventFilter, listener::EventListener},
     robust_provider::IntoRobustProvider,
 };
 
@@ -191,9 +194,9 @@ impl EventScannerBuilder<Unspecified> {
     /// let mut scanner = EventScannerBuilder::historic().connect(robust_provider).await?;
     ///
     /// let filter = EventFilter::new().contract_address(contract_address);
-    /// let mut stream = scanner.subscribe(filter);
-    ///
-    /// scanner.start().await?;
+    /// 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());
@@ -268,9 +271,9 @@ impl EventScannerBuilder<Unspecified> {
     ///     .await?;
     ///
     /// let filter = EventFilter::new().contract_address(contract_address);
-    /// let mut stream = scanner.subscribe(filter);
-    ///
-    /// scanner.start().await?;
+    /// 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 {
@@ -358,9 +361,9 @@ impl EventScannerBuilder<Unspecified> {
     /// let mut scanner = EventScannerBuilder::latest(10).connect(robust_provider).await?;
     ///
     /// let filter = EventFilter::new().contract_address(contract_address);
-    /// let mut stream = scanner.subscribe(filter);
-    ///
-    /// scanner.start().await?;
+    /// 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 {
@@ -568,6 +571,58 @@ impl<Mode> EventScannerBuilder<Mode> {
     }
 }
 
+/// A subscription to scanner events that requires proof the scanner has started.
+///
+/// Created by [`EventScanner::subscribe()`](crate::EventScanner::subscribe), this type holds the
+/// underlying stream but prevents access until [`stream()`](EventSubscription::stream) is called
+/// with a valid [`StartProof`].
+///
+/// This pattern ensures at compile time that [`EventScanner::start()`](crate::EventScanner::start)
+/// is called before attempting to read from the event stream.
+///
+/// # Example
+///
+/// ```ignore
+/// let mut scanner = EventScannerBuilder::live().connect(provider).await?;
+///
+/// // Create subscription (cannot access stream yet)
+/// let subscription = scanner.subscribe(filter);
+///
+/// // Start scanner and get proof
+/// let proof = scanner.start().await?;
+///
+/// // Now access the stream with the proof
+/// let mut stream = subscription.stream(&proof);
+///
+/// while let Some(msg) = stream.next().await {
+///     // process events
+/// }
+/// ```
+pub struct EventSubscription {
+    inner: ReceiverStream<EventScannerResult>,
+}
+
+impl EventSubscription {
+    /// Creates a new subscription wrapping the given stream.
+    pub(crate) fn new(inner: ReceiverStream<EventScannerResult>) -> Self {
+        Self { inner }
+    }
+
+    /// Access the event stream.
+    ///
+    /// Requires a reference to a [`StartProof`] as proof that the scanner
+    /// has been started. The proof is obtained by calling
+    /// `EventScanner::start()`.
+    ///
+    /// # Arguments
+    ///
+    /// * `_proof` - Proof that the scanner has been started
+    #[must_use]
+    pub fn stream(self, _proof: &StartProof) -> ReceiverStream<EventScannerResult> {
+        self.inner
+    }
+}
+
 impl<Mode, N: Network> EventScanner<Mode, N> {
     /// Returns the configured stream buffer capacity.
     #[must_use]
@@ -593,11 +648,43 @@ impl<Mode, N: Network> EventScanner<Mode, N> {
     ///
     /// For scanner to properly stream events, register all subscriptions before calling `start()`.
     #[must_use]
-    pub fn subscribe(&mut self, filter: EventFilter) -> ReceiverStream<EventScannerResult> {
+    pub fn subscribe(&mut self, filter: EventFilter) -> EventSubscription {
         let (sender, receiver) =
             mpsc::channel::<EventScannerResult>(self.block_range_scanner.buffer_capacity());
         self.listeners.push(EventListener { filter, sender });
-        ReceiverStream::new(receiver)
+        EventSubscription::new(ReceiverStream::new(receiver))
+    }
+}
+
+/// Proof that the scanner has been started.
+///
+/// This proof is returned by `EventScanner::start()` and must be passed to
+/// [`EventSubscription::stream()`] to access the event stream. This ensures at compile
+/// time that the scanner is started before attempting to read events.
+///
+/// # Example
+///
+/// ```ignore
+/// let mut scanner = EventScannerBuilder::sync().from_block(0).connect(provider).await?;
+/// let subscription = scanner.subscribe(filter);
+///
+/// // Start the scanner and get the proof
+/// let proof = scanner.start().await?;
+///
+/// // Now we can access the stream
+/// let mut stream = subscription.stream(&proof);
+/// ```
+#[derive(Debug, Clone)]
+pub struct StartProof {
+    /// Private field prevents construction outside this crate
+    _private: (),
+}
+
+impl StartProof {
+    /// Creates a new start proof.
+    #[must_use]
+    pub(crate) fn new() -> Self {
+        Self { _private: () }
     }
 }
 

src/event_scanner/scanner/sync/from_block.rs

@@ -3,7 +3,7 @@ use alloy::{eips::BlockId, network::Network};
 use crate::{
     EventScannerBuilder, ScannerError,
     event_scanner::{
-        EventScanner, SyncFromBlock,
+        EventScanner, StartProof, SyncFromBlock,
         scanner::block_range_handler::{BlockRangeHandler, StreamHandler},
     },
     robust_provider::IntoRobustProvider,
@@ -75,7 +75,7 @@ impl<N: Network> EventScanner<SyncFromBlock, 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` 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,
             block_confirmations = self.config.block_confirmations,
@@ -101,7 +101,7 @@ impl<N: Network> EventScanner<SyncFromBlock, N> {
             handler.handle(stream).await;
         });
 
-        Ok(())
+        Ok(StartProof::new())
     }
 }