ethl 0.1.37

use std::collections::VecDeque;
use std::future::Future;
use std::sync::Once;
use std::time::Duration;

use alloy::{
    network::Ethereum,
    primitives::B256,
    providers::{DynProvider, Provider},
    rpc::types::{BlockNumberOrTag, Filter, Header, Log},
};
use async_stream::stream;
use futures_util::{StreamExt, TryStreamExt, pin_mut, stream::Stream};
use tokio::time::{Instant, sleep, sleep_until};
use tracing::{debug, warn};

use crate::rpc::cursor::{TipGate, TipWaitOutcome, next_cursor_after_batch, wait_for_tip};
use crate::rpc::error_tracking::{
    ErrorCategory, ProviderErrorTracker, find_active_provider, truncate_error_short,
};
use crate::rpc::{
    LogRange, RpcError,
    config::{CaptureTarget, ProviderSettings},
    progress::ScanProgress,
};

/// Maximum block range per WS request to prevent "max results" errors
const MAX_BLOCKS_PER_WS_REQUEST: u64 = 300;

#[derive(Debug, Clone)]
struct HeadCoalescer {
    block_cursor: Option<u64>,
    latest_seen_head: Option<u64>,
    flush_deadline: Option<Instant>,
    /// Coalesce rapid head updates into a bounded polling cadence.
    window: Duration,
    /// Optional scan-progress handle. `None` for every internal test/call site that doesn't
    /// opt in — `advance_cursor` and `record_tip` are no-ops without it.
    progress: Option<ScanProgress>,
}

impl HeadCoalescer {
    fn new(block_cursor: Option<u64>, window: Duration) -> Self {
        Self {
            block_cursor,
            latest_seen_head: None,
            flush_deadline: None,
            window,
            progress: None,
        }
    }

    fn with_progress(mut self, progress: Option<ScanProgress>) -> Self {
        self.progress = progress;
        self
    }

    /// Record a newly observed chain tip on the attached progress handle, if any. Called from
    /// `handle_empty_flush` when its `get_tip` closure resolves `Ok` — the only place in the
    /// live loop that fetches a tip, and only opportunistically (span-capped empty flushes).
    fn record_tip(&self, tip: u64) {
        if let Some(p) = &self.progress {
            p.record_observed_tip(tip);
        }
    }

    fn block_cursor(&self) -> Option<u64> {
        self.block_cursor
    }

    fn observe_head(&mut self, head_number: u64, now: Instant) -> bool {
        let from_block = self.block_cursor.unwrap_or(head_number);
        if head_number < from_block {
            return false;
        }

        self.latest_seen_head = Some(
            self.latest_seen_head
                .map_or(head_number, |latest| latest.max(head_number)),
        );

        if self.flush_deadline.is_none() {
            self.flush_deadline = Some(now + self.window);
        }

        true
    }

    fn pending_range(&self) -> Option<(u64, u64)> {
        let to_block = self.latest_seen_head?;
        let from_block = self.block_cursor.unwrap_or(to_block);
        (to_block >= from_block).then_some((from_block, to_block))
    }

    fn flush_deadline(&self) -> Option<Instant> {
        self.pending_range()?;
        self.flush_deadline
    }

    fn should_flush(&self, now: Instant) -> bool {
        let Some((from_block, to_block)) = self.pending_range() else {
            return false;
        };

        if to_block - from_block + 1 >= MAX_BLOCKS_PER_WS_REQUEST {
            return true;
        }

        self.flush_deadline.is_some_and(|deadline| now >= deadline)
    }

    fn advance_cursor(&mut self, next_block: u64) {
        self.block_cursor = Some(next_block);

        // Every call site passes the *next* block to fetch, never the last-scanned one
        // (`last_block + 1`, `watermark + 1`, `to_block + 1`) — translate back to the
        // highest block actually scanned before recording.
        if let Some(p) = &self.progress {
            p.record_scanned_to(next_block.saturating_sub(1));
        }

        if self
            .latest_seen_head
            .is_some_and(|latest| next_block > latest)
        {
            self.latest_seen_head = None;
            self.flush_deadline = None;
        }
    }

    /// Called after an empty, non-capped flush so the stream waits for the next
    /// head instead of busy-looping the same range. Clears `flush_deadline`
    /// only — the cursor and `latest_seen_head` are untouched, so the next
    /// `observe_head` re-arms the deadline while the span-cap arm of
    /// `should_flush` still fires deadline-free.
    fn defer_flush(&mut self) {
        self.flush_deadline = None;
    }
}

/// Outcome of feeding one head into the live reorg linkage check.
#[derive(Debug, Clone, PartialEq, Eq)]
enum LinkOutcome {
    /// The head extends (or bootstraps) the canonical chain — no reorg.
    Linked,
    /// A reorg whose common ancestor was resolved within the ring.
    Reorg { common_ancestor: u64 },
    /// The head's parent is not the recorded canonical block and the ancestor
    /// could not be resolved from the ring alone (the provider skipped the
    /// intermediate new-chain heads). The caller must fetch the parent block by
    /// hash and walk the new chain back — bounded RPC, reorg branch only — feeding
    /// the fetched `(number, hash)` back through `matches`.
    Detached {
        parent_hash: B256,
        child_number: u64,
    },
    /// Divergence older than the ring floor (`> finality`): catastrophe.
    BeyondFinality,
}

/// A finality-deep ring of canonical `(number, hash)` taken from the head stream,
/// plus a `parent_hash` linkage check. Pure and synchronous — the network is thin
/// glue around it, mirroring `HeadCoalescer`. `depth == 0` disables it: `observe`
/// is then inert and returns `Linked` for every head (zero overhead, the L2 /
/// snapshot default).
#[derive(Debug, Clone)]
struct HeadLinkage {
    ring: VecDeque<(u64, B256)>,
    depth: u64,
}

impl HeadLinkage {
    fn new(depth: u64) -> Self {
        Self {
            ring: VecDeque::new(),
            depth,
        }
    }

    fn enabled(&self) -> bool {
        self.depth > 0
    }

    /// Lowest block number still recorded in the ring (the resolution floor).
    fn floor(&self) -> Option<u64> {
        self.ring.front().map(|&(number, _)| number)
    }

    /// Recorded canonical hash at `number`, if it is within the ring.
    fn hash_at(&self, number: u64) -> Option<B256> {
        let &(front, _) = self.ring.front()?;
        let offset = number.checked_sub(front)? as usize;
        self.ring.get(offset).map(|&(_, hash)| hash)
    }

    /// True if the ring records `hash` as canonical at `number`.
    fn matches(&self, number: u64, hash: B256) -> bool {
        self.hash_at(number) == Some(hash)
    }

    /// Append `(number, hash)` as the new tip, evicting the oldest beyond `depth`.
    fn push(&mut self, number: u64, hash: B256) {
        self.ring.push_back((number, hash));
        while self.ring.len() as u64 > self.depth {
            self.ring.pop_front();
        }
    }

    /// The most recently recorded `(number, hash)` entry (the ring's canonical tip).
    ///
    /// Returns `None` only when the ring is completely empty (before the first
    /// `observe`). When `linkage.enabled()` and a flush fires, this is always `Some`
    /// because `observe` bootstraps or re-anchors the ring at the newest head before
    /// `should_flush` can return true.
    fn latest(&self) -> Option<(u64, B256)> {
        self.ring.back().copied()
    }

    /// Re-anchor the ring at `ancestor` (drop everything above it), then record
    /// the new-chain head `(number, hash)`.
    fn reanchor(&mut self, ancestor: u64, number: u64, hash: B256) {
        while self.ring.back().is_some_and(|&(n, _)| n > ancestor) {
            self.ring.pop_back();
        }
        self.push(number, hash);
    }

    /// Observe one head's `(number, hash, parent_hash)` against the recorded
    /// chain. See `LinkOutcome`. Disabled (`depth == 0`) ⇒ always `Linked`.
    fn observe(&mut self, number: u64, hash: B256, parent_hash: B256) -> LinkOutcome {
        if !self.enabled() {
            return LinkOutcome::Linked;
        }

        let Some(&(tip_number, tip_hash)) = self.ring.back() else {
            // Bootstrap: the first head seeds the ring.
            self.push(number, hash);
            return LinkOutcome::Linked;
        };

        // The common case: a clean forward extension of the tip.
        if number == tip_number + 1 && parent_hash == tip_hash {
            self.push(number, hash);
            return LinkOutcome::Linked;
        }

        // A forward gap (heads skipped ahead — e.g. across a reconnect): the ring cannot
        // assert linkage over the missing heads, so route it through the same bounded
        // backward walk `Detached` already uses (`resolve_reorg_ancestor`) instead of
        // blindly trusting the gap. A walk that reconnects cleanly at the prior tip still
        // reports `Reorg { common_ancestor: <old tip> }` — a zero-depth "reorg" — and the
        // caller's existing terminate-and-resume path already handles that correctly (the
        // consumer's rollback recovery resumes from common_ancestor + 1, exactly where the
        // stream would have continued anyway had nothing gone wrong).
        if number > tip_number + 1 {
            return LinkOutcome::Detached {
                parent_hash,
                child_number: number,
            };
        }

        // `number <= tip_number + 1` and the parent did not extend the tip:
        // a reorg, a duplicate, or a detachment that needs an RPC walk.
        let Some(parent_number) = number.checked_sub(1) else {
            return LinkOutcome::BeyondFinality;
        };
        match self.hash_at(parent_number) {
            // Parent links to a known canonical block.
            Some(canonical_parent) if canonical_parent == parent_hash => {
                // Already recorded exactly → duplicate, no-op.
                if self.hash_at(number) == Some(hash) {
                    return LinkOutcome::Linked;
                }
                // Block `number` was replaced on a chain sharing canonical
                // `parent_number` → that is the common ancestor.
                self.reanchor(parent_number, number, hash);
                LinkOutcome::Reorg {
                    common_ancestor: parent_number,
                }
            }
            // Parent is within the ring but differs → diverges deeper; the caller
            // must walk the new chain back to find the ancestor.
            Some(_) => LinkOutcome::Detached {
                parent_hash,
                child_number: number,
            },
            // Parent is below the ring floor → beyond finality.
            None => LinkOutcome::BeyondFinality,
        }
    }
}

/// Walk backward from `tip_hash` at `tip_number` down to `from_block`, collecting
/// canonical `(number, hash)` pairs in descending order (tip → from_block).
///
/// Used by the backfill boot path to resolve hashes for the reorg-exposed window
/// before handing off to `stream_heads_with_logs`. Does not touch any ring.
///
/// Error semantics: null block → `ReorgBeyondFinality`; transport fault → `ConnectionError`.
pub(crate) async fn collect_window_hashes(
    provider: &DynProvider<Ethereum>,
    tip_hash: B256,
    tip_number: u64,
    from_block: u64,
) -> Result<Vec<(u64, B256)>, RpcError> {
    if tip_number < from_block {
        return Ok(Vec::new());
    }
    let mut collected: Vec<(u64, B256)> = Vec::new();
    let mut current_hash = tip_hash;
    let mut current_number = tip_number;

    loop {
        collected.push((current_number, current_hash));
        if current_number == from_block {
            break;
        }
        let block = match provider.get_block_by_hash(current_hash).await {
            Ok(Some(block)) => block,
            Ok(None) => return Err(RpcError::ReorgBeyondFinality),
            Err(e) => {
                return Err(RpcError::ConnectionError(format!(
                    "window hash walk: block fetch failed: {}",
                    truncate_error_short(e.to_string(), 100)
                )));
            }
        };
        // A by-hash query returning a block with the wrong number is a provider
        // integrity violation — fail closed rather than walking past it.
        if block.header.number != current_number {
            return Err(RpcError::ReorgBeyondFinality);
        }
        let parent_hash = block.header.parent_hash;
        let Some(parent_number) = current_number.checked_sub(1) else {
            return Err(RpcError::ReorgBeyondFinality);
        };
        if parent_number < from_block {
            break;
        }
        current_hash = parent_hash;
        current_number = parent_number;
    }

    Ok(collected)
}

/// Fetch logs for a slice of canonical `(number, hash)` pairs using
/// `at_block_hash` filters — one request per block, ascending.
///
/// `at_block_hash` pins the query to the exact canonical block identified by
/// `hash`, so an orphaned block at that number can never contaminate the result.
/// The caller passes `pairs` in ascending order (from_block → tip).
pub(crate) async fn fetch_logs_by_hash_range(
    providers: &ProviderSettings,
    pairs: &[(u64, B256)],
    base_filter: Option<&Filter>,
) -> Result<Vec<LogRange>, RpcError> {
    let mut ranges: Vec<LogRange> = Vec::with_capacity(pairs.len());

    for &(number, hash) in pairs {
        let query = base_filter
            .cloned()
            .unwrap_or_else(Filter::new)
            .at_block_hash(hash);

        let (_, logs) =
            get_logs_with_http_fallback_retry(providers, &query, 0, number, number, None).await?;

        ranges.push((number, number, logs));
    }

    Ok(ranges)
}

/// Wraps `fut`'s aggregate work in a wall-clock bound so a pathologically slow
/// flush can't block the live head-consuming loop long enough to overflow
/// alloy's internal WS head-buffer (Q-BSC-2 Route B). `max_secs == 0` disables
/// the bound (old, unbounded behavior) — mirrors `tip_wait_max_total_secs`'s
/// own `0`-disables convention.
async fn bound_flush_time<F, T>(max_secs: u64, fut: F) -> Result<T, RpcError>
where
    F: Future<Output = Result<T, RpcError>>,
{
    if max_secs == 0 {
        return fut.await;
    }
    match tokio::time::timeout(Duration::from_secs(max_secs), fut).await {
        Ok(result) => result,
        Err(_) => Err(RpcError::ConnectionError(format!(
            "flush exceeded its {max_secs}s aggregate timeout — provider likely stalled \
             mid-flush; switching provider"
        ))),
    }
}

/// Flush one pending window range from the coalescer by fetching all logs by
/// canonical block hash.
///
/// When the ring is missing hashes for some blocks (boot gap only — a forward-gap
/// reconnect now always terminates the stream via `Detached` before reaching a flush —
/// see `HeadLinkage::observe`), walks backward from `linkage.latest()` to collect
/// canonical `(number, hash)` pairs and uses them directly. Never falls back to a
/// by-number fetch — fail-closed is the invariant. Only called when `linkage.enabled()`.
async fn flush_window_range(
    linkage: &mut HeadLinkage,
    coalescer: &mut HeadCoalescer,
    providers: &ProviderSettings,
    filter: Option<&Filter>,
    provider_index: usize,
    reused: Option<&[DynProvider<Ethereum>]>,
) -> Result<Vec<LogRange>, RpcError> {
    let Some((from_block, pending_to_block)) = coalescer.pending_range() else {
        return Ok(vec![]);
    };
    // Fail closed rather than let the modulo-indexed `http_settings`/`connect_http`
    // lookups below panic on an empty configured set — the same guard `last_block`
    // applies before it does its own `% num_providers` indexing.
    if providers.http_providers.is_empty() {
        return Err(RpcError::ConnectionError(
            "No HTTP providers configured for window flush".to_string(),
        ));
    }
    let to_block = std::cmp::min(from_block + MAX_BLOCKS_PER_WS_REQUEST - 1, pending_to_block);

    // Build the (number, hash) pairs for [from_block, to_block].
    // If any hash is absent from the ring, self-heal: walk backward from the
    // latest known entry to fill the gap. `latest()` is Some whenever the flush
    // fires — `observe` bootstraps or re-anchors at the newest head before
    // `should_flush` can return true.
    let pairs: Vec<(u64, B256)> = if (from_block..=to_block).all(|n| linkage.hash_at(n).is_some()) {
        (from_block..=to_block)
            .map(|n| (n, linkage.hash_at(n).expect("just verified")))
            .collect()
    } else {
        let Some((top_n, top_hash)) = linkage.latest() else {
            // Ring is completely empty — no anchor for the walk.
            return Err(RpcError::ReorgBeyondFinality);
        };
        let walk_provider = resolve_http_provider(providers, reused, provider_index);
        // `collect_window_hashes` walks strictly sequentially: each step needs the
        // previous block's parent hash, which is only known after fetching that
        // block, so there is nothing here for `fetch_concurrently` to parallelize.
        // That is inherent to a backward hash-chain walk, not an oversight — if a
        // very deep backlog ever makes this walk the bottleneck, fixing it needs a
        // different algorithm (e.g. walking by block number instead of by hash, if
        // that is even valid here), which is out of scope for this fix.
        // `top_n < from_block` means the ring's latest known block is already
        // behind the window we need — `collect_window_hashes` below no-ops via
        // its own `tip_number < from_block` guard, so nothing gets walked. Say
        // so plainly rather than reporting a "walking back N blocks" count that
        // would otherwise floor at 1 (via `saturating_sub`) and misdescribe a
        // no-op as a one-block walk in the wrong direction.
        warn!(
            "flush window [{}, {}] missing ring entries — self-heal walking back up \
             to {} blocks from ring anchor {} to {} ({})",
            from_block,
            to_block,
            top_n.saturating_sub(from_block) + 1,
            top_n,
            from_block,
            if top_n < from_block {
                "ring anchor already trails the window's start — the walk \
                 immediately no-ops, nothing is fetched"
            } else {
                "uses the same client-reuse mechanism as the main fetch path when \
                 available, but the walk itself is sequential; not covered by the \
                 concurrent fetch path"
            }
        );
        let mut pairs_desc =
            collect_window_hashes(&walk_provider, top_hash, top_n, from_block).await?;
        // Reverse to ascending (from_block → top_n), then restrict to [from_block, to_block].
        pairs_desc.reverse();
        pairs_desc
            .into_iter()
            .filter(|&(n, _)| n <= to_block)
            .collect()
    };

    // Fetch the entire batch BEFORE advancing the cursor, concurrently (bounded by the
    // provider's own max_concurrency). On any block's failure, the whole batch returns
    // Err with the cursor UNTOUCHED so the whole window range retries on the next
    // provider — never advance past a block whose logs were not delivered to the
    // consumer (fail-closed; no silent gap). Re-fetch on retry is idempotent.
    let last_pending = pairs.last().copied();
    // Two accepted tradeoffs, not oversights:
    // - Concurrent dispatch means a provider outage can now produce up to
    //   `bound * num_http_providers` failed fallback requests before this flush
    //   aborts, versus `num_http_providers` before concurrency — trading a bigger
    //   worst-case fan-out for a bounded wall-clock duration (fail-fast).
    // - `provider_index` here ranges over WS-endpoint indices (it's the live
    //   loop's WS provider position, reused as the HTTP-fallback start index —
    //   pre-existing conflation in this module), not HTTP-provider indices. It
    //   was already used this way for fallback start; it is now also load-bearing
    //   for which HTTP provider's `max_concurrency` gates this flush's dispatch.
    let bound = providers.http_settings(provider_index).max_concurrency as usize;
    let fetch = move |(n, hash): (u64, B256)| async move {
        let query = filter
            .cloned()
            .unwrap_or_else(Filter::new)
            .at_block_hash(hash);
        let (_, logs) =
            get_logs_with_http_fallback_retry(providers, &query, provider_index, n, n, reused)
                .await?;
        Ok::<LogRange, RpcError>((n, n, logs))
    };
    let ranges = sort_ranges_ascending(fetch_concurrently(pairs, bound, fetch).await?);

    if let Some((last, _)) = last_pending {
        coalescer.advance_cursor(last + 1);
    }
    Ok(ranges)
}

