Add robust subscription

examples/historical_scanning/main.rs

@@ -55,7 +55,7 @@ async fn main() -> anyhow::Result<()> {
     let _ = counter_contract.increase().send().await?.get_receipt().await?;
 
     let robust_provider = RobustProviderBuilder::new(provider)
-        .max_timeout(std::time::Duration::from_secs(30))
+        .call_timeout(std::time::Duration::from_secs(30))
         .max_retries(5)
         .min_delay(std::time::Duration::from_millis(500))
         .build()

examples/latest_events_scanning/main.rs

@@ -52,7 +52,7 @@ async fn main() -> anyhow::Result<()> {
         .event(Counter::CountIncreased::SIGNATURE);
 
     let robust_provider = RobustProviderBuilder::new(provider)
-        .max_timeout(std::time::Duration::from_secs(30))
+        .call_timeout(std::time::Duration::from_secs(30))
         .max_retries(5)
         .min_delay(std::time::Duration::from_millis(500))
         .build()

examples/live_scanning/main.rs

@@ -53,7 +53,7 @@ async fn main() -> anyhow::Result<()> {
         .event(Counter::CountIncreased::SIGNATURE);
 
     let robust_provider = RobustProviderBuilder::new(provider)
-        .max_timeout(std::time::Duration::from_secs(30))
+        .call_timeout(std::time::Duration::from_secs(30))
         .max_retries(5)
         .min_delay(std::time::Duration::from_millis(500))
         .build()

examples/sync_from_block_scanning/main.rs

@@ -61,7 +61,7 @@ async fn main() -> anyhow::Result<()> {
     }
 
     let robust_provider = RobustProviderBuilder::new(provider)
-        .max_timeout(Duration::from_secs(30))
+        .call_timeout(Duration::from_secs(30))
         .max_retries(5)
         .min_delay(Duration::from_millis(500))
         .build()

examples/sync_from_latest_scanning/main.rs

@@ -53,7 +53,7 @@ async fn main() -> anyhow::Result<()> {
         .event(Counter::CountIncreased::SIGNATURE);
 
     let robust_provider = RobustProviderBuilder::new(provider)
-        .max_timeout(std::time::Duration::from_secs(30))
+        .call_timeout(std::time::Duration::from_secs(30))
         .max_retries(5)
         .min_delay(std::time::Duration::from_millis(500))
         .build()

src/block_range_scanner.rs

@@ -59,6 +59,7 @@
 //! }
 //! ```
 
+use crate::robust_provider::subscription::RobustSubscription;
 use std::{cmp::Ordering, ops::RangeInclusive};
 use tokio::{
     sync::{mpsc, oneshot},
@@ -77,7 +78,6 @@ use alloy::{
     eips::BlockNumberOrTag,
     network::{BlockResponse, Network, primitives::HeaderResponse},
     primitives::{B256, BlockNumber},
-    pubsub::Subscription,
     transports::{RpcError, TransportErrorKind},
 };
 use tracing::{debug, error, info, warn};
@@ -609,16 +609,29 @@ impl<N: Network> Service<N> {
 
     async fn stream_live_blocks(
         mut range_start: BlockNumber,
-        subscription: Subscription<N::HeaderResponse>,
+        subscription: RobustSubscription<N>,
         sender: mpsc::Sender<Message>,
         block_confirmations: u64,
         max_block_range: u64,
     ) {
         // ensure we start streaming only after the expected_next_block cutoff
         let cutoff = range_start;
-        let mut stream = subscription.into_stream().skip_while(|header| header.number() < cutoff);
+        let mut stream = subscription.into_stream().skip_while(|result| match result {
+            Ok(header) => header.number() < cutoff,
+            Err(_) => false,
+        });
+
+        while let Some(result) = stream.next().await {
+            let incoming_block = match result {
+                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;
+                }
+            };
 
-        while let Some(incoming_block) = stream.next().await {
             let incoming_block_num = incoming_block.number();
             info!(block_number = incoming_block_num, "Received block header");
 

src/robust_provider/builder.rs

@@ -2,41 +2,51 @@ use std::{marker::PhantomData, time::Duration};
 
 use alloy::{network::Network, providers::Provider};
 
-use crate::robust_provider::{Error, IntoProvider, RobustProvider};
+use crate::robust_provider::{
+    Error, IntoProvider, RobustProvider, subscription::DEFAULT_RECONNECT_INTERVAL,
+};
 
 // RPC retry and timeout settings
 /// Default timeout used by `RobustProvider`
-pub const DEFAULT_MAX_TIMEOUT: Duration = Duration::from_secs(60);
+pub const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(60);
+/// Default timeout for subscriptions
+pub const DEFAULT_SUBSCRIPTION_TIMEOUT: Duration = Duration::from_secs(120);
 /// Default maximum number of retry attempts.
 pub const DEFAULT_MAX_RETRIES: usize = 3;
 /// Default base delay between retries.
 pub const DEFAULT_MIN_DELAY: Duration = Duration::from_secs(1);
 
 #[derive(Clone)]
 pub struct RobustProviderBuilder<N: Network, P: IntoProvider<N>> {
-    providers: Vec<P>,
-    max_timeout: Duration,
+    primary_provider: P,
+    fallback_providers: Vec<P>,
+    call_timeout: Duration,
+    subscription_timeout: Duration,
     max_retries: usize,
     min_delay: Duration,
+    reconnect_interval: Duration,
     _network: PhantomData<N>,
 }
 
 impl<N: Network, P: IntoProvider<N>> RobustProviderBuilder<N, P> {
-    /// Create a new `RobustProvider` with default settings.
+    /// Create a new [`RobustProvider`] with default settings.
     ///
     /// The provided provider is treated as the primary provider.
     #[must_use]
     pub fn new(provider: P) -> Self {
         Self {
-            providers: vec![provider],
-            max_timeout: DEFAULT_MAX_TIMEOUT,
+            primary_provider: provider,
+            fallback_providers: vec![],
+            call_timeout: DEFAULT_CALL_TIMEOUT,
+            subscription_timeout: DEFAULT_SUBSCRIPTION_TIMEOUT,
             max_retries: DEFAULT_MAX_RETRIES,
             min_delay: DEFAULT_MIN_DELAY,
+            reconnect_interval: DEFAULT_RECONNECT_INTERVAL,
             _network: PhantomData,
         }
     }
 
-    /// Create a new `RobustProvider` with no retry attempts and only timeout set.
+    /// Create a new [`RobustProvider`] with no retry attempts and only timeout set.
     ///
     /// The provided provider is treated as the primary provider.
     #[must_use]
@@ -49,14 +59,24 @@ impl<N: Network, P: IntoProvider<N>> RobustProviderBuilder<N, P> {
     /// Fallback providers are used when the primary provider times out or fails.
     #[must_use]
     pub fn fallback(mut self, provider: P) -> Self {
-        self.providers.push(provider);
+        self.fallback_providers.push(provider);
         self
     }
 
     /// Set the maximum timeout for RPC operations.
     #[must_use]
-    pub fn max_timeout(mut self, timeout: Duration) -> Self {
-        self.max_timeout = timeout;
+    pub fn call_timeout(mut self, timeout: Duration) -> Self {
+        self.call_timeout = timeout;
+        self
+    }
+
+    /// Set the timeout for subscription operations.
+    ///
+    /// This should be set higher than [`call_timeout`](Self::call_timeout) to accommodate chains
+    /// with slow block times. Default is [`DEFAULT_SUBSCRIPTION_TIMEOUT`].
+    #[must_use]
+    pub fn subscription_timeout(mut self, timeout: Duration) -> Self {
+        self.subscription_timeout = timeout;
         self
     }
 
@@ -74,6 +94,17 @@ impl<N: Network, P: IntoProvider<N>> RobustProviderBuilder<N, P> {
         self
     }
 
+    /// Set the interval for attempting to reconnect to the primary provider.
+    ///
+    /// After a failover to a fallback provider, the subscription will periodically
+    /// attempt to reconnect to the primary provider at this interval.
+    /// Default is [`DEFAULT_RECONNECT_INTERVAL`].
+    #[must_use]
+    pub fn reconnect_interval(mut self, reconnect_interval: Duration) -> Self {
+        self.reconnect_interval = reconnect_interval;
+        self
+    }
+
     /// Build the `RobustProvider`.
     ///
     /// Final builder method: consumes the builder and returns the built [`RobustProvider`].
@@ -82,15 +113,21 @@ impl<N: Network, P: IntoProvider<N>> RobustProviderBuilder<N, P> {
     ///
     /// Returns an error if any of the providers fail to connect.
     pub async fn build(self) -> Result<RobustProvider<N>, Error> {
-        let mut providers = vec![];
-        for p in self.providers {
-            providers.push(p.into_provider().await?.root().to_owned());
+        let primary_provider = self.primary_provider.into_provider().await?.root().to_owned();
+
+        let mut fallback_providers = vec![];
+        for p in self.fallback_providers {
+            fallback_providers.push(p.into_provider().await?.root().to_owned());
         }
+
         Ok(RobustProvider {
-            providers,
-            max_timeout: self.max_timeout,
+            primary_provider,
+            fallback_providers,
+            call_timeout: self.call_timeout,
+            subscription_timeout: self.subscription_timeout,
             max_retries: self.max_retries,
             min_delay: self.min_delay,
+            reconnect_interval: self.reconnect_interval,
         })
     }
 }

src/robust_provider/mod.rs

@@ -2,8 +2,10 @@ pub mod builder;
 pub mod error;
 pub mod provider;
 pub mod provider_conversion;
+pub mod subscription;
 
 pub use builder::*;
 pub use error::Error;
 pub use provider::RobustProvider;
 pub use provider_conversion::{IntoProvider, IntoRobustProvider};
+pub use subscription::RobustSubscription;