ref: rebase onto main

Author: LeoPatOZ

src/block_range_scanner.rs

@@ -67,7 +67,7 @@ use tokio::{
 use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 
 use crate::{
-    IntoRobustProvider, RobustProvider, RobustSubscription, ScannerMessage,
+    IntoRobustProvider, RobustProvider, ScannerMessage,
     error::ScannerError,
     robust_provider::Error as RobustProviderError,
     types::{ScannerStatus, TryStream},
@@ -77,6 +77,7 @@ use alloy::{
     eips::BlockNumberOrTag,
     network::{BlockResponse, Network, primitives::HeaderResponse},
     primitives::{B256, BlockNumber},
+    pubsub::Subscription,
     transports::{RpcError, TransportErrorKind},
 };
 use tracing::{debug, error, info, warn};
@@ -610,29 +611,16 @@ impl<N: Network> Service<N> {
 
     async fn stream_live_blocks(
         mut range_start: BlockNumber,
-        subscription: RobustSubscription<N>,
+        subscription: Subscription<N::HeaderResponse>,
         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(|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;
-                }
-            };
+        let mut stream = subscription.into_stream().skip_while(|header| header.number() < cutoff);
 
+        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,7 +16,4 @@ pub use event_scanner::{
     SyncFromBlock, SyncFromLatestEvents,
 };
 
-pub use robust_provider::{
-    provider::RobustProvider, provider_conversion::IntoRobustProvider,
-    subscription::RobustSubscription,
-};
+pub use robust_provider::{provider::RobustProvider, provider_conversion::IntoRobustProvider};

src/robust_provider/builder.rs

@@ -10,40 +10,36 @@ 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.
 pub const DEFAULT_MIN_DELAY: Duration = Duration::from_secs(1);
 
 #[derive(Clone)]
 pub struct RobustProviderBuilder<N: Network, P: IntoProvider<N>> {
-    pub(crate) providers: Vec<P>,
+    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]
@@ -67,16 +63,6 @@ impl<N: Network, P: IntoProvider<N>> RobustProviderBuilder<N, P> {
         self
     }
 
-    /// Set the timeout for subscription operations.
-    ///
-    /// This should be set higher than `max_timeout` to accommodate chains with slow block times.
-    /// Default is 2 minutes.
-    #[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 {
@@ -106,7 +92,6 @@ 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,6 +2,5 @@ pub mod builder;
 pub mod error;
 pub mod provider;
 pub mod provider_conversion;
-pub mod subscription;
 
 pub use error::Error;

src/robust_provider/provider.rs

@@ -1,216 +1,18 @@
-use std::{fmt::Debug, future::Future, marker::PhantomData, sync::Arc, time::Duration};
+use std::{fmt::Debug, time::Duration};
 
 use alloy::{
-    eips::{BlockId, BlockNumberOrTag},
+    eips::BlockNumberOrTag,
     network::{Ethereum, Network},
-    providers::{
-        DynProvider, Provider, RootProvider,
-        fillers::{FillProvider, TxFiller},
-        layers::{CacheProvider, CallBatchProvider},
-    },
+    providers::{Provider, RootProvider},
     pubsub::Subscription,
     rpc::types::{Filter, Log},
-    transports::{RpcError, TransportErrorKind, http::reqwest::Url},
+    transports::{RpcError, TransportErrorKind},
 };
 use backon::{ExponentialBuilder, Retryable};
-use thiserror::Error;
-use tokio::time::{error as TokioError, timeout};
+use tokio::time::timeout;
 use tracing::{error, info};
 
-#[derive(Error, Debug, Clone)]
-pub enum Error {
-    #[error("Operation timed out")]
-    Timeout,
-    #[error("RPC call failed after exhausting all retry attempts: {0}")]
-    RpcError(Arc<RpcError<TransportErrorKind>>),
-    #[error("Block not found, Block Id: {0}")]
-    BlockNotFound(BlockId),
-}
-
-impl From<RpcError<TransportErrorKind>> for Error {
-    fn from(err: RpcError<TransportErrorKind>) -> Self {
-        Error::RpcError(Arc::new(err))
-    }
-}
-
-impl From<TokioError::Elapsed> for Error {
-    fn from(_: TokioError::Elapsed) -> Self {
-        Error::Timeout
-    }
-}
-
-pub trait IntoProvider<N: Network = Ethereum> {
-    fn into_provider(
-        self,
-    ) -> impl std::future::Future<Output = Result<impl Provider<N>, Error>> + Send;
-}
-
-impl<N: Network> IntoProvider<N> for RobustProvider<N> {
-    async fn into_provider(self) -> Result<impl Provider<N>, Error> {
-        Ok(self.primary().to_owned())
-    }
-}
-
-impl<N: Network> IntoProvider<N> for RootProvider<N> {
-    async fn into_provider(self) -> Result<impl Provider<N>, Error> {
-        Ok(self)
-    }
-}
-
-impl<N: Network> IntoProvider<N> for &str {
-    async fn into_provider(self) -> Result<impl Provider<N>, Error> {
-        Ok(RootProvider::connect(self).await?)
-    }
-}
-
-impl<N: Network> IntoProvider<N> for Url {
-    async fn into_provider(self) -> Result<impl Provider<N>, Error> {
-        Ok(RootProvider::connect(self.as_str()).await?)
-    }
-}
-
-impl<F, P, N> IntoProvider<N> for FillProvider<F, P, N>
-where
-    F: TxFiller<N>,
-    P: Provider<N>,
-    N: Network,
-{
-    async fn into_provider(self) -> Result<impl Provider<N>, Error> {
-        Ok(self)
-    }
-}
-
-impl<P, N> IntoProvider<N> for CacheProvider<P, N>
-where
-    P: Provider<N>,
-    N: Network,
-{
-    async fn into_provider(self) -> Result<impl Provider<N>, Error> {
-        Ok(self)
-    }
-}
-
-impl<N> IntoProvider<N> for DynProvider<N>
-where
-    N: Network,
-{
-    async fn into_provider(self) -> Result<impl Provider<N>, Error> {
-        Ok(self)
-    }
-}
-
-impl<P, N> IntoProvider<N> for CallBatchProvider<P, N>
-where
-    P: Provider<N> + 'static,
-    N: Network,
-{
-    async fn into_provider(self) -> Result<impl Provider<N>, Error> {
-        Ok(self)
-    }
-}
-
-pub trait IntoRobustProvider<N: Network = Ethereum> {
-    fn into_robust_provider(
-        self,
-    ) -> impl std::future::Future<Output = Result<RobustProvider<N>, Error>> + Send;
-}
-
-impl<N: Network, P: IntoProvider<N> + Send> IntoRobustProvider<N> for P {
-    async fn into_robust_provider(self) -> Result<RobustProvider<N>, Error> {
-        RobustProviderBuilder::new(self).build().await
-    }
-}
-
-// RPC retry and timeout settings
-/// Default timeout used by `RobustProvider`
-pub const DEFAULT_MAX_TIMEOUT: Duration = Duration::from_secs(60);
-/// 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,
-    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.
-    ///
-    /// 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,
-            max_retries: DEFAULT_MAX_RETRIES,
-            min_delay: DEFAULT_MIN_DELAY,
-            _network: PhantomData,
-        }
-    }
-
-    /// Create a new `RobustProvider` with no retry attempts and only timeout set.
-    ///
-    /// The provided provider is treated as the primary provider.
-    #[must_use]
-    pub fn fragile(provider: P) -> Self {
-        Self::new(provider).max_retries(0).min_delay(Duration::ZERO)
-    }
-
-    /// Add a fallback provider to the list.
-    ///
-    /// 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
-    }
-
-    /// Set the maximum timeout for RPC operations.
-    #[must_use]
-    pub fn max_timeout(mut self, timeout: Duration) -> Self {
-        self.max_timeout = timeout;
-        self
-    }
-
-    /// Set the maximum number of retry attempts.
-    #[must_use]
-    pub fn max_retries(mut self, max_retries: usize) -> Self {
-        self.max_retries = max_retries;
-        self
-    }
-
-    /// Set the base delay for exponential backoff retries.
-    #[must_use]
-    pub fn min_delay(mut self, min_delay: Duration) -> Self {
-        self.min_delay = min_delay;
-        self
-    }
-
-    /// Build the `RobustProvider`.
-    ///
-    /// Final builder method: consumes the builder and returns the built [`RobustProvider`].
-    ///
-    /// # Errors
-    ///
-    /// 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());
-        }
-        Ok(RobustProvider {
-            providers,
-            max_timeout: self.max_timeout,
-            max_retries: self.max_retries,
-            min_delay: self.min_delay,
-        })
-    }
-}
+use crate::robust_provider::Error;
 
 /// Provider wrapper with built-in retry and timeout mechanisms.
 ///
@@ -219,10 +21,10 @@ impl<N: Network, P: IntoProvider<N>> RobustProviderBuilder<N, P> {
 /// The first provider in the vector is treated as the primary provider.
 #[derive(Clone)]
 pub struct RobustProvider<N: Network = Ethereum> {
-    providers: Vec<RootProvider<N>>,
-    max_timeout: Duration,
-    max_retries: usize,
-    min_delay: Duration,
+    pub(crate) providers: Vec<RootProvider<N>>,
+    pub(crate) max_timeout: Duration,
+    pub(crate) max_retries: usize,
+    pub(crate) min_delay: Duration,
 }
 
 impl<N: Network> RobustProvider<N> {
@@ -458,6 +260,7 @@ impl<N: Network> RobustProvider<N> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::robust_provider::builder::RobustProviderBuilder;
     use alloy::{
         consensus::BlockHeader,
         providers::{ProviderBuilder, WsConnect, ext::AnvilApi},