Merge branch 'main' into add-robust-subscription

Author: LeoPatOZ

README.md

@@ -94,8 +94,8 @@ async fn run_scanner(
             Message::Data(logs) => {
                 println!("Received {} logs: {logs:?}", logs.len());
             }
-            Message::Status(status) => {
-                println!("Status update: {status:?}");
+            Message::Notification(notification) => {
+                println!("Notification received: {notification:?}");
             }
             Message::Error(err) => {
                 eprintln!("Error: {err}");
@@ -226,7 +226,7 @@ let multi_sigs = EventFilter::new()
 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::Status(ScannerStatus)`** – Status notifications from the scanner:
+- **`Message::Notification(Notification)`** – Notifications from the scanner:
 - **`Message::Error(ScannerError)`** – Error notifications if the scanner encounters issues (e.g., RPC failures, connection problems)
 
 Always handle all message types in your stream processing loop to ensure robust error handling and proper reorg detection.

examples/historical_scanning/main.rs

@@ -77,7 +77,7 @@ async fn main() -> anyhow::Result<()> {
             Message::Error(e) => {
                 error!("Received error: {}", e);
             }
-            Message::Status(info) => {
+            Message::Notification(info) => {
                 info!("Received info: {:?}", info);
             }
         }

examples/latest_events_scanning/main.rs

@@ -78,8 +78,8 @@ async fn main() -> anyhow::Result<()> {
             Message::Error(e) => {
                 error!("Received error: {}", e);
             }
-            Message::Status(info) => {
-                info!("Received status: {:?}", info);
+            Message::Notification(info) => {
+                info!("Received notification: {:?}", info);
             }
         }
     }

examples/live_scanning/main.rs

@@ -77,7 +77,7 @@ async fn main() -> anyhow::Result<()> {
             Message::Error(e) => {
                 error!("Received error: {}", e);
             }
-            Message::Status(info) => {
+            Message::Notification(info) => {
                 info!("Received info: {:?}", info);
             }
         }

examples/sync_from_block_scanning/main.rs

@@ -101,8 +101,8 @@ async fn main() -> anyhow::Result<()> {
             Message::Error(e) => {
                 error!("Received error: {}", e);
             }
-            Message::Status(info) => {
-                info!("Received status: {:?}", info);
+            Message::Notification(info) => {
+                info!("Received notification: {:?}", info);
             }
         }
 

examples/sync_from_latest_scanning/main.rs

@@ -85,7 +85,7 @@ async fn main() -> anyhow::Result<()> {
             Message::Error(e) => {
                 error!("Received error: {}", e);
             }
-            Message::Status(info) => {
+            Message::Notification(info) => {
                 info!("Received info: {:?}", info);
             }
         }

src/block_range_scanner.rs

@@ -47,8 +47,8 @@
 //!                     }
 //!                 }
 //!             }
-//!             Message::Status(status) => {
-//!                 info!("Received status message: {:?}", status);
+//!             Message::Notification(notification) => {
+//!                 info!("Received notification: {:?}", notification);
 //!             }
 //!         }
 //!     }
@@ -71,7 +71,7 @@ use crate::{
     ScannerMessage,
     error::ScannerError,
     robust_provider::{Error as RobustProviderError, IntoRobustProvider, RobustProvider},
-    types::{ScannerStatus, TryStream},
+    types::{Notification, TryStream},
 };
 use alloy::{
     consensus::BlockHeader,
@@ -436,7 +436,7 @@ impl<N: Network> Service<N> {
 
             info!("Chain tip reached, switching to live");
 
-            if !sender.try_stream(ScannerStatus::SwitchingToLive).await {
+            if !sender.try_stream(Notification::SwitchingToLive).await {
                 return;
             }
 
@@ -533,15 +533,15 @@ impl<N: Network> Service<N> {
                 }
                 Err(e) => {
                     error!(error = %e, "Terminal RPC call error, shutting down");
-                    _ = sender.try_stream(e);
+                    _ = sender.try_stream(e).await;
                     return;
                 }
             };
 
             if reorged {
                 info!(block_number = %from, hash = %tip_hash, "Reorg detected");
 
-                if !sender.try_stream(ScannerStatus::ReorgDetected).await {
+                if !sender.try_stream(Notification::ReorgDetected).await {
                     break;
                 }
 
@@ -555,7 +555,7 @@ impl<N: Network> Service<N> {
                     }
                     Err(e) => {
                         error!(error = %e, "Terminal RPC call error, shutting down");
-                        _ = sender.try_stream(e);
+                        _ = sender.try_stream(e).await;
                         return;
                     }
                 };
@@ -637,7 +637,7 @@ impl<N: Network> Service<N> {
 
             if incoming_block_num < range_start {
                 warn!("Reorg detected: sending forked range");
-                if !sender.try_stream(ScannerStatus::ReorgDetected).await {
+                if !sender.try_stream(Notification::ReorgDetected).await {
                     return;
                 }
 
@@ -697,7 +697,7 @@ impl<N: Network> Service<N> {
                     }
                 }
                 other => {
-                    // Could be error or status
+                    // Could be error or notification
                     if !sender.try_stream(other).await {
                         break;
                     }

src/event_scanner/scanner/common.rs

@@ -133,9 +133,9 @@ pub fn spawn_log_consumers<N: Network>(
                             break;
                         }
                     }
-                    Ok(BlockRangeMessage::Status(status)) => {
-                        info!(status = ?status, "Received status message");
-                        if !sender.try_stream(status).await {
+                    Ok(BlockRangeMessage::Notification(notification)) => {
+                        info!(notification = ?notification, "Received notification");
+                        if !sender.try_stream(notification).await {
                             break;
                         }
                     }

src/event_scanner/scanner/mod.rs

@@ -167,8 +167,8 @@ impl EventScannerBuilder<Unspecified> {
     ///         Message::Data(logs) => {
     ///             println!("Received {} new events", logs.len());
     ///         }
-    ///         Message::Status(status) => {
-    ///             println!("Status: {:?}", status);
+    ///         Message::Notification(notification) => {
+    ///             println!("Notification received: {:?}", notification);
     ///         }
     ///         Message::Error(e) => {
     ///             eprintln!("Error: {}", e);
@@ -196,12 +196,12 @@ impl EventScannerBuilder<Unspecified> {
     /// # Reorg behavior
     ///
     /// When a reorg is detected:
-    /// 1. Emits [`ScannerStatus::ReorgDetected`][reorg] to all listeners
+    /// 1. Emits [`Notification::ReorgDetected`][reorg] 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
     ///
-    /// [reorg]: crate::types::ScannerStatus::ReorgDetected
+    /// [reorg]: crate::types::Notification::ReorgDetected
     #[must_use]
     pub fn live() -> EventScannerBuilder<Live> {
         EventScannerBuilder::default()
@@ -304,7 +304,7 @@ impl EventScannerBuilder<Unspecified> {
     ///
     /// During the scan, the scanner periodically checks the tip to detect reorgs. On reorg
     /// detection:
-    /// 1. Emits [`ScannerStatus::ReorgDetected`][reorg] to all listeners
+    /// 1. Emits [`Notification::ReorgDetected`][reorg] to all listeners
     /// 2. Resets to the updated tip
     /// 3. Restarts the scan from the new tip
     /// 4. Continues until `count` events are collected
@@ -319,7 +319,7 @@ impl EventScannerBuilder<Unspecified> {
     /// [subscribe]: EventScanner::subscribe
     /// [start]: EventScanner::start
     /// [sync_from_latest]: EventScannerBuilder::from_latest
-    /// [reorg]: crate::ScannerStatus::ReorgDetected
+    /// [reorg]: crate::Notification::ReorgDetected
     #[must_use]
     pub fn latest(count: usize) -> EventScannerBuilder<LatestEvents> {
         EventScannerBuilder::<LatestEvents>::new(count)

src/event_scanner/scanner/sync/from_latest.rs

@@ -9,7 +9,7 @@ use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 use tracing::info;
 
 use crate::{
-    EventScannerBuilder, ScannerError, ScannerStatus,
+    EventScannerBuilder, Notification, ScannerError,
     block_range_scanner::Message as BlockRangeMessage,
     event_scanner::{
         EventScanner,
@@ -106,7 +106,7 @@ impl<N: Network> EventScanner<SyncFromLatestEvents, N> {
             // Use a one-off channel for the notification.
             let (tx, rx) = mpsc::channel::<BlockRangeMessage>(1);
             let stream = ReceiverStream::new(rx);
-            tx.send(BlockRangeMessage::Status(ScannerStatus::SwitchingToLive))
+            tx.send(BlockRangeMessage::Notification(Notification::SwitchingToLive))
                 .await
                 .expect("receiver exists");