Revert "remove subsc. lagged error variant from scannererror"

This reverts commit 8763c45.

Author: 0xNeshi

src/block_range_scanner.rs

@@ -59,7 +59,7 @@
 //! }
 //! ```
 
-use crate::robust_provider::subscription::{self, RobustSubscription};
+use crate::robust_provider::subscription::RobustSubscription;
 use std::{cmp::Ordering, ops::RangeInclusive};
 use tokio::{
     sync::{mpsc, oneshot},
@@ -600,25 +600,9 @@ impl<N: Network> Service<N> {
                 Ok(block) => block,
                 Err(e) => {
                     error!(error = %e, "Error receiving block from stream");
-                    match e {
-                        subscription::Error::Lagged(_) => {
-                            // scanner already accounts for skipped block numbers
-                            // next block will be the actual incoming block
-                            continue;
-                        }
-                        subscription::Error::Timeout => {
-                            _ = sender.try_stream(ScannerError::Timeout).await;
-                            return;
-                        }
-                        subscription::Error::RpcError(rpc_err) => {
-                            _ = sender.try_stream(ScannerError::RpcError(rpc_err)).await;
-                            return;
-                        }
-                        subscription::Error::Closed => {
-                            _ = sender.try_stream(ScannerError::SubscriptionClosed).await;
-                            return;
-                        }
-                    }
+                    // Error from subscription, exit the stream
+                    _ = sender.try_stream(e).await;
+                    return;
                 }
             };
 

src/error.rs

@@ -6,7 +6,12 @@ use alloy::{
 };
 use thiserror::Error;
 
-use crate::{robust_provider::provider::Error as RobustProviderError, types::ScannerResult};
+use crate::{
+    robust_provider::{
+        provider::Error as RobustProviderError, subscription::Error as RobustSubscriptionError,
+    },
+    types::ScannerResult,
+};
 
 #[derive(Error, Debug, Clone)]
 pub enum ScannerError {
@@ -33,6 +38,9 @@ pub enum ScannerError {
 
     #[error("Subscription closed")]
     SubscriptionClosed,
+
+    #[error("Subscription Lagged by {0}")]
+    SubscriptionLagged(u64),
 }
 
 impl From<RobustProviderError> for ScannerError {
@@ -45,12 +53,22 @@ impl From<RobustProviderError> for ScannerError {
     }
 }
 
+impl From<RobustSubscriptionError> for ScannerError {
+    fn from(error: RobustSubscriptionError) -> ScannerError {
+        match error {
+            RobustSubscriptionError::Timeout => ScannerError::Timeout,
+            RobustSubscriptionError::RpcError(err) => ScannerError::RpcError(err),
+            RobustSubscriptionError::Closed => ScannerError::SubscriptionClosed,
+            RobustSubscriptionError::Lagged(count) => ScannerError::SubscriptionLagged(count),
+        }
+    }
+}
+
 impl From<RpcError<TransportErrorKind>> for ScannerError {
     fn from(error: RpcError<TransportErrorKind>) -> Self {
         ScannerError::RpcError(Arc::new(error))
     }
 }
-
 impl<T: Clone> PartialEq<ScannerError> for ScannerResult<T> {
     fn eq(&self, other: &ScannerError) -> bool {
         match self {

src/robust_provider/provider.rs

@@ -26,36 +26,6 @@ pub enum Error {
     BlockNotFound(BlockId),
 }
 
-/// Errors that can occur when using [`RobustProvider`].
-#[derive(Error, Debug, Clone)]
-pub(crate) enum CommonError {
-    #[error("Operation timed out")]
-    Timeout,
-    #[error("RPC call failed after exhausting all retry attempts: {0}")]
-    RpcError(Arc<RpcError<TransportErrorKind>>),
-}
-
-impl From<RpcError<TransportErrorKind>> for CommonError {
-    fn from(err: RpcError<TransportErrorKind>) -> Self {
-        CommonError::RpcError(Arc::new(err))
-    }
-}
-
-impl From<CommonError> for Error {
-    fn from(err: CommonError) -> Self {
-        match err {
-            CommonError::Timeout => Error::Timeout,
-            CommonError::RpcError(e) => Error::RpcError(e),
-        }
-    }
-}
-
-impl From<TokioError::Elapsed> for CommonError {
-    fn from(_: TokioError::Elapsed) -> Self {
-        CommonError::Timeout
-    }
-}
-
 impl From<RpcError<TransportErrorKind>> for Error {
     fn from(err: RpcError<TransportErrorKind>) -> Self {
         Error::RpcError(Arc::new(err))
@@ -164,8 +134,7 @@ impl<N: Network> RobustProvider<N> {
                 move |provider| async move { provider.get_block_number().await },
                 false,
             )
-            .await
-            .map_err(Error::from);
+            .await;
         if let Err(e) = &result {
             error!(error = %e, "eth_getBlockNumber failed");
         }
@@ -253,8 +222,7 @@ impl<N: Network> RobustProvider<N> {
                 move |provider| async move { provider.get_logs(filter).await },
                 false,
             )
-            .await
-            .map_err(Error::from);
+            .await;
         if let Err(e) = &result {
             error!(error = %e, "eth_getLogs failed");
         }
@@ -288,7 +256,7 @@ impl<N: Network> RobustProvider<N> {
             Ok(sub) => Ok(RobustSubscription::new(sub, self.clone())),
             Err(e) => {
                 error!(error = %e, "eth_subscribe failed");
-                Err(e.into())
+                Err(e)
             }
         }
     }
@@ -317,7 +285,7 @@ impl<N: Network> RobustProvider<N> {
         &self,
         operation: F,
         require_pubsub: bool,
-    ) -> Result<T, CommonError>
+    ) -> Result<T, Error>
     where
         F: Fn(RootProvider<N>) -> Fut,
         Fut: Future<Output = Result<T, RpcError<TransportErrorKind>>>,
@@ -338,8 +306,8 @@ impl<N: Network> RobustProvider<N> {
         &self,
         operation: F,
         require_pubsub: bool,
-        last_error: CommonError,
-    ) -> Result<T, CommonError>
+        last_error: Error,
+    ) -> Result<T, Error>
     where
         F: Fn(RootProvider<N>) -> Fut,
         Fut: Future<Output = Result<T, RpcError<TransportErrorKind>>>,
@@ -353,9 +321,9 @@ impl<N: Network> RobustProvider<N> {
         &self,
         operation: F,
         require_pubsub: bool,
-        mut last_error: CommonError,
+        mut last_error: Error,
         start_index: usize,
-    ) -> Result<(T, usize), CommonError>
+    ) -> Result<(T, usize), Error>
     where
         F: Fn(RootProvider<N>) -> Fut,
         Fut: Future<Output = Result<T, RpcError<TransportErrorKind>>>,
@@ -394,7 +362,7 @@ impl<N: Network> RobustProvider<N> {
         &self,
         provider: &RootProvider<N>,
         operation: F,
-    ) -> Result<T, CommonError>
+    ) -> Result<T, Error>
     where
         F: Fn(RootProvider<N>) -> Fut,
         Fut: Future<Output = Result<T, RpcError<TransportErrorKind>>>,
@@ -413,8 +381,8 @@ impl<N: Network> RobustProvider<N> {
                 .sleep(tokio::time::sleep),
         )
         .await
-        .map_err(CommonError::from)?
-        .map_err(CommonError::from)
+        .map_err(Error::from)?
+        .map_err(Error::from)
     }
 
     /// Check if a provider supports pubsub
@@ -516,7 +484,7 @@ mod tests {
 
         let call_count = AtomicUsize::new(0);
 
-        let result: Result<(), CommonError> = provider
+        let result: Result<(), Error> = provider
             .try_operation_with_failover(
                 |_| async {
                     call_count.fetch_add(1, Ordering::SeqCst);
@@ -526,7 +494,7 @@ mod tests {
             )
             .await;
 
-        assert!(matches!(result, Err(CommonError::RpcError(_))));
+        assert!(matches!(result, Err(Error::RpcError(_))));
         assert_eq!(call_count.load(Ordering::SeqCst), 3);
     }
 
@@ -545,7 +513,7 @@ mod tests {
             )
             .await;
 
-        assert!(matches!(result, Err(CommonError::Timeout)));
+        assert!(matches!(result, Err(Error::Timeout)));
     }
 
     #[tokio::test]

src/robust_provider/subscription.rs

@@ -17,7 +17,7 @@ use tokio_stream::Stream;
 use tokio_util::sync::ReusableBoxFuture;
 use tracing::{error, info, warn};
 
-use crate::robust_provider::{RobustProvider, provider::CommonError};
+use crate::robust_provider::{RobustProvider, provider::Error as ProviderError};
 
 /// Errors that can occur when using [`RobustSubscription`].
 #[derive(Error, Debug, Clone)]
@@ -32,11 +32,15 @@ pub enum Error {
     Lagged(u64),
 }
 
-impl From<CommonError> for Error {
-    fn from(err: CommonError) -> Self {
+impl From<ProviderError> for Error {
+    fn from(err: ProviderError) -> Self {
         match err {
-            CommonError::Timeout => Error::Timeout,
-            CommonError::RpcError(e) => Error::RpcError(e),
+            ProviderError::Timeout => Error::Timeout,
+            ProviderError::RpcError(e) => Error::RpcError(e),
+            ProviderError::BlockNotFound(_) => {
+                // This shouldn't happen in subscription context, but we need to handle it
+                Error::RpcError(Arc::new(RpcError::NullResp))
+            }
         }
     }
 }
@@ -180,7 +184,7 @@ impl<N: Network> RobustSubscription<N> {
         }
     }
 
-    async fn switch_to_fallback(&mut self, last_error: CommonError) -> Result<(), Error> {
+    async fn switch_to_fallback(&mut self, last_error: Error) -> Result<(), Error> {
         // If we're on a fallback, try primary first before moving to next fallback
         if self.is_on_fallback() && self.try_reconnect_to_primary(true).await {
             return Ok(());
@@ -198,7 +202,7 @@ impl<N: Network> RobustSubscription<N> {
 
         let (sub, fallback_idx) = self
             .robust_provider
-            .try_fallback_providers_from(&operation, true, last_error, start_index)
+            .try_fallback_providers_from(&operation, true, last_error.into(), start_index)
             .await?;
 
         self.subscription = sub;