taceo_nodes_common/

web3.rs

//! HTTP RPC provider utilities for interacting with Ethereum nodes.
//!
//! This module provides configurable HTTP RPC providers built on top of
//! [`alloy`] transports. It supports:
//!
//! - HTTP RPC with automatic retry and exponential backoff
//! - Multiple HTTP endpoints with automatic failover
//! - Optional wallet integration for transaction signing
//!
//! Use [`HttpRpcProviderBuilder`] to build an HTTP RPC provider.
//! HTTP transports are wrapped with retry and fallback layers to improve
//! reliability when interacting with RPC endpoints.
//!
//! Use [`GetReceiptExt::get_receipt_with_retry`] to reliably wait for a
//! transaction receipt, working around cases where a load-balanced RPC
//! endpoint reports a transaction as confirmed but still returns no receipt
//! for it.
use core::fmt;
use std::{
    future::Future,
    num::NonZeroUsize,
    ops::Deref,
    task::{Context, Poll},
    time::Duration,
};

#[cfg(feature = "web3-asserter")]
use alloy::providers::mock::Asserter;
use alloy::{
    network::{Ethereum, EthereumWallet},
    primitives::ChainId,
    providers::{
        DynProvider, PendingTransactionBuilder, PendingTransactionError, Provider, ProviderBuilder,
        fillers::{BlobGasFiller, ChainIdFiller, NonceManager, SimpleNonceManager},
    },
    rpc::{
        client::RpcClient,
        json_rpc::{RequestPacket, ResponsePacket},
        types::TransactionReceipt,
    },
    transports::{
        RpcError, Transport, TransportError, TransportErrorKind, TransportFut,
        http::{
            Http,
            reqwest::{self, IntoUrl, Url},
        },
        layers::{FallbackLayer, OrRetryPolicyFn, RateLimitRetryPolicy, RetryPolicy},
    },
};
use backon::{BackoffBuilder, ExponentialBuilder, Retryable as _};
use serde::Deserialize;
use tower::{Layer, Service};

use crate::Environment;

pub use backon;

pub mod erc165;
pub mod event_stream;
pub mod signers;

/// A dedicated HTTP RPC provider.
///
/// This provider should be used for regular RPC calls, transaction
/// submission, and helpers such as ERC-165 queries.
#[derive(Clone)]
pub struct HttpRpcProvider(DynProvider);

/// Helper struct to redact the URLs when debug printing the config.
///
/// We don't use secret-string, because we want URL validation during config deserialization to fail early.
#[derive(Clone, Deserialize)]
#[serde(transparent)]
pub struct UrlRedacted(Url);

impl fmt::Debug for UrlRedacted {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("[REDACTED]")
    }
}

/// Configuration for building an [`HttpRpcProvider`].
///
/// Multiple HTTP endpoints can be provided to enable automatic failover.
/// Retry behavior can be tuned via [`RetryPolicyConfig`].
#[derive(Debug, Clone, Deserialize)]
#[non_exhaustive]
pub struct HttpRpcProviderConfig {
    /// List of HTTP RPC endpoints used for requests.
    ///
    /// Uses alloy's [`FallbackService`](https://docs.rs/alloy/latest/alloy/providers/transport/layers/struct.FallbackLayer.html) and configures each endpoint as one potential transport.
    pub http_urls: Vec<UrlRedacted>,
    /// Optional chain ID used by the provider.
    ///
    /// If provided, the [`ChainIdFiller`] will automatically populate
    /// transactions with this value.
    #[serde(default)]
    pub chain_id: Option<ChainId>,
    /// The poll interval for the confirmation heartbeat for alloy.
    ///
    /// Uses alloy's default setting if omitted. For `dev` environment 250ms
    /// and for all other environments 7s.
    #[serde(default)]
    #[serde(with = "humantime_serde")]
    pub confirmations_poll_interval: Option<Duration>,
    /// Retry configuration applied to RPC requests.
    #[serde(default)]
    pub retry_policy_config: RetryPolicyConfig,
}

/// Configuration for RPC retry behavior.
///
/// Requests that fail with retryable errors will be retried using
/// exponential backoff.
#[derive(Debug, Clone, Deserialize)]
#[non_exhaustive]
pub struct RetryPolicyConfig {
    /// Minimum delay between retries.
    ///
    /// Defaults to **1 second**.
    #[serde(default = "RetryPolicyConfig::default_min_delay")]
    #[serde(with = "humantime_serde")]
    pub min_delay: Duration,

    /// Maximum delay between retries.
    ///
    /// Defaults to **8 seconds**.
    #[serde(default = "RetryPolicyConfig::default_max_delay")]
    #[serde(with = "humantime_serde")]
    pub max_delay: Duration,

