Feat/http subscription

Cargo.toml

@@ -27,7 +27,6 @@ backon = "1.6.0"
 tokio-stream = "0.1.17"
 thiserror = "2.0.17"
 tokio-util = "0.7.17"
-futures-util = "0.3"
 
 tracing = { version = "0.1", optional = true }
 serde_json = "1.0.149"

README.md

@@ -163,6 +163,26 @@ while let Some(result) = stream.next().await {
 - When a fallback fails, the primary is tried first before moving to the next fallback.
 - The `Lagged` error indicates the consumer is not keeping pace with incoming blocks.
 
+#### HTTP-based subscriptions (feature flag)
+
+By default, subscriptions use WebSocket/pubsub-capable providers. Normally, HTTP-only providers are skipped during subscription retries. If your environment only exposes HTTP endpoints, you can enable HTTP-based block subscriptions via polling using the `http-subscription` Cargo feature:
+
+```toml
+[dependencies]
+robust-provider = { version = "1.0.0", features = ["http-subscription"] }
+```
+
+With this feature enabled and `allow_http_subscriptions(true)` is set, those HTTP providers can also act as subscription sources via polling, and are treated like regular pubsub-capable providers in the retry/failover logic:
+
+```rust
+let robust = RobustProviderBuilder::new(http_provider)
+    .allow_http_subscriptions(true)
+    // Optional: tune how often to poll for new blocks (defaults to ~12s)
+    .poll_interval(Duration::from_secs(12))
+    .build()
+    .await?;
+```
+
 ---
 
 ## Provider Conversion

src/robust_provider/builder.rs

@@ -153,6 +153,16 @@ impl<N: Network, P: IntoRootProvider<N>> RobustProviderBuilder<N, P> {
     /// # Feature Flag
     ///
     /// This method requires the `http-subscription` feature.
+    ///
+    /// # Example
+    ///
+    /// ```rust,ignore
+    /// let robust = RobustProviderBuilder::new(http_provider)
+    ///     .allow_http_subscriptions(true)
+    ///     .poll_interval(Duration::from_secs(6)) // For faster chains
+    ///     .build()
+    ///     .await?;
+    /// ```
     #[cfg(feature = "http-subscription")]
     #[must_use]
     pub fn poll_interval(mut self, interval: Duration) -> Self {
@@ -169,7 +179,9 @@ impl<N: Network, P: IntoRootProvider<N>> RobustProviderBuilder<N, P> {
     ///
     /// * **Latency**: New blocks detected with up to `poll_interval` delay
     /// * **RPC Load**: Generates one RPC call per `poll_interval`
-    /// * **Missed Blocks**: If `poll_interval` > block time, intermediate blocks may be missed
+    /// * **Intermediate Blocks**: Depending on the node/provider semantics, you may not observe
+    ///   every intermediate block when `poll_interval` is larger than the chain's block time (e.g.
+    ///   if only the latest head is exposed).
     ///
     /// # Feature Flag
     ///
@@ -180,7 +192,6 @@ impl<N: Network, P: IntoRootProvider<N>> RobustProviderBuilder<N, P> {
     /// ```rust,ignore
     /// let robust = RobustProviderBuilder::new(http_provider)
     ///     .allow_http_subscriptions(true)
-    ///     .poll_interval(Duration::from_secs(6)) // For faster chains
     ///     .build()
     ///     .await?;
     /// ```

src/robust_provider/provider.rs

@@ -501,6 +501,12 @@ impl<N: Network> RobustProvider<N> {
     pub async fn subscribe_blocks(&self) -> Result<RobustSubscription<N>, Error> {
         let subscription: SubscriptionBackend<N> = self
             .try_operation_with_failover(move |provider| async move {
+                // if HTTP subscriptions are enabled and the provider currently being tried is HTTP,
+                // we will attempt to connect using it.
+                // Otherwise try subscribing through a PubSub operation, and if the provider is HTTP
+                // just let it fail; the error will be non-retriable, so the algorithm will
+                // automatically switch to the next fallback provider (see
+                // `try_provider_with_timeout`).
                 #[cfg(feature = "http-subscription")]
                 {
                     let not_pubsub = provider.client().pubsub_frontend().is_none();
@@ -513,9 +519,6 @@ impl<N: Network> RobustProvider<N> {
                         });
                     }
                 }
-                // Non-pubsub providers will properly trigger fallback logic without retries because
-                // they return an appropriate RPC error, see the match logic in
-                // `try_provider_with_timeout`.
                 provider
                     .subscribe_blocks()
                     .channel_size(self.subscription_buffer_capacity)
@@ -539,8 +542,6 @@ impl<N: Network> RobustProvider<N> {
     /// If the timeout is exceeded and fallback providers are available, it will
     /// attempt to use each fallback provider in sequence.
     ///
-    /// If `require_pubsub` is true, providers that don't support pubsub will be skipped.
-    ///
     /// # Errors
     ///
     /// * [`CoreError::RpcError`] - if no fallback providers succeeded; contains the last error

src/robust_provider/subscription.rs

@@ -50,18 +50,18 @@ impl<N: Network> From<Subscription<N::HeaderResponse>> for SubscriptionBackend<N
 #[cfg(feature = "http-subscription")]
 impl<N: Network> From<PollerBuilder<(U256,), Vec<BlockHash>>> for SubscriptionBackend<N> {
     fn from(value: PollerBuilder<(U256,), Vec<BlockHash>>) -> Self {
-        use futures_util::{StreamExt, stream};
+        use tokio_stream::StreamExt;
 
         let (sender, receiver) = mpsc::channel(value.channel_size());
 
-        // Spawn a task to forward block hashes to the channel
-        let stream = value.into_stream().flat_map(stream::iter);
+        let mut stream = value.into_stream();
         tokio::spawn(async move {
-            let mut stream = std::pin::pin!(stream);
-            while let Some(hash) = stream.next().await {
-                if sender.send(hash).await.is_err() {
-                    // Receiver dropped, stop polling
-                    break;
+            while let Some(hashes) = stream.next().await {
+                for hash in hashes {
+                    if sender.send(hash).await.is_err() {
+                        // Receiver dropped, stop polling
+                        break;
+                    }
                 }
             }
         });
@@ -204,6 +204,12 @@ impl<N: Network> RobustSubscription<N> {
         let allow_http_subscriptions = self.robust_provider.allow_http_subscriptions;
 
         let operation = move |provider: RootProvider<N>| async move {
+            // if HTTP subscriptions are enabled and the provider currently being tried is HTTP,
+            // we will attempt to connect using it.
+            // Otherwise try subscribing through a PubSub operation, and if the provider is HTTP
+            // just let it fail; the error will be non-retriable, so the algorithm will
+            // automatically switch to the next fallback provider (see
+            // `try_provider_with_timeout`).
             #[cfg(feature = "http-subscription")]
             {
                 let not_pubsub = provider.client().pubsub_frontend().is_none();