/// Runs `fetch` over `items` concurrently, bounded by `bound` in-flight futures at
/// once (clamped to at least 1 — see the `max_concurrency` doc on `HttpRpcSettings`),
/// failing fast on the first error: `try_buffer_unordered` drops the remaining
/// in-flight futures the instant any one resolves `Err`, rather than
/// `try_buffered`'s in-order wait (which can be slower than a sequential loop in
/// the worst case) or a collect-then-check shape that invokes every future before
/// checking any result. Results are returned in completion order, not input order —
/// callers that need input order (e.g. ascending block number) must sort separately.
async fn fetch_concurrently<In, Out, F, Fut>(
    items: Vec<In>,
    bound: usize,
    fetch: F,
) -> Result<Vec<Out>, RpcError>
where
    F: Fn(In) -> Fut,
    Fut: Future<Output = Result<Out, RpcError>>,
{
    futures_util::stream::iter(items.into_iter().map(fetch).map(Ok::<Fut, RpcError>))
        .try_buffer_unordered(bound.max(1))
        .try_collect()
        .await
}

/// Restores ascending block-number order after `fetch_concurrently`'s unordered
/// dispatch. `flush_window_range`'s external contract yields blocks in order; this
/// is the only place that reimposes it — cheap, and only paid on the already-rare
/// "flush was behind" path concurrency exists to fix.
fn sort_ranges_ascending(mut ranges: Vec<LogRange>) -> Vec<LogRange> {
    ranges.sort_by_key(|&(from, _, _)| from);
    ranges
}

/// Walk the new chain backward by hash until a block re-links to the recorded
/// canonical ring, returning the resolved `RpcError::Reorg { common_ancestor }`.
///
/// Invoked only on `LinkOutcome::Detached`, when the provider delivered a new-tip
/// head without re-emitting the intermediate new-chain blocks. Bounded to `depth`
/// `get_block_by_hash` fetches (the ring's own span); exhausting them, or walking
/// below the ring floor, means the divergence is older than finality →
/// `ReorgBeyondFinality`. The pure ring is untouched; only this glue does I/O.
async fn resolve_reorg_ancestor(
    provider: &DynProvider<Ethereum>,
    linkage: &HeadLinkage,
    detached_parent: B256,
    child_number: u64,
    depth: u64,
) -> RpcError {
    // `parent_hash` is the new chain's hash at height `parent_number`.
    let mut parent_hash = detached_parent;
    let Some(mut parent_number) = child_number.checked_sub(1) else {
        return RpcError::ReorgBeyondFinality;
    };
    let floor = linkage.floor().unwrap_or(u64::MAX);

    for _ in 0..depth {
        if linkage.matches(parent_number, parent_hash) {
            return RpcError::Reorg {
                common_ancestor: parent_number,
            };
        }
        if parent_number <= floor {
            // Cannot descend below the ring floor without leaving the finality window.
            return RpcError::ReorgBeyondFinality;
        }
        let block = match provider.get_block_by_hash(parent_hash).await {
            Ok(Some(block)) => block,
            Ok(None) => return RpcError::ReorgBeyondFinality,
            Err(e) => {
                return RpcError::ConnectionError(format!(
                    "reorg ancestor walk: block-by-hash fetch failed: {}",
                    truncate_error_short(e.to_string(), 100)
                ));
            }
        };
        parent_hash = block.header.parent_hash;
        let Some(next) = parent_number.checked_sub(1) else {
            return RpcError::ReorgBeyondFinality;
        };
        parent_number = next;
    }
    RpcError::ReorgBeyondFinality
}

fn fallback_http_index(start_index: usize, attempt: usize, num_http_providers: usize) -> usize {
    (start_index + attempt) % num_http_providers
}

fn fallback_http_exhausted_error(
    from_block: u64,
    to_block: u64,
    attempted_hosts: &[String],
    last_err: &str,
) -> RpcError {
    RpcError::LogFetchError(format!(
        "HTTP fallback exhausted for blocks {}-{} after trying providers [{}]: {}",
        from_block,
        to_block,
        attempted_hosts.join(", "),
        last_err
    ))
}

/// Resolves the HTTP client for `idx`: `Some(set)` reuses an already-built
/// client from `set` (modulo-indexed, the same out-of-range safety
/// `connect_http` has); `None` falls through to `providers.connect_http(idx)`
/// — today's per-call rebuild, unchanged. The seam that lets §5's tests prove
/// reuse vs. rebuild without touching a real network.
pub(crate) fn resolve_http_provider(
    providers: &ProviderSettings,
    reused: Option<&[DynProvider<Ethereum>]>,
    idx: usize,
) -> DynProvider<Ethereum> {
    match reused {
        Some(set) => set[idx % set.len()].clone(),
        None => providers.connect_http(idx),
    }
}

async fn get_logs_with_http_fallback_retry(
    providers: &ProviderSettings,
    query_filter: &Filter,
    start_index: usize,
    from_block: u64,
    to_block: u64,
    reused: Option<&[DynProvider<Ethereum>]>,
) -> Result<(usize, Vec<alloy::rpc::types::Log>), RpcError> {
    let num_http_providers = providers.http_providers.len();
    let mut attempted_hosts: Vec<String> = Vec::with_capacity(num_http_providers);
    let mut last_err = String::new();

    for attempt in 0..num_http_providers {
        let provider_idx = fallback_http_index(start_index, attempt, num_http_providers);
        let provider_host = providers.http_settings(provider_idx).host();
        attempted_hosts.push(provider_host.clone());

        let http_provider = resolve_http_provider(providers, reused, provider_idx);
        match http_provider.get_logs(query_filter).await {
            Ok(logs) => return Ok((provider_idx, logs)),
            Err(err) => {
                last_err = err.to_string();
                warn!(
                    "HTTP fallback {} failed for blocks {}-{}: {}",
                    provider_host,
                    from_block,
                    to_block,
                    truncate_error_short(&last_err, 100)
                );
            }
        }
    }

    Err(fallback_http_exhausted_error(
        from_block,
        to_block,
        &attempted_hosts,
        &last_err,
    ))
}

/// How far a non-empty `getLogs` response actually proves the queried range
/// was covered, decided by `yield_bounds`.
enum YieldBounds {
    /// Watermark == to_block: today's behavior verbatim.
    Full,
    /// Watermark B < to_block: yield (from, B), advance B+1, defer the tail.
    Partial(u64),
    /// No returned log carries a block number: the response proves nothing —
    /// treat it exactly as an empty response (do NOT yield these logs).
    Unattributable,
}

/// Decides how much of `[.., to_block]` a non-empty log batch proves covered.
///
/// The watermark is the highest `block_number` carried by any returned log —
/// a log at block B proves the provider's index reached >= B, which also
/// vouches for the emptiness of every logless block <= B in the range. Only
/// the tail strictly above the watermark is unproven.
fn yield_bounds(logs: &[Log], to_block: u64) -> YieldBounds {
    let Some(watermark) = logs.iter().filter_map(|log| log.block_number).max() else {
        return YieldBounds::Unattributable;
    };

    // A range query cannot legally return a block above what it asked for; a
    // watermark past to_block is a provider integrity violation. Clamp rather
    // than advance the cursor past the requested range.
    if watermark >= to_block {
        YieldBounds::Full
    } else {
        YieldBounds::Partial(watermark)
    }
}

/// Outcome of the empty-batch guard for one flushed chunk, decided by
/// `handle_empty_flush`. Callers map each variant onto their existing control
/// flow (mirrored across the WS and HTTP-fallback sites in
/// `stream_heads_with_logs`).
enum EmptyFlushOutcome {
    /// Non-capped: `defer_flush()` applied. Caller breaks the `'flush` loop.
    Deferred,
    /// Capped, tip >= to_block: caller yields the verified-empty range,
    /// advances the cursor, resets the gate.
    VerifiedEmpty,
    /// Capped, `wait_for_tip` caught up: caller re-queries (continue `'flush`).
    CaughtUp,
    /// Capped, terminal stall: caller yields `ProviderStalled` and returns.
    Stalled {
        to_block: u64,
        reported_tip: u64,
        stalled_cycles: u32,
    },
    /// Tip re-poll failed inside `wait_for_tip`: caller logs and switches provider.
    RepollFailed(String),
    /// Initial tip fetch failed: caller records the error and switches provider.
    TipFetchFailed(String),
}

/// Decides how to handle an empty log batch for one flushed chunk.
///
/// A non-capped chunk defers rather than confirming: a rare false-empty
/// (provider serving `[]` past its own tip) self-heals when the next head's
/// flush re-covers the same span from the unmoved cursor, so confirming every
/// quiet window buys nothing. A span-capped chunk (`to_block - from_block + 1
/// == MAX_BLOCKS_PER_WS_REQUEST`) always confirms against the provider's
/// reported tip — this bounds how far the cursor can advance unverified and is
/// what still catches a frozen or lying backend (the Base-outage failure
/// class), now at the span cap instead of per window.
async fn handle_empty_flush<F, Fut, E>(
    coalescer: &mut HeadCoalescer,
    gate: &mut TipGate,
    providers: &ProviderSettings,
    host: &str,
    to_block: u64,
    span_capped: bool,
    get_tip: F,
) -> EmptyFlushOutcome
where
    F: Fn() -> Fut,
    Fut: Future<Output = Result<u64, E>>,
    E: std::fmt::Display,
{
    if !span_capped {
        coalescer.defer_flush();
        return EmptyFlushOutcome::Deferred;
    }

    let reported_tip = match get_tip().await {
        Ok(tip) => tip,
        Err(e) => return EmptyFlushOutcome::TipFetchFailed(e.to_string()),
    };
    coalescer.record_tip(reported_tip);

    if next_cursor_after_batch(to_block, true, reported_tip).is_err() {
        match wait_for_tip(to_block, reported_tip, gate, providers, host, get_tip).await {
            TipWaitOutcome::CaughtUp => EmptyFlushOutcome::CaughtUp,
            TipWaitOutcome::Terminal {
                to_block,
                reported_tip,
                stalled_cycles,
            } => EmptyFlushOutcome::Stalled {
                to_block,
                reported_tip,
                stalled_cycles,
            },
            TipWaitOutcome::RepollFailed(err) => EmptyFlushOutcome::RepollFailed(err),
        }
    } else {
        EmptyFlushOutcome::VerifiedEmpty
    }
}

async fn stream_heads_provider(
    provider: &DynProvider<Ethereum>,
    provider_host: &str,
) -> impl Stream<Item = Header> {
    stream! {
        let head_stream = provider.subscribe_blocks().await;

        match head_stream {
            Ok(heads) => {
                debug!("WebSocket connected, streaming headers...");
                let mut s = heads.into_stream();
                while let Some(header) = s.next().await {
                    yield header;
                }
            }
            Err(e) => {
                warn!(
                    "Failed to subscribe to blocks on {}: {} - retrying",
                    provider_host,
                    truncate_error_short(e, 100)
                );
            }
        }
    }
}

