remove subsc. lagged error variant from scannererror

Author: 0xNeshi

src/block_range_scanner.rs

@@ -59,7 +59,7 @@
 //! }
 //! ```
 
-use crate::robust_provider::subscription::RobustSubscription;
+use crate::robust_provider::subscription::{self, RobustSubscription};
 use std::{cmp::Ordering, ops::RangeInclusive};
 use tokio::{
     sync::{mpsc, oneshot},
@@ -600,9 +600,25 @@ impl<N: Network> Service<N> {
                 Ok(block) => block,
                 Err(e) => {
                     error!(error = %e, "Error receiving block from stream");
-                    // Error from subscription, exit the stream
-                    _ = sender.try_stream(e).await;
-                    return;
+                    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;
+                        }
+                    }
                 }
             };
 

src/error.rs

@@ -6,12 +6,7 @@ use alloy::{
 };
 use thiserror::Error;
 
-use crate::{
-    robust_provider::{
-        provider::Error as RobustProviderError, subscription::Error as RobustSubscriptionError,
-    },
-    types::ScannerResult,
-};
+use crate::{robust_provider::provider::Error as RobustProviderError, types::ScannerResult};
 
 #[derive(Error, Debug, Clone)]
 pub enum ScannerError {
@@ -38,9 +33,6 @@ pub enum ScannerError {
 
     #[error("Subscription closed")]
     SubscriptionClosed,
-
-    #[error("Subscription Lagged by {0}")]
-    SubscriptionLagged(u64),
 }
 
 impl From<RobustProviderError> for ScannerError {
@@ -53,22 +45,12 @@ 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,6 +26,36 @@ 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))
@@ -134,7 +164,8 @@ impl<N: Network> RobustProvider<N> {
                 move |provider| async move { provider.get_block_number().await },
                 false,
             )
-            .await;
+            .await
+            .map_err(Error::from);
         if let Err(e) = &result {
             error!(error = %e, "eth_getBlockNumber failed");
         }
@@ -222,7 +253,8 @@ impl<N: Network> RobustProvider<N> {
                 move |provider| async move { provider.get_logs(filter).await },
                 false,
             )
-            .await;
+            .await
+            .map_err(Error::from);
         if let Err(e) = &result {
             error!(error = %e, "eth_getLogs failed");
         }
@@ -256,7 +288,7 @@ impl<N: Network> RobustProvider<N> {
             Ok(sub) => Ok(RobustSubscription::new(sub, self.clone())),
             Err(e) => {
                 error!(error = %e, "eth_subscribe failed");
-                Err(e)
+                Err(e.into())
             }
         }
     }
@@ -285,7 +317,7 @@ impl<N: Network> RobustProvider<N> {
         &self,
         operation: F,
         require_pubsub: bool,
-    ) -> Result<T, Error>
+    ) -> Result<T, CommonError>
     where
         F: Fn(RootProvider<N>) -> Fut,
         Fut: Future<Output = Result<T, RpcError<TransportErrorKind>>>,
@@ -306,8 +338,8 @@ impl<N: Network> RobustProvider<N> {
         &self,
         operation: F,
         require_pubsub: bool,
-        last_error: Error,
-    ) -> Result<T, Error>
+        last_error: CommonError,
+    ) -> Result<T, CommonError>
     where
         F: Fn(RootProvider<N>) -> Fut,
         Fut: Future<Output = Result<T, RpcError<TransportErrorKind>>>,
@@ -321,9 +353,9 @@ impl<N: Network> RobustProvider<N> {
         &self,
         operation: F,
         require_pubsub: bool,
-        mut last_error: Error,
+        mut last_error: CommonError,
         start_index: usize,
-    ) -> Result<(T, usize), Error>
+    ) -> Result<(T, usize), CommonError>
     where
         F: Fn(RootProvider<N>) -> Fut,
         Fut: Future<Output = Result<T, RpcError<TransportErrorKind>>>,
@@ -362,7 +394,7 @@ impl<N: Network> RobustProvider<N> {
         &self,
         provider: &RootProvider<N>,
         operation: F,
-    ) -> Result<T, Error>
+    ) -> Result<T, CommonError>
     where
         F: Fn(RootProvider<N>) -> Fut,
         Fut: Future<Output = Result<T, RpcError<TransportErrorKind>>>,
@@ -381,8 +413,8 @@ impl<N: Network> RobustProvider<N> {
                 .sleep(tokio::time::sleep),
         )
         .await
-        .map_err(Error::from)?
-        .map_err(Error::from)
+        .map_err(CommonError::from)?
+        .map_err(CommonError::from)
     }
 
     /// Check if a provider supports pubsub
@@ -484,7 +516,7 @@ mod tests {
 
         let call_count = AtomicUsize::new(0);
 
-        let result: Result<(), Error> = provider
+        let result: Result<(), CommonError> = provider
             .try_operation_with_failover(
                 |_| async {
                     call_count.fetch_add(1, Ordering::SeqCst);
@@ -494,7 +526,7 @@ mod tests {
             )
             .await;
 
-        assert!(matches!(result, Err(Error::RpcError(_))));
+        assert!(matches!(result, Err(CommonError::RpcError(_))));
         assert_eq!(call_count.load(Ordering::SeqCst), 3);
     }
 
@@ -513,7 +545,7 @@ mod tests {
             )
             .await;
 
-        assert!(matches!(result, Err(Error::Timeout)));
+        assert!(matches!(result, Err(CommonError::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::Error as ProviderError};
+use crate::robust_provider::{RobustProvider, provider::CommonError};
 
 /// Errors that can occur when using [`RobustSubscription`].
 #[derive(Error, Debug, Clone)]
@@ -32,15 +32,11 @@ pub enum Error {
     Lagged(u64),
 }
 
-impl From<ProviderError> for Error {
-    fn from(err: ProviderError) -> Self {
+impl From<CommonError> for Error {
+    fn from(err: CommonError) -> Self {
         match err {
-            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))
-            }
+            CommonError::Timeout => Error::Timeout,
+            CommonError::RpcError(e) => Error::RpcError(e),
         }
     }
 }
@@ -184,7 +180,7 @@ impl<N: Network> RobustSubscription<N> {
         }
     }
 
-    async fn switch_to_fallback(&mut self, last_error: Error) -> Result<(), Error> {
+    async fn switch_to_fallback(&mut self, last_error: CommonError) -> 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(());
@@ -202,7 +198,7 @@ impl<N: Network> RobustSubscription<N> {
 
         let (sub, fallback_idx) = self
             .robust_provider
-            .try_fallback_providers_from(&operation, true, last_error.into(), start_index)
+            .try_fallback_providers_from(&operation, true, last_error, start_index)
             .await?;
 
         self.subscription = sub;