espresso_types/v0/impls/

l1.rs

use std::{cmp::Ordering, fmt, num::NonZeroU64, str::FromStr, sync::Arc};
#[cfg(feature = "node")]
use std::{cmp::min, num::NonZeroUsize, pin::Pin, result::Result as StdResult, time::Instant};

#[cfg(feature = "node")]
use alloy::{
    eips::BlockId,
    hex,
    primitives::Address,
    providers::{Provider, ProviderBuilder, WsConnect},
    rpc::{
        client::RpcClient,
        json_rpc::{ErrorPayload, RequestPacket, ResponsePacket},
    },
    transports::{HttpError, RpcError, TransportErrorKind, http::Http},
};
use alloy::{
    primitives::{B256, U256},
    rpc::types::Block,
};
#[cfg(feature = "node")]
use anyhow::Context;
#[cfg(feature = "node")]
use async_broadcast::Sender;
#[cfg(feature = "node")]
use async_trait::async_trait;
use clap::Parser;
use committable::{Commitment, Committable, RawCommitmentBuilder};
#[cfg(feature = "node")]
use espresso_utils::redact::redact_url;
#[cfg(feature = "node")]
use futures::{
    future::{Future, TryFuture, TryFutureExt},
    stream::{self, StreamExt},
};
#[cfg(feature = "node")]
use hotshot_contract_adapter::sol_types::FeeContract;
use hotshot_types::traits::metrics::Metrics;
#[cfg(feature = "node")]
use lru::LruCache;
#[cfg(feature = "node")]
use parking_lot::RwLock;
#[cfg(feature = "node")]
use tokio::{
    spawn,
    sync::{Mutex, MutexGuard, Notify},
    time::{Duration, sleep},
};
#[cfg(feature = "node")]
use tower_service::Service;
#[cfg(feature = "node")]
use tracing::Instrument;
#[cfg(feature = "node")]
use url::Url;

use super::{
    L1BlockInfo, L1SafetyMargin,
    v0_1::{L1BlockInfoWithParent, ParseL1SafetyMarginError},
};
#[cfg(feature = "node")]
use super::{
    L1ClientMetrics, L1State, L1UpdateTask,
    v0_1::{SingleTransport, SingleTransportStatus, SwitchingTransport},
};
use crate::L1ClientOptions;
#[cfg(feature = "node")]
use crate::{FeeInfo, L1Client, L1Event, L1Snapshot};

impl PartialOrd for L1BlockInfo {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for L1BlockInfo {
    fn cmp(&self, other: &Self) -> Ordering {
        self.number.cmp(&other.number)
    }
}

impl From<&Block> for L1BlockInfo {
    fn from(block: &Block) -> Self {
        Self {
            number: block.header.number,
            timestamp: U256::from(block.header.timestamp),
            hash: block.header.hash,
        }
    }
}

impl From<&Block> for L1BlockInfoWithParent {
    fn from(block: &Block) -> Self {
        Self {
            info: block.into(),
            parent_hash: block.header.parent_hash,
        }
    }
}

impl Committable for L1BlockInfo {
    fn commit(&self) -> Commitment<Self> {
        let timestamp: [u8; 32] = self.timestamp.to_le_bytes();

        RawCommitmentBuilder::new(&Self::tag())
            .u64_field("number", self.number)
            // `RawCommitmentBuilder` doesn't have a `u256_field` method, so we simulate it:
            .constant_str("timestamp")
            .fixed_size_bytes(&timestamp)
            .constant_str("hash")
            .fixed_size_bytes(&self.hash.0)
            .finalize()
    }

    fn tag() -> String {
        "L1BLOCK".into()
    }
}

impl L1BlockInfo {
    pub fn number(&self) -> u64 {
        self.number
    }

    pub fn timestamp(&self) -> U256 {
        self.timestamp
    }

    pub fn hash(&self) -> B256 {
        self.hash
    }
}

impl L1SafetyMargin {
    /// Hash-chain verify every block, no matter how old.
    pub const UNLIMITED: Self = Self(None);

    pub fn blocks(self) -> Option<NonZeroU64> {
        self.0
    }
}

impl From<NonZeroU64> for L1SafetyMargin {
    fn from(blocks: NonZeroU64) -> Self {
        Self(Some(blocks))
    }
}

impl FromStr for L1SafetyMargin {
    type Err = ParseL1SafetyMarginError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let s = s.trim();
        if s.eq_ignore_ascii_case("unlimited") {
            return Ok(Self(None));
        }
        let blocks = s.parse().map_err(|_| ParseL1SafetyMarginError {
            input: s.to_string(),
        })?;
        Ok(Self(Some(blocks)))
    }
}

impl fmt::Display for L1SafetyMargin {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self.0 {
            Some(blocks) => write!(f, "{blocks}"),
            None => write!(f, "unlimited"),
        }
    }
}

#[cfg(feature = "node")]
impl Drop for L1UpdateTask {
    fn drop(&mut self) {
        if let Some(task) = self.0.get_mut().take() {
            task.abort();
        }
    }
}

impl Default for L1ClientOptions {
    fn default() -> Self {
        Self::parse_from(std::iter::empty::<String>())
    }
}

impl L1ClientOptions {
    /// Use the given metrics collector to publish metrics related to the L1 client.
    pub fn with_metrics(mut self, metrics: &(impl Metrics + ?Sized)) -> Self {
        self.metrics = Arc::new(metrics.subgroup("l1".into()));
        self
    }

    /// Instantiate an `L1Client` for a given list of provider `Url`s.
    #[cfg(feature = "node")]
    pub fn connect(self, urls: Vec<Url>) -> anyhow::Result<L1Client> {
        // create custom transport
        let t = SwitchingTransport::new(self, urls)
            .with_context(|| "failed to create switching transport")?;
        // Create a new L1 client with the transport
        Ok(L1Client::with_transport(t))
    }

    #[cfg(feature = "node")]
    fn rate_limit_delay(&self) -> Duration {
        self.l1_rate_limit_delay.unwrap_or(self.l1_retry_delay)
    }
}

#[cfg(feature = "node")]
impl L1ClientMetrics {
    fn new(metrics: &(impl Metrics + ?Sized), num_urls: usize) -> Self {
        // Create a counter family for the failures per URL
        let failures = metrics.counter_family("failed_requests".into(), vec!["provider".into()]);

        // Create a counter for each URL
        let mut failure_metrics = Vec::with_capacity(num_urls);
        for url_index in 0..num_urls {
            failure_metrics.push(failures.create(vec![url_index.to_string()]));
        }

        Self {
            head: metrics.create_gauge("head".into(), None).into(),
            finalized: metrics.create_gauge("finalized".into(), None).into(),
            reconnects: metrics
                .create_counter("stream_reconnects".into(), None)
                .into(),
            failovers: metrics.create_counter("failovers".into(), None).into(),
            failures: Arc::new(failure_metrics),
        }
    }
}

#[cfg(feature = "node")]
impl SwitchingTransport {
    /// Create a new `SwitchingTransport` with the given options and URLs
    pub fn new(opt: L1ClientOptions, urls: Vec<Url>) -> anyhow::Result<Self> {
        // Return early if there were no URLs provided
        let Some(first_url) = urls.first().cloned() else {
            return Err(anyhow::anyhow!("No valid URLs provided"));
        };

        // Create the metrics
        let metrics = L1ClientMetrics::new(&**opt.metrics, urls.len());

        // Create a new `SingleTransport` for the first URL
        let first_transport = Arc::new(RwLock::new(SingleTransport::new(&first_url, 0, None)));

        Ok(Self {
            urls: Arc::new(urls),
            current_transport: first_transport,
            opt: Arc::new(opt),
            metrics,
            switch_notify: Arc::new(Notify::new()),
        })
    }

    /// Returns when the transport has been switched
    async fn wait_switch(&self) {
        self.switch_notify.notified().await;
    }

    fn options(&self) -> &L1ClientOptions {
        &self.opt
    }

    fn metrics(&self) -> &L1ClientMetrics {
        &self.metrics
    }
}

#[cfg(feature = "node")]
impl SingleTransportStatus {
    /// Log a successful call to the inner transport
    fn log_success(&mut self) {
        self.consecutive_failures = 0;
        self.consecutive_rate_limits = 0;
    }

    /// Log a rate limit from the inner transport, backing off until `until`. Returns whether or
    /// not the transport should be switched to the next URL.
    ///
    /// One rate limit is not a provider fault: the caller backs off and retries. One that
    /// outlives a backoff is, because every request inside the backoff window short-circuits
    /// without reaching the provider, and nothing else in `call` moves off it: `should_revert`
    /// only reverts *towards* the primary.
    ///
    /// Only a rate limit arriving outside a backoff window counts. Requests already in flight
    /// when the window opened tell us nothing new, and would otherwise reach
    /// [`MAX_CONSECUTIVE_RATE_LIMITS`] in a single burst with no backoff having elapsed at all.
    fn log_rate_limit(&mut self, until: Instant) -> bool {
        let outlived_backoff = self
            .rate_limited_until
            .is_none_or(|prev| Instant::now() >= prev);
        self.rate_limited_until = Some(until);
        if !outlived_backoff {
            return false;
        }

        self.consecutive_rate_limits += 1;
        if self.shutting_down || self.consecutive_rate_limits < MAX_CONSECUTIVE_RATE_LIMITS {
            return false;
        }
        self.shutting_down = true;
        true
    }

    /// Log a failure to call the inner transport. Returns whether or not the transport should be switched to the next URL
    fn log_failure(&mut self, opt: &L1ClientOptions) -> bool {
        // Increment the consecutive failures
        self.consecutive_failures += 1;

        // Check if we should switch to the next URL
        let should_switch = self.should_switch(opt);

        // Update the last failure time
        self.last_failure = Some(Instant::now());

        // Return whether or not we should switch
        should_switch
    }

    /// Whether or not the transport should be switched to the next URL
    fn should_switch(&mut self, opt: &L1ClientOptions) -> bool {
        // If someone else already beat us to switching, return false
        if self.shutting_down {
            return false;
        }

        // If we've reached the max number of consecutive failures, switch to the next URL
        if self.consecutive_failures >= opt.l1_consecutive_failure_tolerance {
            self.shutting_down = true;
            return true;
        }

        // If we've failed recently, switch to the next URL
        let now = Instant::now();
        if let Some(prev) = self.last_failure
            && now.saturating_duration_since(prev) < opt.l1_frequent_failure_tolerance
        {
            self.shutting_down = true;
            return true;
        }

        false
    }

    /// Whether or not the transport should be switched back to the primary URL.
    fn should_revert(&mut self, revert_at: Option<Instant>) -> bool {
        if self.shutting_down {
            // We have already switched away from this transport in another thread.
            return false;
        }
        let Some(revert_at) = revert_at else {
            return false;
        };
        if Instant::now() >= revert_at {
            self.shutting_down = true;
            return true;
        }

        false
    }
}

#[cfg(feature = "node")]
impl SingleTransport {
    /// Create a new `SingleTransport` with the given URL
    fn new(url: &Url, generation: usize, revert_at: Option<Instant>) -> Self {
        Self {
            generation,
            redacted_url: redact_url(url),
            client: Http::new(url.clone()),
            status: Default::default(),
            revert_at,
        }
    }
}

/// alloy returns a non-2xx response with a parseable JSON-RPC body as `Ok`, so the payload, not
/// the `Result` arm, decides whether a provider is healthy.
///
/// This is a partial classification, deliberately. It covers only major error flavors the fleet
/// dumps actually captured. Much remains open: handling the most common providers and RPC flavors,
/// block ranges that adapt to what a provider will serve, separating a dead API key from a
/// transient failure, etc.
///
/// Unless we're fixing urgent bugs we should not add to this list until we have a satisfactory
/// design for robust handling of the whole RPC spectrum, because every provider and RPC flavor has
/// its own quirks.
#[cfg(feature = "node")]
#[derive(Debug)]
enum ResponseOutcome {
    Healthy,
    /// Provider is rate limiting. Back off rather than fail over, up to
    /// [`MAX_CONSECUTIVE_RATE_LIMITS`].
    RateLimited {
        retry_after: Option<Duration>,
    },
    /// The request earned the rejection on its own terms. Scores neither way: every backup
    /// rejects the same request identically, so failing over neither helps nor indicates
    /// anything about the provider.
    RequestRejected,
    Failed,
}

/// How many rate limits, each outliving the backoff the previous one imposed, before a provider
/// is treated as failed.
///
/// A provider still rate limiting after a backoff of its own is not merely busy. Counting it
/// bounds how long the client stays pinned to a provider whose quota is exhausted for the day;
/// without it, `rate_limited_until` short-circuits every request forever and the healthy backup
/// is never reached.
#[cfg(feature = "node")]
const MAX_CONSECUTIVE_RATE_LIMITS: usize = 2;

/// Cap on a server-requested `Retry-After`, which alloy parses verbatim. Honoring a
/// daily-quota delay of hours would stall `wait_for_l1` for its whole duration.
///
/// Matches alloy's own private `MAX_BACKOFF_HINT`, which `RpcErrorExt::backoff_hint` clamps to
/// so "a misbehaving server cannot stall retries"
/// ([alloy-transport-2.4.1/src/error.rs:266-272](https://github.com/alloy-rs/alloy/blob/v2.4.1/crates/transport/src/error.rs#L266-L272)).
#[cfg(feature = "node")]
const MAX_RATE_LIMIT_BACKOFF: Duration = Duration::from_secs(300);

/// The configured delay is operator-set and used as-is; a server-provided one is capped.
#[cfg(feature = "node")]
fn rate_limit_backoff(retry_after: Option<Duration>, configured: Duration) -> Duration {
    match retry_after {
        Some(retry_after) => retry_after.min(MAX_RATE_LIMIT_BACKOFF),
        None => configured,
    }
}

