Scanner notififcation update (#192)

LeoPatOZ · web-flow · commit 25e6e06953d0 · 2025-11-19T22:01:41.000+09:00
diff --git a/README.md b/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.
diff --git a/examples/historical_scanning/main.rs b/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);
             }
         }
diff --git a/examples/latest_events_scanning/main.rs b/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);
             }
         }
     }
diff --git a/examples/live_scanning/main.rs b/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);
             }
         }
diff --git a/examples/sync_from_block_scanning/main.rs b/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);
             }
         }
 
diff --git a/examples/sync_from_latest_scanning/main.rs b/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);
             }
         }
diff --git a/src/block_range_scanner.rs b/src/block_range_scanner.rs
@@ -47,8 +47,8 @@
 //!                     }
 //!                 }
 //!             }
-//!             Message::Status(status) => {
-//!                 info!("Received status message: {:?}", status);
+//!             Message::Notification(notification) => {
+//!                 info!("Received notification: {:?}", notification);
 //!             }
 //!         }
 //!     }
@@ -70,7 +70,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;
             }
 
@@ -541,7 +541,7 @@ impl<N: Network> Service<N> {
             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;
                 }
 
@@ -624,7 +624,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;
                 }
 
@@ -684,7 +684,7 @@ impl<N: Network> Service<N> {
                     }
                 }
                 other => {
-                    // Could be error or status
+                    // Could be error or notification
                     if !sender.try_stream(other).await {
                         break;
                     }
diff --git a/src/event_scanner/scanner/common.rs b/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;
                         }
                     }
diff --git a/src/event_scanner/scanner/mod.rs b/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)
diff --git a/src/event_scanner/scanner/sync/from_latest.rs b/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");
 