/// Yield contract: empty ranges are not yielded per window. A quiet stretch
/// yields nothing until either a non-empty flush (which covers the whole span
/// back to the unmoved cursor) or a span-cap tip-verification (which yields
/// the verified-empty `MAX_BLOCKS_PER_WS_REQUEST`-block chunk). The
/// RangeTooWide HTTP-fallback path is the exception — it still confirms every
/// empty response, per-window. Consumers' cursor persistence coarsens
/// accordingly: expect progress in bursts, not one update per coalesce window.
///
/// A yielded range's `block_end` is the highest block PROVEN by the response
/// — a returned log's block number — which may be below the queried
/// `to_block` when the serving backend's index lagged behind the requested
/// range. The unproven remainder above `block_end` is re-covered like a
/// deferred empty flush: it stays in front of the cursor until the next
/// head's flush picks it up.
/// See [`stream_heads_with_logs`]; additionally records scan progress into `progress` when
/// `Some`, so a consumer can read [`ScanProgress::scanned_to`] / [`ScanProgress::observed_tip`]
/// on demand. `progress: None` is byte-identical to `stream_heads_with_logs`.
pub async fn stream_heads_with_logs_with_progress(
    providers: &ProviderSettings,
    filter: Option<&Filter>,
    progress: Option<&ScanProgress>,
) -> impl Stream<Item = Result<LogRange, RpcError>> {
    let num_providers = providers.wss_endpoints.len();
    let progress = progress.cloned();

    stream! {
        // Initialize per-provider error tracking (same pattern as backfill)
        let mut provider_trackers: Vec<ProviderErrorTracker> = (0..num_providers)
            .map(|i| {
                let host = providers.wss_endpoints[i]
                    .host_str()
                    .unwrap_or("unknown")
                    .to_string();
                ProviderErrorTracker::new(host)
            })
            .collect();
        // One persistent HTTP client per configured provider, built once for this
        // stream invocation and reused across every flush and provider-fallback
        // attempt in it — not rebuilt per provider switch or per flush.
        let reused_http_clients = providers.build_reused_http_clients();

        let mut provider_index = 0;
        let mut coalescer = HeadCoalescer::new(
            filter.and_then(|f| f.get_from_block()),
            Duration::from_millis(providers.live_head_coalesce_window_ms),
        )
        .with_progress(progress);
        // Live-tip reorg detector. `reorg_detect_depth == 0` (every non-mainnet-live
        // consumer) leaves `observe` inert, so this is a no-op on those paths.
        let mut linkage = HeadLinkage::new(providers.reorg_detect_depth);
        let mut gate = TipGate::new();

        loop {
            // Find the next non-suspended provider (same pattern as backfill)
            let active_provider = find_active_provider(&mut provider_trackers, provider_index);
            let Some(idx) = active_provider else {
                yield Err(RpcError::AllProvidersSuspended(format!(
                    "All {} WebSocket providers suspended. Unable to continue streaming.",
                    num_providers
                )));
                return;
            };
            provider_index = idx;

            let provider = match providers.connect_ws(provider_index).await {
                Ok(p) => {
                    provider_trackers[provider_index].record_success();
                    p
                }
                Err(e) => {
                    let suspended = provider_trackers[provider_index].record_error();
                    let backoff = provider_trackers[provider_index].backoff_duration();

                    if suspended {
                        warn!(
                            "WebSocket provider {} suspended after connection failure: {}",
                            provider_trackers[provider_index].identifier(),
                            truncate_error_short(&e, 100)
                        );
                    } else {
                        warn!(
                            "Failed to connect to WebSocket {}: {} - retrying in {:?}",
                            provider_trackers[provider_index].identifier(),
                            truncate_error_short(&e, 100),
                            backoff
                        );
                        sleep(backoff).await;
                    }
                    provider_index = (provider_index + 1) % num_providers;
                    continue;
                }
            };

            // A new provider session must not inherit the previous session's stall
            // state: a stale last_tip makes the new provider's first observation count
            // as a stall instead of a seed (audit F1, early ProviderStalled).
            gate.reset();

            let provider_host = provider_trackers[provider_index].identifier().to_string();
            let s = stream_heads_provider(&provider, &provider_host).await;
            pin_mut!(s);
            let mut should_switch_provider = false;

            loop {
                if coalescer.should_flush(Instant::now()) {
                    // Flush from the cursor through the latest seen head in bounded chunks.
                    'flush: while let Some((from_block, pending_to_block)) = coalescer.pending_range() {
                        let to_block =
                            std::cmp::min(from_block + MAX_BLOCKS_PER_WS_REQUEST - 1, pending_to_block);

                        // When reorg detection is active fetch logs pinned to the canonical
                        // hash so a same-number orphaned block on a stale provider cannot
                        // contaminate the result. When the ring is missing a hash (boot gap
                        // only — a forward-gap reconnect now always terminates the stream via
                        // `Detached` before reaching a flush), `flush_window_range` self-heals
                        // by walking backward from the latest ring entry — never spinning,
                        // never falling back to a by-number fetch inside the window.
                        if linkage.enabled() {
                            match bound_flush_time(
                                providers.flush_max_total_secs,
                                flush_window_range(
                                    &mut linkage,
                                    &mut coalescer,
                                    providers,
                                    filter,
                                    provider_index,
                                    Some(&reused_http_clients),
                                ),
                            )
                            .await
                            {
                                Ok(ranges) => {
                                    for range in ranges {
                                        provider_trackers[provider_index].record_success();
                                        yield Ok(range);
                                    }
                                }
                                Err(e) => {
                                    warn!(
                                        "by-hash window flush failed: {}",
                                        truncate_error_short(&e, 100)
                                    );
                                    should_switch_provider = true;
                                    break 'flush;
                                }
                            }
                            continue;
                        }

                        let query_filter = filter
                            .cloned()
                            .unwrap_or_else(Filter::new)
                            .from_block(from_block)
                            .to_block(to_block);

                        let logs = provider.get_logs(&query_filter).await;

                        match logs {
                            Ok(logs) => {
                                provider_trackers[provider_index].record_success();

                                // A non-empty response only proves coverage up to the highest
                                // block number any returned log carries (the watermark); Partial
                                // means the tail above it is unproven and must not be crossed.
                                let watermark_bounds =
                                    (!logs.is_empty()).then(|| yield_bounds(&logs, to_block));

                                if let Some(YieldBounds::Partial(watermark)) = watermark_bounds {
                                    yield Ok((from_block, watermark, logs));
                                    coalescer.advance_cursor(watermark + 1);
                                    gate.reset();
                                    // Unproven tail [watermark+1, to_block] stays in front of the
                                    // cursor; the next head's flush re-covers it, same as an
                                    // empty flush's deferred-empty pattern.
                                    coalescer.defer_flush();
                                    break 'flush;
                                }

                                // Empty batches are ambiguous: they look identical whether the
                                // range is genuinely event-free or the provider silently returned
                                // [] for blocks past its actual tip. A non-capped chunk defers
                                // instead of confirming (see `handle_empty_flush`); a span-capped
                                // chunk still verifies against the WS provider's reported tip
                                // before advancing the cursor. Unattributable responses (logs
                                // present but none carry a block number) prove nothing and enter
                                // this same guard; the received logs are dropped, not yielded.
                                let treat_as_empty =
                                    !matches!(watermark_bounds, Some(YieldBounds::Full));
                                if treat_as_empty {
                                    let span_capped =
                                        to_block - from_block + 1 == MAX_BLOCKS_PER_WS_REQUEST;
                                    match handle_empty_flush(
                                        &mut coalescer,
                                        &mut gate,
                                        providers,
                                        &provider_host,
                                        to_block,
                                        span_capped,
                                        || provider.get_block_number(),
                                    )
                                    .await
                                    {
                                        EmptyFlushOutcome::Deferred => break 'flush,
                                        EmptyFlushOutcome::VerifiedEmpty => {}
                                        EmptyFlushOutcome::CaughtUp => continue 'flush,
                                        EmptyFlushOutcome::Stalled {
                                            to_block,
                                            reported_tip,
                                            stalled_cycles,
                                        } => {
                                            yield Err(RpcError::ProviderStalled {
                                                to_block,
                                                reported_tip,
                                                stalled_cycles,
                                            });
                                            return;
                                        }
                                        EmptyFlushOutcome::RepollFailed(err) => {
                                            let suspended = provider_trackers
                                                [provider_index]
                                                .record_error();
                                            if suspended {
                                                warn!(
                                                    "WebSocket provider {} suspended after tip re-poll failed during tip-wait for blocks {}-{}: {}",
                                                    provider_trackers[provider_index].identifier(),
                                                    from_block,
                                                    to_block,
                                                    truncate_error_short(&err, 100),
                                                );
                                            } else {
                                                let backoff = provider_trackers
                                                    [provider_index]
                                                    .backoff_duration();
                                                warn!(
                                                    "Tip re-poll failed during wait for blocks {}-{} from {}: {} - backing off {:?}",
                                                    from_block,
                                                    to_block,
                                                    provider_trackers[provider_index].identifier(),
                                                    truncate_error_short(&err, 100),
                                                    backoff
                                                );
                                                sleep(backoff).await;
                                            }
                                            should_switch_provider = true;
                                            break 'flush;
                                        }
                                        EmptyFlushOutcome::TipFetchFailed(err_str) => {
                                            let suspended =
                                                provider_trackers[provider_index].record_error();
                                            if suspended {
                                                warn!(
                                                    "WebSocket provider {} suspended after tip fetch failed during empty-batch guard for blocks {}-{}: {}",
                                                    provider_trackers[provider_index].identifier(),
                                                    from_block, to_block,
                                                    truncate_error_short(&err_str, 100)
                                                );
                                            } else {
                                                let backoff = provider_trackers[provider_index].backoff_duration();
                                                warn!(
                                                    "Tip fetch failed during empty-batch guard for blocks {}-{} from {}: {} - backing off {:?}",
                                                    from_block, to_block,
                                                    provider_trackers[provider_index].identifier(),
                                                    truncate_error_short(&err_str, 100),
                                                    backoff
                                                );
                                                sleep(backoff).await;
                                            }
                                            should_switch_provider = true;
                                            break 'flush;
                                        }
                                    }
                                }

                                let logs = if treat_as_empty { Vec::new() } else { logs };
                                yield Ok((from_block, to_block, logs));
                                coalescer.advance_cursor(to_block + 1);
                                gate.reset();
                            }
                            Err(e) => {
                                let error_msg = e.to_string();
                                let error_category = ErrorCategory::from_error_msg(&error_msg);

                                match error_category {
                                    ErrorCategory::RangeTooWide => {
                                        // Fall back to HTTP provider for this batch
                                        warn!(
                                            "Range too wide for blocks {}-{}, falling back to HTTP RPC: {}",
                                            from_block, to_block, truncate_error_short(&error_msg, 100)
                                        );

                                        match get_logs_with_http_fallback_retry(
                                            providers,
                                            &query_filter,
                                            provider_index,
                                            from_block,
                                            to_block,
                                            Some(&reused_http_clients),
                                        )
                                        .await
                                        {
                                            Ok((http_provider_idx, logs)) => {
                                                provider_trackers[provider_index].record_success();

                                                // Same watermark bound as the WS site: a non-empty
                                                // response only proves coverage up to the highest
                                                // block number a returned log carries. Unlike the
                                                // empty case below, Partial DOES defer here — it
                                                // already advanced the cursor to watermark+1 before
                                                // deferring, so each retry makes forward progress
                                                // and the WS-error -> fallback double-call objection
                                                // (see the comment below) does not apply.
                                                let watermark_bounds = (!logs.is_empty())
                                                    .then(|| yield_bounds(&logs, to_block));

                                                if let Some(YieldBounds::Partial(watermark)) =
                                                    watermark_bounds
                                                {
                                                    yield Ok((from_block, watermark, logs));
                                                    coalescer.advance_cursor(watermark + 1);
                                                    gate.reset();
                                                    coalescer.defer_flush();
                                                    break 'flush;
                                                }

                                                // Empty fallback responses are ambiguous — a lagging
                                                // HTTP provider can legitimately return [] for blocks
                                                // past its own tip. Verify against the HTTP provider
                                                // that served this response, not the WS provider.
                                                // Lazy does NOT apply here: a RangeTooWide error
                                                // followed by an empty fallback is already
                                                // contradictory (WS claimed too many results, HTTP
                                                // returned zero), and deferring would re-trigger the
                                                // WS-error -> fallback double-call on every head
                                                // until the span cap — the opposite of the saving.
                                                // Unattributable responses (logs present but none
                                                // carry a block number) prove nothing and enter this
                                                // same eager guard; the received logs are dropped.
                                                let treat_as_empty = !matches!(
                                                    watermark_bounds,
                                                    Some(YieldBounds::Full)
                                                );
                                                if treat_as_empty {
                                                    let http_provider = providers.connect_http(http_provider_idx);
                                                    let http_host = providers.http_settings(http_provider_idx).host();
                                                    match handle_empty_flush(
                                                        &mut coalescer,
                                                        &mut gate,
                                                        providers,
                                                        &http_host,
                                                        to_block,
                                                        true,
                                                        || http_provider.get_block_number(),
                                                    )
                                                    .await
                                                    {
                                                        EmptyFlushOutcome::Deferred => {
                                                            unreachable!(
                                                                "span_capped: true never defers"
                                                            )
                                                        }
                                                        EmptyFlushOutcome::VerifiedEmpty => {}
                                                        EmptyFlushOutcome::CaughtUp => {
                                                            continue 'flush
                                                        }
                                                        EmptyFlushOutcome::Stalled {
                                                            to_block,
                                                            reported_tip,
                                                            stalled_cycles,
                                                        } => {
                                                            yield Err(
                                                                RpcError::ProviderStalled {
                                                                    to_block,
                                                                    reported_tip,
                                                                    stalled_cycles,
                                                                },
                                                            );
                                                            return;
                                                        }
                                                        EmptyFlushOutcome::RepollFailed(err) => {
                                                            warn!(
                                                                "HTTP fallback {} tip re-poll failed during tip-wait for blocks {}-{}: {}",
                                                                http_host,
                                                                from_block,
                                                                to_block,
                                                                truncate_error_short(&err, 100),
                                                            );
                                                            should_switch_provider = true;
                                                            break 'flush;
                                                        }
                                                        EmptyFlushOutcome::TipFetchFailed(err) => {
                                                            warn!(
                                                                "HTTP fallback {} tip fetch failed during empty-batch guard for blocks {}-{}: {}",
                                                                http_host, from_block, to_block,
                                                                truncate_error_short(&err, 100)
                                                            );
                                                            should_switch_provider = true;
                                                            break 'flush;
                                                        }
                                                    }
                                                }

                                                let logs = if treat_as_empty {
                                                    Vec::new()
                                                } else {
                                                    logs
                                                };
                                                yield Ok((from_block, to_block, logs));
                                                coalescer.advance_cursor(to_block + 1);
                                                gate.reset();
                                            }
                                            Err(http_err) => {
                                                warn!(
                                                    "HTTP fallback retry exhausted for blocks {}-{}: {}",
                                                    from_block,
                                                    to_block,
                                                    truncate_error_short(&http_err, 100)
                                                );
                                                should_switch_provider = true;
                                                break;
                                            }
                                        }
                                    }
                                    ErrorCategory::RateLimit => {
                                        // Rate limit: apply backoff but stay on same provider
                                        provider_trackers[provider_index].record_error();
                                        let backoff = provider_trackers[provider_index].backoff_duration();
                                        warn!(
                                            "Rate limit hit on WS provider {} for blocks {}-{}: {} - backing off {:?}",
                                            provider_trackers[provider_index].identifier(),
                                            from_block, to_block,
                                            truncate_error_short(&error_msg, 100),
                                            backoff
                                        );
                                        sleep(backoff).await;
                                        // Retry same block range after backoff
                                        continue;
                                    }
                                    ErrorCategory::Connection | ErrorCategory::Other => {
                                        // Connection error or unknown: record error and switch provider
                                        let suspended = provider_trackers[provider_index].record_error();

                                        if suspended {
                                            warn!(
                                                "WebSocket provider {} suspended after error on blocks {}-{}: {}",
                                                provider_trackers[provider_index].identifier(),
                                                from_block, to_block,
                                                truncate_error_short(&error_msg, 100)
                                            );
                                        } else {
                                            let backoff = provider_trackers[provider_index].backoff_duration();
                                            warn!(
                                                "Error fetching logs for blocks {}-{} from {}: {} - backing off {:?}",
                                                from_block, to_block,
                                                provider_trackers[provider_index].identifier(),
                                                truncate_error_short(&error_msg, 100),
                                                backoff
                                            );
                                            sleep(backoff).await;
                                        }
                                        should_switch_provider = true;
                                        break;
                                    }
                                }
                            }
                        }
                    }

                    if should_switch_provider {
                        break;
                    }

                    continue;
                }

                let maybe_header = if let Some(deadline) = coalescer.flush_deadline() {
                    tokio::select! {
                        maybe_header = s.next() => maybe_header,
                        _ = sleep_until(deadline) => {
                            continue;
                        }
                    }
                } else {
                    s.next().await
                };

                match maybe_header {
                    Some(header) => {
                        let now = Instant::now();
                        let head_number = header.number;
                        if !coalescer.observe_head(head_number, now) {
                            let from_block = coalescer.block_cursor().unwrap_or(head_number);
                            warn!(
                                "Skipping header {} as it's before the cursor {}",
                                head_number, from_block
                            );
                        }

                        // Reorg detection runs alongside the coalescer's number-based
                        // flush — a parallel `(number, hash, parent_hash)` linkage check.
                        // Disabled (depth 0) ⇒ `observe` is inert and this is a no-op.
                        // On a reorg we emit and end the stream; the consumer decides
                        // resume-from-ancestor (P3 rollback) vs terminate (P2).
                        match linkage.observe(head_number, header.hash, header.parent_hash) {
                            LinkOutcome::Linked => {}
                            LinkOutcome::Reorg { common_ancestor } => {
                                warn!(
                                    "reorg at head {}: common ancestor {} — ending stream for consumer recovery",
                                    head_number, common_ancestor
                                );
                                yield Err(RpcError::Reorg { common_ancestor });
                                return;
                            }
                            LinkOutcome::Detached {
                                parent_hash,
                                child_number,
                            } => {
                                let err = resolve_reorg_ancestor(
                                    &provider,
                                    &linkage,
                                    parent_hash,
                                    child_number,
                                    providers.reorg_detect_depth,
                                )
                                .await;
                                match &err {
                                    RpcError::Reorg { common_ancestor } => warn!(
                                        "reorg at head {} (walked new chain): common ancestor {} — ending stream",
                                        head_number, common_ancestor
                                    ),
                                    _ => warn!(
                                        "reorg at head {} beyond the detector ring — ending stream",
                                        head_number
                                    ),
                                }
                                yield Err(err);
                                return;
                            }
                            LinkOutcome::BeyondFinality => {
                                warn!(
                                    "reorg beyond finality at head {}: no common ancestor within the detector ring — ending stream",
                                    head_number
                                );
                                yield Err(RpcError::ReorgBeyondFinality);
                                return;
                            }
                        }
                    }
                    None => break,
                }
            }

            // Stream ended (connection lost) - record as error and switch provider
            if !should_switch_provider {
                // Stream ended naturally without explicit error
                provider_trackers[provider_index].record_error();
                let backoff = provider_trackers[provider_index].backoff_duration();
                warn!(
                    "WS connection to {} lost or stream ended - reconnecting in {:?}",
                    provider_trackers[provider_index].identifier(),
                    backoff
                );
                sleep(backoff).await;
            }

            // Move to next provider for round-robin (same pattern as backfill)
            provider_index = (provider_index + 1) % num_providers;
        }
    }
}

pub async fn stream_heads_with_logs(
    providers: &ProviderSettings,
    filter: Option<&Filter>,
) -> impl Stream<Item = Result<LogRange, RpcError>> {
    stream_heads_with_logs_with_progress(providers, filter, None).await
}

pub async fn last_block(providers: &ProviderSettings) -> Result<u64, RpcError> {
    let num_providers = providers.http_providers.len();
    if num_providers == 0 {
        return Err(RpcError::ConnectionError(
            "No HTTP providers configured for latest block fetch".to_string(),
        ));
    }

    let mut index: usize = 0;
    loop {
        // Round-robin provider selection so latest-block fetch can fail over.
        let provider_index = index % num_providers;
        let provider = providers.connect_http(provider_index);
        match provider.get_block_number().await {
            Ok(block_number) => return Ok(block_number),
            Err(e) => {
                let provider_host = providers.http_settings(provider_index).host();
                if index >= 10 {
                    return Err(RpcError::ConnectionError(format!(
                        "Failed to fetch latest block number after {} attempts from HTTP provider {}: {}",
                        index, provider_host, e
                    )));
                }
                index += 1;
                warn!(
                    "Failed to fetch latest block number from HTTP provider {} (index {}) - retrying",
                    provider_host, provider_index
                );
                let progressive_sleep = (5 * index).min(50) as u64;
                sleep(Duration::from_secs(progressive_sleep)).await;
            }
        }
    }
}

/// One-shot gate for the capture-policy announcement, one `Once` per policy
/// variant so `Latest`, `Lag`, and `Finalized` are silenced independently.
struct CapturePolicyNotice {
    latest: Once,
    lag: Once,
    finalized: Once,
}

impl CapturePolicyNotice {
    const fn new() -> Self {
        Self {
            latest: Once::new(),
            lag: Once::new(),
            finalized: Once::new(),
        }
    }

