newton-chainio 0.5.2

//! Provider construction helpers.
//!
//! The eigensdk `get_provider` builds a bare HTTP provider with no transport
//! middleware, so a JSON-RPC node returning `HTTP 429 Too Many Requests`
//! surfaces straight to the caller as a hard error. In mainnet we share a single
//! upstream account (~250 req/s on QuickNode) across every gateway; the periodic
//! operator-discovery refresh (`query_registered_operator_and_fill_db`) fans out
//! a burst of contract reads, and whichever gateway loses the race for the
//! shared budget eats the 429, fails to populate its operator pool, and reports
//! a pool size of 0 while its peers report the correct count.
//!
//! [`get_provider_with_retry`] mirrors `eigensdk::common::get_provider` (same
//! `SdkProvider` return type) and attaches two transport layers:
//!
//! * A **proactive** [`ThrottleLayer`] that paces outbound requests to a
//!   per-gateway share of the shared budget (see [`configured_rps`]). This is
//!   the layer that actually keeps a gateway under the upstream ceiling — it
//!   throttles every request, not just retries. A single rate limiter is shared
//!   process-wide (one process == one gateway), so the four providers built per
//!   discovery tick draw from one budget rather than four independent ones.
//! * A **reactive** [`RetryBackoffLayer`] that retries `HTTP 429`/`503` with
//!   backoff and honors a provider-supplied backoff hint when present. This is a
//!   safety net for residual rate-limit responses (e.g. bursts from other RPC
//!   paths that share the account, or an under-provisioned throttle); it does
//!   not by itself bound the shared budget, and its compute-unit pacing only
//!   engages *after* a retryable error inside the retry loop.
//!
//! Retry is layered outside throttle so a retried attempt also waits for a
//! throttle permit rather than bypassing the rate limit.

use alloy::{
    providers::ProviderBuilder,
    rpc::client::RpcClient,
    transports::{
        http::reqwest::Url,
        layers::{RetryBackoffLayer, ThrottleLayer},
    },
};
use eigensdk::common::SdkProvider;
use std::sync::LazyLock;
use tracing::info;

/// Maximum number of rate-limit retries before giving up and surfacing the error.
const MAX_RATE_LIMIT_RETRIES: u32 = 5;

/// Initial backoff in milliseconds applied when no provider backoff hint is present.
const INITIAL_BACKOFF_MS: u64 = 200;

/// Compute-units-per-second budget for the retry layer's post-error pacing.
/// This only affects backoff once a retryable error has occurred; it is not a
/// steady-state request cap (that is the throttle layer's job).
const COMPUTE_UNITS_PER_SECOND: u64 = 4_000;

/// Per-gateway request-rate cap for the production deployment.
///
/// Shared upstream budget (~250 req/s) divided across ~10 prod gateways with
/// headroom for other RPC traffic: `250 * 0.8 / 10 ≈ 20`.
const PROD_RPS: u32 = 20;

/// Per-gateway request-rate cap for non-prod deployments (e.g. `stagef`).
///
/// Same budget across ~2 staging gateways: `250 * 0.8 / 2 ≈ 100`. Local anvil
/// runs also resolve here (default `DEPLOYMENT_ENV`), and 100 req/s is well
/// above what local dev/tests need, so it is effectively a no-op there.
const NON_PROD_RPS: u32 = 100;

/// Build a read-only [`SdkProvider`] with throttle + retry transport layers.
///
/// Replacement for `eigensdk::common::get_provider`, returning `Result` instead
/// of panicking on a bad URL (this runs on every discovery refresh tick, not
/// just at startup, so a transient/misconfigured URL should be a skippable
/// error rather than a panic).
pub fn get_provider_with_retry(rpc_url: &str) -> eyre::Result<SdkProvider> {
    let url = Url::parse(rpc_url).map_err(|e| eyre::eyre!("invalid RPC URL {rpc_url:?}: {e}"))?;

    let retry_layer = RetryBackoffLayer::new(MAX_RATE_LIMIT_RETRIES, INITIAL_BACKOFF_MS, COMPUTE_UNITS_PER_SECOND);
    // Retry is the outer layer so a retried request still passes through (and
    // waits on) the throttle below it.
    let builder = RpcClient::builder().layer(retry_layer);

    let client = match shared_throttle_layer() {
        Some(throttle) => builder.layer(throttle).http(url),
        None => builder.http(url),
    };

    Ok(ProviderBuilder::new().connect_client(client))
}

