Retry logic

Cargo.toml

@@ -33,6 +33,7 @@ tokio-stream = "0.1.17"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
 hex = "0.4"
+backon = "1.5.2"
 
 [package]
 name = "event-scanner"
@@ -66,6 +67,7 @@ alloy-node-bindings.workspace = true
 tokio-stream.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
+backon.workspace = true
 
 [lints]
 workspace = true

src/block_range_scanner.rs

@@ -63,7 +63,7 @@
 //! }
 //! ```
 
-use std::{cmp::Ordering, ops::RangeInclusive};
+use std::{cmp::Ordering, ops::RangeInclusive, time::Duration};
 use tokio::{
     join,
     sync::{mpsc, oneshot},
@@ -72,14 +72,17 @@ use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 
 use crate::{
     error::ScannerError,
+    safe_provider::{
+        DEFAULT_MAX_RETRIES, DEFAULT_MAX_TIMEOUT, DEFAULT_RETRY_INTERVAL, SafeProvider,
+    },
     types::{ScannerMessage, ScannerStatus},
 };
 use alloy::{
     consensus::BlockHeader,
     eips::BlockNumberOrTag,
     network::{BlockResponse, Network, primitives::HeaderResponse},
     primitives::{B256, BlockNumber},
-    providers::{Provider, RootProvider},
+    providers::RootProvider,
     pubsub::Subscription,
     rpc::client::ClientBuilder,
     transports::{
@@ -115,6 +118,9 @@ impl PartialEq<RangeInclusive<BlockNumber>> for Message {
 #[derive(Clone, Copy)]
 pub struct BlockRangeScanner {
     pub max_block_range: u64,
+    pub max_timeout: Duration,
+    pub max_retries: usize,
+    pub retry_interval: Duration,
 }
 
 impl Default for BlockRangeScanner {
@@ -126,7 +132,12 @@ impl Default for BlockRangeScanner {
 impl BlockRangeScanner {
     #[must_use]
     pub fn new() -> Self {
-        Self { max_block_range: DEFAULT_MAX_BLOCK_RANGE }
+        Self {
+            max_block_range: DEFAULT_MAX_BLOCK_RANGE,
+            max_timeout: DEFAULT_MAX_TIMEOUT,
+            max_retries: DEFAULT_MAX_RETRIES,
+            retry_interval: DEFAULT_RETRY_INTERVAL,
+        }
     }
 
     #[must_use]
@@ -135,6 +146,24 @@ impl BlockRangeScanner {
         self
     }
 
+    #[must_use]
+    pub fn with_max_timeout(mut self, rpc_timeout: Duration) -> Self {
+        self.max_timeout = rpc_timeout;
+        self
+    }
+
+    #[must_use]
+    pub fn with_max_retries(mut self, rpc_max_retries: usize) -> Self {
+        self.max_retries = rpc_max_retries;
+        self
+    }
+
+    #[must_use]
+    pub fn with_retry_interval(mut self, rpc_retry_interval: Duration) -> Self {
+        self.retry_interval = rpc_retry_interval;
+        self
+    }
+
     /// Connects to the provider via WebSocket
     ///
     /// # Errors
@@ -169,19 +198,26 @@ impl BlockRangeScanner {
     /// Returns an error if the connection fails
     #[must_use]
     pub fn connect<N: Network>(self, provider: RootProvider<N>) -> ConnectedBlockRangeScanner<N> {
-        ConnectedBlockRangeScanner { provider, max_block_range: self.max_block_range }
+        let safe_provider = SafeProvider::new(provider)
+            .max_timeout(self.max_timeout)
+            .max_retries(self.max_retries)
+            .retry_interval(self.retry_interval);
+        ConnectedBlockRangeScanner {
+            provider: safe_provider,
+            max_block_range: self.max_block_range,
+        }
     }
 }
 
 pub struct ConnectedBlockRangeScanner<N: Network> {
-    provider: RootProvider<N>,
+    provider: SafeProvider<N>,
     max_block_range: u64,
 }
 
 impl<N: Network> ConnectedBlockRangeScanner<N> {
-    /// Returns the underlying Provider.
+    /// Returns the `SafeProvider`
     #[must_use]
-    pub fn provider(&self) -> &RootProvider<N> {
+    pub fn provider(&self) -> &SafeProvider<N> {
         &self.provider
     }
 
@@ -233,7 +269,7 @@ pub enum Command {
 }
 
 struct Service<N: Network> {
-    provider: RootProvider<N>,
+    provider: SafeProvider<N>,
     max_block_range: u64,
     subscriber: Option<mpsc::Sender<Message>>,
     websocket_connected: bool,
@@ -244,7 +280,7 @@ struct Service<N: Network> {
 }
 
 impl<N: Network> Service<N> {
-    pub fn new(provider: RootProvider<N>, max_block_range: u64) -> (Self, mpsc::Sender<Command>) {
+    pub fn new(provider: SafeProvider<N>, max_block_range: u64) -> (Self, mpsc::Sender<Command>) {
         let (cmd_tx, cmd_rx) = mpsc::channel(100);
 
         let service = Self {
@@ -640,9 +676,9 @@ impl<N: Network> Service<N> {
         Ok(())
     }
 
-    async fn stream_live_blocks<P: Provider<N>>(
+    async fn stream_live_blocks(
         mut range_start: BlockNumber,
-        provider: P,
+        provider: SafeProvider<N>,
         sender: mpsc::Sender<Message>,
         block_confirmations: u64,
         max_block_range: u64,
@@ -747,7 +783,7 @@ impl<N: Network> Service<N> {
     }
 
     async fn get_block_subscription(
-        provider: &impl Provider<N>,
+        provider: &SafeProvider<N>,
     ) -> Result<Subscription<N::HeaderResponse>, ScannerError> {
         let ws_stream = provider
             .subscribe_blocks()
@@ -966,6 +1002,7 @@ impl BlockRangeScannerClient {
 #[cfg(test)]
 mod tests {
 
+    use alloy::providers::{Provider, RootProvider};
     use std::time::Duration;
     use tokio::time::timeout;
 
@@ -981,8 +1018,9 @@ mod tests {
     use tokio::sync::mpsc;
     use tokio_stream::StreamExt;
 
-    fn mocked_provider(asserter: Asserter) -> RootProvider<Ethereum> {
-        RootProvider::new(RpcClient::mocked(asserter))
+    fn mocked_provider(asserter: Asserter) -> SafeProvider<Ethereum> {
+        let root_provider = RootProvider::new(RpcClient::mocked(asserter));
+        SafeProvider::new(root_provider)
     }
 
     #[test]

src/event_scanner/modes/common.rs

@@ -3,10 +3,10 @@ use std::ops::RangeInclusive;
 use crate::{
     block_range_scanner::{MAX_BUFFERED_MESSAGES, Message as BlockRangeMessage},
     event_scanner::{filter::EventFilter, listener::EventListener, message::Message},
+    safe_provider::SafeProvider,
 };
 use alloy::{
     network::Network,
-    providers::{Provider, RootProvider},
     rpc::types::{Filter, Log},
     transports::{RpcError, TransportErrorKind},
 };
@@ -25,7 +25,7 @@ pub enum ConsumerMode {
 
 pub async fn handle_stream<N: Network>(
     mut stream: ReceiverStream<BlockRangeMessage>,
-    provider: &RootProvider<N>,
+    provider: &SafeProvider<N>,
     listeners: &[EventListener],
     mode: ConsumerMode,
 ) {
@@ -42,7 +42,7 @@ pub async fn handle_stream<N: Network>(
 }
 
 pub fn spawn_log_consumers<N: Network>(
-    provider: &RootProvider<N>,
+    provider: &SafeProvider<N>,
     listeners: &[EventListener],
     range_tx: &Sender<BlockRangeMessage>,
     mode: ConsumerMode,
@@ -129,7 +129,7 @@ async fn get_logs<N: Network>(
     range: RangeInclusive<u64>,
     event_filter: &EventFilter,
     log_filter: &Filter,
-    provider: &RootProvider<N>,
+    provider: &SafeProvider<N>,
 ) -> Result<Vec<Log>, RpcError<TransportErrorKind>> {
     let log_filter = log_filter.clone().from_block(*range.start()).to_block(*range.end());
 

src/lib.rs

@@ -1,6 +1,7 @@
 pub mod block_range_scanner;
 pub mod error;
 pub mod event_scanner;
+mod safe_provider;
 #[cfg(any(test, feature = "test-utils"))]
 pub mod test_utils;
 pub mod types;