/// Whether the response rejects the request itself: the result set it asks for is over the
/// provider's cap.
///
/// This recurs against a perfectly healthy provider. `l1_events_max_block_range` defaults to
/// 10000, exactly infura's result cap, and `get_finalized_deposits` retries a rejected chunk in
/// a loop at `l1_retry_delay`, so scoring it would burn a generation every couple of seconds and
/// cycle the whole provider list against a request every one of them rejects.
///
/// Only result-count rejections qualify. A block-range cap reads the same way but is a plan or
/// node-config limit, not a property of the request: alchemy caps `eth_getLogs` at 10 blocks on
/// the free tier and not at all on paid ones, so a backup may well serve the identical request
/// and failing over is the right move.
///
/// Matched on the message because infura reuses `-32005` for both a result-count rejection and a
/// rate limit. Claiming it before [`ErrorPayload::is_retry_err`], which calls every `-32005` a
/// rate limit, is what makes deferring to alloy's matcher safe here.
#[cfg(feature = "node")]
fn is_request_rejection(res: &ResponsePacket) -> bool {
    res.as_error()
        .is_some_and(|e| e.message.contains("query returned more than"))
}

/// The JSON-RPC error a response carries, if any.
///
/// Logged in place of the whole `Result`: a scored failure arrives as `Ok(Failure(..))`, which
/// reads as a success in a log search, and the packet holds the provider's entire response body.
#[cfg(feature = "node")]
fn error_payload(
    result: &StdResult<ResponsePacket, RpcError<TransportErrorKind>>,
) -> Option<&ErrorPayload> {
    result.as_ref().ok().and_then(ResponsePacket::as_error)
}

#[cfg(feature = "node")]
impl ResponseOutcome {
    /// Score a transport response for provider health. A batch is scored by the first error in
    /// it; no batch callers exist today.
    fn classify(result: &StdResult<ResponsePacket, RpcError<TransportErrorKind>>) -> Self {
        use ResponseOutcome::*;
        match result {
            // Ordered before the rate-limit check: alloy treats every `-32005` as a rate limit,
            // and infura reuses that code for result-count rejections.
            Ok(res) if is_request_rejection(res) => RequestRejected,
            // alloy returns an HTTP 429 whose body parses as `Ok`, dropping the status and
            // `Retry-After`, so the payload is the only signal left.
            Ok(res) if res.as_error().is_some_and(ErrorPayload::is_retry_err) => {
                RateLimited { retry_after: None }
            },
            Ok(res) if res.is_error() => Failed,
            Ok(_) => Healthy,
            Err(RpcError::Transport(kind))
                if kind
                    .as_http_error()
                    .is_some_and(HttpError::is_rate_limit_err) =>
            {
                RateLimited {
                    retry_after: kind.retry_after(),
                }
            },
            Err(_) => Failed,
        }
    }
}

/// `SwitchingTransport` is an alternative [`Client`](https://docs.rs/alloy/0.12.5/alloy/transports/http/struct.Client.html)
/// which by implementing `tower_service::Service`, traits like [`Transport`](https://docs.rs/alloy/0.12.5/alloy/transports/trait.Transport.html)
/// are auto-derived, thus can be used as an alt [`RpcClient`](https://docs.rs/alloy/0.12.5/alloy/rpc/client/struct.RpcClient.html#method.new)
/// that can be further hooked with `Provider` via `Provider::on_client()`.
#[cfg(feature = "node")]
#[async_trait]
impl Service<RequestPacket> for SwitchingTransport {
    type Error = RpcError<TransportErrorKind>;
    type Response = ResponsePacket;
    type Future =
        Pin<Box<dyn Future<Output = Result<ResponsePacket, RpcError<TransportErrorKind>>> + Send>>;

    fn poll_ready(
        &mut self,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<StdResult<(), Self::Error>> {
        // Just poll the (current) inner client
        self.current_transport.read().clone().client.poll_ready(cx)
    }

    fn call(&mut self, req: RequestPacket) -> Self::Future {
        // Clone ourselves
        let self_clone = self.clone();

        // Pin and box, which turns this into a future
        Box::pin(async move {
            // Clone the current transport
            let mut current_transport = self_clone.current_transport.read().clone();

            // Revert back to the primary transport if it's time.
            let should_revert = current_transport
                .status
                .write()
                .should_revert(current_transport.revert_at);
            if should_revert {
                // Switch to the next generation which maps to index 0.
                let n = self_clone.urls.len();
                // Rounding down to a multiple of n gives us the last generation of the primary transport.
                let prev_primary_gen = (current_transport.generation / n) * n;
                // Adding n jumps to the next generation.
                let next_gen = prev_primary_gen + n;
                current_transport = self_clone.switch_to(next_gen, current_transport);
            }

            // If we've been rate limited, back off until the limit (hopefully) expires.
            let rate_limit_until = current_transport.status.read().rate_limited_until;
            if let Some(t) = rate_limit_until {
                if t > Instant::now() {
                    // Return an error with a non-standard code to indicate client-side rate limit.
                    return Err(RpcError::Transport(TransportErrorKind::Custom(
                        "Rate limit exceeded".into(),
                    )));
                } else {
                    // Reset the rate limit if we are passed it so we don't check every time
                    current_transport.status.write().rate_limited_until = None;
                }
            }

            let result = current_transport.client.call(req).await;
            let outcome = ResponseOutcome::classify(&result);

            if matches!(outcome, ResponseOutcome::Healthy) {
                current_transport.status.write().log_success();
                return result;
            }

            // Leaves the failure counters untouched, in both directions: the request would be
            // rejected by every other provider too, so it is evidence about neither. Warned
            // rather than counted, because the caller has to shrink the request to make progress
            // and the failover metrics would say nothing about that.
            if matches!(outcome, ResponseOutcome::RequestRejected) {
                tracing::warn!(
                    url = %current_transport.redacted_url,
                    error = ?error_payload(&result),
                    "L1 rejected the request"
                );
                return result;
            }

            // Increment the failure metric
            if let Some(f) = self_clone
                .metrics
                .failures
                .get(current_transport.generation % self_clone.urls.len())
            {
                f.add(1);
            }

            // We don't need to worry about race conditions here, since either of these will only
            // return true once.
            let should_switch = if let ResponseOutcome::RateLimited { retry_after } = outcome {
                // Back off on this provider rather than failing over, until it has kept rate
                // limiting through a backoff of its own.
                let until = Instant::now()
                    + rate_limit_backoff(retry_after, self_clone.opt.rate_limit_delay());
                let mut status = current_transport.status.write();
                let should_switch = status.log_rate_limit(until);
                // Warned only when it ends in a failover, which is otherwise unexplainable from
                // production logs: a routine backoff is too frequent to warn about.
                if should_switch {
                    tracing::warn!(
                        url = %current_transport.redacted_url,
                        consecutive = status.consecutive_rate_limits,
                        "L1 still rate limiting after its own backoff, failing over"
                    );
                } else {
                    tracing::debug!(
                        url = %current_transport.redacted_url,
                        consecutive = status.consecutive_rate_limits,
                        "L1 rate limited"
                    );
                }
                should_switch
            } else {
                tracing::warn!(
                    url = %current_transport.redacted_url,
                    ?outcome,
                    error = ?error_payload(&result),
                    transport_err = ?result.as_ref().err(),
                    "L1 client error"
                );
                current_transport
                    .status
                    .write()
                    .log_failure(&self_clone.opt)
            };

            if should_switch {
                // Increment the failovers metric
                self_clone.metrics.failovers.add(1);
                self_clone.switch_to(current_transport.generation + 1, current_transport);
            }

            result
        })
    }
}

#[cfg(feature = "node")]
impl SwitchingTransport {
    fn switch_to(&self, next_gen: usize, current_transport: SingleTransport) -> SingleTransport {
        let next_index = next_gen % self.urls.len();
        let url = self.urls[next_index].clone();
        tracing::info!(url = %redact_url(&url), next_gen, "switch L1 transport");

        let revert_at = if next_gen.is_multiple_of(self.urls.len()) {
            // If we are reverting to the primary transport, clear our scheduled revert time.
            None
        } else if current_transport.generation.is_multiple_of(self.urls.len()) {
            // If we are failing over from the primary transport, schedule a time to automatically
            // revert back.
            Some(Instant::now() + self.opt.l1_failover_revert)
        } else {
            // Otherwise keep the currently scheduled revert time.
            current_transport.revert_at
        };

        // Create a new transport from the next URL and index
        let new_transport = SingleTransport::new(&url, next_gen, revert_at);

        // Switch to the next URL
        *self.current_transport.write() = new_transport.clone();

        // Notify the transport that it has been switched
        self.switch_notify.notify_waiters();

        new_transport
    }
}

/// Which of `L1State`'s independent refresh throttles `throttle_refresh` bounds.
#[cfg(feature = "node")]
enum RefreshKind {
    Head,
    Finalized,
}

#[cfg(feature = "node")]
impl L1Client {
    fn with_transport(transport: SwitchingTransport) -> Self {
        // Create a new provider with that RPC client using the custom transport
        let rpc_client = RpcClient::new(transport.clone(), false);
        let provider = ProviderBuilder::new().connect_client(rpc_client);

        let opt = transport.options().clone();

        let (sender, mut receiver) = async_broadcast::broadcast(opt.l1_events_channel_capacity);
        receiver.set_await_active(false);
        receiver.set_overflow(true);

        Self {
            provider,
            transport,
            state: Arc::new(Mutex::new(L1State::new(opt.l1_blocks_cache_size))),
            sender,
            receiver: receiver.deactivate(),
            update_task: Default::default(),
        }
    }

    /// Construct a new L1 client with the default options.
    pub fn new(url: Vec<Url>) -> anyhow::Result<Self> {
        L1ClientOptions::default().connect(url)
    }

    /// test only
    pub fn anvil(anvil: &alloy::node_bindings::AnvilInstance) -> anyhow::Result<Self> {
        L1ClientOptions {
            l1_ws_provider: Some(vec![anvil.ws_endpoint().parse()?]),
            ..Default::default()
        }
        .connect(vec![anvil.endpoint().parse()?])
    }

    /// Start the background tasks which keep the L1 client up to date.
    pub async fn spawn_tasks(&self) {
        let mut update_task = self.update_task.0.lock().await;
        if update_task.is_none() {
            *update_task = Some(spawn(self.update_loop()));
        }
    }

    /// Shut down background tasks associated with this L1 client.
    ///
    /// The L1 client will still be usable, but will stop updating until [`start`](Self::start) is
    /// called again.
    pub async fn shut_down_tasks(&self) {
        let update_task = self.update_task.0.lock().await.take();
        if let Some(update_task) = update_task {
            update_task.abort();
        }
    }

    fn update_loop(&self) -> impl Future<Output = ()> + use<> {
        let opt = self.options().clone();
        let rpc = self.provider.clone();
        let ws_urls = opt.l1_ws_provider.clone();
        let retry_delay = opt.l1_retry_delay;
        let request_timeout = opt.l1_request_timeout;
        let subscription_timeout = opt.subscription_timeout;
        let state = self.state.clone();
        let sender = self.sender.clone();
        let metrics = self.metrics().clone();
        let polling_interval = opt.l1_polling_interval;
        let transport = self.transport.clone();

        let span = tracing::warn_span!("L1 client update");

        async move {

            for i in 0..usize::MAX {
                let ws;

                // Fetch current L1 head block for the first value of the stream to avoid having
                // to wait for new L1 blocks until the update loop starts processing blocks.
                let l1_head = loop {
                    match tokio::time::timeout(request_timeout, rpc.get_block(BlockId::latest()))
                        .await
                    {
                        Ok(Ok(Some(block))) => break block.header,
                        Ok(Ok(None)) => {
                            tracing::warn!("Failed to fetch L1 head block, will retry");
                        },
                        Ok(Err(err)) => {
                            tracing::warn!("Failed to fetch L1 head block, will retry: err {err}");
                        },
                        Err(_) => {
                            tracing::warn!(
                                "Failed to fetch L1 head block, will retry: timed out"
                            );
                        }
                    }
                    sleep(retry_delay).await;
                };

                // Subscribe to new blocks.
                let mut block_stream = {
                    let res = match &ws_urls {
                        Some(urls) => {
                            // Use a new WebSockets host each time we retry in case there is a
                            // problem with one of the hosts specifically.
                            let provider = i % urls.len();
                            let url = &urls[provider];
                            ws = match ProviderBuilder::new()
                                .connect_ws(WsConnect::new(url.clone()))
                                .await
                            {
                                Ok(ws) => ws,
                                Err(err) => {
                                    tracing::warn!(
                                        provider,
                                        "Failed to connect WebSockets provider: {err:#}"
                                    );
                                    sleep(retry_delay).await;
                                    continue;
                                }
                            };
                            ws.subscribe_blocks().await.map(|stream| {stream::once(async { l1_head.clone() }).chain(stream.into_stream()).boxed()})
                        },
                        None => {
                           rpc
                            .watch_blocks()
                            .await
                            .map(|poller_builder| {
                                // Configure it and get the stream
                                let stream = poller_builder.with_poll_interval(polling_interval).into_stream();

                                let rpc = rpc.clone();

                                // For HTTP, we simulate a subscription by polling. The polling
                                // stream provided by ethers only yields block hashes, so for each
                                // one, we have to go fetch the block itself.
                                stream::once(async { l1_head.clone() })
                                 .chain(
                                    stream.map(stream::iter).flatten().filter_map(move |hash| {
                                        let rpc = rpc.clone();
                                        async move {
                                            match rpc.get_block(BlockId::hash(hash)).await {
                                                Ok(Some(block)) => Some(block.header),
                                                // If we can't fetch the block for some reason, we can
                                                // just skip it.
                                                Ok(None) => {
                                                    tracing::warn!(%hash, "HTTP stream yielded a block hash that was not available");
                                                    None
                                                }
                                                Err(err) => {
                                                    tracing::warn!(
                                                        %hash,
                                                        "Error fetching block from HTTP stream: {err:#}"
                                                    );
                                                    None
                                                }
                                            }
                                        }
                                    }))
                                // Take until the transport is switched, so we will call `watch_blocks` instantly on it
                            }.take_until(transport.wait_switch())
                            .boxed())
                        }
                    };
                    match res {
                        Ok(stream) => stream,
                        Err(err) => {
                            tracing::error!("Error subscribing to L1 blocks: {err:#}");
                            sleep(retry_delay).await;
                            continue;
                        }
                    }
                };

                tracing::info!("Established L1 block stream");
                loop {
                    // Wait for a block, timing out if we don't get one soon enough
                    let block_timeout = tokio::time::timeout(subscription_timeout, block_stream.next()).await;
                    match block_timeout {
                        // We got a block
                        Ok(Some(head)) => {
                            let head = head.number;
                            tracing::debug!(head, "Received L1 block");

                            // Apply the head before touching finalized, so a failing finalized
                            // RPC never stalls the head.
                            {
                                let mut state = state.lock().await;
                                apply_head(&mut state, head, &metrics, &sender).await;
                            }

                            // A new block has been produced. This happens fairly rarely, so it is
                            // now ok to poll to see if a new block has been finalized. One attempt
                            // per head: a failure retries on the next head instead of blocking here.
                            match fetch_finalized_block_from_rpc(&rpc, request_timeout).await {
                                Ok(Some(finalized)) => {
                                    let mut state = state.lock().await;
                                    apply_finalized(&mut state, finalized, &metrics, &sender).await;
                                },
                                Ok(None) => {
                                    tracing::warn!("no finalized block yet");
                                },
                                Err(err) => {
                                    tracing::warn!("Error getting finalized block: {err:#}");
                                },
                            }
                        }
                        // The stream ended
                        Ok(None) => {
                            tracing::error!("L1 block stream ended unexpectedly, trying to re-establish block stream");
                            break;
                        }
                        // We timed out waiting for a block
                        Err(_) => {
                            tracing::error!("No block received for {} seconds, trying to re-establish block stream", subscription_timeout.as_secs());
                            break;
                        }
                    }
                }

                metrics.reconnects.add(1);
            }
        }.instrument(span)
    }