/// Process-wide throttle layer, lazily initialized from [`configured_rps`].
///
/// `ThrottleLayer` owns its rate limiter, so a fresh `ThrottleLayer::new` per
/// provider would give each its own budget. We initialize one limiter for the
/// process so all providers (the four built per discovery tick included) are
/// governed together. `None` means throttling is disabled.
static SHARED_THROTTLE: LazyLock<Option<ThrottleLayer>> = LazyLock::new(|| {
    configured_rps().map(|rps| {
        info!(requests_per_second = rps, "throttling discovery RPC provider");
        ThrottleLayer::new(rps)
    })
});

/// Hand out a clone of the process-wide throttle layer, or `None` if disabled.
///
/// The clone shares the same underlying `Arc` rate limiter, so every provider
/// enforces the one process-wide budget rather than its own.
fn shared_throttle_layer() -> Option<ThrottleLayer> {
    SHARED_THROTTLE.as_ref().map(|layer| ThrottleLayer {
        throttle: layer.throttle.clone(),
    })
}

/// Resolve the per-gateway requests-per-second cap, or `None` to disable.
///
/// Precedence:
/// 1. `RPC_MAX_REQUESTS_PER_SECOND` override — `0` disables throttling, any
///    other positive value is used verbatim (incident retuning without a code
///    change); a non-numeric value is ignored with a warning.
/// 2. Otherwise derived from `DEPLOYMENT_ENV` (`prod` vs everything else).
fn configured_rps() -> Option<u32> {
    if let Ok(raw) = std::env::var("RPC_MAX_REQUESTS_PER_SECOND") {
        match raw.trim().parse::<u32>() {
            Ok(0) => return None,
            Ok(rps) => return Some(rps),
            // Fail closed: a non-numeric override resolves to the conservative
            // prod cap rather than silently falling through to the looser
            // staging default.
            Err(_) => {
                tracing::warn!(
                    value = %raw,
                    rps = PROD_RPS,
                    "non-numeric RPC_MAX_REQUESTS_PER_SECOND; failing closed to prod cap"
                );
                return Some(PROD_RPS);
            }
        }
    }
    // Pass the raw var (empty when unset) straight through; `rps_for_deployment_env`
    // fails closed on anything it doesn't recognize as explicitly non-prod.
    let deployment_env = std::env::var("DEPLOYMENT_ENV").unwrap_or_default();
    Some(rps_for_deployment_env(&deployment_env))
}

/// Map a deployment-env name to its per-gateway requests-per-second cap.
///
/// Fails closed: only env names we explicitly recognize as non-prod get the
/// looser staging budget. Everything else — `prod`, an unset/empty var, case
/// drift (`Prod`), or aliases (`production`, `mainnet`) — resolves to the
/// tightest `PROD_RPS`. Running 5x too hot on a prod misconfig (10×100 blows
/// past the shared ~250/s ceiling exactly when something is already wrong) is
/// the dangerous direction; throttling staging slightly too hard is harmless.
fn rps_for_deployment_env(deployment_env: &str) -> u32 {
    match deployment_env {
        "stagef" | "staget" | "stageg" | "local" | "dev" => NON_PROD_RPS,
        _ => PROD_RPS,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn recognized_non_prod_envs_use_non_prod_rps() {
        for env in ["stagef", "staget", "stageg", "local", "dev"] {
            assert_eq!(rps_for_deployment_env(env), NON_PROD_RPS, "env {env}");
        }
    }

    #[test]
    fn prod_and_unrecognized_envs_fail_closed_to_prod_rps() {
        // prod, unset/empty, case drift, and aliases all resolve to the tight cap.
        for env in ["prod", "", "Prod", "PROD", "production", "mainnet", "garbage"] {
            assert_eq!(rps_for_deployment_env(env), PROD_RPS, "env {env:?}");
        }
    }
}