    /// Maximum number of retry attempts.
    ///
    /// Defaults to **5 retries**.
    #[serde(default = "RetryPolicyConfig::default_max_times")]
    pub max_times: usize,
}

impl HttpRpcProviderConfig {
    /// Creates a new configuration using default retry settings.
    ///
    /// # Errors
    ///
    /// Returns an error if any of the provided URLs cannot be parsed.
    pub fn with_default_values<I, U>(http_urls: I) -> reqwest::Result<Self>
    where
        I: IntoIterator<Item = U>,
        U: IntoUrl,
    {
        let http_urls = http_urls
            .into_iter()
            .map(|x| x.into_url().map(UrlRedacted))
            .collect::<reqwest::Result<Vec<_>>>()?;
        Ok(Self {
            http_urls,
            confirmations_poll_interval: None,
            chain_id: None,
            retry_policy_config: RetryPolicyConfig::default(),
        })
    }
}

impl RetryPolicyConfig {
    /// Default minimum delay between retries: 1 second
    fn default_min_delay() -> Duration {
        Duration::from_secs(1)
    }

    /// Default maximum delay between retries: 8 seconds
    fn default_max_delay() -> Duration {
        Duration::from_secs(8)
    }

    /// Default maximum number of retry attempts: 5
    fn default_max_times() -> usize {
        5
    }

    /// Initialize a `RetryPolicyConfig` with default values
    fn with_default_values() -> Self {
        Self {
            min_delay: Self::default_min_delay(),
            max_delay: Self::default_max_delay(),
            max_times: Self::default_max_times(),
        }
    }
}

impl Default for RetryPolicyConfig {
    fn default() -> Self {
        Self::with_default_values()
    }
}

fn build_transport_stack<S>(
    transports: Vec<S>,
    retry_policy_config: &RetryPolicyConfig,
) -> impl Transport + Clone
where
    S: Service<RequestPacket, Response = ResponsePacket, Error = TransportError>
        + Clone
        + Send
        + Sync
        + 'static,
    S::Future: Send,
{
    let retry_layer = RetryLayer::new(http_retry_policy(), retry_policy_config);
    let retrying_transports = transports
        .into_iter()
        .map(|transport| retry_layer.layer(transport))
        .collect::<Vec<_>>();
    let transport_count =
        NonZeroUsize::new(retrying_transports.len()).expect("transport stack must not be empty");

    // Retry each transport before fallback so JSON-RPC error responses cannot
    // win the fallback race against a slower healthy endpoint.
    FallbackLayer::default()
        .with_active_transport_count(transport_count)
        .layer(retrying_transports)
}

fn http_retry_policy() -> OrRetryPolicyFn {
    // Configure retry policy.
    //
    // The RateLimitRetryPolicy already handles 503 Service Unavailable and other common RPC errors.
    // We additionally check for other common transient errors:
    //   - 403 Forbidden
    //   - 408 Request Timeout
    //   - 502 Bad Gateway
    //   - 504 Gateway Timeout
    RateLimitRetryPolicy::default().or(|error: &TransportError| match error {
        RpcError::Transport(TransportErrorKind::HttpError(e)) => {
            matches!(e.status, 403 | 408 | 502 | 504)
        }
        RpcError::Transport(kind) => kind
            .as_custom()
            .and_then(|error| error.downcast_ref::<reqwest::Error>())
            .is_some_and(reqwest::Error::is_timeout),
        _ => false,
    })
}

/// Builder for constructing an [`HttpRpcProvider`].
///
/// The builder configures retry behavior, fallback transports, optional
/// wallet integration, and provider fillers before creating the provider.
pub struct HttpRpcProviderBuilder {
    http_urls: Vec<UrlRedacted>,
    retry_policy_config: RetryPolicyConfig,
    chain_id: Option<ChainId>,
    confirmations_poll_interval: Option<Duration>,
    is_local: bool,
    wallet: Option<EthereumWallet>,
    reqwest_client: Option<reqwest::Client>,
}

impl From<HttpRpcProviderConfig> for HttpRpcProviderBuilder {
    fn from(value: HttpRpcProviderConfig) -> Self {
        Self::from(&value)
    }
}

impl From<&HttpRpcProviderConfig> for HttpRpcProviderBuilder {
    fn from(value: &HttpRpcProviderConfig) -> Self {
        Self::with_config(value)
    }
}