    /// Get a snapshot from the l1.
    pub async fn snapshot(&self) -> L1Snapshot {
        self.state.lock().await.snapshot
    }

    /// Wait until the highest L1 block number reaches at least `number`.
    ///
    /// This function does not return any information about the block, since the block is not
    /// necessarily finalized when it returns. It is only used to guarantee that some block at
    /// height `number` exists, possibly in the unsafe part of the L1 chain.
    ///
    /// While behind, refreshes the head from the RPC directly rather than only waiting on the
    /// poller.
    pub async fn wait_for_block(&self, number: u64) {
        'outer: loop {
            // Subscribe to events before checking the current state, to ensure we don't miss a
            // relevant event.
            let mut events = self.receiver.activate_cloned();

            // Check if the block we are waiting for already exists.
            {
                let state = self.state.lock().await;
                if state.snapshot.head >= number {
                    return;
                }
                tracing::info!(number, head = state.snapshot.head, "Waiting for l1 block");
            }

            // Wait for the block, refreshing from the RPC on a fixed tick so a lagging poller
            // doesn't stall the wait. The tick runs on its own schedule so a stream of events
            // that don't satisfy `number` can't keep deferring it.
            let mut refresh = tokio::time::interval(self.options().l1_wait_refresh_interval);
            refresh.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
            refresh.tick().await; // the first tick fires immediately

            loop {
                tokio::select! {
                    event = events.next() => {
                        match event {
                            Some(L1Event::NewHead { head }) => {
                                if head >= number {
                                    tracing::info!(number, head, "Got L1 block");
                                    return;
                                }
                                tracing::debug!(number, head, "Waiting for L1 block");
                            },
                            Some(_) => {},
                            None => {
                                // This should not happen: the event stream ended. All we can do
                                // is try again.
                                tracing::warn!(number, "L1 event stream ended unexpectedly; retry");
                                self.retry_delay().await;
                                continue 'outer;
                            },
                        }
                    },
                    _ = refresh.tick() => {
                        self.refresh_head().await;
                        let head = self.state.lock().await.snapshot.head;
                        if head >= number {
                            tracing::info!(number, head, "Got L1 block");
                            return;
                        }
                    },
                }
            }
        }
    }

    /// Get information about the given block.
    ///
    /// If the desired block number is not finalized yet, this function will block until it becomes
    /// finalized.
    pub async fn wait_for_finalized_block(&self, number: u64) -> L1BlockInfo {
        let (state, _) = self.lock_once_finalized_reaches(number).await;
        self.fetch_finalized_block_by_number(state, number).await.1
    }

    /// Wait until the finalized block number reaches at least `number` and return the latest
    /// finalized block, without fetching block `number` itself.
    pub async fn wait_for_latest_finalized_at_least(&self, number: u64) -> L1BlockInfo {
        self.lock_once_finalized_reaches(number).await.1
    }

    /// Wait until the finalized block number reaches at least `number`, returning the locked
    /// state and the latest finalized block.
    ///
    /// While behind, refreshes the finalized block from the RPC directly.
    async fn lock_once_finalized_reaches(
        &self,
        number: u64,
    ) -> (MutexGuard<'_, L1State>, L1BlockInfo) {
        'outer: loop {
            // Subscribe to events before checking the current state, to ensure we don't miss a relevant
            // event.
            let mut events = self.receiver.activate_cloned();

            // Check if the block we are waiting for already exists.
            {
                let state = self.state.lock().await;
                if let Some(finalized) = state.snapshot.finalized {
                    if finalized.number >= number {
                        return (state, finalized);
                    }
                    tracing::info!(
                        number,
                        finalized = ?state.snapshot.finalized,
                        "waiting for l1 finalized block",
                    );
                };
            }

            // Wait for the block, refreshing from the RPC on a fixed tick so a lagging poller
            // doesn't stall the wait. The tick runs on its own schedule so a stream of events
            // that don't satisfy `number` can't keep deferring it.
            let mut refresh = tokio::time::interval(self.options().l1_wait_refresh_interval);
            refresh.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
            refresh.tick().await; // the first tick fires immediately

            loop {
                tokio::select! {
                    event = events.next() => {
                        match event {
                            Some(L1Event::NewFinalized { finalized }) => {
                                let mut state = self.state.lock().await;
                                state.put_finalized(finalized);
                                if finalized.info.number >= number {
                                    tracing::info!(number, ?finalized, "got finalized L1 block");
                                    return (state, finalized.info);
                                }
                                tracing::debug!(number, ?finalized, "waiting for finalized L1 block");
                            },
                            Some(_) => {},
                            None => {
                                // This should not happen: the event stream ended. All we can do
                                // is try again.
                                tracing::warn!(number, "L1 event stream ended unexpectedly; retry");
                                self.retry_delay().await;
                                continue 'outer;
                            },
                        }
                    },
                    _ = refresh.tick() => {
                        self.refresh_finalized().await;
                        let state = self.state.lock().await;
                        if let Some(finalized) = state.snapshot.finalized
                            && finalized.number >= number
                        {
                            tracing::info!(number, ?finalized, "got finalized L1 block");
                            return (state, finalized);
                        }
                    },
                }
            }
        }
    }

    /// Get information about the first finalized block with timestamp greater than or equal
    /// `timestamp`.
    pub async fn wait_for_finalized_block_with_timestamp(&self, timestamp: U256) -> L1BlockInfo {
        // Wait until the finalized block has timestamp >= `timestamp`.
        let (mut state, mut block) = 'outer: loop {
            // Subscribe to events before checking the current state, to ensure we don't miss a
            // relevant event.
            let mut events = self.receiver.activate_cloned();

            // Check if the block we are waiting for already exists.
            {
                let state = self.state.lock().await;
                if let Some(finalized) = state.snapshot.finalized
                    && finalized.timestamp >= timestamp
                {
                    break 'outer (state, finalized);
                }
                tracing::info!(
                    %timestamp,
                    finalized = ?state.snapshot.finalized,
                    "waiting for L1 finalized block",
                );
            }

            // Wait for the block.
            while let Some(event) = events.next().await {
                let L1Event::NewFinalized { finalized } = event else {
                    continue;
                };
                if finalized.info.timestamp >= timestamp {
                    tracing::info!(%timestamp, ?finalized, "got finalized block");
                    break 'outer (self.state.lock().await, finalized.info);
                }
                tracing::debug!(%timestamp, ?finalized, "waiting for L1 finalized block");
            }

            // This should not happen: the event stream ended. All we can do is try again.
            tracing::warn!(%timestamp, "L1 event stream ended unexpectedly; retry",);
            self.retry_delay().await;
        };

        // It is possible there is some earlier block that also has the proper timestamp. Binary
        // search until we find the true earliest block with timestamp >= `timestamp`.
        //
        // Invariants:
        // * `upper_bound <= lower_bound`
        // * `upper_bound = block.number`
        // * Block number `lower_bound - 1` has timestamp < `timestamp` (strictly)
        // * `block` has timestamp >= `timestamp`
        let mut upper_bound = block.number;
        let mut lower_bound = 0;
        while lower_bound < upper_bound {
            let midpoint = (upper_bound + lower_bound) / 2;
            tracing::debug!(
                lower_bound,
                midpoint,
                upper_bound,
                %timestamp,
                ?block,
                "searching for earliest block with sufficient timestamp"
            );

            let (state_lock, midpoint_block) =
                self.fetch_finalized_block_by_number(state, midpoint).await;
            state = state_lock;

            tracing::debug!(?midpoint_block, %timestamp, "pivot on midpoint block");
            if midpoint_block.timestamp < timestamp {
                lower_bound = midpoint + 1;
            } else {
                upper_bound = midpoint;
                block = midpoint_block;
            }
        }

        block
    }

    /// Queries the L1 head directly from the RPC and applies it, throttled by
    /// [`Self::throttle_refresh`]. Bounded to `l1_wait_refresh_timeout` so a hung request can't
    /// park a waiter past the point where the poller has already published the head.
    async fn refresh_head(&self) {
        if !self.throttle_refresh(RefreshKind::Head).await {
            return;
        }
        let timeout = self.options().l1_wait_refresh_timeout;
        match tokio::time::timeout(timeout, self.provider.get_block_number()).await {
            Ok(Ok(head)) => {
                let mut state = self.state.lock().await;
                apply_head(&mut state, head, self.metrics(), &self.sender).await;
            },
            Ok(Err(err)) => {
                tracing::debug!("Error refreshing L1 head from RPC: {err:#}");
            },
            Err(_) => {
                tracing::debug!("Timed out refreshing L1 head from RPC");
            },
        }
    }

    /// Queries the L1 finalized block directly from the RPC and applies it, throttled by
    /// [`Self::throttle_refresh`]. Bounded to `l1_wait_refresh_timeout` so a hung request can't
    /// park a waiter past the point where the poller has already published the finalized block.
    async fn refresh_finalized(&self) {
        if !self.throttle_refresh(RefreshKind::Finalized).await {
            return;
        }
        let timeout = self.options().l1_wait_refresh_timeout;
        match fetch_finalized_block_from_rpc(&self.provider, timeout).await {
            Ok(Some(finalized)) => {
                let mut state = self.state.lock().await;
                apply_finalized(&mut state, finalized, self.metrics(), &self.sender).await;
            },
            Ok(None) => {
                tracing::debug!("no finalized block yet");
            },
            Err(err) => {
                tracing::debug!("Error refreshing L1 finalized block from RPC: {err:#}");
            },
        }
    }

    /// Bounds `refresh_head` or `refresh_finalized` (per `kind`) to one RPC call per
    /// `l1_wait_refresh_interval`, shared across every concurrent waiter of that kind, so a burst
    /// of waiters on a stalled L1 doesn't turn into a busy loop of RPC calls. Head and finalized
    /// are throttled independently, so a wait on one can't starve a wait on the other.
    async fn throttle_refresh(&self, kind: RefreshKind) -> bool {
        let interval = self.options().l1_wait_refresh_interval;
        let mut state = self.state.lock().await;
        let now = Instant::now();
        let last_refresh = match kind {
            RefreshKind::Head => &mut state.last_head_refresh,
            RefreshKind::Finalized => &mut state.last_finalized_refresh,
        };
        if last_refresh.is_some_and(|last| now.saturating_duration_since(last) < interval) {
            return false;
        }
        *last_refresh = Some(now);
        true
    }

    async fn fetch_finalized_block_by_number<'a>(
        &'a self,
        mut state: MutexGuard<'a, L1State>,
        number: u64,
    ) -> (MutexGuard<'a, L1State>, L1BlockInfo) {
        let latest_finalized = state
            .snapshot
            .finalized
            .expect("get_finalized_block called before any blocks are finalized");
        assert!(
            number <= latest_finalized.number,
            "requesting a finalized block {number} that isn't finalized; snapshot: {:?}",
            state.snapshot,
        );

        if let Some(safety_margin) = self.options().l1_finalized_safety_margin.blocks()
            && number < latest_finalized.number.saturating_sub(safety_margin.get())
        {
            // If the requested block height is so old that we can assume all L1 providers have
            // finalized it, we don't need to worry about failing over to a lagging L1 provider
            // which has yet to finalize the block, so we don't need to bother with the
            // expensive hash chaining logic below. Just look up the block by number and assume
            // the response is finalized.
            tracing::debug!(
                number,
                ?latest_finalized,
                "skipping hash check for old finalized block"
            );
            let (state, block) = self
                .load_and_cache_finalized_block(state, number.into())
                .await;
            return (state, block.info);
        }

        // To get this block and be sure we are getting the correct finalized block, we first need
        // to find an equal or later block so we can find the expected hash of this block. If we
        // were to just look up the block by number, there could be problems if we failed over to a
        // different (lagging) L1 provider, which has yet to finalize this block and reports a
        // different block with the same number.
        let mut successor_number = number;
        let mut successor = loop {
            if let Some(block) = state.finalized.get(&successor_number) {
                break *block;
            }
            successor_number += 1;
            if successor_number > latest_finalized.number {
                // We don't have any cached finalized block after the requested one; fetch the
                // current finalized block from the network.
                // Don't hold state lock while fetching from network.
                drop(state);
                let block = loop {
                    match fetch_finalized_block_from_rpc(
                        &self.provider,
                        self.options().l1_request_timeout,
                    )
                    .await
                    {
                        Ok(Some(block)) => {
                            break block;
                        },
                        Ok(None) => {
                            tracing::warn!(
                                "no finalized block even though finalized snapshot is Some; this \
                                 can be caused by an L1 client failover"
                            );
                            self.retry_delay().await;
                        },
                        Err(err) => {
                            tracing::warn!("Error getting finalized block: {err:#}");
                            self.retry_delay().await;
                        },
                    }
                };
                state = self.state.lock().await;
                state.put_finalized(block);
                break block;
            }
        };

        // Work backwards from the known finalized successor, fetching blocks by parent hash so we
        // know we are getting the correct block.
        while successor.info.number > number {
            tracing::debug!(
                number,
                ?successor,
                "checking hash chaining for finalized block"
            );
            (state, successor) = self
                .load_and_cache_finalized_block(state, successor.parent_hash.into())
                .await;
        }

        (state, successor.info)
    }

    async fn load_and_cache_finalized_block<'a>(
        &'a self,
        mut state: MutexGuard<'a, L1State>,
        id: BlockId,
    ) -> (MutexGuard<'a, L1State>, L1BlockInfoWithParent) {
        // Don't hold state lock while fetching from network.
        drop(state);
        let request_timeout = self.options().l1_request_timeout;
        let block = loop {
            let block =
                match tokio::time::timeout(request_timeout, self.provider.get_block(id)).await {
                    Ok(Ok(Some(block))) => block,
                    Ok(Ok(None)) => {
                        tracing::warn!(
                            %id,
                            "provider error: finalized L1 block should always be available"
                        );
                        self.retry_delay().await;
                        continue;
                    },
                    Ok(Err(err)) => {
                        tracing::warn!(%id, "failed to get finalized L1 block: {err:#}");
                        self.retry_delay().await;
                        continue;
                    },
                    Err(_) => {
                        tracing::warn!(%id, "timed out getting finalized L1 block");
                        self.retry_delay().await;
                        continue;
                    },
                };
            break (&block).into();
        };
        state = self.state.lock().await;
        state.put_finalized(block);
        (state, block)
    }

    /// Get fee info for each `Deposit` occurring between `prev`
    /// and `new`. Returns `Vec<FeeInfo>`
    pub async fn get_finalized_deposits(
        &self,
        fee_contract_address: Address,
        prev_finalized: Option<u64>,
        new_finalized: u64,
    ) -> Vec<FeeInfo> {
        // No new blocks have been finalized, therefore there are no
        // new deposits.
        if prev_finalized >= Some(new_finalized) {
            return vec![];
        }

        let opt = self.options();

        // `prev` should have already been processed unless we
        // haven't processed *any* blocks yet.
        let prev = prev_finalized.map(|prev| prev + 1).unwrap_or(0);

        // Divide the range `prev_finalized..=new_finalized` into chunks of size
        // `events_max_block_range`.
        let mut start = prev;
        let end = new_finalized;
        let chunk_size = opt.l1_events_max_block_range;
        let chunks = std::iter::from_fn(move || {
            let chunk_end = min(start + chunk_size - 1, end);
            if chunk_end < start {
                return None;
            }

            let chunk = (start, chunk_end);
            start = chunk_end + 1;
            Some(chunk)
        });

        // Fetch events for each chunk.
        let events = stream::iter(chunks).then(|(from, to)| {
            let retry_delay = opt.l1_retry_delay;
            let fee_contract = FeeContract::new(fee_contract_address, self.provider.clone());
            async move {
                tracing::debug!(from, to, "fetch events in range");

                // query for deposit events, loop until successful.
                loop {
                    match fee_contract
                        .Deposit_filter()
                        .address(*fee_contract.address())
                        .from_block(from)
                        .to_block(to)
                        .query()
                        .await
                    {
                        Ok(events) => break stream::iter(events),
                        Err(err) => {
                            tracing::warn!(from, to, %err, "Fee L1Event Error");
                            sleep(retry_delay).await;
                        },
                    }
                }
            }
        });
        events
            .flatten()
            .map(|(deposit, _)| FeeInfo::from(deposit))
            .collect()
            .await
    }

    /// Check if the given address is a proxy contract.
    pub async fn is_proxy_contract(&self, proxy_address: Address) -> anyhow::Result<bool> {
        // confirm that the proxy_address is a proxy
        // using the implementation slot, 0x360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc, which is the keccak-256 hash of "eip1967.proxy.implementation" subtracted by 1
        let hex_bytes =
            hex::decode("360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc")
                .expect("Failed to decode hex string");
        let implementation_slot = B256::from_slice(&hex_bytes);
        let storage = self
            .provider
            .get_storage_at(proxy_address, implementation_slot.into())
            .await?;

        let implementation_address = Address::from_slice(&storage.to_be_bytes::<32>()[12..]);

        // when the implementation address is not equal to zero, it's a proxy
        Ok(implementation_address != Address::ZERO)
    }

    pub async fn retry_on_all_providers<Fut>(
        &self,
        op: impl Fn() -> Fut,
    ) -> Result<Fut::Ok, Fut::Error>
    where
        Fut: TryFuture,
    {
        let transport = &self.transport;
        let start = transport.current_transport.read().generation % transport.urls.len();
        let end = start + transport.urls.len();
        loop {
            match TryFutureExt::into_future(op()).await {
                Ok(res) => return Ok(res),
                Err(err) => {
                    if transport.current_transport.read().generation >= end {
                        return Err(err);
                    } else {
                        self.retry_delay().await;
                    }
                },
            }
        }
    }

    pub(crate) fn options(&self) -> &L1ClientOptions {
        self.transport.options()
    }

    fn metrics(&self) -> &L1ClientMetrics {
        self.transport.metrics()
    }

    async fn retry_delay(&self) {
        sleep(self.options().l1_retry_delay).await;
    }
}

