Add robust subscription (#172)

Co-authored-by: Nenad <[email protected]>

Author: LeoPatOZ

Cargo.lock

@@ -60,6 +60,7 @@ alloy-node-bindings.workspace = true
 tokio-stream.workspace = true
 tracing.workspace = true
 backon.workspace = true
+tokio-util = "0.7.17"
 
 [dev-dependencies]
 tracing-subscriber.workspace = true

Cargo.toml

@@ -270,6 +270,44 @@ All examples spin up a local `anvil` instance, deploy a demo counter contract, a
 
 ---
 
+## Robust Provider
+
+`event-scanner` ships with a `robust_provider` module that wraps Alloy providers with:
+
+- bounded per-call timeouts and exponential backoff retries
+- automatic failover from a primary provider to one or more fallbacks
+- resilient WebSocket block subscriptions with timeout handling and reconnection.
+
+The main entry point is `robust_provider::RobustProviderBuilder`, which accepts a wide
+range of provider types (URLs, `RootProvider`, layered providers, etc.) through the
+`IntoRobustProvider` and `IntoRobustProvider` traits.
+
+A typical setup looks like:
+
+```rust
+use alloy::providers::ProviderBuilder;
+use event_scanner::robust_provider::RobustProviderBuilder;
+use std::time::Duration;
+
+async fn example() -> anyhow::Result<()> {
+    let ws = ProviderBuilder::new().connect("ws://localhost:8545").await?;
+    let http = ProviderBuilder::new().connect_http("http://localhost:8545".parse()?);
+
+    let provider = RobustProviderBuilder::new(ws)
+        .fallback(http)
+        .call_timeout(Duration::from_secs(30))
+        .subscription_timeout(Duration::from_secs(120))
+        .build()
+        .await?;
+    Ok(())
+}
+```
+
+You can then pass this `robust` provider into `EventScannerBuilder::connect` just like
+any other provider.
+
+---
+
 ## Testing
 
 (We recommend using [nextest](https://crates.io/crates/cargo-nextest) to run the tests)

README.md

@@ -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/historical_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/latest_events_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/live_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_block_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_latest_scanning/main.rs

@@ -59,6 +59,7 @@
 //! }
 //! ```
 
+use crate::robust_provider::subscription::{self, RobustSubscription};
 use std::{cmp::Ordering, ops::RangeInclusive};
 use tokio::{
     sync::{mpsc, oneshot},
@@ -68,7 +69,7 @@ use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 
 use crate::{
     ScannerError, ScannerMessage,
-    robust_provider::{Error as RobustProviderError, IntoRobustProvider, RobustProvider},
+    robust_provider::{IntoRobustProvider, RobustProvider, provider::Error as RobustProviderError},
     types::{IntoScannerResult, Notification, ScannerResult, TryStream},
 };
 
@@ -77,7 +78,6 @@ use alloy::{
     eips::BlockId,
     network::{BlockResponse, Network, primitives::HeaderResponse},
     primitives::{B256, BlockNumber},
-    pubsub::Subscription,
 };
 use tracing::{debug, error, info, warn};
 
@@ -583,16 +583,45 @@ impl<N: Network> Service<N> {
 
     async fn stream_live_blocks(
         mut range_start: BlockNumber,
-        subscription: Subscription<N::HeaderResponse>,
+        subscription: RobustSubscription<N>,
         sender: mpsc::Sender<BlockScannerResult>,
         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");
+                    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;
+                        }
+                    }
+                }
+            };
 
-        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/block_range_scanner.rs

@@ -6,7 +6,7 @@ use alloy::{
 };
 use thiserror::Error;
 
-use crate::{robust_provider::Error as RobustProviderError, types::ScannerResult};
+use crate::{robust_provider::provider::Error as RobustProviderError, types::ScannerResult};
 
 #[derive(Error, Debug, Clone)]
 pub enum ScannerError {
@@ -30,6 +30,9 @@ pub enum ScannerError {
 
     #[error("Max block range must be greater than 0")]
     InvalidMaxBlockRange,
+
+    #[error("Subscription closed")]
+    SubscriptionClosed,
 }
 
 impl From<RobustProviderError> for ScannerError {
@@ -47,6 +50,7 @@ impl From<RpcError<TransportErrorKind>> for ScannerError {
         ScannerError::RpcError(Arc::new(error))
     }
 }
+
 impl<T: Clone> PartialEq<ScannerError> for ScannerResult<T> {
     fn eq(&self, other: &ScannerError) -> bool {
         match self {