    /// The message to log the FIRST time this policy VARIANT is resolved in this process;
    /// `None` on every later call for that variant. Gating is per-variant, not per-value: a
    /// process resolving `Lag(64)` then later `Lag(200)` announces only the first — there is no
    /// real-process case where the configured target changes value mid-run, so this is
    /// deliberately the simpler of the two designs. `Latest`, `Lag`, and `Finalized` each have
    /// their own gate, so a process that resolves more than one variant (e.g. a config change
    /// between restarts within one long-lived process) announces each variant once.
    fn first_message(&self, target: CaptureTarget, block: u64) -> Option<String> {
        let gate = match target {
            CaptureTarget::Latest => &self.latest,
            CaptureTarget::Lag(_) => &self.lag,
            CaptureTarget::Finalized => &self.finalized,
        };
        let mut won = false;
        gate.call_once(|| won = true);
        if !won {
            return None;
        }
        Some(match target {
            CaptureTarget::Latest => format!(
                "ethl capture target is `Latest` (block {}): capturing to the raw chain \
                 tip with NO reorg protection — a reorg can rewrite recently captured \
                 blocks. Reorg-prone chains should opt into `Lag`/`Finalized`.",
                block,
            ),
            CaptureTarget::Lag(n) => format!(
                "ethl DEEP LAG ACTIVE (reorg-safety): capturing to tip - {} = block {}.\n\
                 This build AVOIDS reorgs by lagging; events within {} blocks of tip are DELAYED.",
                n, block, n,
            ),
            CaptureTarget::Finalized => format!(
                "ethl DEEP LAG ACTIVE (reorg-safety): capturing to `finalized` block {}.\n\
                 (~2 epochs ≈ 64 blocks ≈ 13 min behind chain tip).\n\
                 This build AVOIDS reorgs by lagging; events within ~64 blocks of tip are DELAYED.",
                block
            ),
        })
    }
}

static CAPTURE_POLICY_NOTICE: CapturePolicyNotice = CapturePolicyNotice::new();

