ref: introduce ChannelState enum for explicit channel closure handling (#223)

<!-- Append the issue number -->
Resolves #222 

## Description

This PR addresses the issue of ambiguous channel closure handling by
introducing a `ChannelState` enum with explicit `Open` and `Closed`
variants that makes the code intent clear:

```rust
pub enum ChannelState {
    /// The channel is open and the message was successfully sent.
    Open,
    /// The channel is closed (receiver dropped), no further messages can be sent.
    Closed,
}
```
### Before vs After

**Before:**
```rust
if !sender.try_stream(Notification::ReorgDetected).await {
    return; // Channel closed
}
```

**After:**
```rust
if sender.try_stream(Notification::ReorgDetected).await.is_closed() {
    return;
}
```

### Testing

- Added comprehensive tests for `ChannelState` enum
- Added tests for `TryStream` trait behavior with open/closed channels

---

Author: yug49

src/block_range_scanner/common.rs

@@ -7,7 +7,7 @@ use tokio_stream::StreamExt;
 use crate::{
     ScannerError, ScannerMessage,
     block_range_scanner::{range_iterator::RangeIterator, reorg_handler::ReorgHandler},
-    types::{IntoScannerResult, Notification, ScannerResult, TryStream},
+    types::{ChannelState, IntoScannerResult, Notification, ScannerResult, TryStream},
 };
 use alloy::{
     consensus::BlockHeader,
@@ -83,7 +83,9 @@ pub(crate) async fn stream_live_blocks<N: Network>(
     // only once the first relevant block is received from the subscription, and not before that;
     // otherwise callers might perform certain operations expecting the relevant blocks to start
     // coming, when in fact they are not.
-    if notify_after_first_block && !sender.try_stream(Notification::SwitchingToLive).await {
+    if notify_after_first_block &&
+        sender.try_stream(Notification::SwitchingToLive).await.is_closed()
+    {
         return;
     }
 
@@ -258,7 +260,8 @@ async fn stream_blocks_continuously<
         };
 
         if let Some(common_ancestor) = common_ancestor {
-            if !handle_reorg_detected(common_ancestor, stream_start, state, sender).await {
+            if handle_reorg_detected(common_ancestor, stream_start, state, sender).await.is_closed()
+            {
                 return; // Channel closed
             }
         } else {
@@ -268,7 +271,7 @@ async fn stream_blocks_continuously<
 
         // Stream the next batch of confirmed blocks
         let batch_end_num = incoming_block.saturating_sub(block_confirmations);
-        if !stream_next_batch(
+        if stream_next_batch(
             batch_end_num,
             state,
             stream_start,
@@ -278,20 +281,21 @@ async fn stream_blocks_continuously<
             reorg_handler,
         )
         .await
+        .is_closed()
         {
             return; // Channel closed
         }
     }
 }
 
-/// Handles a detected reorg by notifying and adjusting the streaming state
-/// Returns false if the channel is closed
+/// Handles a detected reorg by notifying and adjusting the streaming state.
+/// Returns `ChannelState::Closed` if the channel is closed, `ChannelState::Open` otherwise.
 async fn handle_reorg_detected<N: Network>(
     common_ancestor: N::BlockResponse,
     stream_start: BlockNumber,
     state: &mut LiveStreamingState<N>,
     sender: &mpsc::Sender<BlockScannerResult>,
-) -> bool {
+) -> ChannelState {
     let ancestor_num = common_ancestor.header().number();
 
     info!(
@@ -300,8 +304,11 @@ async fn handle_reorg_detected<N: Network>(
         "Reorg detected during live streaming"
     );
 
-    if !sender.try_stream(Notification::ReorgDetected { common_ancestor: ancestor_num }).await {
-        return false;
+    let channel_state =
+        sender.try_stream(Notification::ReorgDetected { common_ancestor: ancestor_num }).await;
+
+    if channel_state.is_closed() {
+        return ChannelState::Closed;
     }
 
     // Reset streaming position based on common ancestor
@@ -325,11 +332,11 @@ async fn handle_reorg_detected<N: Network>(
         state.previous_batch_end = Some(common_ancestor);
     }
 
-    true
+    ChannelState::Open
 }
 
 /// Streams the next batch of blocks up to `batch_end_num`.
-/// Returns false if the channel is closed
+/// Returns `ChannelState::Closed` if the channel is closed, `ChannelState::Open` otherwise.
 async fn stream_next_batch<N: Network>(
     batch_end_num: BlockNumber,
     state: &mut LiveStreamingState<N>,
@@ -338,10 +345,10 @@ async fn stream_next_batch<N: Network>(
     sender: &mpsc::Sender<BlockScannerResult>,
     provider: &RobustProvider<N>,
     reorg_handler: &mut ReorgHandler<N>,
-) -> bool {
+) -> ChannelState {
     if batch_end_num < state.batch_start {
         // No new confirmed blocks to stream yet
-        return true;
+        return ChannelState::Open;
     }
 
     // The minimum common ancestor is the block before the stream start
@@ -360,13 +367,13 @@ async fn stream_next_batch<N: Network>(
 
     if state.previous_batch_end.is_none() {
         // Channel closed
-        return false;
+        return ChannelState::Closed;
     }
 
     // SAFETY: Overflow cannot realistically happen
     state.batch_start = batch_end_num + 1;
 
-    true
+    ChannelState::Open
 }
 
 /// Tracks the current state of live streaming
@@ -413,7 +420,7 @@ pub(crate) async fn stream_historical_range<N: Network>(
 
     for range in RangeIterator::forward(start, finalized_batch_end, max_block_range) {
         trace!(range_start = *range.start(), range_end = *range.end(), "Streaming finalized range");
-        if !sender.try_stream(range).await {
+        if sender.try_stream(range).await.is_closed() {
             return None; // channel closed
         }
     }
@@ -484,7 +491,7 @@ pub(crate) async fn stream_range_with_reorg_handling<N: Network>(
             }
         };
 
-        if !sender.try_stream(batch).await {
+        if sender.try_stream(batch).await.is_closed() {
             return None; // channel closed
         }
 
@@ -503,7 +510,8 @@ pub(crate) async fn stream_range_with_reorg_handling<N: Network>(
                 common_ancestor = common_ancestor,
                 "Reorg detected during historical streaming, resetting range iterator"
             );
-            if !sender.try_stream(Notification::ReorgDetected { common_ancestor }).await {
+            if sender.try_stream(Notification::ReorgDetected { common_ancestor }).await.is_closed()
+            {
                 return None;
             }
             let reset_to = (common_ancestor + 1).max(min_common_ancestor);

src/block_range_scanner/rewind_handler.rs

@@ -15,7 +15,7 @@ use crate::{
         common::BlockScannerResult, range_iterator::RangeIterator, reorg_handler::ReorgHandler,
         ring_buffer::RingBufferCapacity,
     },
-    types::TryStream,
+    types::{ChannelState, TryStream},
 };
 
 pub(crate) struct RewindHandler<N: Network> {
@@ -126,7 +126,7 @@ impl<N: Network> RewindHandler<N> {
         let mut iter = RangeIterator::reverse(from, to, max_block_range);
         for range in &mut iter {
             // stream the range regularly, i.e. from smaller block number to greater
-            if !sender.try_stream(range).await {
+            if sender.try_stream(range).await.is_closed() {
                 break;
             }
 
@@ -146,14 +146,15 @@ impl<N: Network> RewindHandler<N> {
                         tip = tip.header().number(),
                         "Reorg detected during rewind, rescanning affected blocks"
                     );
-                    if !Self::handle_reorg_rescan(
+                    if Self::handle_reorg_rescan(
                         &mut tip,
                         common_ancestor,
                         max_block_range,
                         sender,
                         provider,
                     )
                     .await
+                    .is_closed()
                     {
                         return;
                     }
@@ -164,14 +165,15 @@ impl<N: Network> RewindHandler<N> {
 
     /// Handles re-scanning of reorged blocks.
     ///
-    /// Returns `true` on success, `false` if stream closed or terminal error occurred.
+    /// Returns `ChannelState::Closed` if stream closed or terminal error occurred,
+    /// `ChannelState::Open` on success.
     async fn handle_reorg_rescan(
         tip: &mut N::BlockResponse,
         common_ancestor: N::BlockResponse,
         max_block_range: u64,
         sender: &mpsc::Sender<BlockScannerResult>,
         provider: &RobustProvider<N>,
-    ) -> bool {
+    ) -> ChannelState {
         let tip_number = tip.header().number();
         let common_ancestor = common_ancestor.header().number();
 
@@ -182,8 +184,8 @@ impl<N: Network> RewindHandler<N> {
             "Rescanning reorged blocks"
         );
 
-        if !sender.try_stream(Notification::ReorgDetected { common_ancestor }).await {
-            return false;
+        if sender.try_stream(Notification::ReorgDetected { common_ancestor }).await.is_closed() {
+            return ChannelState::Closed;
         }
 
         // Get the new tip block (same height as original tip, but new hash)
@@ -203,7 +205,7 @@ impl<N: Network> RewindHandler<N> {
                     );
                 }
                 _ = sender.try_stream(e).await;
-                return false;
+                return ChannelState::Closed;
             }
         };
 
@@ -212,11 +214,11 @@ impl<N: Network> RewindHandler<N> {
 
         for batch in RangeIterator::forward(rescan_from, tip_number, max_block_range) {
             trace!(range_start = *batch.start(), range_end = *batch.end(), "Rescanning batch");
-            if !sender.try_stream(batch).await {
-                return false;
+            if sender.try_stream(batch).await.is_closed() {
+                return ChannelState::Closed;
             }
         }
 
-        true
+        ChannelState::Open
     }
 }

src/block_range_scanner/sync_handler.rs

@@ -198,7 +198,7 @@ impl<N: Network> SyncHandler<N> {
             }
         };
 
-        if !sender.try_stream(Notification::SwitchingToLive).await {
+        if sender.try_stream(Notification::SwitchingToLive).await.is_closed() {
             debug!("Channel closed before live streaming could start");
             return;
         }

src/event_scanner/block_range_handler.rs

@@ -123,7 +123,7 @@ impl<N: Network> StreamHandler<N> {
                             continue;
                         }
 
-                        if !listener.sender.try_stream(result).await {
+                        if listener.sender.try_stream(result).await.is_closed() {
                             return;
                         }
                     }
@@ -306,12 +306,12 @@ impl<N: Network> LatestEventsHandler<N> {
                                 // since logs haven't been sent yet
                             }
                             Ok(ScannerMessage::Notification(notification)) => {
-                                if !listener.sender.try_stream(notification).await {
+                                if listener.sender.try_stream(notification).await.is_closed() {
                                     return;
                                 }
                             }
                             Err(e) => {
-                                if !listener.sender.try_stream(e).await {
+                                if listener.sender.try_stream(e).await.is_closed() {
                                     return;
                                 }
                             }

src/types.rs

@@ -4,6 +4,26 @@ use tokio::sync::mpsc;
 
 use crate::ScannerError;
 
+/// Represents the state of a channel after attempting to send a message.
+///
+/// This enum provides explicit semantics for channel operations, making it clear
+/// whether the downstream receiver is still listening or has been dropped.
+#[derive(Debug, Clone, Copy)]
+pub(crate) enum ChannelState {
+    /// The channel is open and the message was successfully sent.
+    Open,
+    /// The channel is closed (receiver dropped), no further messages can be sent.
+    Closed,
+}
+
+impl ChannelState {
+    /// Returns `true` if the channel is closed.
+    #[must_use]
+    pub(crate) fn is_closed(self) -> bool {
+        matches!(self, ChannelState::Closed)
+    }
+}
+
 /// Messages streamed by the scanner to subscribers.
 ///
 /// Each message represents either data or a notification about the scanner's state or behavior.
@@ -109,12 +129,93 @@ impl<T: Clone> IntoScannerResult<T> for Notification {
 
 /// Internal helper for attempting to forward a stream item through an `mpsc` channel.
 pub(crate) trait TryStream<T: Clone> {
-    async fn try_stream<M: IntoScannerResult<T>>(&self, msg: M) -> bool;
+    async fn try_stream<M: IntoScannerResult<T>>(&self, msg: M) -> ChannelState;
 }
 
 impl<T: Clone + Debug> TryStream<T> for mpsc::Sender<ScannerResult<T>> {
-    async fn try_stream<M: IntoScannerResult<T>>(&self, msg: M) -> bool {
+    async fn try_stream<M: IntoScannerResult<T>>(&self, msg: M) -> ChannelState {
         let item = msg.into_scanner_message_result();
-        self.send(item).await.is_ok()
+        if self.send(item).await.is_err() {
+            return ChannelState::Closed;
+        }
+        ChannelState::Open
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::ops::RangeInclusive;
+
+    use crate::ScannerError;
+
+    /// Type alias for test results.
+    type TestResult = Result<ScannerMessage<RangeInclusive<u64>>, ScannerError>;
+
+    mod channel_state_enum {
+        use super::*;
+
+        #[test]
+        fn is_closed_returns_false_for_open_state() {
+            assert!(!ChannelState::Open.is_closed());
+        }
+
+        #[test]
+        fn is_closed_returns_true_for_closed_state() {
+            assert!(ChannelState::Closed.is_closed());
+        }
+
+        #[test]
+        fn channel_state_is_copy() {
+            let state = ChannelState::Open;
+            let copied = state; // Copy, not move
+            assert!(!state.is_closed()); // Both are still valid
+            assert!(!copied.is_closed());
+        }
+
+        #[test]
+        fn channel_state_debug_format() {
+            assert_eq!(format!("{:?}", ChannelState::Open), "Open");
+            assert_eq!(format!("{:?}", ChannelState::Closed), "Closed");
+        }
+    }
+
+    mod try_stream {
+        use super::*;
+
+        #[tokio::test]
+        async fn try_stream_returns_open_when_receiver_exists() {
+            let (tx, _rx) = mpsc::channel::<TestResult>(10);
+
+            let result = tx.try_stream(Notification::SwitchingToLive).await;
+
+            assert!(!result.is_closed());
+        }
+
+        #[tokio::test]
+        async fn try_stream_returns_closed_when_receiver_dropped() {
+            let (tx, rx) = mpsc::channel::<TestResult>(10);
+            drop(rx); // Drop the receiver to close the channel
+
+            let result = tx.try_stream(Notification::SwitchingToLive).await;
+
+            assert!(result.is_closed());
+        }
+
+        #[tokio::test]
+        async fn try_stream_sends_message_successfully() {
+            let (tx, mut rx) = mpsc::channel::<TestResult>(10);
+
+            let result = tx.try_stream(Notification::SwitchingToLive).await;
+
+            assert!(!result.is_closed());
+
+            // Verify the message was actually sent
+            let received = rx.recv().await.unwrap();
+            assert!(matches!(
+                received,
+                Ok(ScannerMessage::Notification(Notification::SwitchingToLive))
+            ));
+        }
     }
 }