#[cfg(feature = "node")]
impl L1State {
    fn new(cache_size: NonZeroUsize) -> Self {
        Self {
            snapshot: Default::default(),
            finalized: LruCache::new(cache_size),
            last_finalized: None,
            last_head_refresh: None,
            last_finalized_refresh: None,
        }
    }

    fn put_finalized(&mut self, block: L1BlockInfoWithParent) {
        assert!(
            self.snapshot.finalized.is_some()
                && block.info.number <= self.snapshot.finalized.unwrap().number,
            "inserting a finalized block {block:?} that isn't finalized; snapshot: {:?}",
            self.snapshot,
        );

        if Some(block.info.number()) > self.last_finalized {
            self.last_finalized = Some(block.info.number());
        }

        if let Some((old_number, old_block)) = self.finalized.push(block.info.number, block)
            && old_number == block.info.number
            && block != old_block
        {
            tracing::error!(
                ?old_block,
                ?block,
                "got different info for the same finalized height; something has gone very wrong \
                 with the L1",
            );
        }
    }
}

/// Applies `head` to `state` if it advances the snapshot, updating the metric and broadcasting
/// `NewHead`. Never lowers the head.
#[cfg(feature = "node")]
async fn apply_head(
    state: &mut L1State,
    head: u64,
    metrics: &L1ClientMetrics,
    sender: &Sender<L1Event>,
) {
    if head <= state.snapshot.head {
        return;
    }
    tracing::debug!(head, old_head = state.snapshot.head, "L1 head updated");
    metrics.head.set(head as usize);
    state.snapshot.head = head;
    // Emit an event about the new L1 head. Ignore send errors; it just means no one is
    // listening to events right now.
    sender
        .broadcast_direct(L1Event::NewHead { head })
        .await
        .ok();
}

/// Applies `finalized` to `state` if it advances the snapshot, updating the metric and
/// broadcasting `NewFinalized`. Never lowers the finalized block.
///
/// Also raises the head to the finalized block's number if the head is behind it, since a
/// finalized block always exists on L1 and thus is truthfully also the head, or below it.
#[cfg(feature = "node")]
async fn apply_finalized(
    state: &mut L1State,
    finalized: L1BlockInfoWithParent,
    metrics: &L1ClientMetrics,
    sender: &Sender<L1Event>,
) {
    if Some(finalized.info) <= state.snapshot.finalized {
        return;
    }
    tracing::info!(
        ?finalized,
        old_finalized = ?state.snapshot.finalized,
        "L1 finalized updated",
    );
    if state.snapshot.head < finalized.info.number {
        apply_head(state, finalized.info.number, metrics, sender).await;
    }
    metrics.finalized.set(finalized.info.number as usize);
    state.snapshot.finalized = Some(finalized.info);
    state.put_finalized(finalized);
    sender
        .broadcast_direct(L1Event::NewFinalized { finalized })
        .await
        .ok();
}

/// Bounded by `timeout`, so a provider that never answers can't park a caller indefinitely.
#[cfg(feature = "node")]
async fn fetch_finalized_block_from_rpc(
    rpc: &impl Provider,
    timeout: Duration,
) -> anyhow::Result<Option<L1BlockInfoWithParent>> {
    let Some(block) = tokio::time::timeout(timeout, rpc.get_block(BlockId::finalized()))
        .await
        .context("L1 request timed out")??
    else {
        // This can happen in rare cases where the L1 chain is very young and has not finalized a
        // block yet. This is more common in testing and demo environments. In any case, we proceed
        // with a null L1 block rather than wait for the L1 to finalize a block, which can take a
        // long time. Callers log at a level appropriate to their context.
        return Ok(None);
    };

    Ok(Some((&block).into()))
}

#[cfg(test)]
mod test {
    use std::{
        borrow::Cow,
        ops::Add,
        sync::atomic::{AtomicUsize, Ordering},
        time::Duration,
    };

    use alloy::{
        eips::BlockNumberOrTag,
        node_bindings::{Anvil, AnvilInstance},
        primitives::utils::parse_ether,
        providers::{ext::AnvilApi, layers::AnvilProvider},
    };
    use espresso_contract_deployer::{Contracts, deploy_fee_contract_proxy};
    use hotshot_types::traits::metrics::NoMetrics;
    use serde_json::{Value, json};
    use time::OffsetDateTime;
    use warp::Filter;

    use super::*;

