fix: rebase past changes

Author: LeoPatOZ

src/block_range_scanner.rs

@@ -67,7 +67,7 @@ use tokio::{
 use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 
 use crate::{
-    IntoRobustProvider, RobustProvider, ScannerMessage,
+    IntoRobustProvider, RobustProvider, RobustSubscription, ScannerMessage,
     error::ScannerError,
     robust_provider::Error as RobustProviderError,
     types::{ScannerStatus, TryStream},
@@ -77,7 +77,6 @@ use alloy::{
     eips::BlockNumberOrTag,
     network::{BlockResponse, Network, primitives::HeaderResponse},
     primitives::{B256, BlockNumber},
-    pubsub::Subscription,
     transports::{RpcError, TransportErrorKind},
 };
 use tracing::{debug, error, info, warn};
@@ -611,16 +610,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/lib.rs

@@ -16,4 +16,7 @@ pub use event_scanner::{
     SyncFromBlock, SyncFromLatestEvents,
 };
 
-pub use robust_provider::{provider::RobustProvider, provider_conversion::IntoRobustProvider};
+pub use robust_provider::{
+    provider::RobustProvider, provider_conversion::IntoRobustProvider,
+    subscription::RobustSubscription,
+};

src/robust_provider/builder.rs

@@ -10,6 +10,8 @@ use crate::{
 // RPC retry and timeout settings
 /// Default timeout used by `RobustProvider`
 pub const DEFAULT_MAX_TIMEOUT: Duration = Duration::from_secs(60);
+/// Default timeout for subscriptions (longer to accommodate slow block times)
+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.
@@ -19,27 +21,29 @@ pub const DEFAULT_MIN_DELAY: Duration = Duration::from_secs(1);
 pub struct RobustProviderBuilder<N: Network, P: IntoProvider<N>> {
     providers: Vec<P>,
     max_timeout: Duration,
+    subscription_timeout: Duration,
     max_retries: usize,
     min_delay: 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,
+            subscription_timeout: DEFAULT_SUBSCRIPTION_TIMEOUT,
             max_retries: DEFAULT_MAX_RETRIES,
             min_delay: DEFAULT_MIN_DELAY,
             _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]
@@ -63,6 +67,16 @@ impl<N: Network, P: IntoProvider<N>> RobustProviderBuilder<N, P> {
         self
     }
 
+    /// Set the timeout for subscription operations.
+    ///
+    /// This should be set higher than [`max_timeout`](Self::max_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
+    }
+
     /// Set the maximum number of retry attempts.
     #[must_use]
     pub fn max_retries(mut self, max_retries: usize) -> Self {
@@ -92,6 +106,7 @@ impl<N: Network, P: IntoProvider<N>> RobustProviderBuilder<N, P> {
         Ok(RobustProvider {
             providers,
             max_timeout: self.max_timeout,
+            subscription_timeout: self.subscription_timeout,
             max_retries: self.max_retries,
             min_delay: self.min_delay,
         })

src/robust_provider/mod.rs

@@ -2,5 +2,6 @@ pub mod builder;
 pub mod error;
 pub mod provider;
 pub mod provider_conversion;
+pub mod subscription;
 
 pub use error::Error;