impl HttpRpcProviderBuilder {
    /// Creates a new builder from the given configuration.
    ///
    /// # Panics
    ///
    /// Panics if `config.http_urls` is empty. At least one HTTP endpoint
    /// must be provided so that a transport stack can be constructed.
    #[must_use]
    pub fn with_config(config: &HttpRpcProviderConfig) -> Self {
        assert!(!config.http_urls.is_empty(), "http URLs must not be empty");
        Self {
            http_urls: config.http_urls.clone(),
            retry_policy_config: config.retry_policy_config.clone(),
            chain_id: config.chain_id,
            is_local: false,
            wallet: None,
            confirmations_poll_interval: config.confirmations_poll_interval,
            reqwest_client: None,
        }
    }

    /// Creates a new builder using default retry settings.
    ///
    /// # Errors
    ///
    /// Returns an error if any of the provided URLs cannot be parsed.
    ///
    /// # Example
    ///
    /// ```
    /// use taceo_nodes_common::web3::HttpRpcProviderBuilder;
    ///
    /// let builder = HttpRpcProviderBuilder::with_default_values(["http://127.0.0.1:8545"])?;
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    pub fn with_default_values<I, U>(http_urls: I) -> reqwest::Result<Self>
    where
        I: IntoIterator<Item = U>,
        U: IntoUrl,
    {
        Ok(Self::with_config(
            &HttpRpcProviderConfig::with_default_values(http_urls)?,
        ))
    }

    /// Explicitly sets the underling `reqwest::Client` for alloy.
    ///
    /// If not set, the builder will build a client with default configuration.
    #[must_use]
    pub fn reqwest_client(mut self, reqwest_client: reqwest::Client) -> Self {
        self.reqwest_client = Some(reqwest_client);
        self
    }

    /// Configures the environment used by the provider.
    #[must_use]
    pub fn environment(mut self, environment: Environment) -> Self {
        self.is_local = environment.is_dev();
        self
    }

    /// Sets the poll interval in which alloy fetches blocks for transaction confirmations.
    #[must_use]
    pub fn confirmations_poll_interval(mut self, confirmations_poll_interval: Duration) -> Self {
        self.confirmations_poll_interval = Some(confirmations_poll_interval);
        self
    }

    /// Sets the chain ID used by the provider.
    #[must_use]
    pub fn chain_id(mut self, chain_id: ChainId) -> Self {
        self.chain_id = Some(chain_id);
        self
    }

    /// Configures the retry behavior for HTTP RPC requests.
    #[must_use]
    pub fn retry_policy(mut self, retry_policy_config: RetryPolicyConfig) -> Self {
        self.retry_policy_config = retry_policy_config;
        self
    }

    /// Adds a wallet used for signing transactions.
    #[must_use]
    pub fn wallet(mut self, wallet: EthereumWallet) -> Self {
        self.wallet = Some(wallet);
        self
    }

    /// Builds the [`HttpRpcProvider`].
    ///
    /// Uses [`SimpleNonceManager::default()`] for nonce management. Use
    /// [`Self::build_with_nonce_manager`] to provide a custom nonce manager.
    ///
    /// # Errors
    ///
    /// Returns a [`TransportError`] if the HTTP transport stack cannot be
    /// initialized, including failures to create the underlying reqwest client.
    pub fn build(self) -> Result<HttpRpcProvider, TransportError> {
        self.build_with_nonce_manager(SimpleNonceManager::default())
    }

    /// Builds the [`HttpRpcProvider`] using the provided nonce manager.
    ///
    /// This allows callers to customize how transaction nonces are tracked
    /// while keeping the rest of the builder configuration unchanged.
    ///
    /// # Errors
    ///
    /// Returns a [`TransportError`] if the HTTP transport stack cannot be
    /// initialized, including failures to create the underlying reqwest client.
    pub fn build_with_nonce_manager<N: NonceManager + 'static>(
        self,
        nonce_manager: N,
    ) -> Result<HttpRpcProvider, TransportError> {
        let HttpRpcProviderBuilder {
            http_urls,
            retry_policy_config,
            chain_id,
            is_local,
            wallet,
            confirmations_poll_interval,
            reqwest_client,
        } = self;

        let reqwest = if let Some(reqwest_client) = reqwest_client {
            reqwest_client
        } else {
            reqwest::ClientBuilder::new()
                .build()
                .map_err(TransportErrorKind::custom)?
        };

        let transports = http_urls
            .into_iter()
            .map(|url| Http::with_client(reqwest.clone(), url.0))
            .collect::<Vec<_>>();
        let transport = build_transport_stack(transports, &retry_policy_config);

        let client = RpcClient::builder().transport(transport, is_local);
        let client = if let Some(confirmations_poll_interval) = confirmations_poll_interval {
            client.with_poll_interval(confirmations_poll_interval)
        } else {
            client
        };

        let http_provider_builder = ProviderBuilder::new()
            .filler(ChainIdFiller::new(chain_id))
            .filler(BlobGasFiller::default())
            .with_nonce_management(nonce_manager)
            .with_gas_estimation();

        let provider = if let Some(wallet) = wallet {
            http_provider_builder
                .wallet(wallet)
                .connect_client(client)
                .erased()
        } else {
            http_provider_builder.connect_client(client).erased()
        };

        Ok(HttpRpcProvider(provider))
    }
}

