Fix the retry layer

crates/node/src/provider_utils/fallback_gas_filler.rs

@@ -1,5 +1,3 @@
-use std::{borrow::Cow, fmt::Display};
-
 use alloy::{
     eips::BlockNumberOrTag,
     network::{Network, TransactionBuilder},
@@ -111,28 +109,23 @@ where
                             },
                             state_overrides: Default::default(),
                             block_overrides: Default::default(),
+                            tx_index: Default::default(),
                         },
                     )
                     .await?
                     .try_into_call_frame()
                     .map_err(|err| {
-                        RpcError::LocalUsageError(
-                            FallbackGasFillerError::new(format!(
-                                "Expected a callframe trace, but got: {err:?}"
-                            ))
-                            .boxed(),
+                        RpcError::local_usage_str(
+                            format!("Expected a callframe trace, but got: {err:?}").as_str(),
                         )
                     })?;
 
                 let gas_used = u64::try_from(trace.gas_used).map_err(|_| {
-                    RpcError::LocalUsageError(
-                        FallbackGasFillerError::new(
-                            "Transaction trace returned a value of gas used that exceeds u64",
-                        )
-                        .boxed(),
+                    RpcError::local_usage_str(
+                        "Transaction trace returned a value of gas used that exceeds u64",
                     )
                 })?;
-                let gas_limit = gas_used.saturating_mul(120) / 100;
+                let gas_limit = gas_used.saturating_mul(2);
 
                 if let Some(gas_price) = tx.gas_price() {
                     return Ok(GasFillable::Legacy {
@@ -174,24 +167,3 @@ impl Default for FallbackGasFiller {
         Self::new()
     }
 }
-
-#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
-struct FallbackGasFillerError(Cow<'static, str>);
-
-impl FallbackGasFillerError {
-    pub fn new(string: impl Into<Cow<'static, str>>) -> Self {
-        Self(string.into())
-    }
-
-    pub fn boxed(self) -> Box<Self> {
-        Box::new(self)
-    }
-}
-
-impl Display for FallbackGasFillerError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        Display::fmt(&self.0, f)
-    }
-}
-
-impl std::error::Error for FallbackGasFillerError {}

crates/node/src/provider_utils/provider.rs

@@ -1,4 +1,4 @@
-use std::{sync::LazyLock, time::Duration};
+use std::sync::LazyLock;
 
 use alloy::{
     network::{Network, NetworkWallet, TransactionBuilder4844},
@@ -10,16 +10,14 @@ use alloy::{
 };
 use anyhow::{Context, Result};
 
-use crate::provider_utils::{
-    ConcurrencyLimiterLayer, FallbackGasFiller, ReceiptRetryLayer, ReceiptRetryProvider,
-};
+use crate::provider_utils::{ConcurrencyLimiterLayer, FallbackGasFiller, ReceiptRetryLayer};
 
 pub type ConcreteProvider<N, W> = FillProvider<
     JoinFill<
         JoinFill<JoinFill<JoinFill<Identity, FallbackGasFiller>, ChainIdFiller>, NonceFiller>,
         WalletFiller<W>,
     >,
-    ReceiptRetryProvider<RootProvider<N>, N>,
+    RootProvider<N>,
     N,
 >;
 
@@ -48,6 +46,7 @@ where
 
     let client = ClientBuilder::default()
         .layer(GLOBAL_CONCURRENCY_LIMITER_LAYER.clone())
+        .layer(ReceiptRetryLayer::default())
         .connect(rpc_url)
         .await
         .context("Failed to construct the RPC client")?;
@@ -55,11 +54,6 @@ where
     let provider = ProviderBuilder::new()
         .disable_recommended_fillers()
         .network::<N>()
-        .layer(
-            ReceiptRetryLayer::default()
-                .with_polling_duration(Duration::from_secs(90))
-                .with_polling_interval(Duration::from_millis(500)),
-        )
         .filler(fallback_gas_filler)
         .filler(chain_id_filler)
         .filler(nonce_filler)

crates/node/src/provider_utils/receipt_retry_layer.rs

@@ -1,12 +1,12 @@
-use std::{marker::PhantomData, time::Duration};
+use std::time::Duration;
 
 use alloy::{
-    network::Network,
-    primitives::TxHash,
-    providers::{Provider, ProviderCall, ProviderLayer, RootProvider},
-    transports::{RpcError, TransportErrorKind},
+    network::{AnyNetwork, Network},
+    rpc::json_rpc::{RequestPacket, ResponsePacket},
+    transports::{TransportError, TransportErrorKind, TransportFut},
 };
 use tokio::time::{interval, timeout};
+use tower::{Layer, Service};
 
 /// A layer that allows for automatic retries for getting the receipt.
 ///
@@ -57,95 +57,100 @@ impl Default for ReceiptRetryLayer {
     }
 }
 
-impl<P, N> ProviderLayer<P, N> for ReceiptRetryLayer
-where
-    P: Provider<N>,
-    N: Network,
-{
-    type Provider = ReceiptRetryProvider<P, N>;
+impl<S> Layer<S> for ReceiptRetryLayer {
+    type Service = ReceiptRetryService<S>;
 
-    fn layer(&self, inner: P) -> Self::Provider {
-        ReceiptRetryProvider::new(self.polling_duration, self.polling_interval, inner)
+    fn layer(&self, inner: S) -> Self::Service {
+        ReceiptRetryService {
+            service: inner,
+            polling_duration: self.polling_duration,
+            polling_interval: self.polling_interval,
+        }
     }
 }
 
-#[derive(Debug, Clone)]
-pub struct ReceiptRetryProvider<P, N> {
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct ReceiptRetryService<S> {
+    /// The internal service.
+    service: S,
+
     /// The amount of time to keep polling for the receipt before considering it a timeout.
     polling_duration: Duration,
 
     /// The interval of time to wait between each poll for the receipt.
     polling_interval: Duration,
-
-    /// Inner provider.
-    inner: P,
-
-    /// Phantom data
-    phantom: PhantomData<N>,
 }
 
-impl<P, N> ReceiptRetryProvider<P, N>
+impl<S> Service<RequestPacket> for ReceiptRetryService<S>
 where
-    P: Provider<N>,
-    N: Network,
+    S: Service<RequestPacket, Future = TransportFut<'static>, Error = TransportError>
+        + Send
+        + 'static
+        + Clone,
 {
-    /// Instantiate a new cache provider.
-    pub const fn new(polling_duration: Duration, polling_interval: Duration, inner: P) -> Self {
-        Self {
-            inner,
-            polling_duration,
-            polling_interval,
-            phantom: PhantomData,
-        }
-    }
-}
+    type Response = ResponsePacket;
+    type Error = TransportError;
+    type Future = TransportFut<'static>;
 
-impl<P, N> Provider<N> for ReceiptRetryProvider<P, N>
-where
-    P: Provider<N>,
-    N: Network,
-{
-    #[inline(always)]
-    fn root(&self) -> &RootProvider<N> {
-        self.inner.root()
+    fn poll_ready(
+        &mut self,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
+        self.service.poll_ready(cx)
     }
 
-    fn get_transaction_receipt(
-        &self,
-        hash: TxHash,
-    ) -> ProviderCall<(TxHash,), Option<<N as Network>::ReceiptResponse>> {
-        let client = self.inner.weak_client();
-        let polling_duration = self.polling_duration;
+    #[allow(clippy::nonminimal_bool)]
+    fn call(&mut self, req: RequestPacket) -> Self::Future {
+        type ReceiptOutput = <AnyNetwork as Network>::ReceiptResponse;
+
+        let mut service = self.service.clone();
         let polling_interval = self.polling_interval;
+        let polling_duration = self.polling_duration;
 
-        ProviderCall::BoxedFuture(Box::pin(async move {
-            let client = client
-                .upgrade()
-                .ok_or_else(|| TransportErrorKind::custom_str("RPC client dropped"))?;
+        Box::pin(async move {
+            let request = req.as_single().ok_or_else(|| {
+                TransportErrorKind::custom_str("Retry layer doesn't support batch requests")
+            })?;
+            let method = request.method();
+            let requires_retries = method == "eth_getTransactionReceipt"
+                || (method.contains("debug") && method.contains("trace"));
 
-            let receipt = timeout(polling_duration, async move {
+            if !requires_retries {
+                return service.call(req).await;
+            }
+
+            timeout(polling_duration, async {
                 let mut interval = interval(polling_interval);
 
                 loop {
-                    let result = client
-                        .request::<(TxHash,), Option<<N as Network>::ReceiptResponse>>(
-                            "eth_getTransactionReceipt",
-                            (hash,),
-                        )
-                        .await;
-                    if let Ok(Some(receipt)) = result {
-                        return receipt;
+                    interval.tick().await;
+
+                    let Ok(resp) = service.call(req.clone()).await else {
+                        continue;
+                    };
+                    let response = resp.as_single().expect("Can't fail");
+                    if response.is_error() {
+                        continue;
                     }
 
-                    interval.tick().await;
+                    if method == "eth_getTransactionReceipt"
+                        && response
+                            .payload()
+                            .clone()
+                            .deserialize_success::<ReceiptOutput>()
+                            .ok()
+                            .and_then(|resp| resp.try_into_success().ok())
+                            .is_some()
+                        || method != "eth_getTransactionReceipt"
+                    {
+                        return resp;
+                    } else {
+                        continue;
+                    }
                 }
             })
             .await
-            .map_err(|_| {
-                RpcError::local_usage_str("Timeout when waiting for transaction receipt")
-            })?;
-
-            Ok(Some(receipt))
-        }))
+            .map_err(|_| TransportErrorKind::custom_str("Timeout when retrying request"))
+        })
     }
 }