    /// JSON-RPC bodies recovered from production log dumps, except where marked synthetic. The
    /// alchemy bodies are raw captures (including the `***` redaction applied by the telemetry
    /// pipeline); the infura envelopes are reconstructed around the exact code/message/data
    /// captured.
    mod fixtures {
        pub const ALCHEMY_APP_INACTIVE: &str = r#"{"jsonrpc":"2.0","id":399193,"error":{"code":-32600,"message":"App is inactive. Please create a new app or contact support at https://dashboard.alchemy.com/***"}}"#;
        pub const ALCHEMY_10_BLOCK_RANGE: &str = r#"{"jsonrpc":"2.0","id":1005,"error":{"code":-32600,"message":"Under the Free tier plan, you can make eth_getLogs requests with up to a 10 block range. Based on your parameters, this block range should work: [0x1735fc9, 0x1735fd2]. Upgrade to PAYG for expanded block range."}}"#;
        pub const BLOCK_RANGE_TOO_LARGE: &str = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32062,"message":"Block range is too large"}}"#;
        /// Synthetic: telemetry has never captured a JSON-RPC body carrying code 429 from any
        /// provider. The message is alchemy's real throughput-limit text; the code is invented,
        /// to cover a 429 body that alloy parses and therefore hands back as `Ok`.
        pub const ALCHEMY_RATE_LIMIT: &str = r#"{"jsonrpc":"2.0","id":1,"error":{"code":429,"message":"Your app has exceeded its concurrent requests capacity. If you have retries enabled, you can safely ignore this message. If not, check out https://docs.alchemy.com/reference/throughput. Reach out to us if you'd like to increase your limits: https://dashboard.alchemy.com/support"}}"#;
        /// Infura's 429 body is a bare error object, not a JSON-RPC response, so alloy fails to
        /// parse it and it arrives as `Err(HttpError)`.
        pub const INFURA_RATE_LIMIT: &str = r#"{"code":-32005,"message":"Too Many Requests","data":{"see":"https://infura.io/dashboard"}}"#;
        /// Infura's per-second rate limit, as a well-formed JSON-RPC response, so alloy parses it
        /// and it arrives as `Ok`. Not a production capture: body from
        /// <https://github.com/INFURA/infura/issues/201>.
        pub const INFURA_RATE_EXCEEDED: &str = r#"{"jsonrpc":"2.0","id":3419,"error":{"code":-32005,"message":"project ID request rate exceeded","data":{"rate":{"allowed_rps":50,"backoff_seconds":0,"current_rps":52.3},"see":"https://infura.io/docs/ethereum/json-rpc/ratelimits"}}}"#;
        /// What infura returns for the rest of the UTC day once the credit quota is spent. Not a
        /// production capture: the message is the one alloy's `is_retry_err` special-cases as
        /// "thrown by infura if out of budget for the day and ratelimited".
        pub const INFURA_DAILY_QUOTA: &str = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32005,"message":"daily request count exceeded, request rate limited","data":{"see":"https://infura.io/dashboard"}}}"#;
        /// Infura load-balancer artifact: the node that served the request had not yet seen the
        /// head block. Not a production capture: the message is the one alloy's `is_retry_err`
        /// special-cases as "a load balancer issue".
        pub const INFURA_HEADER_NOT_FOUND: &str =
            r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"header not found"}}"#;
        pub const INFURA_TOO_MANY_RESULTS: &str = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32005,"message":"query returned more than 10000 results. Try with this block range [0x1500000, 0x15000FA].","data":{"from":"0x1500000","limit":10000,"to":"0x15000FA"}}}"#;
        pub const INFURA_UNAVAILABLE: &str = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32603,"message":"service temporarily unavailable"}}"#;
        pub const SUCCESS: &str = r#"{"jsonrpc":"2.0","id":1,"result":"0x1"}"#;
    }

    fn ok_packet(body: &str) -> StdResult<ResponsePacket, RpcError<TransportErrorKind>> {
        Ok(serde_json::from_str(body).expect("valid JSON-RPC response fixture"))
    }

    #[test]
    fn test_response_outcome_healthy_on_success() {
        assert!(matches!(
            ResponseOutcome::classify(&ok_packet(fixtures::SUCCESS)),
            ResponseOutcome::Healthy
        ));
    }

    #[test]
    fn test_response_outcome_alchemy_app_inactive_is_failed() {
        assert!(matches!(
            ResponseOutcome::classify(&ok_packet(fixtures::ALCHEMY_APP_INACTIVE)),
            ResponseOutcome::Failed
        ));
    }

    /// A block-range cap is a plan limit, not a request property: a paid backup serves the
    /// identical request, so this must stay scored and fail over.
    #[test]
    fn test_response_outcome_alchemy_10_block_range_is_failed() {
        assert!(matches!(
            ResponseOutcome::classify(&ok_packet(fixtures::ALCHEMY_10_BLOCK_RANGE)),
            ResponseOutcome::Failed
        ));
    }

    #[test]
    fn test_response_outcome_block_range_too_large_is_failed() {
        assert!(matches!(
            ResponseOutcome::classify(&ok_packet(fixtures::BLOCK_RANGE_TOO_LARGE)),
            ResponseOutcome::Failed
        ));
    }

    #[test]
    fn test_response_outcome_alchemy_rate_limit_body_is_rate_limited() {
        assert!(matches!(
            ResponseOutcome::classify(&ok_packet(fixtures::ALCHEMY_RATE_LIMIT)),
            ResponseOutcome::RateLimited { retry_after: None }
        ));
    }

    /// The shape a node hits when its infura credits run out: 429 for the rest of the UTC day.
    /// `RateLimited` rather than `Failed` keeps the backoff, and `MAX_CONSECUTIVE_RATE_LIMITS`
    /// still gets the node onto a working provider.
    #[test]
    fn test_response_outcome_infura_daily_quota_is_rate_limited() {
        assert!(matches!(
            ResponseOutcome::classify(&ok_packet(fixtures::INFURA_DAILY_QUOTA)),
            ResponseOutcome::RateLimited { retry_after: None }
        ));
    }

    /// Deferring to alloy's matcher pulls this in, and `RateLimited` is the handling it wants:
    /// the node that answered is behind the head, so back off instead of hammering it, and fail
    /// over if it stays behind past the backoff.
    #[test]
    fn test_response_outcome_infura_header_not_found_is_rate_limited() {
        assert!(matches!(
            ResponseOutcome::classify(&ok_packet(fixtures::INFURA_HEADER_NOT_FOUND)),
            ResponseOutcome::RateLimited { retry_after: None }
        ));
    }

    // Same -32005 as the result-count rejection, and it parses, so only the message separates
    // this from a request the backup would reject too.
    #[test]
    fn test_response_outcome_infura_rate_exceeded_is_rate_limited() {
        assert!(matches!(
            ResponseOutcome::classify(&ok_packet(fixtures::INFURA_RATE_EXCEEDED)),
            ResponseOutcome::RateLimited { retry_after: None }
        ));
    }

    #[test]
    fn test_infura_rate_limit_body_is_not_a_json_rpc_response() {
        serde_json::from_str::<ResponsePacket>(fixtures::INFURA_RATE_LIMIT).unwrap_err();
    }

    // Same -32005 as infura's rate limit, different message: a result-count rejection is the
    // request's fault, not the provider's.
    #[test]
    fn test_response_outcome_infura_too_many_results_is_request_rejected() {
        assert!(matches!(
            ResponseOutcome::classify(&ok_packet(fixtures::INFURA_TOO_MANY_RESULTS)),
            ResponseOutcome::RequestRejected
        ));
    }

    #[test]
    fn test_response_outcome_infura_unavailable_is_failed() {
        assert!(matches!(
            ResponseOutcome::classify(&ok_packet(fixtures::INFURA_UNAVAILABLE)),
            ResponseOutcome::Failed
        ));
    }

    // No captured provider response has ever carried JSON-RPC code 429 (alchemy sends -32600,
    // infura sends -32005); `ErrorResp` is not rate-limit signal here regardless of code.
    #[test]
    fn test_response_outcome_error_resp_is_failed() {
        let result: StdResult<ResponsePacket, RpcError<TransportErrorKind>> =
            Err(RpcError::ErrorResp(alloy::rpc::json_rpc::ErrorPayload {
                code: 429,
                message: "Too Many Requests".into(),
                data: None,
            }));
        assert!(matches!(
            ResponseOutcome::classify(&result),
            ResponseOutcome::Failed
        ));
    }

    #[test]
    fn test_response_outcome_http_error_429_is_rate_limited() {
        let result: StdResult<ResponsePacket, RpcError<TransportErrorKind>> =
            Err(RpcError::Transport(TransportErrorKind::HttpError(
                alloy::transports::HttpError {
                    status: 429,
                    body: fixtures::INFURA_RATE_LIMIT.to_owned(),
                },
            )));
        assert!(matches!(
            ResponseOutcome::classify(&result),
            ResponseOutcome::RateLimited { retry_after: None }
        ));
    }

    #[test]
    fn test_response_outcome_http_error_with_retry_after_is_rate_limited() {
        let result: StdResult<ResponsePacket, RpcError<TransportErrorKind>> = Err(
            RpcError::Transport(TransportErrorKind::HttpErrorWithRetryAfter {
                error: alloy::transports::HttpError {
                    status: 429,
                    body: fixtures::INFURA_RATE_LIMIT.to_owned(),
                },
                retry_after: Duration::from_secs(52),
            }),
        );
        assert!(matches!(
            ResponseOutcome::classify(&result),
            ResponseOutcome::RateLimited { retry_after: Some(d) } if d == Duration::from_secs(52)
        ));
    }

    #[test]
    fn test_rate_limit_backoff_honors_server_value_under_cap() {
        assert_eq!(
            rate_limit_backoff(Some(Duration::from_secs(52)), Duration::from_secs(1)),
            Duration::from_secs(52)
        );
    }

    #[test]
    fn test_rate_limit_backoff_clamps_server_value_over_cap() {
        assert_eq!(
            rate_limit_backoff(Some(Duration::from_secs(86400)), Duration::from_secs(1)),
            MAX_RATE_LIMIT_BACKOFF
        );
    }

    /// The configured delay is operator-set, so it is used as-is even beyond the cap.
    #[test]
    fn test_rate_limit_backoff_without_server_value_uses_configured_delay() {
        let configured = MAX_RATE_LIMIT_BACKOFF + Duration::from_secs(1);
        assert_eq!(rate_limit_backoff(None, configured), configured);
    }

    #[test]
    fn test_response_outcome_http_error_403_is_failed() {
        let result: StdResult<ResponsePacket, RpcError<TransportErrorKind>> =
            Err(RpcError::Transport(TransportErrorKind::HttpError(
                alloy::transports::HttpError {
                    status: 403,
                    body: String::new(),
                },
            )));
        assert!(matches!(
            ResponseOutcome::classify(&result),
            ResponseOutcome::Failed
        ));
    }

    #[test]
    fn test_l1_client_options_default_safety_margin() {
        assert_eq!(
            L1ClientOptions::default()
                .l1_finalized_safety_margin
                .blocks(),
            NonZeroU64::new(100)
        );
    }

    #[test]
    fn test_l1_safety_margin_from_str() {
        assert_eq!(
            "100".parse::<L1SafetyMargin>().unwrap().blocks(),
            NonZeroU64::new(100)
        );
        assert_eq!(
            "unlimited".parse::<L1SafetyMargin>().unwrap().blocks(),
            None
        );
        assert_eq!(
            "  Unlimited  ".parse::<L1SafetyMargin>().unwrap().blocks(),
            None
        );
        "garbage".parse::<L1SafetyMargin>().unwrap_err();
        "0".parse::<L1SafetyMargin>().unwrap_err();
    }

    #[test]
    fn test_l1_safety_margin_display_round_trip() {
        for margin in [
            L1SafetyMargin::from(NonZeroU64::new(100).unwrap()),
            L1SafetyMargin::UNLIMITED,
        ] {
            assert_eq!(
                margin.to_string().parse::<L1SafetyMargin>().unwrap(),
                margin
            );
        }
    }

    #[test]
    fn test_switching_transport_debug_hides_credentials() {
        let opt = L1ClientOptions {
            l1_ws_provider: Some(vec!["wss://u:p@ws.invalid/v2/WS_SECRET".parse().unwrap()]),
            ..Default::default()
        };
        let transport = SwitchingTransport::new(
            opt,
            vec!["https://u:p@rpc.invalid/v2/HTTP_SECRET".parse().unwrap()],
        )
        .expect("switching transport constructs");

        let debug = format!("{transport:?}");
        assert!(!debug.contains("WS_SECRET"), "{debug}");
        assert!(!debug.contains("HTTP_SECRET"), "{debug}");
        assert!(!debug.contains("u:p"), "{debug}");
        assert!(debug.contains("rpc.invalid"), "{debug}");
    }

    #[test]
    fn test_node_state_debug_hides_credentials() {
        let l1 = L1Client::new(vec![
            "https://u:p@rpc.invalid/v2/HTTP_SECRET".parse().unwrap(),
        ])
        .expect("L1 client constructs");
        let node_state = crate::NodeState::mock().with_l1(l1);

        let debug = format!("{node_state:?}");
        assert!(!debug.contains("HTTP_SECRET"), "{debug}");
        assert!(!debug.contains("u:p"), "{debug}");
    }

    async fn new_l1_client_opt(
        anvil: &Arc<AnvilInstance>,
        f: impl FnOnce(&mut L1ClientOptions),
    ) -> L1Client {
        let mut opt = L1ClientOptions {
            l1_events_max_block_range: 1,
            l1_polling_interval: Duration::from_secs(1),
            subscription_timeout: Duration::from_secs(5),
            ..Default::default()
        };
        f(&mut opt);

        let l1_client = opt
            .connect(vec![anvil.endpoint_url()])
            .expect("Failed to create L1 client");

        l1_client.spawn_tasks().await;
        l1_client
    }

    async fn new_l1_client(anvil: &Arc<AnvilInstance>, include_ws: bool) -> L1Client {
        new_l1_client_opt(anvil, |opt| {
            if include_ws {
                opt.l1_ws_provider = Some(vec![anvil.ws_endpoint_url()]);
            }
        })
        .await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_get_finalized_deposits() -> anyhow::Result<()> {
        let num_deposits = 5;

        let anvil = Anvil::new().spawn();
        let wallet = anvil.wallet().unwrap();
        let deployer = wallet.default_signer().address();
        let inner_provider = ProviderBuilder::new()
            .wallet(wallet)
            .connect_http(anvil.endpoint_url());
        // a provider that holds both anvil (to avoid accidental drop) and wallet-enabled L1 provider
        let provider = AnvilProvider::new(inner_provider, Arc::new(anvil));
        // cache store for deployed contracts
        let mut contracts = Contracts::new();

        // init and kick off the L1Client which wraps around standard L1 provider with more app-specific state management
        let l1_client = new_l1_client(provider.anvil(), false).await;

        // Initialize a contract with some deposits
        let fee_proxy_addr = deploy_fee_contract_proxy(&provider, &mut contracts, deployer).await?;
        let fee_proxy = FeeContract::new(fee_proxy_addr, &provider);
        let num_tx_for_deploy = provider.get_block_number().await?;

        // make some deposits.
        for n in 1..=num_deposits {
            // Varied amounts are less boring.
            let amount = n as f32 / 10.0;
            let receipt = fee_proxy
                .deposit(deployer)
                .value(parse_ether(&amount.to_string())?)
                .send()
                .await?
                .get_receipt()
                .await?;
            assert!(receipt.inner.is_success());
        }

        let cur_height = provider.get_block_number().await?;
        assert_eq!(num_deposits + num_tx_for_deploy, cur_height);

        // Set prev deposits to `None` so `Filter` will start at block
        // 0. The test would also succeed if we pass `0` (b/c first
        // block did not deposit).
        let pending = l1_client
            .get_finalized_deposits(fee_proxy_addr, None, cur_height)
            .await;

        assert_eq!(num_deposits as usize, pending.len(), "{pending:?}");
        assert_eq!(deployer, pending[0].account().0);
        assert_eq!(
            U256::from(1500000000000000000u64),
            pending
                .iter()
                .fold(U256::from(0), |total, info| total.add(info.amount().0))
        );

        // check a few more cases
        let pending = l1_client
            .get_finalized_deposits(fee_proxy_addr, Some(0), cur_height)
            .await;
        assert_eq!(num_deposits as usize, pending.len());

        let pending = l1_client
            .get_finalized_deposits(fee_proxy_addr, Some(0), 0)
            .await;
        assert_eq!(0, pending.len());

        let pending = l1_client
            .get_finalized_deposits(fee_proxy_addr, Some(0), 1)
            .await;
        assert_eq!(0, pending.len());

        let pending = l1_client
            .get_finalized_deposits(fee_proxy_addr, Some(num_tx_for_deploy), num_tx_for_deploy)
            .await;
        assert_eq!(0, pending.len());

        let pending = l1_client
            .get_finalized_deposits(
                fee_proxy_addr,
                Some(num_tx_for_deploy),
                num_tx_for_deploy + 1,
            )
            .await;
        assert_eq!(1, pending.len());

        // what happens if `new_finalized` is `0`?
        let pending = l1_client
            .get_finalized_deposits(fee_proxy_addr, Some(num_tx_for_deploy), 0)
            .await;
        assert_eq!(0, pending.len());

        Ok(())
    }

    async fn test_wait_for_finalized_block_helper(ws: bool) {
        let anvil = Arc::new(Anvil::new().block_time_f64(0.1).spawn());
        let l1_client = new_l1_client(&anvil, ws).await;
        let provider = &l1_client.provider;

        // Wait for a block 10 blocks in the future.
        let block_height = provider.get_block_number().await.unwrap();
        let block = l1_client.wait_for_finalized_block(block_height + 10).await;
        assert_eq!(block.number, block_height + 10);

        // Compare against underlying provider.
        let true_block = provider
            .get_block(BlockId::Number(BlockNumberOrTag::Number(block_height + 10)))
            .full()
            .await
            .unwrap()
            .unwrap();

        assert_eq!(
            block.timestamp.to::<u64>(),
            true_block.header.inner.timestamp
        );
        assert_eq!(block.hash, true_block.header.hash);
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_finalized_block_ws() {
        test_wait_for_finalized_block_helper(true).await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_finalized_block_http() {
        test_wait_for_finalized_block_helper(false).await
    }

    async fn test_wait_for_old_finalized_block_helper(ws: bool) {
        let anvil = Arc::new(Anvil::new().block_time_f64(0.2).spawn());
        let l1_client = new_l1_client_opt(&anvil, |opt| {
            if ws {
                opt.l1_ws_provider = Some(vec![anvil.ws_endpoint_url()]);
            }
            opt.l1_finalized_safety_margin = NonZeroU64::new(1).unwrap().into();
        })
        .await;
        let provider = &l1_client.provider;

        // Wait for anvil to finalize a few blocks.
        l1_client.wait_for_finalized_block(2).await;

        // Get an old finalized block.
        let block = l1_client.wait_for_finalized_block(0).await;

        // Compare against underlying provider.
        let true_block = provider.get_block(0.into()).await.unwrap().unwrap();
        assert_eq!(block.hash, true_block.header.hash);
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_old_finalized_block_ws() {
        test_wait_for_old_finalized_block_helper(true).await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_old_finalized_block_http() {
        test_wait_for_old_finalized_block_helper(false).await
    }

    async fn test_wait_for_finalized_block_by_timestamp_helper(ws: bool) {
        let anvil = Arc::new(Anvil::new().block_time_f64(0.2).spawn());
        let l1_client = new_l1_client(&anvil, ws).await;
        let provider = &l1_client.provider;

        // Wait for a block 5 blocks in the future.
        let timestamp = U256::from(OffsetDateTime::now_utc().unix_timestamp() as u64 + 5);
        let block = l1_client
            .wait_for_finalized_block_with_timestamp(timestamp)
            .await;
        assert!(
            block.timestamp >= timestamp,
            "wait_for_finalized_block_with_timestamp({timestamp}) returned too early a block: \
             {block:?}",
        );
        let parent = provider
            .get_block(BlockId::Number(BlockNumberOrTag::Number(block.number - 1)))
            .full()
            .await
            .unwrap()
            .unwrap();
        assert!(
            parent.header.inner.timestamp < timestamp.to::<u64>(),
            "wait_for_finalized_block_with_timestamp({timestamp}) did not return the earliest \
             possible block: returned {block:?}, but earlier block {parent:?} has an acceptable \
             timestamp too",
        );

        // Compare against underlying provider.
        let true_block = provider
            .get_block(BlockId::Number(BlockNumberOrTag::Number(block.number)))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(
            block.timestamp.to::<u64>(),
            true_block.header.inner.timestamp
        );
        assert_eq!(block.hash, true_block.header.hash);
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_finalized_block_by_timestamp_ws() {
        test_wait_for_finalized_block_by_timestamp_helper(true).await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_finalized_block_by_timestamp_http() {
        test_wait_for_finalized_block_by_timestamp_helper(false).await
    }

    async fn test_wait_for_old_finalized_block_by_timestamp_helper(ws: bool) {
        let anvil = Arc::new(Anvil::new().block_time_f64(0.2).spawn());
        let l1_client = new_l1_client(&anvil, ws).await;

        // Get the timestamp of the first block.
        let true_block = l1_client.wait_for_finalized_block(0).await;
        let timestamp = true_block.timestamp;

        // Wait for some more blocks to be produced.
        l1_client.wait_for_finalized_block(10).await;

        // Get the old block by timestamp.
        let block = l1_client
            .wait_for_finalized_block_with_timestamp(U256::from(timestamp))
            .await;
        assert_eq!(block, true_block);
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_old_finalized_block_by_timestamp_ws() {
        test_wait_for_old_finalized_block_by_timestamp_helper(true).await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_old_finalized_block_by_timestamp_http() {
        test_wait_for_old_finalized_block_by_timestamp_helper(false).await
    }

    async fn test_wait_for_block_helper(ws: bool) {
        let anvil = Arc::new(Anvil::new().block_time_f64(0.1).spawn());
        let l1_client = new_l1_client(&anvil, ws).await;
        let provider = &l1_client.provider;

        // Wait for a block 10 blocks in the future.
        let block_height = provider.get_block_number().await.unwrap();
        l1_client.wait_for_block(block_height + 10).await;

        let new_block_height = provider.get_block_number().await.unwrap();
        assert!(
            new_block_height >= block_height + 10,
            "wait_for_block returned too early; initial height = {block_height}, new height = \
             {new_block_height}",
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_block_ws() {
        test_wait_for_block_helper(true).await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_block_http() {
        test_wait_for_block_helper(false).await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_block_refreshes_without_poller() {
        let anvil = Anvil::new().arg("--no-mining").spawn();
        let l1_client = L1ClientOptions::default()
            .connect(vec![anvil.endpoint_url()])
            .expect("Failed to create L1 client");

        let head = l1_client.get_block_number().await.unwrap();
        l1_client.anvil_mine(Some(1), None).await.unwrap();

        tokio::time::timeout(Duration::from_secs(3), l1_client.wait_for_block(head + 1))
            .await
            .expect("wait_for_block did not refresh from the RPC without the poller");
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_finalized_block_refreshes_without_poller() {
        let anvil = Anvil::new()
            .args(["--no-mining", "--slots-in-an-epoch", "1"])
            .spawn();
        let l1_client = L1ClientOptions::default()
            .connect(vec![anvil.endpoint_url()])
            .expect("Failed to create L1 client");

        // One slot per epoch advances finalization quickly; five blocks is comfortably enough.
        l1_client.anvil_mine(Some(5), None).await.unwrap();
        let new_finalized = fetch_finalized_block_from_rpc(
            &l1_client.provider,
            l1_client.options().l1_request_timeout,
        )
        .await
        .unwrap()
        .expect("anvil finalized a block after mining");

        let block = tokio::time::timeout(
            Duration::from_secs(3),
            l1_client.wait_for_finalized_block(new_finalized.info.number),
        )
        .await
        .expect("wait_for_finalized_block did not refresh from the RPC without the poller");
        assert_eq!(block.hash, new_finalized.info.hash);
    }

    /// An RPC that pruned old history returns null for block `number`; waiting for a later
    /// finalized block must not request it.
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_latest_finalized_at_least_skips_pruned_block() {
        let anvil = Anvil::new()
            .args(["--no-mining", "--slots-in-an-epoch", "1"])
            .spawn();
        let by_number_requests = Arc::new(AtomicUsize::new(0));
        let proxy_url =
            serve_pruned_history_proxy(anvil.endpoint_url(), by_number_requests.clone()).await;
        // Margin 1 routes `pruned_number` through the by-number fetch, so a regression shows in
        // the counter.
        let l1_client = L1ClientOptions {
            l1_finalized_safety_margin: NonZeroU64::new(1).unwrap().into(),
            l1_wait_refresh_interval: Duration::from_millis(50),
            ..Default::default()
        }
        .connect(vec![proxy_url])
        .expect("Failed to create L1 client");

        l1_client.anvil_mine(Some(5), None).await.unwrap();
        let new_finalized = fetch_finalized_block_from_rpc(
            &l1_client.provider,
            l1_client.options().l1_request_timeout,
        )
        .await
        .unwrap()
        .expect("anvil finalized a block after mining");
        let pruned_number = new_finalized.info.number - 2;

        let block = tokio::time::timeout(
            Duration::from_secs(3),
            l1_client.wait_for_latest_finalized_at_least(pruned_number),
        )
        .await
        .expect("wait_for_latest_finalized_at_least did not return");
        assert_eq!(block, new_finalized.info);
        assert_eq!(by_number_requests.load(Ordering::SeqCst), 0);

        let old_path = tokio::time::timeout(
            Duration::from_millis(500),
            l1_client.wait_for_finalized_block(pruned_number),
        )
        .await;
        old_path.expect_err("wait_for_finalized_block returned a block the RPC pruned");
        assert!(by_number_requests.load(Ordering::SeqCst) > 0);
    }

    /// Proxies every JSON-RPC request to `target` except `eth_getBlockByNumber` with a numeric
    /// block, which returns null as on an RPC that pruned old history, counting those requests.
    async fn serve_pruned_history_proxy(target: Url, by_number_requests: Arc<AtomicUsize>) -> Url {
        let provider = ProviderBuilder::new().connect_http(target);
        let route = warp::post()
            .and(warp::body::json::<Value>())
            .then(move |req: Value| {
                let provider = provider.clone();
                let by_number_requests = by_number_requests.clone();
                async move {
                    let id = req["id"].clone();
                    let is_by_number_query = req["method"].as_str() == Some("eth_getBlockByNumber")
                        && req["params"]
                            .get(0)
                            .and_then(Value::as_str)
                            .is_some_and(|block| block.starts_with("0x"));
                    if is_by_number_query {
                        by_number_requests.fetch_add(1, Ordering::SeqCst);
                        return jsonrpc_result(id, Value::Null);
                    }

                    let method = req["method"]
                        .as_str()
                        .expect("JSON-RPC request has method")
                        .to_string();
                    let result: Value = provider
                        .raw_request(Cow::Owned(method), req["params"].clone())
                        .await
                        .expect("proxy request to anvil");
                    jsonrpc_result(id, result)
                }
            });
        test_server::serve_on_random_port(route).await
    }

    /// Never responds, to prove a hung L1 RPC can't stall a bounded call indefinitely.
    async fn serve_hanging() -> Url {
        let route = warp::post()
            .and(warp::body::json::<Value>())
            .then(|req: Value| async move {
                tokio::time::sleep(Duration::from_secs(60)).await;
                jsonrpc_result(req["id"].clone(), json!("0x0"))
            });
        test_server::serve_on_random_port(route).await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_fetch_finalized_block_from_rpc_bounded_by_request_timeout() {
        let url = serve_hanging().await;
        let l1_client = L1ClientOptions {
            l1_request_timeout: Duration::from_millis(200),
            ..Default::default()
        }
        .connect(vec![url])
        .expect("Failed to create L1 client");

        let result = tokio::time::timeout(
            Duration::from_secs(2),
            fetch_finalized_block_from_rpc(
                &l1_client.provider,
                l1_client.options().l1_request_timeout,
            ),
        )
        .await
        .expect("l1_request_timeout did not bound a hung finalized-block RPC call");

        assert!(result.is_err(), "a hung RPC call must time out as an error");
    }

    /// `throttle_refresh` must bound `Head` and `Finalized` independently, so a wait on one kind
    /// can't starve a wait on the other.
    ///
    /// `throttle_refresh` gates on `std::time::Instant`, not tokio time, so this uses a short
    /// configured interval and a real sleep rather than `tokio::time::pause`/`advance`.
    #[test_log::test(tokio::test)]
    async fn test_throttle_refresh_bounds_each_kind_independently() {
        let interval = Duration::from_millis(20);
        let l1_client = L1ClientOptions {
            l1_wait_refresh_interval: interval,
            ..Default::default()
        }
        .connect(vec!["http://localhost:0".parse().unwrap()])
        .expect("Failed to create L1 client");

        assert!(
            l1_client.throttle_refresh(RefreshKind::Head).await,
            "first head refresh must not be throttled"
        );
        assert!(
            l1_client.throttle_refresh(RefreshKind::Finalized).await,
            "first finalized refresh must not be throttled"
        );
        assert!(
            !l1_client.throttle_refresh(RefreshKind::Head).await,
            "a second immediate head refresh must be throttled"
        );
        assert!(
            !l1_client.throttle_refresh(RefreshKind::Finalized).await,
            "a second immediate finalized refresh must be throttled"
        );

        sleep(interval).await;

        assert!(
            l1_client.throttle_refresh(RefreshKind::Head).await,
            "head refresh must be allowed again once the interval elapses"
        );
    }

    /// Wraps `result` in a JSON-RPC 2.0 success envelope for `id`.
    fn jsonrpc_result(id: Value, result: Value) -> warp::reply::Json {
        warp::reply::json(&json!({ "jsonrpc": "2.0", "id": id, "result": result }))
    }

    /// Answers every JSON-RPC request with head `0x0`, counting how many requests it receives.
    async fn serve_counting_head(counter: Arc<AtomicUsize>) -> Url {
        let route = warp::post()
            .and(warp::body::json::<Value>())
            .then(move |req: Value| {
                let counter = counter.clone();
                async move {
                    counter.fetch_add(1, Ordering::SeqCst);
                    jsonrpc_result(req["id"].clone(), json!("0x0"))
                }
            });
        test_server::serve_on_random_port(route).await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_block_refresh_is_throttled() {
        let counter = Arc::new(AtomicUsize::new(0));
        let url = serve_counting_head(counter.clone()).await;
        let l1_client = L1ClientOptions::default()
            .connect(vec![url])
            .expect("Failed to create L1 client");

        let wait = Duration::from_secs(2);
        let waiters = (0..10).map(|_| {
            let l1_client = &l1_client;
            async move {
                tokio::time::timeout(wait, l1_client.wait_for_block(u64::MAX))
                    .await
                    .ok();
            }
        });
        futures::future::join_all(waiters).await;

        let interval = l1_client.options().l1_wait_refresh_interval;
        let max_requests = (wait.as_millis() / interval.as_millis()) as usize + 1;
        let requests = counter.load(Ordering::SeqCst);
        assert!(
            requests >= 1,
            "10 concurrent waiters made no refresh request at all"
        );
        assert!(
            requests <= max_requests,
            "throttle allowed {requests} refresh requests from 10 concurrent waiters, expected at \
             most {max_requests}",
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_block_refresh_broadcasts_new_head() {
        let anvil = Anvil::new().arg("--no-mining").spawn();
        let l1_client = L1ClientOptions::default()
            .connect(vec![anvil.endpoint_url()])
            .expect("Failed to create L1 client");
        let mut events = l1_client.receiver.activate_cloned();

        let head = l1_client.get_block_number().await.unwrap();
        l1_client.anvil_mine(Some(1), None).await.unwrap();
        tokio::time::timeout(Duration::from_secs(3), l1_client.wait_for_block(head + 1))
            .await
            .expect("wait_for_block did not refresh from the RPC without the poller");

        assert_eq!(l1_client.snapshot().await.head, head + 1);

        tokio::time::timeout(Duration::from_secs(1), async {
            loop {
                if let L1Event::NewHead { head: h } =
                    events.next().await.expect("event stream open")
                    && h == head + 1
                {
                    return;
                }
            }
        })
        .await
        .expect("refresh did not broadcast NewHead for the refreshed head");
    }

    /// A refresh hitting `Rate limit exceeded` must not panic or busy-loop.
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_wait_for_block_refresh_recovers_from_rate_limit() {
        let counter = Arc::new(AtomicUsize::new(0));
        let rate_limited = test_server::serve_on_random_port(warp::any().map({
            let counter = counter.clone();
            move || {
                counter.fetch_add(1, Ordering::SeqCst);
                warp::reply::with_header(
                    warp::reply::with_status(
                        fixtures::ALCHEMY_RATE_LIMIT,
                        test_server::StatusCode::TOO_MANY_REQUESTS,
                    ),
                    "content-type",
                    "application/json",
                )
            }
        }))
        .await;
        let anvil = Anvil::new().block_time(1).spawn();

        let l1_client = L1ClientOptions {
            l1_frequent_failure_tolerance: Duration::from_millis(0),
            l1_consecutive_failure_tolerance: 1,
            l1_rate_limit_delay: Some(Duration::ZERO),
            ..Default::default()
        }
        .connect(vec![rate_limited, anvil.endpoint_url()])
        .expect("Failed to create L1 client");

        tokio::time::timeout(Duration::from_secs(10), l1_client.wait_for_block(3))
            .await
            .expect(
                "wait_for_block did not recover once the refresh failed over off the rate-limited \
                 provider",
            );

        // The rate-limited provider must stop receiving requests once refresh fails over to the
        // healthy one: `MAX_CONSECUTIVE_RATE_LIMITS` requests trigger the failover, then every
        // later tick goes to anvil instead.
        let requests = counter.load(Ordering::SeqCst);
        assert_eq!(
            requests, MAX_CONSECUTIVE_RATE_LIMITS,
            "rate-limited provider kept receiving requests after failover"
        );
    }

    /// Must not trip `put_finalized`'s assertion that a finalized block is already known.
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_refresh_finalized_handles_missing_finalized_block() {
        let route = warp::post()
            .and(warp::body::json::<Value>())
            .then(|req: Value| async move { jsonrpc_result(req["id"].clone(), Value::Null) });
        let url = test_server::serve_on_random_port(route).await;
        let l1_client = L1ClientOptions::default()
            .connect(vec![url])
            .expect("Failed to create L1 client");

        l1_client.refresh_finalized().await;

        assert!(l1_client.snapshot().await.finalized.is_none());
    }

    #[test_log::test(tokio::test)]
    async fn test_apply_head_never_lowers_snapshot() {
        let mut state = L1State::new(NonZeroUsize::new(10).unwrap());
        state.snapshot.head = 10;
        let metrics = L1ClientMetrics::new(&NoMetrics, 1);
        let (sender, mut receiver) = async_broadcast::broadcast(1);

        apply_head(&mut state, 5, &metrics, &sender).await;

        assert_eq!(state.snapshot.head, 10);
        assert!(
            receiver.try_recv().is_err(),
            "a non-advancing head must not broadcast NewHead"
        );
    }

    #[test_log::test(tokio::test)]
    async fn test_apply_finalized_never_lowers_snapshot() {
        let mut state = L1State::new(NonZeroUsize::new(10).unwrap());
        let current = L1BlockInfoWithParent {
            info: L1BlockInfo {
                number: 10,
                timestamp: U256::from(1),
                hash: B256::repeat_byte(1),
            },
            parent_hash: B256::ZERO,
        };
        state.snapshot.finalized = Some(current.info);
        let metrics = L1ClientMetrics::new(&NoMetrics, 1);
        let (sender, mut receiver) = async_broadcast::broadcast(1);

        let lower = L1BlockInfoWithParent {
            info: L1BlockInfo {
                number: 5,
                timestamp: U256::from(2),
                hash: B256::repeat_byte(2),
            },
            parent_hash: B256::ZERO,
        };
        apply_finalized(&mut state, lower, &metrics, &sender).await;

        assert_eq!(state.snapshot.finalized, Some(current.info));
        assert!(
            receiver.try_recv().is_err(),
            "a non-advancing finalized block must not broadcast NewFinalized"
        );
    }

    #[test_log::test(tokio::test)]
    async fn test_apply_finalized_rejects_equal_height_different_hash() {
        let mut state = L1State::new(NonZeroUsize::new(10).unwrap());
        let current = L1BlockInfoWithParent {
            info: L1BlockInfo {
                number: 10,
                timestamp: U256::from(1),
                hash: B256::repeat_byte(1),
            },
            parent_hash: B256::ZERO,
        };
        state.snapshot.finalized = Some(current.info);
        let metrics = L1ClientMetrics::new(&NoMetrics, 1);
        let (sender, mut receiver) = async_broadcast::broadcast(1);

        let same_height_different_hash = L1BlockInfoWithParent {
            info: L1BlockInfo {
                number: 10,
                timestamp: U256::from(2),
                hash: B256::repeat_byte(2),
            },
            parent_hash: B256::ZERO,
        };
        apply_finalized(&mut state, same_height_different_hash, &metrics, &sender).await;

        assert_eq!(state.snapshot.finalized, Some(current.info));
        assert!(
            receiver.try_recv().is_err(),
            "an equal-height block with a different hash must not overwrite the snapshot or \
             broadcast"
        );
    }

    #[test_log::test(tokio::test)]
    async fn test_apply_finalized_applies_higher() {
        let mut state = L1State::new(NonZeroUsize::new(10).unwrap());
        // Head already covers the finalized height, so this test exercises only the finalized
        // update; head-raising is covered by `test_apply_finalized_raises_head_to_match`.
        state.snapshot.head = 10;
        let current = L1BlockInfoWithParent {
            info: L1BlockInfo {
                number: 5,
                timestamp: U256::from(1),
                hash: B256::repeat_byte(1),
            },
            parent_hash: B256::ZERO,
        };
        state.snapshot.finalized = Some(current.info);
        let metrics = L1ClientMetrics::new(&NoMetrics, 1);
        let (sender, mut receiver) = async_broadcast::broadcast(1);

        let higher = L1BlockInfoWithParent {
            info: L1BlockInfo {
                number: 10,
                timestamp: U256::from(2),
                hash: B256::repeat_byte(2),
            },
            parent_hash: B256::ZERO,
        };
        apply_finalized(&mut state, higher, &metrics, &sender).await;

        assert_eq!(state.snapshot.finalized, Some(higher.info));
        let event = receiver
            .try_recv()
            .expect("a higher finalized block broadcasts NewFinalized");
        assert!(matches!(event, L1Event::NewFinalized { finalized } if finalized == higher));
    }

    #[test_log::test(tokio::test)]
    async fn test_apply_finalized_raises_head_to_match() {
        let mut state = L1State::new(NonZeroUsize::new(10).unwrap());
        state.snapshot.head = 10;
        let metrics = L1ClientMetrics::new(&NoMetrics, 1);
        let (sender, mut receiver) = async_broadcast::broadcast(2);

        let finalized = L1BlockInfoWithParent {
            info: L1BlockInfo {
                number: 20,
                timestamp: U256::from(1),
                hash: B256::repeat_byte(1),
            },
            parent_hash: B256::ZERO,
        };
        apply_finalized(&mut state, finalized, &metrics, &sender).await;

        assert_eq!(state.snapshot.head, 20);
        assert_eq!(state.snapshot.finalized, Some(finalized.info));

        let head_event = receiver
            .try_recv()
            .expect("a finalized block above the head broadcasts NewHead");
        assert!(matches!(head_event, L1Event::NewHead { head } if head == 20));

        let finalized_event = receiver
            .try_recv()
            .expect("the finalized block also broadcasts NewFinalized");
        assert!(
            matches!(finalized_event, L1Event::NewFinalized { finalized: f } if f == finalized)
        );
    }

    async fn test_reconnect_update_task_helper(ws: bool) {
        // Use port 0 to let OS assign a port, avoiding race conditions
        let anvil = Arc::new(Anvil::new().block_time(1).port(0u16).spawn());
        let port = anvil.port();
        let client = new_l1_client(&anvil, ws).await;

        let initial_state = client.snapshot().await;
        tracing::info!(?initial_state, "initial state");

        // Check the state is updating.
        let mut retry = 0;
        let updated_state = loop {
            assert!(retry < 10, "state did not update in time");

            let updated_state = client.snapshot().await;
            if updated_state.head > initial_state.head {
                break updated_state;
            }
            tracing::info!(retry, "waiting for state update");
            sleep(Duration::from_secs(1)).await;
            retry += 1;
        };
        tracing::info!(?updated_state, "state updated");

        // Disconnect the WebSocket and reconnect it. Technically this spawns a whole new Anvil
        // chain, but for the purposes of this test it should look to the client like an L1 server
        // closing a WebSocket connection.
        drop(anvil);

        // Let the connection stay down for a little while: Ethers internally tries to reconnect,
        // and starting up to fast again might hit that and cause a false positive. The problem is,
        // Ethers doesn't try very hard, and if we wait a bit, we will test the worst possible case
        // where the internal retry logic gives up and just kills the whole provider.
        tracing::info!("sleep 5");
        sleep(Duration::from_secs(5)).await;

        // Once a connection is reestablished, the state will eventually start to update again.
        tracing::info!("restarting L1");
        let _anvil = Anvil::new().block_time(1).port(port).spawn();

        let mut retry = 0;
        let final_state = loop {
            assert!(retry < 5, "state did not update in time");

            let final_state = client.snapshot().await;
            if final_state.head > updated_state.head {
                break final_state;
            }
            tracing::info!(retry, "waiting for state update");
            sleep(Duration::from_secs(1)).await;
            retry += 1;
        };
        tracing::info!(?final_state, "state updated");
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_reconnect_update_task_ws() {
        test_reconnect_update_task_helper(true).await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_reconnect_update_task_http() {
        test_reconnect_update_task_helper(false).await
    }

    // #[test_log::test(tokio::test(flavor = "multi_thread"))]
    // async fn test_fetch_stake_table() -> anyhow::Result<()> {

    //     let anvil = Anvil::new().spawn();
    //     let wallet = anvil.wallet().unwrap();
    //     let inner_provider = ProviderBuilder::new()
    //         .wallet(wallet)
    //         .on_http(anvil.endpoint_url());
    //     let provider = AnvilProvider::new(inner_provider, Arc::new(anvil));

    //     let l1_client = new_l1_client(provider.anvil(), false).await;
    //     let mut contracts = Contracts::new();

    //     let stake_table_addr =
    //         deploy_permissioned_stake_table(&provider, &mut contracts, vec![]).await?;
    //     let stake_table_contract = PermissionedStakeTable::new(stake_table_addr, &provider);

    //     let mut rng = rand::thread_rng();
    //     let node = NodeInfoSol::rand(&mut rng);

    //     let new_nodes: Vec<NodeInfoSol> = vec![node];
    //     stake_table_contract
    //         .update(vec![], new_nodes)
    //         .send()
    //         .await?
    //         .watch()
    //         .await?;

    //     let block = l1_client.get_block(BlockId::latest()).await?.unwrap();
    //     let nodes = l1_client
    //         .get_stake_table(stake_table_addr, block.header.inner.number)
    //         .await?;

    //     let result = nodes.stake_table.0[0].clone();
    //     assert_eq!(result.stake_table_entry.stake_amount.to::<u64>(), 1);
    //     Ok(())
    // }

    /// Requests already in flight when the backoff window opened are not new evidence about the
    /// provider: without this, two concurrent callers fail over on a single burst.
    #[test]
    fn test_rate_limits_inside_the_backoff_window_count_once() {
        let mut status = SingleTransportStatus::default();
        let until = Instant::now() + Duration::from_secs(3600);

        assert!(!status.log_rate_limit(until));
        for _ in 0..MAX_CONSECUTIVE_RATE_LIMITS + 1 {
            assert!(!status.log_rate_limit(until), "still inside the window");
        }
        assert_eq!(status.consecutive_rate_limits, 1);
    }

    /// The window elapsing between them is what makes rate limits count towards a failover.
    #[test]
    fn test_rate_limits_outliving_their_backoff_switch() {
        let mut status = SingleTransportStatus::default();

        for _ in 0..MAX_CONSECUTIVE_RATE_LIMITS - 1 {
            assert!(!status.log_rate_limit(Instant::now()));
        }
        assert!(status.log_rate_limit(Instant::now()));
    }

    /// A success in between clears the count, so rate limits a provider recovers from never
    /// accumulate into a failover.
    #[test]
    fn test_success_resets_the_rate_limit_count() {
        let mut status = SingleTransportStatus::default();

        for _ in 0..MAX_CONSECUTIVE_RATE_LIMITS * 2 {
            assert!(!status.log_rate_limit(Instant::now()));
            status.log_success();
        }
        assert_eq!(status.consecutive_rate_limits, 0);
    }

    /// A helper function to get the index of the current provider in the failover list.
    fn get_failover_index(provider: &L1Client) -> usize {
        let transport = &provider.transport;
        provider.transport.current_transport.read().generation % transport.urls.len()
    }

    /// A provider that always answers with a parseable JSON-RPC error must be scored unhealthy
    /// and trigger failover, not be treated as healthy (see [`ResponseOutcome`]).
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_failover_on_dead_provider_with_json_rpc_error_body() {
        let dead_provider = test_server::serve_fixed(
            test_server::StatusCode::FORBIDDEN,
            "application/json",
            fixtures::ALCHEMY_APP_INACTIVE,
        )
        .await;
        let anvil = Anvil::new().block_time(1).spawn();

        let provider = L1ClientOptions {
            l1_frequent_failure_tolerance: Duration::from_millis(0),
            l1_consecutive_failure_tolerance: 3,
            ..Default::default()
        }
        .connect(vec![dead_provider, anvil.endpoint_url()])
        .expect("Failed to create L1 client");

        for _ in 0..2 {
            provider.get_block_number().await.unwrap_err();
            assert_eq!(get_failover_index(&provider), 0);
        }

        provider.get_block_number().await.unwrap_err();
        assert_eq!(
            get_failover_index(&provider),
            1,
            "client should have failed over from the dead provider"
        );
        provider
            .get_block_number()
            .await
            .expect("requests succeed from the healthy provider");
    }

    /// A rate-limited provider that a failover-happy config would abandon on the first error.
    /// The backoff is long enough that every call after the first short-circuits client-side.
    async fn rate_limited_client(rate_limit_delay: Duration) -> (L1Client, AnvilInstance) {
        let rate_limited = test_server::serve_fixed(
            test_server::StatusCode::TOO_MANY_REQUESTS,
            "application/json",
            fixtures::ALCHEMY_RATE_LIMIT,
        )
        .await;
        let anvil = Anvil::new().block_time(1).spawn();

        let client = L1ClientOptions {
            l1_frequent_failure_tolerance: Duration::from_millis(0),
            l1_consecutive_failure_tolerance: 1,
            l1_rate_limit_delay: Some(rate_limit_delay),
            ..Default::default()
        }
        .connect(vec![rate_limited, anvil.endpoint_url()])
        .expect("Failed to create L1 client");

        (client, anvil)
    }

    /// A rate limit backs off on the current provider instead of failing over.
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_no_failover_on_rate_limit_json_rpc_error_body() {
        let (provider, _anvil) = rate_limited_client(Duration::from_secs(3600)).await;

        let err = provider.get_block_number().await.unwrap_err();
        assert_eq!(get_failover_index(&provider), 0);
        assert!(
            !err.to_string().contains("Rate limit exceeded"),
            "first call reaches the provider: {err}"
        );

        // Inside the backoff window, so this never leaves the client, and still does not fail
        // over even though a single failure would.
        let err = provider.get_block_number().await.unwrap_err();
        assert_eq!(get_failover_index(&provider), 0);
        assert!(
            err.to_string().contains("Rate limit exceeded"),
            "second call short-circuits on the backoff: {err}"
        );
    }

    /// A provider still rate limiting after its own backoff is treated as failed. Otherwise
    /// `rate_limited_until` short-circuits every later request and the healthy backup, which the
    /// client already holds, is never reached.
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_failover_on_persistent_rate_limit() {
        let (provider, _anvil) = rate_limited_client(Duration::ZERO).await;

        for _ in 0..MAX_CONSECUTIVE_RATE_LIMITS {
            provider.get_block_number().await.unwrap_err();
        }

        assert_eq!(get_failover_index(&provider), 1);
        provider
            .get_block_number()
            .await
            .expect("requests succeed from the healthy provider");
    }

    /// A range rejection is the request's fault: every backup rejects it identically, so it must
    /// not burn a generation. `l1_consecutive_failure_tolerance: 1` fails over on any scored
    /// failure, so a stable index is the whole assertion.
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_no_failover_on_request_rejection() {
        let rejecting = test_server::serve_fixed(
            test_server::StatusCode::OK,
            "application/json",
            fixtures::INFURA_TOO_MANY_RESULTS,
        )
        .await;
        let anvil = Anvil::new().block_time(1).spawn();

        let provider = L1ClientOptions {
            l1_consecutive_failure_tolerance: 1,
            ..Default::default()
        }
        .connect(vec![rejecting, anvil.endpoint_url()])
        .expect("Failed to create L1 client");

        for _ in 0..3 {
            provider.get_block_number().await.unwrap_err();
            assert_eq!(get_failover_index(&provider), 0);
        }
    }

    async fn test_failover_update_task_helper(ws: bool) {
        let anvil = Anvil::new().block_time(1).spawn();

        // Create an L1 client with fake providers, and check that the state is still updated after
        // it correctly fails over to the real providers.
        let client = L1ClientOptions {
            l1_polling_interval: Duration::from_secs(1),
            // Use a very long subscription timeout, so that we only succeed by triggering a
            // failover.
            subscription_timeout: Duration::from_secs(1000),
            l1_ws_provider: if ws {
                Some(vec![
                    "ws://notarealurl:1234".parse().unwrap(),
                    anvil.ws_endpoint_url(),
                ])
            } else {
                None
            },
            ..Default::default()
        }
        .connect(vec![
            "http://notarealurl:1234".parse().unwrap(),
            anvil.endpoint_url(),
        ])
        .expect("Failed to create L1 client");

        client.spawn_tasks().await;

        let initial_state = client.snapshot().await;
        tracing::info!(?initial_state, "initial state");

        // Check the state is updating.
        let mut retry = 0;
        let updated_state = loop {
            assert!(retry < 10, "state did not update in time");

            let updated_state = client.snapshot().await;
            if updated_state.head > initial_state.head {
                break updated_state;
            }
            tracing::info!(retry, "waiting for state update");
            sleep(Duration::from_secs(1)).await;
            retry += 1;
        };
        tracing::info!(?updated_state, "state updated");
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_failover_update_task_ws() {
        test_failover_update_task_helper(true).await;
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_failover_update_task_http() {
        test_failover_update_task_helper(false).await;
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_failover_consecutive_failures() {
        let anvil = Anvil::new().block_time(1).spawn();

        let l1_options = L1ClientOptions {
            l1_polling_interval: Duration::from_secs(1),
            l1_frequent_failure_tolerance: Duration::from_millis(0),
            l1_consecutive_failure_tolerance: 3,
            ..Default::default()
        };

        let provider = l1_options
            .connect(vec![
                "http://notarealurl:1234".parse().unwrap(),
                anvil.endpoint_url(),
            ])
            .expect("Failed to create L1 client");

        // Make just enough failed requests not to trigger a failover.
        for _ in 0..2 {
            provider.get_block_number().await.unwrap_err();
            assert!(get_failover_index(&provider) == 0);
        }

        // The final request triggers failover.
        provider.get_block_number().await.unwrap_err();
        assert!(get_failover_index(&provider) == 1);

        // Now requests succeed.
        provider.get_block_number().await.unwrap();
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_failover_frequent_failures() {
        let anvil = Anvil::new().block_time(1).spawn();
        let provider = L1ClientOptions {
            l1_polling_interval: Duration::from_secs(1),
            l1_frequent_failure_tolerance: Duration::from_millis(100),
            ..Default::default()
        }
        .connect(vec![
            "http://notarealurl:1234".parse().unwrap(),
            anvil.endpoint_url(),
        ])
        .expect("Failed to create L1 client");

        // Two failed requests that are not within the tolerance window do not trigger a failover.
        provider.get_block_number().await.unwrap_err();
        sleep(Duration::from_secs(1)).await;
        provider.get_block_number().await.unwrap_err();

        // Check that we didn't fail over.
        assert!(get_failover_index(&provider) == 0);

        // Reset the window.
        sleep(Duration::from_secs(1)).await;

        // Two failed requests in a row trigger failover.
        provider.get_block_number().await.unwrap_err();
        provider.get_block_number().await.unwrap_err();
        provider.get_block_number().await.unwrap();
        assert!(get_failover_index(&provider) == 1);
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_failover_revert() {
        let anvil = Anvil::new().block_time(1).spawn();
        let provider = L1ClientOptions {
            l1_polling_interval: Duration::from_secs(1),
            l1_consecutive_failure_tolerance: 1,
            l1_failover_revert: Duration::from_secs(2),
            ..Default::default()
        }
        .connect(vec![
            "http://notarealurl:1234".parse().unwrap(),
            anvil.endpoint_url(),
        ])
        .expect("Failed to create L1 client");

        // The first request fails and triggers a failover.
        provider.get_block_number().await.unwrap_err();
        assert_eq!(get_failover_index(&provider), 1);

        // The next request succeeds from the other provider.
        provider.get_block_number().await.unwrap();

        // Eventually we revert back to the primary and requests fail again.
        sleep(Duration::from_millis(2100)).await;
        provider.get_block_number().await.unwrap_err();
    }

    /// Proxies every JSON-RPC request to `target` except `eth_getBlockByNumber("finalized", ..)`,
    /// which always errors. Lets a test drive a real anvil chain while forcing the finalized RPC
    /// to fail.
    async fn serve_finalized_failing_proxy(target: Url) -> Url {
        let provider = ProviderBuilder::new().connect_http(target);
        let route = warp::post()
            .and(warp::body::json::<Value>())
            .then(move |req: Value| {
                let provider = provider.clone();
                async move {
                    let id = req["id"].clone();
                    let is_finalized_query = req["method"].as_str() == Some("eth_getBlockByNumber")
                        && req["params"].get(0).and_then(Value::as_str) == Some("finalized");
                    if is_finalized_query {
                        return warp::reply::json(&json!({
                            "jsonrpc": "2.0",
                            "id": id,
                            "error": {"code": -32000, "message": "finalized RPC unavailable"},
                        }));
                    }

                    let method = req["method"].as_str().unwrap_or_default().to_string();
                    let result: Value = provider
                        .raw_request(Cow::Owned(method), req["params"].clone())
                        .await
                        .expect("proxy request to anvil");
                    jsonrpc_result(id, result)
                }
            });
        test_server::serve_on_random_port(route).await
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_head_not_gated_by_finalized() {
        let anvil = Anvil::new().block_time(1).spawn();
        let proxy_url = serve_finalized_failing_proxy(anvil.endpoint_url()).await;

        let client = L1ClientOptions {
            l1_polling_interval: Duration::from_millis(200),
            ..Default::default()
        }
        .connect(vec![proxy_url])
        .expect("Failed to create L1 client");
        client.spawn_tasks().await;

        let initial = client.snapshot().await;
        let mut retry = 0;
        loop {
            assert!(
                retry < 30,
                "head did not advance despite a failing finalized RPC"
            );
            let snapshot = client.snapshot().await;
            if snapshot.head > initial.head {
                break;
            }
            sleep(Duration::from_millis(200)).await;
            retry += 1;
        }

        assert!(
            client.snapshot().await.finalized.is_none(),
            "finalized RPC always fails in this test"
        );
    }

    // Checks that the L1 client initialized the state on startup even
    // if the L1 is not currently mining blocks. It's useful for testing that we
    // don't require an L1 that is continuously mining blocks.
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_update_loop_initializes_l1_state() {
        // Use port 0 to let OS assign a port, avoiding race conditions
        let anvil = Arc::new(Anvil::new().port(0u16).spawn());
        let l1_client = new_l1_client(&anvil, true).await;

        for _try in 0..10 {
            let mut state = l1_client.state.lock().await;
            let has_snapshot = state.snapshot.finalized.is_some();
            let has_cache = state.finalized.get(&0).is_some();
            drop(state);
            if has_snapshot && has_cache {
                return;
            }
            sleep(Duration::from_millis(200)).await;
        }
        panic!("L1 state of L1Client not initialized");
    }
}