impl HttpRpcProvider {
    /// Returns the HTTP RPC provider.
    #[must_use]
    #[inline]
    pub fn inner(&self) -> DynProvider {
        self.0.clone()
    }

    /// Creates a provider backed by the given mocked [`Asserter`].
    ///
    /// This is intended for tests and uses Alloy's mocked provider
    /// infrastructure. See the Alloy mocking provider example for details:
    /// <https://alloy.rs/examples/providers/mocking/>.
    ///
    /// This method does not build a retry-layer. It must only be used for tests.
    #[cfg(feature = "web3-asserter")]
    #[must_use]
    pub fn with_mock_asserter(asserter: Asserter) -> Self {
        Self(
            ProviderBuilder::new()
                .connect_mocked_client(asserter)
                .erased(),
        )
    }
}

#[cfg(feature = "web3-asserter")]
impl From<Asserter> for HttpRpcProvider {
    fn from(value: Asserter) -> Self {
        Self::with_mock_asserter(value)
    }
}

impl AsRef<DynProvider> for HttpRpcProvider {
    fn as_ref(&self) -> &DynProvider {
        self
    }
}

impl Deref for HttpRpcProvider {
    type Target = DynProvider;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

#[derive(Debug, Clone)]
struct RetryLayer {
    policy: OrRetryPolicyFn,
    backoff: ExponentialBuilder,
}

impl RetryLayer {
    /// Creates a new retry layer using the provided retry policy and configuration.
    ///
    /// The retry behavior is implemented using exponential backoff with jitter.
    ///
    /// The following parameters are taken from [`RetryPolicyConfig`]:
    ///
    /// - minimum retry delay
    /// - maximum retry delay
    /// - maximum number of retry attempts
    pub fn new(policy: OrRetryPolicyFn, config: &RetryPolicyConfig) -> Self {
        let backoff = ExponentialBuilder::default()
            .with_min_delay(config.min_delay)
            .with_max_delay(config.max_delay)
            .with_max_times(config.max_times)
            .with_jitter();
        Self { policy, backoff }
    }
}

impl<S> Layer<S> for RetryLayer {
    type Service = RetryService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        RetryService {
            inner,
            policy: self.policy.clone(),
            backoff: self.backoff,
        }
    }
}

/// Tower service that wraps each request in a retry loop with exponential backoff.
#[derive(Debug, Clone)]
struct RetryService<S> {
    inner: S,
    policy: OrRetryPolicyFn,
    backoff: ExponentialBuilder,
}

impl<S> Service<RequestPacket> for RetryService<S>
where
    S: Service<RequestPacket, Response = ResponsePacket, Error = TransportError>
        + Clone
        + Send
        + Sync
        + 'static,
    S::Future: Send,
{
    type Response = ResponsePacket;
    type Error = TransportError;
    type Future = TransportFut<'static>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, request: RequestPacket) -> Self::Future {
        let service = self.clone();
        let backoff = self.backoff;
        let policy = self.policy.clone();

        Box::pin(async move {
            (|| service.clone().call_and_parse_error(request.clone()))
                .retry(backoff)
                .sleep(tokio::time::sleep)
                .when(|e| policy.should_retry(e))
                .notify(|err, duration| {
                    tracing::warn!(
                        ?err,
                        "Retrying RPC request after: {duration:?}. Reason: {err}"
                    );
                })
                // Adjust the backoff duration based on the policy and the current hint:
                // - If `dur` is `None`, we stop retrying (max attempts reached).
                // - If `dur` is `Some(d)` and the policy provides a backoff hint, use the policy hint.
                // - If `dur` is `Some(d)` and the policy hint is `None`, use the original `d`.
                .adjust(|e, dur| dur.and_then(|d| policy.backoff_hint(e).or(Some(d))))
                .await
        })
    }
}