/// Resolve the block number to use as the capture target based on `settings.capture_target`.
///
/// - `Latest` → current chain tip (existing behavior).
/// - `Lag(n)` → tip saturating_sub n.
/// - `Finalized` → the `finalized` tagged block; returns `Err` if the provider returns none.
///
/// Emits a one-time `warn!` per capture policy variant, for every variant
/// including `Latest`, announcing the capture decision once per process.
pub async fn resolve_capture_target(settings: &ProviderSettings) -> Result<u64, RpcError> {
    match settings.capture_target {
        CaptureTarget::Latest => {
            let tip = last_block(settings).await?;
            if let Some(message) = CAPTURE_POLICY_NOTICE.first_message(settings.capture_target, tip)
            {
                warn!("{message}");
            }
            Ok(tip)
        }
        CaptureTarget::Lag(n) => {
            let tip = last_block(settings).await?;
            let target_block = tip.saturating_sub(n);
            if let Some(message) =
                CAPTURE_POLICY_NOTICE.first_message(settings.capture_target, target_block)
            {
                warn!("{message}");
            }
            Ok(target_block)
        }
        CaptureTarget::Finalized => {
            let provider = settings.connect_http(0);
            let result = provider
                .get_block_by_number(BlockNumberOrTag::Finalized)
                .await
                .map_err(|e| {
                    RpcError::ConnectionError(format!("finalized block fetch failed: {e}"))
                })?;
            match result {
                Some(block) => {
                    let finalized = block.header.number;
                    if let Some(message) =
                        CAPTURE_POLICY_NOTICE.first_message(settings.capture_target, finalized)
                    {
                        warn!("{message}");
                    }
                    Ok(finalized)
                }
                None => Err(RpcError::Other(
                    "Provider returned no finalized block; cannot determine safe capture target. \
                     Ensure the provider supports the `finalized` tag (mainnet only)."
                        .to_string(),
                )),
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        CapturePolicyNotice, EmptyFlushOutcome, HeadCoalescer, HeadLinkage, LinkOutcome,
        MAX_BLOCKS_PER_WS_REQUEST, RpcError, TipGate, YieldBounds, bound_flush_time,
        collect_window_hashes, fallback_http_exhausted_error, fallback_http_index,
        fetch_concurrently, fetch_logs_by_hash_range, flush_window_range, handle_empty_flush,
        resolve_http_provider, resolve_reorg_ancestor, sort_ranges_ascending, yield_bounds,
    };
    use crate::rpc::config::{CaptureTarget, HttpRpcSettings, ProviderSettings};
    use alloy::primitives::B256;
    use alloy::providers::{Provider, ProviderBuilder};
    use alloy::rpc::types::Log;
    use alloy::transports::mock::Asserter;
    use std::time::Duration;
    use tokio::time::Instant;

    /// Coalesce window for tests that previously anchored timing assertions on
    /// the module's `LIVE_HEAD_COALESCE_WINDOW` const, before it became the
    /// per-instance `HeadCoalescer::window` field.
    const WINDOW: Duration = Duration::from_millis(500);

    /// Distinct synthetic block hash; `byte` keeps old/new chains separable.
    fn hb(byte: u8) -> B256 {
        B256::repeat_byte(byte)
    }

    #[test]
    fn fallback_http_index_wraps_round_robin() {
        assert_eq!(fallback_http_index(0, 0, 3), 0);
        assert_eq!(fallback_http_index(0, 1, 3), 1);
        assert_eq!(fallback_http_index(0, 2, 3), 2);
        assert_eq!(fallback_http_index(0, 3, 3), 0);
        assert_eq!(fallback_http_index(2, 1, 3), 0);
    }

    #[test]
    fn fallback_http_exhausted_error_includes_hosts() {
        let err = fallback_http_exhausted_error(
            100,
            120,
            &["rpc.ankr.com".to_string(), "mainnet.infura.io".to_string()],
            "timeout",
        )
        .to_string();
        assert!(err.contains("rpc.ankr.com"));
        assert!(err.contains("mainnet.infura.io"));
        assert!(err.contains("timeout"));
    }

    #[test]
    fn head_coalescer_ignores_stale_heads_before_cursor() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(42), WINDOW);

        assert!(!coalescer.observe_head(41, now));
        assert_eq!(coalescer.pending_range(), None);
        assert_eq!(coalescer.flush_deadline(), None);
    }

    #[test]
    fn head_coalescer_keeps_first_flush_deadline_while_heads_accumulate() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);

        assert!(coalescer.observe_head(100, now));
        let deadline = coalescer.flush_deadline().unwrap();

        assert!(coalescer.observe_head(105, now + Duration::from_millis(50)));
        assert_eq!(coalescer.pending_range(), Some((100, 105)));
        assert_eq!(coalescer.flush_deadline(), Some(deadline));
        assert!(!coalescer.should_flush(now + WINDOW - Duration::from_millis(1)));
        assert!(coalescer.should_flush(now + WINDOW));
    }

    #[test]
    fn head_coalescer_flushes_immediately_when_pending_range_reaches_chunk_limit() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);

        assert!(coalescer.observe_head(MAX_BLOCKS_PER_WS_REQUEST, now));
        assert_eq!(
            coalescer.pending_range(),
            Some((1, MAX_BLOCKS_PER_WS_REQUEST))
        );
        assert!(coalescer.should_flush(now));
    }

    #[test]
    fn head_coalescer_preserves_pending_range_across_partial_cursor_advances() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(10), WINDOW);

        assert!(coalescer.observe_head(25, now));
        coalescer.advance_cursor(20);
        assert_eq!(coalescer.pending_range(), Some((20, 25)));
        assert!(coalescer.flush_deadline().is_some());

        coalescer.advance_cursor(26);
        assert_eq!(coalescer.pending_range(), None);
        assert_eq!(coalescer.flush_deadline(), None);
        assert_eq!(coalescer.block_cursor(), Some(26));
    }

    /// G.2.2 §2.2/§4.3 — `advance_cursor` is the single translation point shared by all five
    /// live-path call sites (`:429`, `:832`, `:938`, `:980`, `:1069` at `EXPECT_TIP`): every
    /// caller passes the *next* block to fetch, never the last-scanned one, so `advance_cursor`
    /// must record `next_block - 1`. Exercised directly here (no live-provider scaffolding
    /// needed — `HeadCoalescer` and `ScanProgress` are both plain synchronous types); the five
    /// call sites themselves are not reachable by the existing mock harness (see
    /// `test_event_streamer_head`'s `#[ignore]`), which is the disclosed residual in the REPORT.
    #[test]
    fn advance_cursor_records_scanned_to_as_next_block_minus_one() {
        let progress = crate::rpc::progress::ScanProgress::default();
        let mut coalescer =
            HeadCoalescer::new(Some(10), WINDOW).with_progress(Some(progress.clone()));

        assert_eq!(progress.scanned_to(), None);

        coalescer.advance_cursor(101);
        assert_eq!(progress.scanned_to(), Some(100));

        // Monotonic: an out-of-order lower advance must not move progress backwards.
        coalescer.advance_cursor(51);
        assert_eq!(progress.scanned_to(), Some(100));
    }

    /// `record_tip` is the shared point behind `handle_empty_flush`'s `get_tip()` resolution.
    #[test]
    fn record_tip_populates_progress_when_attached_and_is_a_no_op_otherwise() {
        let progress = crate::rpc::progress::ScanProgress::default();
        let with_progress = HeadCoalescer::new(None, WINDOW).with_progress(Some(progress.clone()));
        with_progress.record_tip(500);
        assert_eq!(progress.observed_tip(), Some(500));

        // No progress attached: must not panic, and there is nothing to observe.
        let without_progress = HeadCoalescer::new(None, WINDOW);
        without_progress.record_tip(500);
    }

    #[test]
    fn head_coalescer_handles_mainnet_slow_blocks() {
        // Ethereum mainnet: ~12 s/block. Each head must flush in a single 500 ms coalesce window;
        // the chunk-limit guard (MAX_BLOCKS_PER_WS_REQUEST) must never trigger for live cadence.
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(1_000_000), WINDOW);

        // First mainnet head observed.
        assert!(coalescer.observe_head(1_000_000, now));
        assert_eq!(coalescer.pending_range(), Some((1_000_000, 1_000_000)));

        // Must not flush before the coalesce window expires.
        assert!(!coalescer.should_flush(now + Duration::from_millis(499)));
        // Must flush once the window expires.
        assert!(coalescer.should_flush(now + WINDOW));

        // Advance cursor; simulate the next block arriving ~12 s later.
        coalescer.advance_cursor(1_000_001);
        assert_eq!(coalescer.pending_range(), None);

        let t12s = now + Duration::from_secs(12);
        assert!(coalescer.observe_head(1_000_001, t12s));
        assert_eq!(coalescer.pending_range(), Some((1_000_001, 1_000_001)));

        // One block in range: chunk limit (300) must not fire.
        assert!(!coalescer.should_flush(t12s));
        assert!(coalescer.should_flush(t12s + WINDOW));
    }

    /// The coalesce window is read from the constructed instance, not a fixed
    /// constant: two coalescers built with different windows must each flush at
    /// their own deadline, not a shared hard-coded value.
    #[test]
    fn head_coalescer_honors_its_own_window() {
        let now = Instant::now();
        let short_window = Duration::from_millis(100);
        let long_window = Duration::from_secs(5);

        let mut short = HeadCoalescer::new(Some(1), short_window);
        assert!(short.observe_head(1, now));
        assert!(!short.should_flush(now + short_window - Duration::from_millis(1)));
        assert!(short.should_flush(now + short_window));

        let mut long = HeadCoalescer::new(Some(1), long_window);
        assert!(long.observe_head(1, now));
        assert!(!long.should_flush(now + long_window - Duration::from_millis(1)));
        assert!(long.should_flush(now + long_window));
    }

    #[test]
    fn head_coalescer_defer_flush_clears_deadline_only() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        assert!(coalescer.observe_head(300, now));
        assert_eq!(coalescer.pending_range(), Some((1, 300)));
        assert!(coalescer.flush_deadline().is_some());

        coalescer.defer_flush();
        assert_eq!(
            coalescer.flush_deadline(),
            None,
            "defer_flush must clear the deadline"
        );
        assert_eq!(
            coalescer.pending_range(),
            Some((1, 300)),
            "defer_flush must not touch the pending range"
        );

        // A 300-wide span still should_flush()es with no deadline armed — the
        // span-cap arm of should_flush is deadline-free.
        assert!(coalescer.should_flush(now));
    }

    /// After a `Partial(b)` yield advances the cursor to `b + 1` and defers,
    /// the deferred-empty pattern applies verbatim: the deadline clears, the
    /// pending range starts at the unproven tail, and the next head re-arms.
    #[test]
    fn coalescer_partial_defer_reserves_unproven_tail_and_rearms() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);
        coalescer.observe_head(105, now);
        assert_eq!(coalescer.pending_range(), Some((100, 105)));

        // Simulate a Partial(102) yield: cursor advances only through the
        // proven watermark, then the unproven tail defers.
        coalescer.advance_cursor(103);
        coalescer.defer_flush();

        assert_eq!(
            coalescer.flush_deadline(),
            None,
            "defer_flush must clear the deadline after a Partial yield"
        );
        assert_eq!(
            coalescer.pending_range(),
            Some((103, 105)),
            "pending range must start at watermark+1, the unproven tail"
        );

        // The next head's flush re-arms the deadline over the still-unproven tail.
        assert!(coalescer.observe_head(105, now + Duration::from_millis(1)));
        assert!(coalescer.flush_deadline().is_some());
    }

    // ==================== log-watermark bound (yield_bounds) ====================

    /// Builds a `Log` fixture via the JSON shape already used for the
    /// `fetch_logs_by_hash_range` test (`"blockNumber": format!("0x{:x}", n)`),
    /// nulling `blockNumber` for the `None` case (the field is required by
    /// `Log`'s deserializer, so it cannot simply be omitted).
    fn log_fixture(block_number: Option<u64>) -> Log {
        let value = serde_json::json!({
            "address": "0x0000000000000000000000000000000000000001",
            "topics": [],
            "data": "0x",
            "blockHash": hb(0).to_string(),
            "blockNumber": block_number.map(|n| format!("0x{:x}", n)),
            "transactionHash": hb(0).to_string(),
            "transactionIndex": "0x0",
            "logIndex": "0x0",
            "removed": false
        });
        serde_json::from_value(value).expect("valid Log fixture")
    }

    #[test]
    fn yield_bounds_full_when_tail_has_logs() {
        let logs = vec![log_fixture(Some(101)), log_fixture(Some(102))];
        assert!(matches!(yield_bounds(&logs, 102), YieldBounds::Full));
    }

    #[test]
    fn yield_bounds_partial_stops_at_watermark() {
        let logs = vec![log_fixture(Some(101))];
        assert!(matches!(
            yield_bounds(&logs, 102),
            YieldBounds::Partial(101)
        ));
    }

    /// Logs at 101 and 104 with `to = 105`: blocks 102-103 are vouched by the
    /// 104 log (a log at block B proves the provider's index reached >= B);
    /// only 105, above the watermark, is unproven.
    #[test]
    fn yield_bounds_interior_gaps_are_proven() {
        let logs = vec![log_fixture(Some(101)), log_fixture(Some(104))];
        assert!(matches!(
            yield_bounds(&logs, 105),
            YieldBounds::Partial(104)
        ));
    }

    #[test]
    fn yield_bounds_mixed_none_uses_some_only() {
        let logs = vec![log_fixture(Some(101)), log_fixture(None)];
        assert!(matches!(
            yield_bounds(&logs, 102),
            YieldBounds::Partial(101)
        ));
    }

    #[test]
    fn yield_bounds_all_none_is_unattributable() {
        let logs = vec![log_fixture(None), log_fixture(None)];
        assert!(matches!(
            yield_bounds(&logs, 999),
            YieldBounds::Unattributable
        ));
    }

    /// A log claiming a block above `to_block` is a provider integrity
    /// violation (a range query cannot legally return higher blocks); clamp
    /// to `Full` rather than advance the cursor past the requested range.
    #[test]
    fn yield_bounds_watermark_above_range_clamps() {
        let logs = vec![log_fixture(Some(103))];
        assert!(matches!(yield_bounds(&logs, 102), YieldBounds::Full));
    }

    // ==================== lazy empty-verification (handle_empty_flush) ====================

    /// Non-capped empty flush defers without calling the tip at all. A reverted
    /// per-window confirm would call `get_block_number` and, against an asserter
    /// with zero queued responses, panic on the exhausted queue instead of
    /// returning `Deferred`.
    #[tokio::test]
    async fn lazy_empty_non_capped_defers_without_tip_call() {
        let asserter = Asserter::new(); // zero queued responses
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);
        coalescer.observe_head(105, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((100, 105)));

        let mut gate = TipGate::new();
        let outcome = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host",
            105,
            false,
            || http_provider.get_block_number(),
        )
        .await;

        assert!(
            matches!(outcome, EmptyFlushOutcome::Deferred),
            "expected Deferred"
        );
        assert_eq!(coalescer.flush_deadline(), None);
        assert_eq!(
            coalescer.pending_range(),
            Some((100, 105)),
            "cursor must stay unmoved by a deferred flush"
        );
    }

    /// After a deferred flush, the next `observe_head` re-arms the deadline and
    /// the pending range still starts at the ORIGINAL (unmoved) cursor — the
    /// self-heal shape: the next flush re-covers the deferred blocks.
    #[test]
    fn lazy_deferred_flush_rearms_and_spans_from_unmoved_cursor() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);
        coalescer.observe_head(105, now);
        assert_eq!(coalescer.pending_range(), Some((100, 105)));

        coalescer.defer_flush();
        assert_eq!(coalescer.flush_deadline(), None);

        coalescer.observe_head(110, now + Duration::from_millis(10));
        assert_eq!(
            coalescer.pending_range(),
            Some((100, 110)),
            "deferred span must extend from the original unmoved cursor"
        );
        assert!(
            coalescer.flush_deadline().is_some(),
            "the next observe_head must re-arm the deadline"
        );
    }

    /// Span-capped chunk, tip already at or past `to_block`: verified empty on
    /// the first tip check, no `wait_for_tip` episode.
    #[tokio::test]
    async fn span_cap_empty_tip_current_is_verified_empty() {
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!("0x64")); // get_block_number → 100
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        coalescer.observe_head(100, Instant::now());
        let mut gate = TipGate::new();

        let outcome = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host",
            100,
            true,
            || http_provider.get_block_number(),
        )
        .await;

        assert!(
            matches!(outcome, EmptyFlushOutcome::VerifiedEmpty),
            "expected VerifiedEmpty"
        );
    }

    /// Span-capped chunk, tip behind `to_block`: enters `wait_for_tip`, which
    /// catches up on the first re-poll. `from_mock`'s `tip_wait_backoff_secs: 0`
    /// keeps the wait sleepless.
    #[tokio::test]
    async fn span_cap_empty_tip_behind_enters_tip_wait() {
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!("0x63")); // initial tip → 99 (< to_block)
        asserter.push_success(&serde_json::json!("0x64")); // re-poll → 100 (caught up)
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        coalescer.observe_head(100, Instant::now());
        let mut gate = TipGate::new();

        let outcome = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host",
            100,
            true,
            || http_provider.get_block_number(),
        )
        .await;

        assert!(
            matches!(outcome, EmptyFlushOutcome::CaughtUp),
            "expected CaughtUp"
        );
    }

    /// Span-capped chunk, tip flat across every poll: `wait_for_tip` terminates
    /// via the stall gate. This plus the previous test is the Base-outage
    /// invariant — reverting the cap guard (making the span-capped arm defer
    /// too) fails both.
    #[tokio::test]
    async fn span_cap_empty_flat_tip_stalls() {
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!("0x5a")); // initial tip → 90
        asserter.push_success(&serde_json::json!("0x5a")); // re-poll cycle 1 → 90 (flat)
        asserter.push_success(&serde_json::json!("0x5a")); // re-poll cycle 2 → 90 (flat, hits cap)
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        )
        .with_tip_wait_max_stall_cycles(2);
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        coalescer.observe_head(100, Instant::now());
        let mut gate = TipGate::new();

        let outcome = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host",
            100,
            true,
            || http_provider.get_block_number(),
        )
        .await;

        assert!(
            matches!(
                outcome,
                EmptyFlushOutcome::Stalled {
                    to_block: 100,
                    reported_tip: 90,
                    stalled_cycles: 2,
                }
            ),
            "expected Stalled(stalled_cycles=2), got non-matching outcome"
        );
    }

    /// F1 (0.1.29 core soundness audit): a `TipGate` is shared across
    /// provider sessions within one stream. Phase 1 reproduces the hazard —
    /// session A leaves the gate dirty (`RepollFailed`, the provider-switch
    /// path), and session B, run against the SAME never-reset gate, has its
    /// first observation miscounted as a regression instead of a fresh seed,
    /// so it hits `Stall(5)` on only 5 total tip polls instead of the 6 a
    /// freshly-seeded gate would need — one cycle early. Phase 2 shows the
    /// fix (`gate.reset()` between sessions, exactly what the connect-site
    /// change does): session B now correctly consumes 6 polls before
    /// `Stall(5)`. Each phase queues EXACTLY the responses its session B
    /// consumes, so an extra (correct-behavior) poll fails on an unqueued
    /// response rather than silently passing with the wrong arithmetic.
    #[tokio::test]
    async fn handle_empty_flush_stale_gate_stalls_one_cycle_early_without_reset() {
        // ---- Phase 1: no reset between sessions -- reproduce the miscount.
        let asserter = Asserter::new();
        // Session A: tip 990 (below to_block 999), then a transport error on
        // the wait-for-tip re-poll -- the RepollFailed provider-switch path
        // that leaves the gate dirty (audit F1 step 4-5).
        asserter.push_success(&serde_json::json!("0x3de")); // 990
        asserter.push_failure_msg("transport blip");
        // Session B, same gate, NOT reset: flat tip 988 (below A's stale
        // last_tip=990), so B's first observation counts as a regression
        // instead of a seed. 5 total get_tip calls (1 initial + 4 re-polls)
        // reach Stall(5) here -- one call short of a fresh gate's 6.
        for _ in 0..5 {
            asserter.push_success(&serde_json::json!("0x3dc")); // 988
        }
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        coalescer.observe_head(999, Instant::now());
        let mut gate = TipGate::new();

        let session_a = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host-a",
            999,
            true,
            || http_provider.get_block_number(),
        )
        .await;
        assert!(
            matches!(session_a, EmptyFlushOutcome::RepollFailed(_)),
            "expected RepollFailed for session A"
        );

        // Provider switch happens here in production, WITHOUT gate.reset() --
        // this is the bug this fix corrects.
        let session_b = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host-b",
            999,
            true,
            || http_provider.get_block_number(),
        )
        .await;
        assert!(
            matches!(
                session_b,
                EmptyFlushOutcome::Stalled {
                    to_block: 999,
                    reported_tip: 988,
                    stalled_cycles: 5,
                }
            ),
            "expected Stalled(stalled_cycles=5) one cycle early from the stale gate"
        );

        // ---- Phase 2: gate.reset() between sessions -- the fixed contract.
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!("0x3de")); // 990
        asserter.push_failure_msg("transport blip");
        // Session B now gets a fresh seed observation, so it needs one MORE
        // flat poll (6 total: 1 initial + 5 re-polls) before Stall(5).
        for _ in 0..6 {
            asserter.push_success(&serde_json::json!("0x3dc")); // 988
        }
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        coalescer.observe_head(999, Instant::now());
        let mut gate = TipGate::new();

        let session_a = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host-a",
            999,
            true,
            || http_provider.get_block_number(),
        )
        .await;
        assert!(
            matches!(session_a, EmptyFlushOutcome::RepollFailed(_)),
            "expected RepollFailed for session A"
        );

        // The connect-site fix: reset the gate before the new session.
        gate.reset();

        let session_b = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host-b",
            999,
            true,
            || http_provider.get_block_number(),
        )
        .await;
        assert!(
            matches!(
                session_b,
                EmptyFlushOutcome::Stalled {
                    to_block: 999,
                    reported_tip: 988,
                    stalled_cycles: 5,
                }
            ),
            "expected Stalled(stalled_cycles=5) reached only after the full \
             6-poll budget once the gate is reset"
        );
    }

    /// Synthetic block with explicit `number` and `parentHash`.
    ///
    /// Use this for tests that exercise `collect_window_hashes`, which now
    /// cross-checks `block.header.number == current_number` (fail-closed).
    fn block_with_parent_at(number: u64, parent: B256) -> serde_json::Value {
        let zero = hb(0).to_string();
        serde_json::json!({
            "hash":             zero,
            "parentHash":       parent.to_string(),
            "sha3Uncles":       zero,
            "miner":            "0x0000000000000000000000000000000000000000",
            "stateRoot":        zero,
            "transactionsRoot": zero,
            "receiptsRoot":     zero,
            "logsBloom":        format!("0x{}", "0".repeat(512)),
            "difficulty":       "0x0",
            "number":           format!("0x{:x}", number),
            "gasLimit":         "0x0",
            "gasUsed":          "0x0",
            "timestamp":        "0x0",
            "extraData":        "0x",
            "mixHash":          zero,
            "nonce":            "0x0000000000000000",
            "transactions":     [],
            "uncles":           []
        })
    }

    /// Synthetic block whose only meaningful field is `parentHash`.
    ///
    /// For `resolve_reorg_ancestor` tests, the block number is not checked, so
    /// passing 0 is fine. Use `block_with_parent_at` when `collect_window_hashes`
    /// is under test (it cross-checks the block number).
    fn block_with_parent(parent: B256) -> serde_json::Value {
        block_with_parent_at(0, parent)
    }

    /// Seed a linkage with a clean canonical chain `[lo, hi]` (hash = `hb(n)`).
    fn seed_chain(depth: u64, lo: u64, hi: u64) -> HeadLinkage {
        let mut linkage = HeadLinkage::new(depth);
        for n in lo..=hi {
            let parent = if n == lo {
                hb((lo - 1) as u8)
            } else {
                hb((n - 1) as u8)
            };
            assert_eq!(linkage.observe(n, hb(n as u8), parent), LinkOutcome::Linked);
        }
        linkage
    }

    #[test]
    fn head_linkage_disabled_is_inert() {
        let mut linkage = HeadLinkage::new(0);
        // A blatant parent mismatch still returns Linked when depth == 0.
        assert_eq!(linkage.observe(100, hb(1), hb(9)), LinkOutcome::Linked);
        assert_eq!(linkage.observe(101, hb(2), hb(7)), LinkOutcome::Linked);
        assert_eq!(linkage.floor(), None);
    }

    #[test]
    fn head_linkage_clean_chain_advances_and_bounds_ring() {
        let linkage = seed_chain(3, 100, 103);
        // Ring is bounded at depth = 3; the oldest entry (100) is evicted.
        assert_eq!(linkage.floor(), Some(101));
        assert_eq!(linkage.hash_at(100), None);
        assert_eq!(linkage.hash_at(103), Some(hb(103)));
    }

    #[test]
    fn head_linkage_one_block_reorg_reports_parent_as_ancestor() {
        let mut linkage = seed_chain(8, 100, 103);
        // 103' replaces 103, still building on canonical 102 → ancestor 102.
        assert_eq!(
            linkage.observe(103, hb(203), hb(102)),
            LinkOutcome::Reorg {
                common_ancestor: 102
            }
        );
        // Ring re-anchored: 103 now records the new hash, clean extension resumes.
        assert_eq!(linkage.hash_at(103), Some(hb(203)));
        assert_eq!(linkage.observe(104, hb(204), hb(203)), LinkOutcome::Linked);
    }

    #[test]
    fn head_linkage_deep_reorg_via_reemit_reports_deeper_ancestor() {
        let mut linkage = seed_chain(8, 100, 105);
        // Provider re-emits the new chain from 101 (parent = canonical 100): a
        // five-block-deep reorg, resolved in-ring at common ancestor 100.
        assert_eq!(
            linkage.observe(101, hb(201), hb(100)),
            LinkOutcome::Reorg {
                common_ancestor: 100
            }
        );
    }

    #[test]
    fn head_linkage_skipped_intermediate_heads_detach() {
        let mut linkage = seed_chain(8, 100, 103);
        // Provider jumps to new tip 104' whose parent 103' (hb(203)) we never saw;
        // the parent at 103 differs from canonical → Detached, needs an RPC walk.
        match linkage.observe(104, hb(204), hb(203)) {
            LinkOutcome::Detached {
                parent_hash,
                child_number,
            } => {
                assert_eq!(parent_hash, hb(203));
                assert_eq!(child_number, 104);
            }
            other => panic!("expected Detached, got {other:?}"),
        }
    }

    #[test]
    fn head_linkage_divergence_below_floor_is_beyond_finality() {
        // Ring covers [103, 105]; floor = 103.
        let mut linkage = seed_chain(3, 100, 105);
        assert_eq!(linkage.floor(), Some(103));
        // A head whose parent (102) is below the ring floor cannot be resolved,
        // and no ancestor is fabricated.
        assert_eq!(
            linkage.observe(103, hb(203), hb(202)),
            LinkOutcome::BeyondFinality
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_walks_skipped_chain_to_common_ancestor() {
        // Canonical ring [100, 103]; new chain shares ancestor 101.
        let linkage = seed_chain(8, 100, 103);
        // New tip 104' detached at parent 103' (hb(203)); the walk fetches
        // 103' (parent 102' = hb(202)) then 102' (parent 101 = hb(101) ∈ ring).
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(202)));
        asserter.push_success(&block_with_parent(hb(101)));
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();

        let err = resolve_reorg_ancestor(&provider, &linkage, hb(203), 104, 8).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 101
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_exhausts_ring_as_beyond_finality() {
        // Shallow ring [102, 103] (depth 2); floor = 102.
        let linkage = seed_chain(2, 100, 103);
        assert_eq!(linkage.floor(), Some(102));
        // New tip 104' detached at 103' (hb(203)); the walk fetches 103' (parent
        // 102' = hb(202)), which differs from canonical 102 at the floor.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(202)));
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();

        let err = resolve_reorg_ancestor(&provider, &linkage, hb(203), 104, 2).await;
        assert!(matches!(err, RpcError::ReorgBeyondFinality), "got {err:?}");
    }

    // ── Task A: pure observe shapes ──────────────────────────────────────────

    #[test]
    fn head_linkage_back_to_back_reorgs_ring_stays_coherent() {
        // Seed canonical [100, 105]; depth=8 so all fit and no eviction yet.
        let mut linkage = seed_chain(8, 100, 105);

        // ForkA: 104' replaces 104, building on canonical 103 → ancestor 103.
        assert_eq!(
            linkage.observe(104, hb(204), hb(103)),
            LinkOutcome::Reorg {
                common_ancestor: 103
            }
        );
        // Ring re-anchored at 103; 104 now records new-chain hash.
        assert_eq!(linkage.hash_at(104), Some(hb(204)));

        // Extend forkA's chain by one clean block.
        assert_eq!(linkage.observe(105, hb(205), hb(204)), LinkOutcome::Linked);

        // ForkB: a competing 105'' builds on forkA's 104 (canonical ancestor 104).
        // Use hb(49) for forkB's hash — a byte distinct from forkA's range (hb(200+)).
        assert_eq!(
            linkage.observe(105, hb(49), hb(204)),
            LinkOutcome::Reorg {
                common_ancestor: 104
            }
        );
        // Ring re-anchored at 104; 105 records forkB hash; ring stays bounded.
        assert_eq!(linkage.hash_at(105), Some(hb(49)));
        assert_eq!(linkage.hash_at(104), Some(hb(204))); // unchanged by forkB
        assert!(linkage.ring.len() as u64 <= 8);
    }

    #[test]
    fn head_linkage_multi_header_same_height_duplicate_vs_replace() {
        let mut linkage = seed_chain(8, 100, 104);
        // Re-observing the canonical tip with its exact hash → no-op (Linked).
        assert_eq!(linkage.observe(104, hb(104), hb(103)), LinkOutcome::Linked);
        assert_eq!(linkage.hash_at(104), Some(hb(104)));
        // A competing block at the same height with a different hash, same parent
        // → replace, ancestor = 103.
        assert_eq!(
            linkage.observe(104, hb(204), hb(103)),
            LinkOutcome::Reorg {
                common_ancestor: 103
            }
        );
        assert_eq!(linkage.hash_at(104), Some(hb(204)));
    }

    #[test]
    fn head_linkage_identical_hash_no_op() {
        // The `hash_at(number) == Some(hash)` branch: re-observing an already
        // recorded (number, hash) pair returns Linked and leaves the ring alone.
        let mut linkage = seed_chain(8, 100, 104);
        let floor_before = linkage.floor();
        let depth_before = linkage.ring.len();
        assert_eq!(linkage.observe(104, hb(104), hb(103)), LinkOutcome::Linked);
        assert_eq!(linkage.floor(), floor_before);
        assert_eq!(linkage.ring.len(), depth_before);
        assert_eq!(linkage.hash_at(104), Some(hb(104)));
    }

    #[test]
    fn head_linkage_forward_gap_becomes_detached_for_verification() {
        // `number > tip + 1`: heads skipped ahead (e.g. across a reconnect).
        // The gap can no longer be trusted blindly — it routes through `Detached`
        // so the caller walks it back via `resolve_reorg_ancestor`, and the ring
        // is left untouched (no clear, no re-anchor) for that walk to consult.
        let mut linkage = seed_chain(8, 100, 103); // tip = 103
        assert_eq!(
            linkage.observe(107, hb(107), hb(106)),
            LinkOutcome::Detached {
                parent_hash: hb(106),
                child_number: 107,
            }
        );
        // Old ring state preserved — `seed_chain(8, 100, 103)` builds [100,101,102,103]
        // with no eviction at depth 8, so floor() (the ring front's number) is 100.
        assert_eq!(linkage.floor(), Some(100));
        assert_eq!(linkage.hash_at(103), Some(hb(103)));
    }

    #[test]
    fn head_linkage_idempotent_reanchor_bounds_ring() {
        // After a reorg, a clean chain links cleanly and the ring stays bounded.
        let mut linkage = seed_chain(5, 100, 106); // ring [102,103,104,105,106] (evicted 100,101)
        assert_eq!(linkage.floor(), Some(102));

        // Reorg: 105' replaces 105, ancestor 104.
        assert_eq!(
            linkage.observe(105, hb(205), hb(104)),
            LinkOutcome::Reorg {
                common_ancestor: 104
            }
        );
        assert_eq!(linkage.hash_at(105), Some(hb(205)));

        // Extend cleanly from the new chain tip.
        assert_eq!(linkage.observe(106, hb(206), hb(205)), LinkOutcome::Linked);
        assert_eq!(linkage.observe(107, hb(207), hb(206)), LinkOutcome::Linked);

        // Ring still bounded at depth 5; oldest entry evicted.
        assert!(linkage.ring.len() as u64 <= 5);
        assert_eq!(linkage.hash_at(105), Some(hb(205))); // new-chain hash retained
        assert_eq!(linkage.hash_at(102), None); // evicted
    }

    // ── Task B: resolve_reorg_ancestor walk-depth matrix ────────────────────

    // Shared ring for the in-window rows: depth=15, canonical [100,109], floor=100,
    // tip=109. Each test detaches at child=111 (parent_number=110 ∉ ring).
    // k pushes → common ancestor at 110-k. Rings are distinct from catastrophe rows.

    #[tokio::test]
    async fn resolve_reorg_ancestor_walk_depth_1() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(109))); // 110'→canonical 109
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 109
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_walk_depth_2() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(201))); // 110'→109'
        asserter.push_success(&block_with_parent(hb(108))); // 109'→canonical 108
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 108
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_walk_depth_3() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(201))); // 110'→109'
        asserter.push_success(&block_with_parent(hb(202))); // 109'→108'
        asserter.push_success(&block_with_parent(hb(107))); // 108'→canonical 107
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 107
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_walk_depth_7() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        // Six intermediate new-chain blocks (hb(201)..hb(206)), then canonical 103.
        asserter.push_success(&block_with_parent(hb(201)));
        asserter.push_success(&block_with_parent(hb(202)));
        asserter.push_success(&block_with_parent(hb(203)));
        asserter.push_success(&block_with_parent(hb(204)));
        asserter.push_success(&block_with_parent(hb(205)));
        asserter.push_success(&block_with_parent(hb(206)));
        asserter.push_success(&block_with_parent(hb(103))); // final→canonical 103
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 103
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_deepest_in_ring_resolves() {
        // Ring depth=5, canonical [100,104], floor=100. Child=105 → 4 fetches →
        // ancestor=100 (the floor). With depth=5 the loop runs 5 iterations: the
        // match check at the start of iter 4 fires after 4 prior fetches bring
        // parent_number to 100. Using child=106 would need 6 iterations (off by one).
        let linkage = seed_chain(5, 100, 104);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(201))); // 104'→103'
        asserter.push_success(&block_with_parent(hb(202))); // 103'→102'
        asserter.push_success(&block_with_parent(hb(203))); // 102'→101'
        asserter.push_success(&block_with_parent(hb(100))); // 101'→canonical 100
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 105, 5).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 100
                }
            ),
            "got {err:?}"
        );
    }

    /// End-to-end wiring proof: a forward gap that turns out to hide a genuine
    /// reorg, walked back and correctly detected — `observe` → `Detached` →
    /// `resolve_reorg_ancestor` → `Reorg`, not just `observe` in isolation.
    #[tokio::test]
    async fn forward_gap_walks_back_and_detects_reorg() {
        // Canonical ring [100, 103], depth 8.
        let linkage = seed_chain(8, 100, 103);

        // A forward-gap head arrives: 107, whose parent (106') is NOT canonical —
        // a reorg happened somewhere during the connection gap.
        let mut linkage = linkage;
        let outcome = linkage.observe(107, hb(207), hb(206));
        assert_eq!(
            outcome,
            LinkOutcome::Detached {
                parent_hash: hb(206),
                child_number: 107,
            },
            "forward gap must route through Detached, not blindly re-anchor"
        );

        // Walk back: 106'→105'→104'→ canonical 103 (ancestor).
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(205)));
        asserter.push_success(&block_with_parent(hb(204)));
        asserter.push_success(&block_with_parent(hb(103)));
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();

        let err = resolve_reorg_ancestor(&provider, &linkage, hb(206), 107, 8).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 103
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_loop_exhausted_is_beyond_finality() {
        // Depth cap (depth=3) prevents resolution: the loop exhausts its 3 iterations
        // without finding a matching canonical ancestor. Uses a distinct ring from the
        // in-window rows (depth=3, seed [100,109] → ring [107,108,109], floor=107).
        let linkage = seed_chain(3, 100, 109);
        assert_eq!(linkage.floor(), Some(107));
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(201)));
        asserter.push_success(&block_with_parent(hb(202)));
        asserter.push_success(&block_with_parent(hb(203))); // loop ends after this fetch
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 3).await;
        assert!(matches!(err, RpcError::ReorgBeyondFinality), "got {err:?}");
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_parent_at_floor_is_beyond_finality() {
        // `parent_number <= floor` fires before any fetch: detached_parent sits
        // exactly at the ring floor, which the walk cannot descend below.
        let linkage = seed_chain(10, 100, 109); // floor = 100
        let asserter = Asserter::new(); // no pushes needed
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        // child=101 → parent_number=100 = floor; first iteration hits the floor check.
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 101, 10).await;
        assert!(matches!(err, RpcError::ReorgBeyondFinality), "got {err:?}");
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_null_response_is_beyond_finality() {
        // `Ok(None)` from `get_block_by_hash` (provider claims it doesn't have the
        // block) is treated conservatively as `> finality`, not a transport error.
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!(null));
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(matches!(err, RpcError::ReorgBeyondFinality), "got {err:?}");
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_transport_error_is_connection_error() {
        // A mid-walk transport error is recoverable (the stream can restart and
        // re-attempt); it must NOT collapse into `ReorgBeyondFinality` and trigger
        // do-not-serve. Assert `ConnectionError`, not `ReorgBeyondFinality`.
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_failure_msg("transport blip");
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(matches!(err, RpcError::ConnectionError(_)), "got {err:?}");
    }

    // ── P3.4: by-hash recovery for the reorg-exposed tail ───────────────────

    /// Canonical chain [100, 103]; `collect_window_hashes` walks backward from
    /// tip=103 (hash=hb(103)) to from_block=100 using `get_block_by_hash`.
    /// Each block response carries the parent hash. Then `fetch_logs_by_hash_range`
    /// issues one `get_logs` per block pinned to the canonical hash (ascending).
    /// Assert: four LogRanges covering [100,100]..[103,103], one log each.
    #[tokio::test]
    async fn by_hash_tail_fetch_over_mock() {
        // Asserter for the walk: three get_block_by_hash calls (103→102→101; 100 == from_block,
        // loop ends without fetching 100's parent). Must use block_with_parent_at so the
        // number cross-check in collect_window_hashes passes.
        let walk_asserter = Asserter::new();
        walk_asserter.push_success(&block_with_parent_at(103, hb(102))); // block 103 → parent 102
        walk_asserter.push_success(&block_with_parent_at(102, hb(101))); // block 102 → parent 101
        walk_asserter.push_success(&block_with_parent_at(101, hb(100))); // block 101 → parent 100
        let walk_provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(walk_asserter)
            .erased();

        let pairs_desc = collect_window_hashes(&walk_provider, hb(103), 103, 100)
            .await
            .expect("walk should succeed");

        // Pairs are in descending order: [(103,hb(103)), (102,hb(102)), (101,hb(101)), (100,hb(100))].
        assert_eq!(pairs_desc.len(), 4);
        assert_eq!(pairs_desc[0], (103, hb(103)));
        assert_eq!(pairs_desc[1], (102, hb(102)));
        assert_eq!(pairs_desc[2], (101, hb(101)));
        assert_eq!(pairs_desc[3], (100, hb(100)));

        // Reverse to ascending for the log fetch.
        let mut pairs = pairs_desc;
        pairs.reverse();

        // Asserter for the log fetches: four get_logs calls, one per block.
        // Each returns a single synthetic log tagged with the block number.
        let log_asserter = Asserter::new();
        for n in 100u64..=103 {
            log_asserter.push_success(&serde_json::json!([{
                "address": "0x0000000000000000000000000000000000000001",
                "topics": [],
                "data": "0x",
                "blockHash": hb(n as u8).to_string(),
                "blockNumber": format!("0x{:x}", n),
                "transactionHash": hb(0).to_string(),
                "transactionIndex": "0x0",
                "logIndex": "0x0",
                "removed": false
            }]));
        }
        let log_provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(log_asserter)
            .erased();
        let settings = ProviderSettings::from_mock(log_provider);

        let ranges = fetch_logs_by_hash_range(&settings, &pairs, None)
            .await
            .expect("log fetch should succeed");

        // Four LogRanges, each a single block with one log.
        assert_eq!(ranges.len(), 4, "expected one range per block");
        for (i, (from, to, logs)) in ranges.iter().enumerate() {
            let block_n = 100u64 + i as u64;
            assert_eq!(*from, block_n, "from_block mismatch at index {i}");
            assert_eq!(*to, block_n, "to_block mismatch at index {i}");
            assert_eq!(logs.len(), 1, "expected one log at block {block_n}");
            assert_eq!(
                logs[0].block_number,
                Some(block_n),
                "log block number mismatch"
            );
        }
    }

    /// `flush_window_range` fetches in-window blocks by canonical hash only.
    ///
    /// Setup: ring holds canonical hb(200) for block 200. The mock has exactly one
    /// log response — the canonical log tagged with hb(200). The mock also has an
    /// "orphan" log queued second (tagged with a different hash). If by-number were
    /// used, the asserter would return responses in order and the test would get the
    /// canonical log (first) for the by-number call — so instead we prove by-number
    /// is never invoked structurally: `flush_window_range` is the by-hash-only path;
    /// the mock has only ONE response queued; if a second call (by-number fallback)
    /// happened, `flush_window_range` would panic on an exhausted asserter.
    #[tokio::test]
    async fn orphan_rejection() {
        let canonical_log = serde_json::json!([{
            "address": "0xCAFECAFECAFECAFECAFECAFECAFECAFECAFECAFE",
            "topics": [],
            "data": "0x01",
            "blockHash": hb(200).to_string(),
            "blockNumber": "0xc8",
            "transactionHash": hb(1).to_string(),
            "transactionIndex": "0x0",
            "logIndex": "0x0",
            "removed": false
        }]);
        // One response only. A by-number fallback would exhaust the asserter and panic.
        let asserter = Asserter::new();
        asserter.push_success(&canonical_log);
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );

        // Ring holds canonical hash for block 200; coalescer has [200, 200] pending.
        let mut linkage = HeadLinkage::new(8);
        linkage.observe(200, hb(200), hb(199)); // bootstraps ring with (200, hb(200))
        let mut coalescer = HeadCoalescer::new(Some(200), WINDOW);
        coalescer.observe_head(200, Instant::now());

        let ranges = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None)
            .await
            .expect("flush should succeed");

        assert_eq!(ranges.len(), 1, "one range per block");
        let (from, to, logs) = &ranges[0];
        assert_eq!((*from, *to), (200, 200));
        assert_eq!(logs.len(), 1, "canonical log returned");
        assert_eq!(
            logs[0].block_hash,
            Some(hb(200)),
            "log must carry the canonical block hash, not an orphan's"
        );
        // Cursor advanced past block 200.
        assert_eq!(coalescer.pending_range(), None);
    }

    /// A null block response from `get_block_by_hash` during the hash walk is
    /// classified as `ReorgBeyondFinality` (fail-closed). A transport error is
    /// classified as `ConnectionError` (not `ReorgBeyondFinality`) so the caller
    /// can distinguish a recoverable network fault from a genuine finality violation.
    #[tokio::test]
    async fn hash_mismatch_null_block_fail_closed() {
        // Null block → ReorgBeyondFinality.
        {
            let asserter = Asserter::new();
            asserter.push_success(&serde_json::json!(null));
            let provider = ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased();
            // Walk from block 5 (hash hb(5)) down to block 1; the first fetch returns null.
            let err = collect_window_hashes(&provider, hb(5), 5, 1)
                .await
                .expect_err("null block must be an error");
            assert!(
                matches!(err, RpcError::ReorgBeyondFinality),
                "null block must yield ReorgBeyondFinality, got {err:?}"
            );
        }
        // Transport error → ConnectionError (not ReorgBeyondFinality).
        {
            let asserter = Asserter::new();
            asserter.push_failure_msg("connection reset");
            let provider = ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased();
            let err = collect_window_hashes(&provider, hb(5), 5, 1)
                .await
                .expect_err("transport error must be an error");
            assert!(
                matches!(err, RpcError::ConnectionError(_)),
                "transport fault must yield ConnectionError, got {err:?}"
            );
        }
    }

    /// When `depth == 0` the by-hash path is never entered.
    /// `HeadLinkage::enabled()` returns false; `observe` returns `Linked` for every
    /// head regardless of parent mismatch. No `get_block_by_hash` call is issued.
    #[test]
    fn dormancy_at_depth_zero() {
        let mut linkage = HeadLinkage::new(0);
        assert!(!linkage.enabled(), "depth=0 must be disabled");

        // Any head — even one with a bogus parent — must return Linked.
        assert_eq!(
            linkage.observe(100, hb(1), hb(99)),
            LinkOutcome::Linked,
            "depth=0 must not trigger reorg detection"
        );
        assert_eq!(
            linkage.observe(101, hb(2), hb(55)), // obviously wrong parent
            LinkOutcome::Linked,
            "depth=0 must not trigger reorg detection on bad parent"
        );
        // Ring stays empty: no state accrues when disabled.
        assert_eq!(
            linkage.floor(),
            None,
            "disabled linkage must not record state"
        );
    }

    /// After the boot gap is flushed by `flush_window_range`, the ring holds the
    /// first observed head. The production path for the NEXT head is a clean
    /// `observe` extension — no false reorg fires.
    ///
    /// In production: `stream_heads_with_logs` creates a fresh empty ring, then
    /// `observe(first_head)` bootstraps it. `flush_window_range` handles the gap
    /// [cursor, first_head] using the walked pairs directly (no ring seeding).
    /// The second head (first_head + 1) then links cleanly via the normal parent
    /// check — because `observe(first_head+1)` checks ring.back() = first_head.
    #[test]
    fn boot_tail_seeds_ring() {
        // Production state after first head 105 arrives: ring bootstrapped to {105}.
        let mut linkage = HeadLinkage::new(8);
        assert_eq!(
            linkage.observe(105, hb(105), hb(104)),
            LinkOutcome::Linked,
            "bootstrap observe"
        );
        assert_eq!(linkage.hash_at(105), Some(hb(105)));

        // After flush_window_range handles [cursor=101, 105], the cursor moves to 106.
        // The next WS head is 106 with correct parent hb(105) — must link cleanly.
        assert_eq!(
            linkage.observe(106, hb(106), hb(105)),
            LinkOutcome::Linked,
            "first post-flush head must link without false reorg"
        );
        assert_eq!(linkage.hash_at(106), Some(hb(106)));

        // Subsequent heads also link cleanly.
        assert_eq!(linkage.observe(107, hb(107), hb(106)), LinkOutcome::Linked);
    }

    // ── P3.4 addendum: live flush self-heal + orphan rejection ──────────────

    /// REGRESSION — live flush with a head gap must not wedge.
    ///
    /// Scenario: depth=8, cursor=101, first WS head is 105 (gap [101,104]).
    /// Before the fix the flush found `hash_at(101) = None`, broke out of the for
    /// loop, and `continue`d the `'flush while` with the cursor unmoved — an
    /// infinite spin. After the fix `flush_window_range` self-heals by walking from
    /// ring.latest() (105) back to 101, then fetches logs for each block by hash.
    ///
    /// The test is bounded by tokio's async executor; a spin would deadlock it.
    #[tokio::test]
    async fn live_flush_gap_self_heals() {
        // Ring: observe(105) bootstraps to {105}. Cursor: 101.
        let mut linkage = HeadLinkage::new(8);
        assert_eq!(linkage.observe(105, hb(105), hb(104)), LinkOutcome::Linked);
        let mut coalescer = HeadCoalescer::new(Some(101), WINDOW);
        coalescer.observe_head(105, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((101, 105)));

        // Mock: walk 105→104→103→102 (4 get_block_by_hash calls; 101==from_block so no
        // fetch for 101 itself). Then 5 get_logs (at_block_hash) for blocks 101..=105.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent_at(105, hb(104)));
        asserter.push_success(&block_with_parent_at(104, hb(103)));
        asserter.push_success(&block_with_parent_at(103, hb(102)));
        asserter.push_success(&block_with_parent_at(102, hb(101)));
        for _ in 101u64..=105 {
            asserter.push_success(&serde_json::json!([]));
        }
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );

        let ranges = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None)
            .await
            .expect("flush must succeed, not spin");

        assert_eq!(ranges.len(), 5, "one LogRange per block 101..=105");
        for (i, (from, to, _)) in ranges.iter().enumerate() {
            let n = 101u64 + i as u64;
            assert_eq!((*from, *to), (n, n));
        }
        // Cursor advanced past all flushed blocks.
        assert_eq!(coalescer.pending_range(), None);
    }

    /// `collect_window_hashes`: a by-hash response whose `header.number` does not
    /// match the expected number is a provider integrity violation — fail closed.
    #[tokio::test]
    async fn collect_window_hashes_number_mismatch() {
        // Walk from tip=5 to from_block=1; mock returns a block claiming number=99.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent_at(99, hb(4))); // wrong number
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();

        let err = collect_window_hashes(&provider, hb(5), 5, 1)
            .await
            .expect_err("number mismatch must be an error");
        assert!(
            matches!(err, RpcError::ReorgBeyondFinality),
            "number mismatch must yield ReorgBeyondFinality, got {err:?}"
        );
    }

    /// `flush_window_range`: when `latest()` is None (ring completely empty),
    /// the function returns `ReorgBeyondFinality` rather than panicking.
    #[tokio::test]
    async fn flush_window_range_empty_ring_is_fail_closed() {
        let asserter = Asserter::new(); // no responses needed
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let mut linkage = HeadLinkage::new(8); // empty ring (no observe yet)
        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);
        coalescer.observe_head(100, Instant::now());

        let err = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None)
            .await
            .expect_err("empty ring must be fail-closed");
        assert!(matches!(err, RpcError::ReorgBeyondFinality), "got {err:?}");
    }

    /// Minimal `tracing::Subscriber` enabled for every callsite, recording
    /// nothing. `warn!`'s arguments are only evaluated when some subscriber is
    /// both installed (via `tracing::subscriber::set_default`) and interested
    /// (`enabled` returning `true`, plus `tracing::callsite::rebuild_interest_cache`
    /// so a stale "nobody's listening" cache from an earlier test doesn't keep
    /// suppressing evaluation) — with neither, a regression test that only
    /// asserts "the call didn't panic" can pass even when the logged expression
    /// itself would panic, because it never actually runs. Installing this for
    /// the scope of a test forces that evaluation, so the assertion is real.
    struct AlwaysOnSubscriber;

    impl tracing::Subscriber for AlwaysOnSubscriber {
        fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool {
            true
        }
        fn new_span(&self, _span: &tracing::span::Attributes<'_>) -> tracing::span::Id {
            tracing::span::Id::from_u64(1)
        }
        fn record(&self, _span: &tracing::span::Id, _values: &tracing::span::Record<'_>) {}
        fn record_follows_from(&self, _span: &tracing::span::Id, _follows: &tracing::span::Id) {}
        fn event(&self, _event: &tracing::Event<'_>) {}
        fn enter(&self, _span: &tracing::span::Id) {}
        fn exit(&self, _span: &tracing::span::Id) {}
    }

    /// REGRESSION — `flush_window_range`'s self-heal walk when the ring's only
    /// entry (`top_n`) is *older* than the coalescer's `from_block` (a stale
    /// ring lagging a cursor that has already advanced past it). Before the
    /// fix, the self-heal `warn!` computed `top_n - from_block + 1` before
    /// reaching `collect_window_hashes`'s own `tip_number < from_block` guard —
    /// on a build where the `warn!` argument is evaluated (any binary with a
    /// subscriber installed at a level that enables it, e.g. the CLI's default
    /// INFO subscriber), that subtraction panics. This test drives exactly
    /// that ring/cursor shape end-to-end and asserts the call returns cleanly
    /// (empty ranges, since nothing above the stale ring's tip is resolvable)
    /// rather than panicking.
    ///
    /// `tracing::warn!` does not evaluate its argument expressions unless a
    /// subscriber is installed and interested — with none installed (the
    /// default in this test binary), the underflow-prone expression would
    /// never run and this test would pass identically whether or not the fix
    /// is in place. `AlwaysOnSubscriber` forces real evaluation for the scope
    /// of this call.
    #[tokio::test]
    async fn flush_window_range_self_heal_with_stale_ring_does_not_panic() {
        let asserter = Asserter::new(); // no responses needed — the walk short-circuits
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );

        // Ring holds only an old block (top_n = 50)...
        let mut linkage = HeadLinkage::new(8);
        assert_eq!(linkage.observe(50, hb(50), hb(49)), LinkOutcome::Linked);

        // ...but the coalescer's cursor is already ahead of it (from_block = 100).
        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);
        coalescer.observe_head(100, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((100, 100)));

        // Force the self-heal `warn!`'s arguments to actually be evaluated —
        // see `AlwaysOnSubscriber`'s doc comment.
        let subscriber_guard = tracing::subscriber::set_default(AlwaysOnSubscriber);
        tracing::callsite::rebuild_interest_cache();

        let ranges = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None)
            .await
            .expect(
                "a stale ring must not panic — collect_window_hashes's own guard \
                 no-ops the walk",
            );
        assert!(
            ranges.is_empty(),
            "no hashes are resolvable above the stale ring's tip"
        );

        // Restore the default and rebuild the cache again so no later test in
        // this binary inherits an "always interested" callsite cache from here.
        drop(subscriber_guard);
        tracing::callsite::rebuild_interest_cache();
    }

    /// REGRESSION — the self-heal walk's own `resolve_http_provider(providers,
    /// reused, provider_index)` call must resolve its client from the
    /// caller-supplied `reused` set, the same way the subsequent by-hash log
    /// fetch already does — not rebuild a fresh one via `connect_http`.
    ///
    /// Proven the same way `flush_window_range_dispatches_concurrently_and_
    /// restores_ascending_order` proves the log-fetch side's reuse:
    /// `settings`' own configured HTTP endpoint is unreachable, so a
    /// `connect_http` rebuild fails the walk's very first `get_block_by_hash`
    /// call outright, while `reused`'s working mock lets the flush succeed —
    /// success is only possible if the walk actually used the reused client.
    #[tokio::test]
    async fn flush_window_range_self_heal_walk_uses_reused_http_client() {
        // Ring: observe(105) bootstraps to {105}. Cursor: 101 — same shape as
        // `live_flush_gap_self_heals`, forcing the self-heal walk.
        let mut linkage = HeadLinkage::new(8);
        assert_eq!(linkage.observe(105, hb(105), hb(104)), LinkOutcome::Linked);
        let mut coalescer = HeadCoalescer::new(Some(101), WINDOW);
        coalescer.observe_head(105, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((101, 105)));

        // The reused (working) client: walk 105→104→103→102 (4 get_block_by_hash
        // calls; 101==from_block so no fetch for 101 itself), then 5 get_logs.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent_at(105, hb(104)));
        asserter.push_success(&block_with_parent_at(104, hb(103)));
        asserter.push_success(&block_with_parent_at(103, hb(102)));
        asserter.push_success(&block_with_parent_at(102, hb(101)));
        for _ in 101u64..=105 {
            asserter.push_success(&serde_json::json!([]));
        }
        let reused = vec![
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        ];

        // `connect_http`, if the self-heal walk fell back to it, would build a
        // real client against this unreachable host and fail the walk.
        let settings = ProviderSettings::new(
            vec![HttpRpcSettings::new("http://unused".to_string(), 1, 100)],
            vec!["ws://127.0.0.1:9".parse().expect("static url")],
        );

        let ranges = flush_window_range(
            &mut linkage,
            &mut coalescer,
            &settings,
            None,
            0,
            Some(&reused),
        )
        .await
        .expect(
            "self-heal walk must resolve its client from `reused`, not rebuild via connect_http",
        );

        assert_eq!(ranges.len(), 5, "one LogRange per block 101..=105");
        for (i, (from, to, _)) in ranges.iter().enumerate() {
            let n = 101u64 + i as u64;
            assert_eq!((*from, *to), (n, n));
        }
    }

    /// REGRESSION — `flush_window_range` with zero configured HTTP providers must
    /// return `Err`, not panic. Before the fix, the concurrent path's
    /// `providers.http_settings(provider_index)` call did `idx % http_providers.len()`
    /// with no guard, so an empty `http_providers` divided by zero and panicked —
    /// on an input for which the pre-fix sequential code (and `last_block`'s own
    /// `num_providers == 0` guard, the convention this mirrors) both cleanly
    /// returned `Err`. The ring is fully populated here (no self-heal walk) so this
    /// isolates the `http_settings` guard specifically.
    #[tokio::test]
    async fn flush_window_range_empty_http_providers_is_fail_closed_not_a_panic() {
        let mut linkage = HeadLinkage::new(8);
        assert_eq!(linkage.observe(100, hb(100), hb(99)), LinkOutcome::Linked);
        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);
        coalescer.observe_head(100, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((100, 100)));

        let settings = ProviderSettings::new(
            vec![], // zero configured HTTP providers
            vec!["ws://127.0.0.1:9".parse().expect("static url")],
        );

        let err = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None)
            .await
            .expect_err("zero HTTP providers must be fail-closed, not panic");
        assert!(matches!(err, RpcError::ConnectionError(_)), "got {err:?}");
    }

    /// The backfill window boundary: blocks strictly below `tip - depth` are
    /// fetched by number (the normal backfill path); blocks `>= tip - depth` (within
    /// the reorg-exposed window) are fetched by hash.
    ///
    /// This test exercises the predicate logic directly: given a specific tip and
    /// depth, verify that `from_block >= tip.saturating_sub(depth)` correctly
    /// partitions the range.
    #[test]
    fn window_boundary() {
        // tip = 1000, depth = 50: window starts at 950.
        let tip: u64 = 1000;
        let depth: u64 = 50;
        let window_floor = tip.saturating_sub(depth); // 950

        // Blocks below the window floor are served by the backfill-by-number path.
        for n in [0u64, 500, 949] {
            assert!(
                n < window_floor,
                "block {n} should be below the window floor {window_floor}"
            );
        }

        // The window floor itself and everything above it is in the by-hash zone.
        for n in [950u64, 975, 1000] {
            assert!(
                n >= window_floor,
                "block {n} should be inside the by-hash window (floor {window_floor})"
            );
        }

        // Saturating_sub at depth == 0 yields u64::MAX (no window) — the by-hash
        // path is never entered, matching the dormancy_at_depth_zero invariant.
        let no_window_floor = tip.saturating_sub(0);
        assert_eq!(no_window_floor, tip, "depth=0 must not create a window");

        // Depth larger than tip saturates at 0: the entire history is in the window.
        let huge_depth_floor = tip.saturating_sub(tip + 1);
        assert_eq!(
            huge_depth_floor, 0,
            "depth > tip must floor at 0 (all blocks in window)"
        );
    }

    /// REGRESSION — a partial failure in a multi-block flush must not silently
    /// skip earlier blocks.
    ///
    /// Before the fix, `flush_window_range` advanced `coalescer.advance_cursor`
    /// inside the fetch loop. If block 201's fetch failed after 200 succeeded,
    /// the cursor advanced past 200 and its logs were discarded. On reconnect the
    /// stream resumed from 201, silently skipping block 200.
    ///
    /// After the fix the cursor is only advanced once the ENTIRE batch succeeds.
    /// This test fails against the pre-fix code (cursor at 201) and passes after
    /// (cursor stays at 200 — the whole range retries on the next provider).
    #[tokio::test]
    async fn flush_window_range_partial_failure_leaves_cursor_unmoved() {
        // Ring: blocks 200 and 201 observed (clean chain).
        let mut linkage = HeadLinkage::new(8);
        assert_eq!(linkage.observe(200, hb(200), hb(199)), LinkOutcome::Linked);
        assert_eq!(linkage.observe(201, hb(201), hb(200)), LinkOutcome::Linked);

        // Coalescer: cursor at 200, head 201 seen → pending [200, 201].
        let mut coalescer = HeadCoalescer::new(Some(200), WINDOW);
        coalescer.observe_head(201, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((200, 201)));

        // Single mocked HTTP provider (from_mock): one queued success (block 200's
        // get_logs), then one queued failure (block 201's get_logs). With a single
        // provider, one failure exhausts all retry attempts.
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!([])); // block 200 logs succeed
        asserter.push_failure_msg("transient error"); // block 201 logs fail

        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );

        let result =
            flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None).await;

        // Must return Err — the batch did not complete.
        assert!(result.is_err(), "partial failure must return Err");

        // The cursor must NOT have advanced past block 200 — the whole range must
        // retry on the next provider so block 200 is not silently skipped.
        assert_eq!(
            coalescer.pending_range(),
            Some((200, 201)),
            "cursor must stay at 200 so block 200 is not silently skipped on reconnect"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn bound_flush_time_disabled_at_zero_awaits_normally() {
        let result: Result<u32, RpcError> = bound_flush_time(0, async {
            tokio::time::sleep(Duration::from_millis(50)).await;
            Ok(42)
        })
        .await;
        assert_eq!(
            result.unwrap(),
            42,
            "max_secs=0 must not time out, however long the future takes"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn bound_flush_time_fires_on_a_slow_future() {
        let result: Result<u32, RpcError> = bound_flush_time(1, async {
            tokio::time::sleep(Duration::from_millis(1200)).await;
            Ok(42)
        })
        .await;
        assert!(
            matches!(result, Err(RpcError::ConnectionError(_))),
            "a future exceeding max_secs must yield ConnectionError, got {result:?}"
        );
    }

    /// Hermetic wiring proof for the live consumption loop:
    ///   observe_head → observe → should_flush → flush_window_range → [ranges]
    ///   → observe_reorg → LinkOutcome::Reorg
    ///
    /// This directly sequences the operations `stream_heads_with_logs` performs
    /// in its inner loop and asserts the correct outcomes at each step. It covers
    /// the two behaviors that had no hermetic test: the gap self-heal path and the
    /// reorg-terminate path.
    ///
    /// Residual: the outer WS-reconnect/failover loop in `stream_heads_with_logs`
    /// is exercised only by the `#[ignore]`d network test (`test_event_streamer_head`
    /// in `tests/events_test.rs`). A full seam extraction was evaluated and deferred
    /// to avoid restructuring the core stream at a release gate.
    #[tokio::test]
    async fn live_loop_wiring_gap_then_reorg() {
        // Coalescer: cursor at 101 (live filter from_block = 101).
        // Linkage: depth 8.
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(101), WINDOW);
        let mut linkage = HeadLinkage::new(8);

        // Step 1: head 105 arrives with a gap (101-104 skipped).
        // observe_head extends the coalescer's pending range to [101, 105].
        assert!(coalescer.observe_head(105, now));
        assert_eq!(coalescer.pending_range(), Some((101, 105)));

        // observe bootstraps the ring to {105} (bootstrap path: first observe on an
        // empty ring pushes and returns Linked).
        assert_eq!(linkage.observe(105, hb(105), hb(104)), LinkOutcome::Linked);
        assert_eq!(linkage.latest(), Some((105, hb(105))));

        // Step 2: simulate time advancing past the coalesce window by querying
        // should_flush with a synthetic instant past the deadline — no real or
        // paused sleep needed since this step never awaits.
        let after_window = now + WINDOW + Duration::from_millis(1);
        assert!(
            coalescer.should_flush(after_window),
            "coalesce window expired — flush must fire"
        );

        // Step 3: flush_window_range self-heals the gap [101, 105].
        // Walk: get_block_by_hash for hb(105)→hb(104)→hb(103)→hb(102) (4 fetches;
        // 101==from_block so the walk stops after pushing block 101).
        // Logs: 5 get_logs at_block_hash for blocks 101..=105.
        // Both walk and log calls use the same mocked HTTP provider.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent_at(105, hb(104)));
        asserter.push_success(&block_with_parent_at(104, hb(103)));
        asserter.push_success(&block_with_parent_at(103, hb(102)));
        asserter.push_success(&block_with_parent_at(102, hb(101)));
        for _ in 101u64..=105 {
            asserter.push_success(&serde_json::json!([]));
        }
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );

        let ranges = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None)
            .await
            .expect("gap self-heal must succeed");

        assert_eq!(ranges.len(), 5, "one range per block 101..=105");
        for (i, &(from, to, _)) in ranges.iter().enumerate() {
            let n = 101u64 + i as u64;
            assert_eq!((from, to), (n, n), "range at index {i} must be block {n}");
        }
        // Cursor advanced past the full batch — no silent gap.
        assert_eq!(
            coalescer.pending_range(),
            None,
            "cursor must be at 106 after flush"
        );

        // Step 4: two more clean heads arrive and link.
        assert_eq!(linkage.observe(106, hb(106), hb(105)), LinkOutcome::Linked);
        assert_eq!(linkage.observe(107, hb(107), hb(106)), LinkOutcome::Linked);

        // Step 5: a competing block 107 with the same parent (hb(106)) but a
        // different hash (hb(250)) triggers a reorg. The ring has canonical hb(107)
        // at 107; a new block with parent matching canonical hb(106) at 106 produces
        // Reorg { common_ancestor: 106 }. `stream_heads_with_logs` would yield
        // Err(RpcError::Reorg { common_ancestor: 106 }) and return here.
        let outcome = linkage.observe(107, hb(250), hb(106));
        assert!(
            matches!(
                outcome,
                LinkOutcome::Reorg {
                    common_ancestor: 106
                }
            ),
            "competing block at 107 with canonical parent must produce Reorg{{106}}, got {outcome:?}"
        );
    }

    #[test]
    fn capture_policy_notice_speaks_once_per_policy() {
        // A fresh instance — never the process-global static, which `Once`
        // makes a one-shot resource shared across every test in this binary.
        let notice = CapturePolicyNotice::new();

        let latest_block = 1_000u64;
        let latest_first = notice.first_message(CaptureTarget::Latest, latest_block);
        assert!(
            latest_first.is_some(),
            "first Latest resolution must announce"
        );
        let latest_message = latest_first.unwrap();
        assert!(
            !latest_message.contains("DEEP LAG"),
            "Latest must not claim deep lag: {latest_message}"
        );
        assert!(
            notice
                .first_message(CaptureTarget::Latest, latest_block)
                .is_none(),
            "second Latest resolution must be silent"
        );
        assert!(
            notice
                .first_message(CaptureTarget::Latest, latest_block)
                .is_none(),
            "third Latest resolution must be silent"
        );

        let lag_block = 900u64;
        assert!(
            notice
                .first_message(CaptureTarget::Lag(64), lag_block)
                .is_some(),
            "first Lag resolution must announce (Latest already having spoken must not \
             consume Lag's independent gate)"
        );
        assert!(
            notice
                .first_message(CaptureTarget::Lag(64), lag_block)
                .is_none(),
            "second Lag resolution must be silent"
        );
        assert!(
            notice
                .first_message(CaptureTarget::Lag(64), lag_block)
                .is_none(),
            "third Lag resolution must be silent"
        );
        assert!(
            notice
                .first_message(CaptureTarget::Lag(200), lag_block)
                .is_none(),
            "gating is per-variant, not per-value: a distinct Lag(n) must not re-arm the gate"
        );

        let finalized_block = 500u64;
        let finalized_first = notice.first_message(CaptureTarget::Finalized, finalized_block);
        assert!(
            finalized_first.is_some(),
            "first Finalized resolution must announce (Latest and Lag having already \
             spoken must not consume Finalized's independent gate)"
        );
        let finalized_message = finalized_first.unwrap();
        assert!(
            finalized_message.contains("DEEP LAG ACTIVE"),
            "Finalized message must claim deep lag: {finalized_message}"
        );
        assert!(
            finalized_message.contains(&finalized_block.to_string()),
            "Finalized message must report the resolved block: {finalized_message}"
        );
        assert!(
            notice
                .first_message(CaptureTarget::Finalized, finalized_block)
                .is_none(),
            "second Finalized resolution must be silent"
        );
        assert!(
            notice
                .first_message(CaptureTarget::Finalized, finalized_block)
                .is_none(),
            "third Finalized resolution must be silent"
        );
    }

    // ── B.7 addendum: reused HTTP clients + concurrent flush dispatch ───────

    /// `resolve_http_provider`: `Some(reused)` at the same index returns the
    /// identical client handle on every call; `None` rebuilds a fresh one every
    /// call — today's `connect_http` behavior, unchanged. Built against a real
    /// (non-mock) `ProviderSettings`: `from_mock`'s `injected_http` short-circuit
    /// would make both branches return the same cloned instance, making this
    /// distinction unobservable (see the settled decision this test proves).
    #[test]
    fn resolve_http_provider_reuses_vs_rebuilds() {
        let settings = ProviderSettings::new(
            vec![HttpRpcSettings::new(
                "http://127.0.0.1:9".to_string(),
                1,
                100,
            )],
            vec!["ws://127.0.0.1:9".parse().expect("static url")],
        );
        let reused = settings.build_reused_http_clients();
        assert_eq!(reused.len(), 1, "one client per configured HTTP provider");

        // Some(&reused) at the same index: identical handle every call.
        let a = resolve_http_provider(&settings, Some(&reused), 0);
        let b = resolve_http_provider(&settings, Some(&reused), 0);
        assert!(
            std::sync::Weak::ptr_eq(&a.weak_client(), &b.weak_client()),
            "Some(reused) at the same index must return the identical client"
        );

        // None: a fresh client every call.
        let c = resolve_http_provider(&settings, None, 0);
        let d = resolve_http_provider(&settings, None, 0);
        assert!(
            !std::sync::Weak::ptr_eq(&c.weak_client(), &d.weak_client()),
            "None must rebuild a fresh client on every call, not reuse one"
        );
    }

    /// `fetch_concurrently`: dispatch is genuinely concurrent, not one-at-a-time.
    /// Fails against a revert to a sequential (e.g. `for` loop) dispatch shape,
    /// which would never see more than one in-flight fetch.
    #[tokio::test(start_paused = true)]
    async fn fetch_concurrently_runs_items_in_parallel_up_to_bound() {
        use std::sync::Arc;
        use std::sync::atomic::{AtomicUsize, Ordering};

        let bound = 4usize;
        let n = 10usize;
        let in_flight = Arc::new(AtomicUsize::new(0));
        let peak = Arc::new(AtomicUsize::new(0));

        let items: Vec<usize> = (0..n).collect();
        let in_flight_for_fetch = in_flight.clone();
        let peak_for_fetch = peak.clone();
        let fetch = move |_: usize| {
            let in_flight = in_flight_for_fetch.clone();
            let peak = peak_for_fetch.clone();
            async move {
                let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
                peak.fetch_max(now, Ordering::SeqCst);
                tokio::time::sleep(Duration::from_millis(50)).await;
                in_flight.fetch_sub(1, Ordering::SeqCst);
                Ok::<usize, RpcError>(0)
            }
        };

        let result = fetch_concurrently(items, bound, fetch).await;
        assert!(result.is_ok(), "all items succeed: {result:?}");

        let observed_peak = peak.load(Ordering::SeqCst);
        assert!(
            observed_peak > 1,
            "expected concurrent dispatch (peak > 1), got peak={observed_peak}"
        );
        assert!(
            observed_peak <= bound,
            "peak in-flight {observed_peak} exceeded the bound {bound}"
        );
    }

    /// `fetch_concurrently`: fail-fast, not collect-then-check. An early item's
    /// error must short-circuit before every item is invoked — distinguishes
    /// `try_buffer_unordered` from a `join_all`-style shape that always invokes
    /// and awaits all N items before checking any result (the Concurrency test
    /// above doesn't by itself rule this out: a collect-then-check dispatch can
    /// still run several items concurrently, just never stop early).
    #[tokio::test]
    async fn fetch_concurrently_fails_fast_not_collect_then_check() {
        use std::sync::Arc;
        use std::sync::atomic::{AtomicUsize, Ordering};

        let bound = 3usize;
        let n = 10usize;
        let items: Vec<usize> = (0..n).collect();

        let invoked = Arc::new(AtomicUsize::new(0));
        let invoked_for_fetch = invoked.clone();
        let fetch = move |i: usize| {
            invoked_for_fetch.fetch_add(1, Ordering::SeqCst);
            async move {
                if i == 0 {
                    Err(RpcError::LogFetchError("early failure".to_string()))
                } else {
                    Ok::<usize, RpcError>(i)
                }
            }
        };

        let result = fetch_concurrently(items, bound, fetch).await;
        assert!(result.is_err(), "an early failure must surface as Err");

        let invoked_count = invoked.load(Ordering::SeqCst);
        assert!(
            invoked_count < n,
            "expected fewer than {n} closures invoked under fail-fast, got {invoked_count}"
        );
    }

    /// `fetch_concurrently`: fail-fast even when an earlier-indexed neighbor is
    /// slow. Distinguishes `try_buffer_unordered` (surfaces the first error the
    /// instant it occurs, dropping remaining in-flight futures) from
    /// `try_buffered` (holds a `FuturesOrdered`, so an error at index 1 would
    /// wait for index 0 to resolve first) — the Fail-fast test above alone
    /// can't tell the two apart, since an error at an early index surfaces
    /// quickly under either combinator.
    #[tokio::test(start_paused = true)]
    async fn fetch_concurrently_fails_fast_under_early_slow_neighbor() {
        let bound = 2usize;
        let items = vec![0usize, 1usize];
        let long_sleep = Duration::from_secs(3600);

        let fetch = move |i: usize| async move {
            if i == 0 {
                tokio::time::sleep(long_sleep).await;
                Ok::<usize, RpcError>(0)
            } else {
                Err(RpcError::LogFetchError("fast failure".to_string()))
            }
        };

        let start = Instant::now();
        let result = fetch_concurrently(items, bound, fetch).await;
        let elapsed = start.elapsed();

        assert!(result.is_err(), "overall result must be Err");
        assert!(
            elapsed < long_sleep,
            "Err must be observable before item 0's sleep elapses, elapsed={elapsed:?}"
        );
    }

    /// `sort_ranges_ascending`: restores ascending block-number order from
    /// synthetic out-of-order input. Cheaper and more direct than proving the same
    /// thing by driving `flush_window_range` end-to-end (see
    /// `flush_window_range_dispatches_concurrently_and_restores_ascending_order`
    /// below for that integration-level proof).
    #[test]
    fn sort_ranges_ascending_orders_by_block_number() {
        let unordered: Vec<(u64, u64, Vec<Log>)> =
            vec![(105, 105, vec![]), (100, 100, vec![]), (103, 103, vec![])];

        let sorted = sort_ranges_ascending(unordered);

        let numbers: Vec<u64> = sorted.iter().map(|&(from, _, _)| from).collect();
        assert_eq!(numbers, vec![100, 103, 105]);
    }

    /// `flush_window_range` end-to-end at a real concurrency bound
    /// (`max_concurrency > 1`), which every other mock-backed test in this file
    /// cannot exercise: they all build `ProviderSettings` via `from_mock`, which
    /// pins `max_concurrency: 1` (see the comment on `from_mock` in `config.rs`).
    /// Reverting `flush_window_range`'s dispatch to the old sequential `for` loop
    /// (no `fetch_concurrently`, no `sort_ranges_ascending`) makes this test fail:
    /// peak in-flight would never exceed 1, and the deliberately-out-of-order
    /// completion this test forces could never happen.
    ///
    /// Uses a small hand-rolled `Provider` double instead of the `Asserter`/mock
    /// transport this file otherwise uses: `Asserter`'s FIFO queue is content-blind
    /// and resolves each request synchronously with no real async delay, so it
    /// can't demonstrate genuine overlap between concurrently in-flight fetches —
    /// see the settled decision this test proves directly instead.
    #[tokio::test(start_paused = true)]
    async fn flush_window_range_dispatches_concurrently_and_restores_ascending_order() {
        use alloy::network::Ethereum;
        use alloy::providers::{DynProvider, RootProvider};
        use alloy::rpc::types::Filter;
        use alloy::transports::TransportResult;
        use std::sync::Arc;
        use std::sync::Mutex;
        use std::sync::atomic::{AtomicUsize, Ordering};

        /// `root()` delegates to an unused stub provider — never queried, since
        /// this test's ring covers the whole flushed range, so `flush_window_range`
        /// never takes the self-heal `get_block_by_hash` walk. `get_logs` is
        /// overridden to inject a controllable async delay per call and record (a)
        /// peak concurrently in-flight calls and (b) the order calls complete in —
        /// proof of real concurrency and real out-of-order resolution, not just
        /// that the code compiles against a bound > 1.
        #[derive(Clone)]
        struct DelayedLogsProvider {
            stub_root: DynProvider<Ethereum>,
            call_order: Arc<AtomicUsize>,
            delays_by_call_order: Arc<Vec<Duration>>,
            in_flight: Arc<AtomicUsize>,
            peak_in_flight: Arc<AtomicUsize>,
            completion_order: Arc<Mutex<Vec<usize>>>,
        }

        #[async_trait::async_trait]
        impl Provider<Ethereum> for DelayedLogsProvider {
            fn root(&self) -> &RootProvider<Ethereum> {
                self.stub_root.root()
            }

            async fn get_logs(&self, _filter: &Filter) -> TransportResult<Vec<Log>> {
                let call_index = self.call_order.fetch_add(1, Ordering::SeqCst);
                let now_in_flight = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
                self.peak_in_flight
                    .fetch_max(now_in_flight, Ordering::SeqCst);
                let delay = self
                    .delays_by_call_order
                    .get(call_index)
                    .copied()
                    .unwrap_or(Duration::ZERO);
                tokio::time::sleep(delay).await;
                self.in_flight.fetch_sub(1, Ordering::SeqCst);
                self.completion_order.lock().unwrap().push(call_index);
                Ok(Vec::new())
            }
        }

        // Ring: blocks 300..=303 observed (clean chain — no self-heal walk needed).
        let mut linkage = HeadLinkage::new(8);
        for n in 300u64..=303 {
            assert_eq!(
                linkage.observe(n, hb(n as u8), hb((n - 1) as u8)),
                LinkOutcome::Linked
            );
        }
        let mut coalescer = HeadCoalescer::new(Some(300), WINDOW);
        coalescer.observe_head(303, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((300, 303)));

        // Dispatch order follows ascending block order (300, 301, 302, 303 — see
        // `pairs` construction in `flush_window_range`). Block 300's fetch, the
        // first dispatched, is deliberately the slowest: under the old sequential
        // dispatch nothing could ever finish before it. Under concurrent dispatch,
        // 301-303 must resolve first.
        let delays_by_call_order = vec![
            Duration::from_millis(500), // block 300
            Duration::from_millis(10),  // block 301
            Duration::from_millis(20),  // block 302
            Duration::from_millis(30),  // block 303
        ];
        let peak_in_flight = Arc::new(AtomicUsize::new(0));
        let completion_order = Arc::new(Mutex::new(Vec::new()));
        let unused_asserter = Asserter::new(); // never queried
        let double = DelayedLogsProvider {
            stub_root: ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(unused_asserter)
                .erased(),
            call_order: Arc::new(AtomicUsize::new(0)),
            delays_by_call_order: Arc::new(delays_by_call_order),
            in_flight: Arc::new(AtomicUsize::new(0)),
            peak_in_flight: peak_in_flight.clone(),
            completion_order: completion_order.clone(),
        };
        let reused = vec![double.erased()];

        // max_concurrency 4 (>= the 4 pending blocks) so every fetch launches at
        // once — the bound this test needs to prove overlap, unreachable through
        // `from_mock`'s pinned `max_concurrency: 1`.
        let settings = ProviderSettings::new(
            vec![HttpRpcSettings::new("http://unused".to_string(), 4, 100)],
            vec!["ws://127.0.0.1:9".parse().expect("static url")],
        );

        let ranges = flush_window_range(
            &mut linkage,
            &mut coalescer,
            &settings,
            None,
            0,
            Some(&reused),
        )
        .await
        .expect("flush must succeed");

        assert_eq!(
            peak_in_flight.load(Ordering::SeqCst),
            4,
            "all 4 blocks' fetches must be in flight together, not one at a time"
        );

        let observed_completion_order = completion_order.lock().unwrap().clone();
        assert_eq!(
            &observed_completion_order[0..3],
            &[1, 2, 3],
            "the 3 fast fetches (301-303) must resolve before the slow one (300), \
             got order {observed_completion_order:?}"
        );
        assert_eq!(
            observed_completion_order[3], 0,
            "the slow fetch (300) must resolve last, proving it did not block the \
             others, got order {observed_completion_order:?}"
        );

        // Despite resolving out of order above, the output `flush_window_range`
        // yields is still restored to ascending block-number order.
        assert_eq!(ranges.len(), 4);
        let numbers: Vec<u64> = ranges.iter().map(|&(from, _, _)| from).collect();
        assert_eq!(numbers, vec![300, 301, 302, 303]);
    }

    /// `flush_window_range`: peak in-flight dispatch tracks the *configured*
    /// `max_concurrency`, not a value hardcoded inside `flush_window_range`.
    /// The test above (4 pending blocks, `max_concurrency: 4`) can't
    /// distinguish "bound comes from configuration" from "bound is hardcoded
    /// to something >= 4" — a dispatch bound hardcoded to, say, 64 would still
    /// pass it. Here the pending-block count (6) and `max_concurrency` (2)
    /// differ, and peak in-flight must land on exactly 2 — not 4 (the other
    /// test's value), not 6 (dispatch-all), not 1 (a sequential fallback).
    #[tokio::test(start_paused = true)]
    async fn flush_window_range_peak_in_flight_tracks_configured_max_concurrency() {
        use alloy::network::Ethereum;
        use alloy::providers::{DynProvider, RootProvider};
        use alloy::rpc::types::Filter;
        use alloy::transports::TransportResult;
        use std::sync::Arc;
        use std::sync::atomic::{AtomicUsize, Ordering};

        /// Tracks only peak concurrently in-flight calls — this test needs no
        /// per-call delay staggering or completion-order proof, just the bound.
        #[derive(Clone)]
        struct PeakTrackingLogsProvider {
            stub_root: DynProvider<Ethereum>,
            in_flight: Arc<AtomicUsize>,
            peak_in_flight: Arc<AtomicUsize>,
        }

        #[async_trait::async_trait]
        impl Provider<Ethereum> for PeakTrackingLogsProvider {
            fn root(&self) -> &RootProvider<Ethereum> {
                self.stub_root.root()
            }

            async fn get_logs(&self, _filter: &Filter) -> TransportResult<Vec<Log>> {
                let now_in_flight = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
                self.peak_in_flight
                    .fetch_max(now_in_flight, Ordering::SeqCst);
                tokio::time::sleep(Duration::from_millis(20)).await;
                self.in_flight.fetch_sub(1, Ordering::SeqCst);
                Ok(Vec::new())
            }
        }

        // Ring: 6 pending blocks, clean chain, no self-heal walk needed.
        let mut linkage = HeadLinkage::new(8);
        for n in 400u64..=405 {
            assert_eq!(
                linkage.observe(n, hb(n as u8), hb((n - 1) as u8)),
                LinkOutcome::Linked
            );
        }
        let mut coalescer = HeadCoalescer::new(Some(400), WINDOW);
        coalescer.observe_head(405, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((400, 405)));

        let peak_in_flight = Arc::new(AtomicUsize::new(0));
        let unused_asserter = Asserter::new(); // never queried
        let double = PeakTrackingLogsProvider {
            stub_root: ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(unused_asserter)
                .erased(),
            in_flight: Arc::new(AtomicUsize::new(0)),
            peak_in_flight: peak_in_flight.clone(),
        };
        let reused = vec![double.erased()];

        // max_concurrency 2, strictly below the 6 pending blocks.
        let settings = ProviderSettings::new(
            vec![HttpRpcSettings::new("http://unused".to_string(), 2, 100)],
            vec!["ws://127.0.0.1:9".parse().expect("static url")],
        );

        let ranges = flush_window_range(
            &mut linkage,
            &mut coalescer,
            &settings,
            None,
            0,
            Some(&reused),
        )
        .await
        .expect("flush must succeed");

        assert_eq!(
            peak_in_flight.load(Ordering::SeqCst),
            2,
            "peak in-flight must equal the configured max_concurrency (2), not \
             the pending block count (6) or any other hardcoded bound"
        );
        assert_eq!(ranges.len(), 6);
    }
}