diff --git a/src/event_scanner/scanner/sync/mod.rs b/src/event_scanner/scanner/sync/mod.rs
@@ -16,7 +16,7 @@ impl EventScannerBuilder<Synchronize> {
     ///
     /// 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 [`ScannerStatus::SwitchingToLive`][switch_to_live] to
+    /// 2. **Automatic transition**: Emits [`Notification::SwitchingToLive`][switch_to_live] to
     ///    signal the mode change
     /// 3. **Live streaming phase**: Continuously monitors and streams new events as they arrive
     ///    on-chain
@@ -48,9 +48,9 @@ impl EventScannerBuilder<Synchronize> {
     ///         Message::Data(logs) => {
     ///             println!("Received {} events", logs.len());
     ///         }
-    ///         Message::Status(status) => {
-    ///             println!("Status update: {:?}", status);
-    ///             // You'll see ScannerStatus::SwitchingToLive when transitioning
+    ///         Message::Notification(notification) => {
+    ///             println!("Notification received: {:?}", notification);
+    ///             // You'll see Notification::SwitchingToLive when transitioning
     ///         }
     ///         Message::Error(e) => {
     ///             eprintln!("Error: {}", e);
@@ -91,17 +91,17 @@ impl EventScannerBuilder<Synchronize> {
     /// # Detailed reorg behavior
     ///
     /// - **Latest events phase**: Restart the scanner. On detecting a reorg, emits
-    ///   [`ScannerStatus::ReorgDetected`][reorg], resets the rewind start to the new tip, and
+    ///   [`Notification::ReorgDetected`][reorg], 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 [`ScannerStatus::ReorgDetected`][reorg], adjusts the next
+    ///   confirmations. On reorg, emits [`Notification::ReorgDetected`][reorg], adjusts the next
     ///   confirmed window (possibly re-emitting confirmed portions), and continues streaming.
     ///
     /// [subscribe]: crate::EventScanner::subscribe
     /// [start]: crate::event_scanner::EventScanner::start
-    /// [reorg]: crate::types::ScannerStatus::ReorgDetected
-    /// [switch_to_live]: crate::types::ScannerStatus::SwitchingToLive
+    /// [reorg]: crate::types::Notification::ReorgDetected
+    /// [switch_to_live]: crate::types::Notification::SwitchingToLive
     #[must_use]
     pub fn from_latest(self, count: usize) -> EventScannerBuilder<SyncFromLatestEvents> {
         EventScannerBuilder::<SyncFromLatestEvents>::new(count)
@@ -114,7 +114,7 @@ impl EventScannerBuilder<Synchronize> {
     ///
     /// 1. **Historical sync phase**: Streams events from `from_block` up to the current confirmed
     ///    tip
-    /// 2. **Automatic transition**: Emits [`ScannerStatus::SwitchingToLive`][switch_to_live] to
+    /// 2. **Automatic transition**: Emits [`Notification::SwitchingToLive`][switch_to_live] to
     ///    signal the mode change
     /// 3. **Live streaming phase**: Continuously monitors and streams new events as they arrive
     ///    on-chain
@@ -146,9 +146,9 @@ impl EventScannerBuilder<Synchronize> {
     ///         Message::Data(logs) => {
     ///             println!("Received {} events", logs.len());
     ///         }
-    ///         Message::Status(status) => {
-    ///             println!("Status update: {:?}", status);
-    ///             // You'll see ScannerStatus::SwitchingToLive when transitioning
+    ///         Message::Notification(notification) => {
+    ///             println!("Notification received: {:?}", notification);
+    ///             // You'll see Notification::SwitchingToLive when transitioning
     ///         }
     ///         Message::Error(e) => {
     ///             eprintln!("Error: {}", e);
@@ -202,13 +202,13 @@ impl EventScannerBuilder<Synchronize> {
     ///
     /// - **Historical sync phase**: Streams events in chronological order without reorg detection
     /// - **Live streaming phase**: Respects the configured block confirmations. On reorg, emits
-    ///   [`ScannerStatus::ReorgDetected`][reorg], adjusts the next confirmed window (possibly
+    ///   [`Notification::ReorgDetected`][reorg], adjusts the next confirmed window (possibly
     ///   re-emitting confirmed portions), and continues streaming.
     ///
     /// [subscribe]: crate::EventScanner::subscribe
     /// [start]: crate::event_scanner::EventScanner::start
-    /// [reorg]: crate::types::ScannerStatus::ReorgDetected
-    /// [switch_to_live]: crate::types::ScannerStatus::SwitchingToLive
+    /// [reorg]: crate::types::Notification::ReorgDetected
+    /// [switch_to_live]: crate::types::Notification::SwitchingToLive
     #[must_use]
     pub fn from_block(
         self,
diff --git a/src/lib.rs b/src/lib.rs
@@ -9,7 +9,7 @@ mod event_scanner;
 mod types;
 
 pub use error::ScannerError;
-pub use types::{ScannerMessage, ScannerStatus};
+pub use types::{Notification, ScannerMessage};
 
 pub use event_scanner::{
     EventFilter, EventScanner, EventScannerBuilder, Historic, LatestEvents, Live, Message,
diff --git a/src/test_utils/macros.rs b/src/test_utils/macros.rs
@@ -257,8 +257,8 @@ pub async fn assert_event_sequence<S: Stream<Item = Message> + Unpin>(
 /// * **Gap or overlap**: A streamed range doesn't start exactly at the next expected block number,
 ///   indicating a gap or overlap in coverage.
 /// * **Out of bounds**: A streamed range extends beyond the end of the expected range.
-/// * **Wrong message type**: The stream yields a non-`Data` message (e.g., `Error` or `Status`)
-///   when a block range is expected.
+/// * **Wrong message type**: The stream yields a non-`Data` message (e.g., `Error` or
+///   `Notification`) when a block range is expected.
 /// * **Stream closed early**: The stream ends before the entire expected range is covered.
 ///
 /// On panic, the error message includes the expected remaining range and all previously
diff --git a/src/types.rs b/src/types.rs
@@ -7,24 +7,28 @@ use tracing::{info, warn};
 pub enum ScannerMessage<T: Clone, E: Error + Clone> {
     Data(T),
     Error(E),
-    Status(ScannerStatus),
+    Notification(Notification),
 }
 
 #[derive(Copy, Debug, Clone, PartialEq)]
-pub enum ScannerStatus {
+pub enum Notification {
     SwitchingToLive,
     ReorgDetected,
 }
 
-impl<T: Clone, E: Error + Clone> From<ScannerStatus> for ScannerMessage<T, E> {
-    fn from(value: ScannerStatus) -> Self {
-        ScannerMessage::Status(value)
+impl<T: Clone, E: Error + Clone> From<Notification> for ScannerMessage<T, E> {
+    fn from(value: Notification) -> Self {
+        ScannerMessage::Notification(value)
     }
 }
 
-impl<T: Clone, E: Error + Clone> PartialEq<ScannerStatus> for ScannerMessage<T, E> {
-    fn eq(&self, other: &ScannerStatus) -> bool {
-        if let ScannerMessage::Status(status) = self { status == other } else { false }
+impl<T: Clone, E: Error + Clone> PartialEq<Notification> for ScannerMessage<T, E> {
+    fn eq(&self, other: &Notification) -> bool {
+        if let ScannerMessage::Notification(notification) = self {
+            notification == other
+        } else {
+            false
+        }
     }
 }
 
diff --git a/tests/block_range_scanner.rs b/tests/block_range_scanner.rs
@@ -5,7 +5,7 @@ use alloy::{
 };
 use alloy_node_bindings::Anvil;
 use event_scanner::{
-    ScannerError, ScannerStatus, assert_closed, assert_empty, assert_next, assert_range_coverage,
+    Notification, ScannerError, assert_closed, assert_empty, assert_next, assert_range_coverage,
     block_range_scanner::BlockRangeScanner,
 };
 
@@ -131,7 +131,7 @@ async fn shallow_block_confirmation_does_not_mitigate_reorg() -> anyhow::Result<
     // mint additional blocks
     provider.anvil_mine(Some(3), None).await?;
 
-    assert_next!(stream, ScannerStatus::ReorgDetected);
+    assert_next!(stream, Notification::ReorgDetected);
     assert_range_coverage!(stream, 0..=10);
     assert_empty!(stream);
 
@@ -163,7 +163,7 @@ async fn historical_emits_correction_range_when_reorg_below_end() -> anyhow::Res
     assert_next!(stream, 30..=59);
     assert_next!(stream, 60..=89);
     assert_next!(stream, 90..=110);
-    assert_next!(stream, ScannerStatus::ReorgDetected);
+    assert_next!(stream, Notification::ReorgDetected);
     assert_next!(stream, 105..=110);
     assert_closed!(stream);
 
@@ -197,7 +197,7 @@ async fn historical_emits_correction_range_when_end_num_reorgs() -> anyhow::Resu
     assert_next!(stream, 30..=59);
     assert_next!(stream, 60..=89);
     assert_next!(stream, 90..=120);
-    assert_next!(stream, ScannerStatus::ReorgDetected);
+    assert_next!(stream, Notification::ReorgDetected);
     assert_next!(stream, 120..=120);
     assert_closed!(stream);
 
diff --git a/tests/live/reorg.rs b/tests/live/reorg.rs
diff --git a/tests/sync/from_block.rs b/tests/sync/from_block.rs
diff --git a/tests/sync/from_latest.rs b/tests/sync/from_latest.rs

Original file line number	Diff line number	Diff line change
`@@ -77,7 +77,7 @@ async fn main() -> anyhow::Result<()> {`
`77`	`77`	`Message::Error(e) => {`
`78`	`78`	`error!("Received error: {}", e);`
`79`	`79`	`}`
`80`		`- Message::Status(info) => {`
	`80`	`+ Message::Notification(info) => {`
`81`	`81`	`info!("Received info: {:?}", info);`
`82`	`82`	`}`
`83`	`83`	`}`
Original file line number	Diff line number	Diff line change
`@@ -78,8 +78,8 @@ async fn main() -> anyhow::Result<()> {`
`78`	`78`	`Message::Error(e) => {`
`79`	`79`	`error!("Received error: {}", e);`
`80`	`80`	`}`
`81`		`- Message::Status(info) => {`
`82`		`- info!("Received status: {:?}", info);`
	`81`	`+ Message::Notification(info) => {`
	`82`	`+ info!("Received notification: {:?}", info);`
`83`	`83`	`}`
`84`	`84`	`}`
`85`	`85`	`}`
Original file line number	Diff line number	Diff line change
`@@ -101,8 +101,8 @@ async fn main() -> anyhow::Result<()> {`
`101`	`101`	`Message::Error(e) => {`
`102`	`102`	`error!("Received error: {}", e);`
`103`	`103`	`}`
`104`		`- Message::Status(info) => {`
`105`		`- info!("Received status: {:?}", info);`
	`104`	`+ Message::Notification(info) => {`
	`105`	`+ info!("Received notification: {:?}", info);`
`106`	`106`	`}`
`107`	`107`	`}`
`108`	`108`
Original file line number	Diff line number	Diff line change
`@@ -85,7 +85,7 @@ async fn main() -> anyhow::Result<()> {`
`85`	`85`	`Message::Error(e) => {`
`86`	`86`	`error!("Received error: {}", e);`
`87`	`87`	`}`
`88`		`- Message::Status(info) => {`
	`88`	`+ Message::Notification(info) => {`
`89`	`89`	`info!("Received info: {:?}", info);`
`90`	`90`	`}`
`91`	`91`	`}`
Original file line number	Diff line number	Diff line change
`@@ -133,9 +133,9 @@ pub fn spawn_log_consumers<N: Network>(`
`133`	`133`	`break;`
`134`	`134`	`}`
`135`	`135`	`}`
`136`		`- Ok(BlockRangeMessage::Status(status)) => {`
`137`		`- info!(status = ?status, "Received status message");`
`138`		`- if !sender.try_stream(status).await {`
	`136`	`+ Ok(BlockRangeMessage::Notification(notification)) => {`
	`137`	`+ info!(notification = ?notification, "Received notification");`
	`138`	`+ if !sender.try_stream(notification).await {`
`139`	`139`	`break;`
`140`	`140`	`}`
`141`	`141`	`}`