impl<S> RetryService<S>
where
    S: Service<RequestPacket, Response = ResponsePacket, Error = TransportError>
        + Clone
        + Send
        + Sync
        + 'static,
    S::Future: Send,
{
    async fn call_and_parse_error(
        mut self,
        request: RequestPacket,
    ) -> Result<ResponsePacket, RpcError<TransportErrorKind>> {
        let resp = self.inner.call(request).await?;
        if let Some(e) = resp.as_error() {
            Err(TransportError::ErrorResp(e.to_owned()))
        } else {
            Ok(resp)
        }
    }
}

/// Extension trait adding [`get_receipt_with_retry`](GetReceiptExt::get_receipt_with_retry)
/// to [`PendingTransactionBuilder`].
pub trait GetReceiptExt: Sized {
    /// Waits for the receipt of an already-broadcast transaction, re-polling
    /// by transaction hash if the RPC does not serve the receipt right away.
    ///
    /// This works around a gap in `alloy`'s own
    /// [`PendingTransactionBuilder::get_receipt`]: it retries while a
    /// transaction is unconfirmed, but once its confirmation watcher sees the
    /// transaction confirmed and the immediately-following
    /// `eth_getTransactionReceipt` call returns `null` anyway, it gives up
    /// with no further retry. This is common behind a load-balanced RPC
    /// endpoint where the backend serving that second call hasn't caught up
    /// yet -- the transaction is already on chain, so treating a missing
    /// receipt as a failed submission is wrong.
    ///
    /// Tries [`PendingTransactionBuilder::get_receipt`] first, which already
    /// waits for the transaction to confirm. If that fails, falls back to
    /// retrying [`PendingTransactionBuilder::get_receipt`] again on a fresh
    /// builder for the same transaction hash and required confirmations,
    /// according to the given `builder` (e.g.
    /// [`ConstantBuilder`](backon::ConstantBuilder) or [`ExponentialBuilder`])
    /// until it stops yielding backoff durations.
    ///
    /// # Errors
    ///
    /// Returns the last [`PendingTransactionError`] seen once retries are
    /// exhausted.
    fn get_receipt_with_retry(
        self,
        builder: impl BackoffBuilder,
    ) -> impl Future<Output = Result<TransactionReceipt, PendingTransactionError>> + Send;

    /// Waits for the receipt of an already-broadcast transaction, re-polling
    /// by transaction hash if the RPC does not serve the receipt right away.
    ///
    /// This works around a gap in `alloy`'s own
    /// [`PendingTransactionBuilder::get_receipt`]: it retries while a
    /// transaction is unconfirmed, but once its confirmation watcher sees the
    /// transaction confirmed and the immediately-following
    /// `eth_getTransactionReceipt` call returns `null` anyway, it gives up
    /// with no further retry. This is common behind a load-balanced RPC
    /// endpoint where the backend serving that second call hasn't caught up
    /// yet -- the transaction is already on chain, so treating a missing
    /// receipt as a failed submission is wrong.
    ///
    /// Tries [`PendingTransactionBuilder::get_receipt`] first, which already
    /// waits for the transaction to confirm. If that fails, falls back to
    /// retrying [`PendingTransactionBuilder::get_receipt`] again on a fresh
    /// builder for the same transaction hash and required confirmations,
    /// 10 times with a 2-second delay between attempts.
    ///
    /// # Errors
    ///
    /// Returns the last [`PendingTransactionError`] seen once retries are
    /// exhausted.
    fn get_receipt_with_default_retry(
        self,
    ) -> impl Future<Output = Result<TransactionReceipt, PendingTransactionError>> + Send {
        self.get_receipt_with_retry(
            backon::ConstantBuilder::new()
                .with_delay(Duration::from_secs(2))
                .with_max_times(10),
        )
    }
}

impl GetReceiptExt for PendingTransactionBuilder<Ethereum> {
    async fn get_receipt_with_retry(
        self,
        builder: impl BackoffBuilder,
    ) -> Result<TransactionReceipt, PendingTransactionError> {
        let tx_hash = *self.tx_hash();
        let provider = self.provider().clone();
        let config = self.inner().clone();

        match self.get_receipt().await {
            Ok(receipt) => return Ok(receipt),
            Err(err) => {
                tracing::warn!("no receipt for transaction {tx_hash} yet ({err}), re-polling");
            }
        }

        let poll = || async {
            let pending = PendingTransactionBuilder::from_config(provider.clone(), config.clone());
            pending.get_receipt().await
        };

        poll.retry(builder)
            .sleep(tokio::time::sleep)
            .notify(|err, dur| {
                tracing::warn!(
                    "failed to fetch receipt for transaction {tx_hash} ({err}), retrying in {dur:?}"
                );
            })
            .await
    }
}

#[cfg(test)]
pub(crate) mod tests;