ethl 0.1.44

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, info, 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, 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;

/// Bound on the head await when no flush deadline is pending, so a provider
/// that accepts the WS subscription and then goes silent — half-open socket,
/// dead subscription that never closes the stream — can't park the session
/// forever. Not a staleness or liveness signal: it makes no claim about
/// whether the chain is progressing, only that this one subscription has
/// gone quiet long enough to be worth abandoning. Expiry rotates to the next
/// provider; it never records a provider error. A consecutive expiry on
/// EVERY configured provider — a full quiet rotation with no header seen in
/// between — surfaces `RpcError::AllProvidersIdle` instead of rotating
/// again; see `IdleRotationCounter`.
///
/// 120s so rotation is attempted before booda's per-chain lag CRITICAL pages
/// the operator (288s on ethereum, 500s on base): the alarm should report
/// exhaustion, not a condition nothing has tried yet to fix.
/// `IdleRotationCounter::expire` thresholds on the configured provider
/// count, so a two-provider chain reaches `AllProvidersIdle` at 240s.
/// Robinhood configures a single provider, so 120s is its entire budget
/// rather than half of one — accepted 2026-09-04 as the correct asymmetry,
/// not a bug to route around.
const IDLE_HEAD_AWAIT_SECS: u64 = 120;

/// Steps the reorg-ring top-up walks per flush (`flush_window_range`,
/// `HeadLinkage::extend_downward`). Bounded, not a config knob: the walk is a
/// sequential hash chain (one RPC per step) running inside
/// `flush_window_range`'s own `bound_flush_time` budget — 4 s on BSC — which
/// exists to keep this loop under alloy's 16-slot WS head buffer (~7.2 s to
/// overflow at ~0.45 s/block, `bsc.rs:83`). A one-shot walk to the full 96
/// depth is ~95 sequential round trips, ~42 ms each against public BSC RPC —
/// it would blow the 4 s bound on its own. `LIVE_ROUTER_COALESCE_WINDOW_MS`
/// is 0 in CDK, so a flush runs on essentially every head (~0.45–0.75 s
/// apart); at 8 steps/flush that consumes roughly 500 ms of headroom out of
/// the 4 s budget against the wall-clock realities the budget is sized for —
/// and the ring reaches full depth in ~12 blocks (5–9 s), about 8× faster
/// than the forward-observation rate it replaces.
///
/// Happy-path cost only. The walk shares the flush's aggregate bound, not a
/// per-step one, and each step can itself fan out over every configured HTTP
/// provider on a miss (`get_block_by_hash_with_http_fallback`) — worst case
/// `8 × N` RPCs, not 8, and per-call timeouts are configured larger than this
/// aggregate bound, so one hung call can individually exceed the whole
/// budget on its own. Accepted (settled decision 7): the top-up shares the
/// flush's bound rather than getting its own, and a timed-out top-up is
/// fail-closed like any other stalled flush (see `flush_window_range`'s
/// placement comment on `top_up_reorg_ring`).
const REORG_RING_TOP_UP_STEPS: u64 = 8;

/// Length bound on `flush_window_range`'s self-heal walk, in blocks.
///
/// That walk is sequential and fail-closed — one `get_block_by_hash` per block,
/// no by-number fallback — so its cost is linear in how far the flush cursor
/// trails the ring's floor, and nothing else caps it. `flush_max_total_secs`
/// defaults to `0` (disabled), so on any chain that has not opted in this is the
/// walk's only bound, and `MAX_BLOCKS_PER_WS_REQUEST` caps a single fetch batch
/// rather than this cursor-to-tip span.
///
/// Sized about an order of magnitude above the deepest recovery window in
/// service (96 on BSC and mainnet), because the trigger that fires at roughly
/// `recovery_window_depth + 1` blocks — a flush failing repeatedly while heads
/// keep arriving — needs room to accumulate several such cycles before the bound
/// closes on it; a bound at or under the window would fail closed on that
/// trigger's very first attempt, during ordinary degraded operation. At BSC's
/// ~0.45 s blocks, 1,024 blocks is ~7.5 minutes of missed flushing, and ~43 s of
/// walking at the ~42 ms per sequential step measured for
/// `REORG_RING_TOP_UP_STEPS`: far enough out that only a genuinely stuck stream
/// reaches it, near enough that reaching it does not itself become the outage.
/// Added to the configured window rather than maximized against it, so the bound
/// always carries this constant's full headroom above the backlog it must cover
/// — never merely equal to it, whatever depth a chain configures.
const SELF_HEAL_WALK_MAX_BLOCKS: u64 = 1_024;

#[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,
    },
    /// Heads were skipped: the arriving head sits above `ring_tip + 1`, so the
    /// ring cannot assert linkage over the missing blocks. Missed observation,
    /// not a fork — the caller reports the break in our own view and hands
    /// recovery to the consumer. Kept distinct from `NoRecordedParent` so the
    /// caller can log the ring tip and the gap it left.
    Gap { ring_tip: u64 },
    /// Nothing is recorded at the head's parent height — the parent sits below
    /// the ring floor, the ring never reached there, or the head is block 0 and
    /// has no parent at all. Absence of a record is not a claim about the
    /// chain: there was nothing here to disagree with.
    NoRecordedParent,
}

/// A ring of canonical `(number, hash)` taken from the head stream, as deep as
/// the configured recovery window, 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)
    }

    /// Lowest block number the ring may hold while honoring `depth`, given
    /// the current tip. `None` on an empty ring, mirroring `latest()`. Kept
    /// as a method rather than exposing `depth` itself (a private field) to
    /// callers that need to know how far the ring should reach — the
    /// top-up's stop condition: once `floor()` reaches this, the ring
    /// already spans `depth` and there is nothing left to walk.
    fn lowest_worth_recording(&self) -> Option<u64> {
        let (latest, _) = self.latest()?;
        Some(latest.saturating_sub(self.depth.saturating_sub(1)))
    }

    /// The recovery window this ring is configured to retain, in blocks.
    /// Distinct from `lowest_worth_recording`, which answers where the ring
    /// should reach given its current tip; this is the configured span itself,
    /// which the self-heal walk's length bound must clear.
    fn recovery_window_depth(&self) -> u64 {
        self.depth
    }

    /// 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();
        }
    }

    /// Extend the ring downward with `pairs` in ascending `(number, hash)`
    /// order — the shape a backward hash-chain walk yields once reversed.
    /// Not a `push`: these sit below the ring's current front, so appending
    /// would corrupt the ring's ordering and its floor.
    ///
    /// Pairs at or above the current floor are already recorded and are
    /// skipped — the normal case for a walk's own anchor pair. The kept
    /// pairs must be contiguous with the front and with each other: `hash_at`
    /// indexes by offset from the front, so a gap would silently return the
    /// wrong hash for every entry above it. Never carries the floor below
    /// `lowest_worth_recording`, so the ring never exceeds `depth`. A batch
    /// that fails either check records nothing. Inert when `!enabled()`.
    fn extend_downward(&mut self, pairs: &[(u64, B256)]) {
        if !self.enabled() {
            return;
        }
        let Some(front) = self.floor() else {
            return; // nothing to anchor a downward extension against
        };
        let deepest = self.lowest_worth_recording().unwrap_or(front);
        let below: Vec<(u64, B256)> = pairs
            .iter()
            .copied()
            .filter(|&(n, _)| (deepest..front).contains(&n))
            .collect();
        let Some(&(top, _)) = below.last() else {
            return; // nothing below the floor, or nothing within depth
        };
        let contiguous =
            top + 1 == front && below.windows(2).all(|pair| pair[1].0 == pair[0].0 + 1);
        if !contiguous {
            return;
        }
        for &(number, hash) in below.iter().rev() {
            self.ring.push_front((number, hash));
        }
    }

    /// 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 head the ring already records at this exact height is not new
        // information, wherever in the ring it sits. Tested before the ring is
        // consulted for the parent: a re-observed head whose parent has since
        // been evicted would otherwise read as an absent record and end the
        // stream over a block we already hold.
        if self.hash_at(number) == Some(hash) {
            return LinkOutcome::Linked;
        }

        // A forward gap (heads skipped ahead — e.g. across a reconnect) is
        // missed observation, not a fork. Its length is outage duration times
        // block rate and bears no relation to how deep two chains part, so a
        // backward walk sized from the ring's fork window would spend a
        // fork-sized budget on a question it was never asked and then report a
        // divergence it never observed. Report the break in our own view and
        // let the consumer re-anchor.
        if number > tip_number + 1 {
            return LinkOutcome::Gap {
                ring_tip: tip_number,
            };
        }

        // `number <= tip_number + 1`, not a duplicate, and the parent did not
        // extend the tip: a reorg, or a detachment that needs an RPC walk.
        let Some(parent_number) = number.checked_sub(1) else {
            return LinkOutcome::NoRecordedParent;
        };
        match self.hash_at(parent_number) {
            // Parent links to a known canonical block, and `number` is not the
            // block we recorded there: it was replaced on a chain sharing
            // canonical `parent_number` → that is the common ancestor.
            Some(canonical_parent) if canonical_parent == parent_hash => {
                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,
            },
            // Nothing recorded at the parent's height. `hash_at` returned
            // `None`, so no comparison happened here at all — a parent below
            // the floor and a ring that never reached there are the same
            // absence, and neither disagrees with anything.
            None => LinkOutcome::NoRecordedParent,
        }
    }
}

/// Sequentially asks the configured HTTP providers for `hash`, starting at
/// `start_index` and rotating exactly as `get_logs_with_http_fallback_retry`
/// does — never a concurrent fan-out. We trust our providers: a response
/// whose own header satisfies `is_match` IS the requested block, no second
/// source needed, no corroboration. Null, a transport error, and an identity
/// mismatch (wrong hash, or for `collect_window_hashes` also wrong number)
/// are the same thing — "this provider did not answer" — and simply advance
/// to the next one. `skip_first` excludes `start_index` itself from the
/// rotation, for callers that already queried it directly through their own
/// already-connected client.
///
/// Returns the answering provider's index alongside the header — the same
/// shape `get_logs_with_http_fallback_retry` returns — so a caller that steps
/// repeatedly can start its next call where this one left off, instead of
/// paying a full rotation from `start_index` again.
///
/// Exhausting every attempt without a match is a claim about our OWN view —
/// no provider answered — never about the chain; the caller decides what
/// that means. The returned error mirrors `fallback_http_exhausted_error`'s
/// shape: the hash and the hosts tried.
async fn get_block_by_hash_with_http_fallback(
    caller: &str,
    providers: &ProviderSettings,
    reused: Option<&[DynProvider<Ethereum>]>,
    start_index: usize,
    skip_first: bool,
    hash: B256,
    is_match: impl Fn(&Header) -> bool,
) -> Result<(usize, Box<Header>), RpcError> {
    let n = providers.http_providers.len();
    let mut attempted_hosts: Vec<String> = Vec::new();
    let mut last_err = String::new();

    for step in usize::from(skip_first)..n {
        let idx = fallback_http_index(start_index, step, n);
        let host = providers.http_settings(idx).host();
        let http_provider = resolve_http_provider(providers, reused, idx);
        let outcome = match http_provider.get_block_by_hash(hash).await {
            Ok(Some(block)) if is_match(&block.header) => return Ok((idx, Box::new(block.header))),
            Ok(Some(_)) => "answered a non-matching block".to_string(),
            Ok(None) => "has no block".to_string(),
            Err(e) => truncate_error_short(e, 100),
        };
        warn!("{caller}: provider {host} did not answer for block {hash}: {outcome}");
        last_err = outcome;
        attempted_hosts.push(host);
    }

    Err(RpcError::ConnectionError(if attempted_hosts.is_empty() {
        format!("{caller}: no provider left to try for block {hash}")
    } else {
        format!(
            "{caller}: no provider answered for block {hash} after trying [{}]: {}",
            attempted_hosts.join(", "),
            last_err
        )
    }))
}

/// `N ≥ 1`, and when a reused-client slice is supplied its length matches `N` —
/// the shared precondition both walk entry points enforce before any HTTP
/// resolution, the same fail-closed posture `flush_window_range`'s own
/// empty-set guard already uses. Production upholds `reused.len() == N` via
/// `build_reused_http_clients`; a mismatch, like `N == 0`, is a configuration
/// fault, not a walk outcome to keep walking past.
fn provider_set_preconditions(
    n: usize,
    reused: Option<&[DynProvider<Ethereum>]>,
) -> Option<RpcError> {
    if n == 0 {
        return Some(RpcError::ConnectionError(
            "no HTTP providers configured to resolve reorg-walk steps".to_string(),
        ));
    }
    if let Some(set) = reused
        && set.len() != n
    {
        return Some(RpcError::ConnectionError(format!(
            "reused HTTP client count {} does not match configured provider count {n}",
            set.len(),
        )));
    }
    None
}

/// 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.
///
/// Trust rule: a response whose hash AND number match what was asked is the
/// requested block, verified outright — no second source. Null, a transport
/// error, a wrong hash, or a wrong number are all "this provider did not
/// answer"; the primary's own miss is retried against the other configured
/// HTTP providers in sequence, and exhausting them yields `ConnectionError`
/// (a claim about our view, never the chain).
pub(crate) async fn collect_window_hashes(
    provider: &DynProvider<Ethereum>,
    providers: &ProviderSettings,
    reused: Option<&[DynProvider<Ethereum>]>,
    primary_index: usize,
    tip_hash: B256,
    tip_number: u64,
    from_block: u64,
) -> Result<Vec<(u64, B256)>, RpcError> {
    let n = providers.http_providers.len();
    if let Some(err) = provider_set_preconditions(n, reused) {
        return Err(err);
    }
    if tip_number < from_block {
        return Ok(Vec::new());
    }
    let primary_index = primary_index % n;
    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 hash = current_hash;
        let header = match provider.get_block_by_hash(hash).await {
            // Matching hash and number: trusted outright, no fallback, no
            // extra RPC.
            Ok(Some(block))
                if block.header.hash == hash && block.header.number == current_number =>
            {
                Box::new(block.header)
            }
            // Null, wrong hash, wrong number, or a transport error: the
            // primary did not answer — try the other configured providers.
            _ => {
                let (_, header) = get_block_by_hash_with_http_fallback(
                    "collect_window_hashes",
                    providers,
                    reused,
                    primary_index,
                    true,
                    hash,
                    |h| h.hash == hash && h.number == current_number,
                )
                .await?;
                header
            }
        };
        let parent_hash = header.parent_hash;
        let Some(parent_number) = current_number.checked_sub(1) else {
            // Walked down to block 0, which has no parent to continue from.
            // Running out of chain says nothing about a fork.
            return Err(RpcError::Gap);
        };
        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?;

        // One block per range, fetched by the hash that pins it: the canonical
        // identity of `to_block` is known here, so the range carries it.
        ranges.push(LogRange {
            from_block: number,
            to_block: number,
            logs,
            tip_hash: Some(hash),
        });
    }

    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.
///
/// `label` identifies the providers the flush was working against — the
/// helper is provider-agnostic and puts the caller's label straight into the
/// timeout message, ahead of the prose, so identity survives even a tight
/// downstream cap. It claims nothing about which of them stalled: nothing
/// observes an in-flight provider today.
async fn bound_flush_time<F, T>(max_secs: u64, label: &str, 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!(
            "{label}: flush exceeded its {max_secs}s aggregate timeout — provider \
             likely stalled mid-flush; switching provider"
        ))),
    }
}

/// The structured event name the live loop stamps on a `Gap`. Named here
/// rather than written inline at the raise site so a hermetic test can pin
/// it: a consumer's log-metric filter matches the literal string, so a
/// rename or typo would otherwise leave every gate green while the filter
/// silently stopped matching. Same reason `flush_failure_log_line` exists.
fn live_head_gap_event() -> &'static str {
    "live_head_gap"
}

/// The structured event name the live loop's reorg walk stamps on
/// `ReorgBeyondRing` — the one exit from that walk that is a claim about the
/// chain, not about a provider. The walk's other errors keep the walk's
/// existing free-text line; naming only this one is what lets a filter alarm
/// on a chain claim without also counting a stalled provider.
fn live_head_reorg_beyond_ring_event() -> &'static str {
    "live_head_reorg_beyond_ring"
}

/// Which structured event, if any, the reorg walk's dispatch site stamps on
/// `err`. `ReorgBeyondRing` is the walk's one exit that is a claim about the
/// chain; everything else — a spent budget, a transport fault reaching this
/// walk — keeps the walk's unnamed free-text line, so a filter on the named
/// event can never miscount a stalled provider as a chain claim.
///
/// Extracted, like `classify_flush_failure` and `flush_failure_log_line`, so
/// a hermetic test can assert the split directly: the live loop's own
/// dispatch arm has no hermetic seam (`TODO G.15`).
fn walk_failure_event(err: &RpcError) -> Option<&'static str> {
    match err {
        RpcError::ReorgBeyondRing => Some(live_head_reorg_beyond_ring_event()),
        _ => None,
    }
}

/// What the live loop must do with a failed by-hash flush.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FlushFailure {
    /// Yield the error and end the stream.
    EndStream,
    /// Retry the same window; the flush's HTTP fetch restarts at the same
    /// index, so this does not reach a different provider.
    SwitchProvider,
}

/// Separate a flush failure worth retrying from one no retry can fix.
///
/// A failed flush moves neither the cursor nor the pending span, so a retry
/// re-runs the identical window. For a transport fault that is the point —
/// retrying may still succeed once the provider tracker's backoff clears,
/// even though the retry's HTTP fetch restarts at the same index and never
/// reaches a different provider. `Gap` and
/// `ReorgBeyondRing` deny that premise: the answer is a property of the span
/// and the ring, not of who was asked, and a successful reconnect clears the
/// provider tracker's backoff on the way round, so nothing throttles the
/// repeat. Those claims belong to the consumer, the only party that can
/// re-anchor and resume.
///
/// `AllProvidersSuspended` and `AllProvidersIdle` belong with the transport
/// class, not the deny-the-premise one: both are self-clearing conditions, not
/// permanent claims about every provider. `ProviderErrorTracker`'s
/// `cleanup_old_errors`/`is_suspended` (`error_tracking.rs`) un-suspend a
/// provider once its errors age out of the sliding window, and
/// `AllProvidersIdle`'s own doc comment states it is retryable by contract —
/// a rotation-worthy transport signal, not a claim about the span or the ring.
///
/// Neither variant is constructed anywhere in `flush_window_range`'s call
/// graph today — the live generator raises both itself, at a
/// `yield Err(...); return;` site outside this dispatch, verified by tracing
/// every construction site. That is a fact about today's call graph, latent
/// rather than exercised; it says nothing about which `FlushFailure` would be
/// correct if a caller ever routed either through this classifier.
///
/// Exhaustive by design: a variant added later must be classified here rather
/// than inherit the retry default, which is how a claim came to be retried.
fn classify_flush_failure(err: &RpcError) -> FlushFailure {
    match err {
        RpcError::Gap | RpcError::ReorgBeyondRing | RpcError::Reorg { .. } => {
            FlushFailure::EndStream
        }
        RpcError::ConnectionError(_)
        | RpcError::SubscriptionError(_)
        | RpcError::LogFetchError(_)
        | RpcError::TransportError(_)
        | RpcError::CursorPastTip { .. }
        | RpcError::ProviderStalled { .. }
        | RpcError::AllProvidersSuspended(_)
        | RpcError::AllProvidersIdle { .. }
        | RpcError::Other(_) => FlushFailure::SwitchProvider,
    }
}

/// Builds the by-hash flush failure log line, bounded at the house limit
/// (`MAX_ERROR_MSG_LEN`) rather than the 100-byte cap other sites in this file
/// use for a foreign provider error string — `err`'s own text already carries
/// the provider identity `bound_flush_time` embedded, and a cap that fits
/// neither would cut it back off. Extracted so a hermetic test can assert on
/// the emitted line directly; this module has no log-capture dependency.
///
/// Naming the disposition is what makes the line diagnostic: the same error
/// text ends the stream or retries depending on `failure`, and an operator
/// reading only the error cannot tell which happened. Building it here also
/// leaves the dispatch site's match arms pure control flow — see the live
/// loop's `Err` arm for why that matters.
///
/// `SwitchProvider` says only "retrying the same window" — no claim about
/// which provider services the retry. The live loop's call site always
/// invokes `flush_window_range` with HTTP `provider_index` `0`, and
/// `get_logs_with_http_fallback_retry` restarts its fallback from that same
/// index on every call, so a retried flush's HTTP fetch never rotates, no
/// matter how many HTTP or WS providers are configured. What does rotate on
/// `SwitchProvider` is the live loop's WS session — a separate index this
/// flush never reads.
fn flush_failure_log_line(err: &RpcError, failure: FlushFailure) -> String {
    let disposition = match failure {
        FlushFailure::EndStream => "ending stream for consumer recovery",
        FlushFailure::SwitchProvider => "retrying the same window",
    };
    format!(
        "by-hash window flush failed: {} — {disposition}",
        truncate_error(err)
    )
}

/// Names the two providers a by-hash flush is working against: the WS stream's
/// own host, and the host its HTTP fallback starts from. Reads the fallback
/// through `first()` rather than `http_settings(0)`, whose modulo indexing
/// panics on an empty configured set — `flush_window_range`'s own guard is
/// what turns that configuration fault into a fail-closed error, and this
/// label is built before the flush can reach it.
fn flush_timeout_label(providers: &ProviderSettings, ws_host: &str) -> String {
    let fallback = providers
        .http_providers
        .first()
        .map(|settings| settings.host())
        .unwrap_or_else(|| "<none>".to_string());
    format!("ws={ws_host} fallback={fallback}")
}

/// 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 — a boot gap, where the ring
/// bootstraps at the first live head while the flush cursor still sits where
/// backfill left it, or a flush that has failed repeatedly while heads kept
/// arriving — walks backward from `linkage.latest()` to collect
/// canonical `(number, hash)` pairs, records them into the ring (`extend_downward`),
/// and uses them directly. Never falls back to a by-number fetch — fail-closed is
/// the invariant. On every flush, whether or not that self-heal walk ran, also
/// runs a small bounded top-up that walks the ring downward from its own floor
/// toward its configured recovery window depth — see `REORG_RING_TOP_UP_STEPS`
/// — so a ring left shallow by a stream restart deepens across the next
/// several flushes instead of staying wherever forward observation alone last
/// left it. 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
    // `poll_providers_for_block_number` applies before 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 {
        // Anchored at the live head (`linkage.latest()`), never a backfill
        // cursor: only the live head is contiguous with the head stream by
        // construction. A backfill cursor is several heights stale by the
        // time the WS subscription delivers its first head, so seeding from
        // it would make the very next head arrive as `Detached` — a healthy
        // chain killing its own stream on every boot.
        let Some((top_n, top_hash)) = linkage.latest() else {
            // Ring is completely empty — no anchor for the walk, and no record
            // anywhere to disagree with.
            return Err(RpcError::Gap);
        };
        // The walk is sequential and fail-closed, so its wall-clock cost grows
        // with the shortfall it is asked to cover. Refuse a span that would park
        // the live loop instead of healing it, and say only what that refusal
        // establishes: we did not reconstruct our own view, not that the chain
        // did anything.
        let walk_blocks = top_n.saturating_sub(from_block) + 1;
        let walk_bound = SELF_HEAL_WALK_MAX_BLOCKS.saturating_add(linkage.recovery_window_depth());
        if walk_blocks > walk_bound {
            warn!(
                "flush window [{}, {}] would need a {}-block self-heal walk from ring anchor \
                 {} — beyond the {}-block bound; ending the flush rather than walking it",
                from_block, to_block, walk_blocks, top_n, walk_bound
            );
            return Err(RpcError::Gap);
        }
        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,
            providers,
            reused,
            provider_index,
            top_hash,
            top_n,
            from_block,
        )
        .await?;
        // Reverse to ascending (from_block → top_n). Record the walk's pairs into
        // the ring before restricting to this flush's own window — they are
        // already fetched, and this is the only place the ring's floor ever
        // moves down from its live-head anchor toward the flush cursor, at
        // zero added RPC cost (the top-up below pays for everything deeper).
        // `extend_downward` discards the walk's own anchor pair (already
        // recorded at the ring's back, `top_n`/`top_hash` above) and anything
        // that would break the ring's contiguity.
        pairs_desc.reverse();
        linkage.extend_downward(&pairs_desc);
        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?;
        // Same by-hash rule as `fetch_logs_by_hash_range`: this range is one
        // block, pinned to the hash it was fetched with.
        Ok::<LogRange, RpcError>(LogRange {
            from_block: n,
            to_block: n,
            logs,
            tip_hash: Some(hash),
        })
    };
    let ranges = sort_ranges_ascending(fetch_concurrently(pairs, bound, fetch).await?);

    // Bounded top-up (settled decision 6: runs on every flush, not only when
    // the self-heal branch above fires). Placed here — after the batch
    // fetch, before the cursor advance — deliberately, and this call site's
    // position is the one thing in this function you must not move:
    // `bound_flush_time` (at this function's own call sites) is a
    // `tokio::time::timeout` that drops this whole future on expiry, and
    // `coalescer` is a `&mut` borrow that outlives the drop while the local
    // `ranges` does not. A top-up placed after `advance_cursor` would, on
    // timeout, leave the cursor moved past blocks whose logs were never
    // yielded — the silent gap the fail-closed invariant above forbids. Here
    // a timeout instead discards `ranges` with the cursor untouched,
    // identical to today's stalled-provider behavior. Pinned by
    // `flush_window_range_timeout_mid_top_up_drops_cursor_advance`.
    top_up_reorg_ring(linkage, providers, provider_index, reused).await;

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

/// Extend `linkage`'s ring downward toward its recovery window floor by up to
/// `REORG_RING_TOP_UP_STEPS` blocks. Errors are logged and swallowed
/// (settled decision 8) — the caller's own ranges are already fetched and
/// unaffected; the ring simply regrows at the pre-change rate on the next
/// flush. See `flush_window_range`'s call site for why this call's position
/// relative to the cursor advance is load-bearing, not stylistic.
async fn top_up_reorg_ring(
    linkage: &mut HeadLinkage,
    providers: &ProviderSettings,
    provider_index: usize,
    reused: Option<&[DynProvider<Ethereum>]>,
) {
    let (Some(floor), Some(target)) = (linkage.floor(), linkage.lowest_worth_recording()) else {
        return;
    };
    if floor <= target {
        return;
    }
    let floor_hash = linkage
        .hash_at(floor)
        .expect("the ring's own floor is always recorded at its own number");
    let walk_from = target.max(floor.saturating_sub(REORG_RING_TOP_UP_STEPS));
    let walk_provider = resolve_http_provider(providers, reused, provider_index);
    match collect_window_hashes(
        &walk_provider,
        providers,
        reused,
        provider_index,
        floor_hash,
        floor,
        walk_from,
    )
    .await
    {
        Ok(mut pairs_desc) => {
            pairs_desc.reverse();
            linkage.extend_downward(&pairs_desc);
            info!(
                "reorg ring top-up: floor {floor} -> {} (target {target})",
                linkage.floor().unwrap_or(floor)
            );
        }
        Err(err) => {
            // Degraded, not broken: the caller's own ranges are already
            // fetched and unaffected — the ring simply regrows at the
            // pre-change rate. A provider outage severe enough to matter
            // already surfaces through the live loop's own by-hash flush
            // failure `warn!` (fired when `flush_window_range` itself times
            // out or errors), so this stays at debug! rather than paging
            // someone for a top-up alone.
            debug!("reorg ring top-up failed, will retry next flush: {err}");
        }
    }
}

/// 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.
///
/// Written as a plain `fn` returning a named `impl Future + Send` rather than an
/// `async fn`, so its `Send`-ness is declared here instead of left to inference —
/// see the `#[allow]` below for why that distinction matters.
#[allow(clippy::manual_async_fn)]
// This must stay a named `-> impl Future + Send` and NOT collapse back to `async
// fn`: an `async fn`'s returned future is opaque, so whether it is `Send` is left
// to rustc's inference at each call site. A caller that borrows locals into a
// spawned stream needs that future to be `Send` for ANY lifetime the `fetch`
// closure borrows over — a higher-ranked obligation inference cannot discharge
// through an opaque type. That failure surfaces as an "implementation of `Send`
// is not general enough" error at the *consumer's* call site, not here, so
// clippy's own suggestion to simplify this to `async fn` reintroduces the bug
// invisibly to every gate in this crate.
fn fetch_concurrently<In, Out, F, Fut>(
    items: Vec<In>,
    bound: usize,
    fetch: F,
) -> impl Future<Output = Result<Vec<Out>, RpcError>> + Send
where
    In: Send,
    Out: Send,
    F: Fn(In) -> Fut + Send,
    Fut: Future<Output = Result<Out, RpcError>> + Send,
{
    async move {
        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(|range| range.from_block);
    ranges
}

/// Walk the new chain backward by hash until a block re-links to the recorded
/// canonical ring, returning the common ancestor's block number.
///
/// Invoked only on `LinkOutcome::Detached`, when the provider delivered a new-tip
/// head without re-emitting the intermediate new-chain blocks. Bounded to `depth`
/// sequential `get_block_by_hash` fetches against `provider` (the ring's own
/// span), plus at most `N - 1` more per step when a step has to fall back to
/// the other configured HTTP providers.
///
/// Only one exit here claims the chain: a fetched hash compared against a
/// recorded canonical hash at the ring's floor, and they differed
/// (`ReorgBeyondRing`). Every other exit — a spent budget, block 0, a height
/// the ring never recorded — reports `Gap`, which says only that we
/// could not establish what happened. The pure ring is untouched; only this
/// glue does I/O.
///
/// The walk is a PATH, not a search: every block has one parent, so each step
/// is determined, and hash identity is what makes it sound to take consecutive
/// steps from different providers — a block that hashes to the requested hash
/// is that block regardless of who served it, so its `parent_hash` is the same
/// either way. Which provider answers can change whether we SEE the next step,
/// never which step is next.
///
/// `provider` is the WS-style session that detected the detachment. We trust
/// our providers: its answer is either the requested block, verified outright,
/// or it is not an answer (`Ok(None)`, a transport error, or a hash mismatch)
/// and the step falls back to the `N` configured HTTP providers in sequence —
/// `provider` itself contributes no answer; it is the suspect, and its index
/// is a WS-endpoint position that does not map onto the HTTP ring.
async fn walk_to_common_ancestor(
    provider: &DynProvider<Ethereum>,
    providers: &ProviderSettings,
    reused: Option<&[DynProvider<Ethereum>]>,
    linkage: &HeadLinkage,
    detached_parent: B256,
    child_number: u64,
    depth: u64,
) -> Result<u64, RpcError> {
    let n = providers.http_providers.len();
    if let Some(err) = provider_set_preconditions(n, reused) {
        return Err(err);
    }

    // `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 {
        // The detached head is block 0 and has no parent to walk from.
        return Err(RpcError::Gap);
    };
    let floor = linkage.floor().unwrap_or(u64::MAX);
    // `detached_parent` is the incoming push's own claim about its parent,
    // never a header we fetched and verified. Only from the first successful
    // step onward does `parent_hash` carry a hash-confirmed value, and only
    // then may a disagreement at the floor be reported as a fact about the
    // chain rather than about the provider that told us.
    let mut confirmed_a_step = false;
    // Carries which HTTP provider last answered, so a stalling provider costs
    // this walk one timeout rather than one per step. Seeded at 0 since the
    // first step has no prior answer to inherit; left alone on a step the WS
    // `provider` answers directly, since there is no fallback outcome to
    // advance onto.
    let mut http_cursor = 0;

    for _ in 0..depth {
        if linkage.matches(parent_number, parent_hash) {
            return Ok(parent_number);
        }
        if parent_number <= floor {
            // At the floor the ring does hold a record and `matches` just
            // rejected it, so the disagreement is real — provided the hash that
            // disagreed was one we fetched. Below the floor, and on the first
            // iteration, nothing was compared that we can stand behind.
            return if parent_number == floor && confirmed_a_step {
                Err(RpcError::ReorgBeyondRing)
            } else {
                Err(RpcError::Gap)
            };
        }
        let hash = parent_hash;
        let header = match provider.get_block_by_hash(hash).await {
            // Matching hash: trusted outright, no fallback, no extra RPC.
            Ok(Some(block)) if block.header.hash == hash => Box::new(block.header),
            // Null, wrong hash, or a transport error: the WS session did not
            // answer — try the configured HTTP providers in sequence, starting
            // where the previous step's fallback left off.
            _ => {
                let (answered_index, header) = get_block_by_hash_with_http_fallback(
                    "walk_to_common_ancestor",
                    providers,
                    reused,
                    http_cursor,
                    false,
                    hash,
                    |h| h.hash == hash,
                )
                .await?;
                http_cursor = answered_index;
                header
            }
        };
        parent_hash = header.parent_hash;
        confirmed_a_step = true;
        let Some(next) = parent_number.checked_sub(1) else {
            // Walked down to block 0 without re-linking.
            return Err(RpcError::Gap);
        };
        parent_number = next;
    }
    // The budget ran out with the ring intact and no disagreement observed:
    // a statement about how far we were willing to look, not about the chain.
    Err(RpcError::Gap)
}

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> {
    assert!(
        !providers.http_providers.is_empty(),
        "resolve_http_provider: no HTTP providers configured"
    );
    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
    }
}

/// The head await's three distinguishable outcomes, once bounded by
/// [`IDLE_HEAD_AWAIT_SECS`].
enum IdleHeadAwait {
    /// A header arrived within the bound.
    Header(Box<Header>),
    /// The stream ended (`None` from the underlying stream).
    StreamEnded,
    /// No header arrived before the bound elapsed.
    Expired,
}

/// Awaits the next head, bounded by [`IDLE_HEAD_AWAIT_SECS`] — the testable
/// seam for the `else` arm of the head-await select, which otherwise has no
/// way to reach a hermetic test without a live WS connection.
async fn await_head_within_idle_bound<S>(s: &mut S) -> IdleHeadAwait
where
    S: Stream<Item = Header> + Unpin,
{
    match tokio::time::timeout(Duration::from_secs(IDLE_HEAD_AWAIT_SECS), s.next()).await {
        Ok(Some(header)) => IdleHeadAwait::Header(Box::new(header)),
        Ok(None) => IdleHeadAwait::StreamEnded,
        Err(_) => IdleHeadAwait::Expired,
    }
}

/// Counts consecutive `IdleHeadAwait::Expired` events across provider
/// rotations, so a full quiet rotation — every configured WS provider tried
/// in turn, none delivering a head — can be told apart from one provider's
/// ordinary silence. Any received header resets the count, including one
/// below the cursor: it is evidence the chain is not quiet, regardless of
/// what the coalescer does with it.
struct IdleRotationCounter {
    consecutive_expiries: usize,
}

/// Decision returned by `IdleRotationCounter::expire` for one idle-bound
/// expiry.
enum IdleRotation {
    /// Fewer than `threshold` consecutive expiries — keep rotating.
    Continue,
    /// `threshold` consecutive expiries with no header in between.
    AllIdle,
}

impl IdleRotationCounter {
    fn new() -> Self {
        Self {
            consecutive_expiries: 0,
        }
    }

    fn reset(&mut self) {
        self.consecutive_expiries = 0;
    }

    /// Called once per idle-bound expiry, before rotating to the next
    /// provider. `threshold` is the number of configured WS providers — a
    /// single-provider deployment reaches `AllIdle` on its first expiry,
    /// converting what would otherwise be a silent self-reconnect loop into
    /// a visible, consumer-policied signal.
    fn expire(&mut self, threshold: usize) -> IdleRotation {
        self.consecutive_expiries += 1;
        if self.consecutive_expiries >= threshold.max(1) {
            IdleRotation::AllIdle
        } else {
            IdleRotation::Continue
        }
    }
}

async fn stream_heads_provider(
    provider: &DynProvider<Ethereum>,
    provider_host: &str,
) -> impl Stream<Item = Header> + Send {
    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`. The returned
/// stream is `Send` and may be spawned onto an executor (e.g. `tokio::spawn`).
pub async fn stream_heads_with_logs_with_progress(
    providers: &ProviderSettings,
    filter: Option<&Filter>,
    progress: Option<&ScanProgress>,
) -> impl Stream<Item = Result<LogRange, RpcError>> + Send {
    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. `recovery_window_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.recovery_window_depth);
        let mut gate = TipGate::new();
        // Scoped to the whole stream, not one provider session: a full
        // quiet rotation is defined across provider switches.
        let mut idle_rotation = IdleRotationCounter::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 — a boot
                        // gap, or a flush that has failed repeatedly while heads kept
                        // arriving — `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() {
                            // The label's fallback host tracks the `provider_index`
                            // the flush is called with below — `0`, the first
                            // configured provider. Naming both providers makes a
                            // timeout attributable without asserting which stalled.
                            let flush_timeout_label =
                                flush_timeout_label(providers, &provider_host);
                            match bound_flush_time(
                                providers.flush_max_total_secs,
                                &flush_timeout_label,
                                flush_window_range(
                                    &mut linkage,
                                    &mut coalescer,
                                    providers,
                                    filter,
                                    0,
                                    Some(&reused_http_clients),
                                ),
                            )
                            .await
                            {
                                Ok(ranges) => {
                                    for range in ranges {
                                        provider_trackers[provider_index].record_success();
                                        yield Ok(range);
                                    }
                                }
                                // Both arms are bare control flow on purpose: the
                                // generator reaches its provider through
                                // `connect_ws`, which needs a live WebSocket, so
                                // nothing hermetic can drive this dispatch and
                                // prove the two arms are not transposed.
                                // Everything testable — the classification and
                                // the log line — is therefore kept out of the
                                // arms, leaving the untestable remainder as small
                                // as it can be. Extracting a seam is `TODO G.15`.
                                Err(e) => {
                                    let failure = classify_flush_failure(&e);
                                    warn!("{}", flush_failure_log_line(&e, failure));
                                    match failure {
                                        FlushFailure::EndStream => {
                                            yield Err(e);
                                            return;
                                        }
                                        FlushFailure::SwitchProvider => {
                                            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 {
                                    // The by-number path: this range was fetched by block
                                    // number and never learned its top block's hash.
                                    yield Ok(LogRange {
                                        from_block,
                                        to_block: watermark,
                                        logs,
                                        tip_hash: None,
                                    });
                                    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(LogRange {
                                    from_block,
                                    to_block,
                                    logs,
                                    tip_hash: None,
                                });
                                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,
                                            0,
                                            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
                                                {
                                                    // By-number HTTP fallback: no
                                                    // canonical hash was learned.
                                                    yield Ok(LogRange {
                                                        from_block,
                                                        to_block: watermark,
                                                        logs,
                                                        tip_hash: None,
                                                    });
                                                    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(LogRange {
                                                    from_block,
                                                    to_block,
                                                    logs,
                                                    tip_hash: None,
                                                });
                                                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 {
                    match await_head_within_idle_bound(&mut s).await {
                        IdleHeadAwait::Header(header) => Some(*header),
                        IdleHeadAwait::StreamEnded => None,
                        IdleHeadAwait::Expired => match idle_rotation.expire(num_providers) {
                            IdleRotation::Continue => {
                                warn!(
                                    "No head from {} in {}s - idle bound expired, rotating provider",
                                    provider_trackers[provider_index].identifier(),
                                    IDLE_HEAD_AWAIT_SECS
                                );
                                should_switch_provider = true;
                                break;
                            }
                            IdleRotation::AllIdle => {
                                warn!(
                                    "No head from any of {} configured provider(s) within {}s - a \
                                     full quiet rotation",
                                    num_providers, IDLE_HEAD_AWAIT_SECS
                                );
                                yield Err(RpcError::AllProvidersIdle {
                                    providers_tried: num_providers,
                                    idle_secs: IDLE_HEAD_AWAIT_SECS,
                                });
                                return;
                            }
                        },
                    }
                };

                match maybe_header {
                    Some(header) => {
                        idle_rotation.reset();
                        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::Gap { ring_tip } => {
                                // The event a consumer's log-metric filter keys
                                // on. Its three fields are the observed facts
                                // the error deliberately does not carry, read
                                // from the ring's own tip and the head that
                                // arrived.
                                warn!(
                                    event = live_head_gap_event(),
                                    ring_tip,
                                    head = head_number,
                                    gap = head_number.saturating_sub(ring_tip),
                                    "live heads skipped ahead of the recovery ring — ending stream for consumer recovery"
                                );
                                yield Err(RpcError::Gap);
                                return;
                            }
                            LinkOutcome::Detached {
                                parent_hash,
                                child_number,
                            } => {
                                // The parent is inside the ring and differs, so
                                // there is fork evidence to resolve. Only an
                                // `Ok` here names an ancestor; the walk's own
                                // errors already distinguish what they observed
                                // from what they merely failed to establish.
                                match walk_to_common_ancestor(
                                    &provider,
                                    providers,
                                    Some(&reused_http_clients),
                                    &linkage,
                                    parent_hash,
                                    child_number,
                                    providers.recovery_window_depth,
                                )
                                .await
                                {
                                    Ok(ancestor) => {
                                        warn!(
                                            "reorg at head {} (walked new chain): common ancestor {} — ending stream",
                                            head_number, ancestor
                                        );
                                        yield Err(RpcError::Reorg { common_ancestor: ancestor });
                                        return;
                                    }
                                    Err(err) => {
                                        match walk_failure_event(&err) {
                                            Some(event_name) => warn!(
                                                event = event_name,
                                                head = head_number,
                                                "reorg walk at head {head_number} found a canonical hash disagreement beyond the recovery ring — ending stream for consumer recovery"
                                            ),
                                            None => warn!(
                                                "reorg walk at head {} ended the stream: {}",
                                                head_number,
                                                truncate_error(&err)
                                            ),
                                        }
                                        yield Err(err);
                                        return;
                                    }
                                }
                            }
                            LinkOutcome::NoRecordedParent => {
                                warn!(
                                    "head {} has no canonical record at its parent's height — ending stream for consumer recovery",
                                    head_number
                                );
                                yield Err(RpcError::Gap);
                                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;
        }
    }
}

/// The returned stream is `Send` and may be spawned onto an executor (e.g. `tokio::spawn`).
pub async fn stream_heads_with_logs(
    providers: &ProviderSettings,
    filter: Option<&Filter>,
) -> impl Stream<Item = Result<LogRange, RpcError>> + Send {
    stream_heads_with_logs_with_progress(providers, filter, None).await
}

/// Round-robin `providers` for a block number, failing over to the next
/// configured provider on any error rather than ending the call. `fetch`
/// extracts the target block number from a connected provider; its
/// `Err(String)` covers both a transport failure and a provider that
/// answered without one but produced no usable result — either counts as
/// that provider's failed attempt and rotates. `empty_set_phrase` and
/// `attempt_phrase` name what's being fetched, in the empty-provider-set
/// guard and in the per-attempt/exhaustion messages respectively.
///
/// Eleven attempts (index 0..=10), `index % num_providers` rotation,
/// progressive backoff `(5 * index).min(50)` seconds between attempts:
/// `last_block`'s ladder, shared here so every caller fails over the same
/// way.
async fn poll_providers_for_block_number<F, Fut>(
    providers: &ProviderSettings,
    empty_set_phrase: &str,
    attempt_phrase: &str,
    fetch: F,
) -> Result<u64, RpcError>
where
    F: Fn(DynProvider<Ethereum>) -> Fut,
    Fut: Future<Output = Result<u64, String>>,
{
    let num_providers = providers.http_providers.len();
    if num_providers == 0 {
        return Err(RpcError::ConnectionError(format!(
            "No HTTP providers configured for {empty_set_phrase}"
        )));
    }

    let mut index: usize = 0;
    loop {
        // Round-robin provider selection so the fetch can fail over.
        let provider_index = index % num_providers;
        let provider = providers.connect_http(provider_index);
        match fetch(provider).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 {attempt_phrase} after {} attempts from HTTP provider {}: {}",
                        index, provider_host, e
                    )));
                }
                index += 1;
                warn!(
                    "Failed to fetch {attempt_phrase} 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;
            }
        }
    }
}

pub async fn last_block(providers: &ProviderSettings) -> Result<u64, RpcError> {
    poll_providers_for_block_number(
        providers,
        "latest block fetch",
        "latest block number",
        |provider| async move { provider.get_block_number().await.map_err(|e| e.to_string()) },
    )
    .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, failing over across every
///   configured provider; `Err` only once every attempt is spent.
///
/// 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 finalized = poll_providers_for_block_number(
                settings,
                "finalized block fetch",
                "finalized block number",
                |provider| async move {
                    let result = provider
                        .get_block_by_number(BlockNumberOrTag::Finalized)
                        .await
                        .map_err(|e| e.to_string())?;
                    result.map(|block| block.header.number).ok_or_else(|| {
                        "Provider returned no finalized block; cannot determine safe capture target. \
                         Ensure the provider supports the `finalized` tag (mainnet only)."
                            .to_string()
                    })
                },
            )
            .await?;
            if let Some(message) =
                CAPTURE_POLICY_NOTICE.first_message(settings.capture_target, finalized)
            {
                warn!("{message}");
            }
            Ok(finalized)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        CapturePolicyNotice, EmptyFlushOutcome, FlushFailure, HeadCoalescer, HeadLinkage,
        IDLE_HEAD_AWAIT_SECS, IdleHeadAwait, IdleRotation, IdleRotationCounter, LinkOutcome,
        MAX_BLOCKS_PER_WS_REQUEST, REORG_RING_TOP_UP_STEPS, RpcError, SELF_HEAL_WALK_MAX_BLOCKS,
        TipGate, YieldBounds, await_head_within_idle_bound, bound_flush_time,
        classify_flush_failure, collect_window_hashes, fallback_http_exhausted_error,
        fallback_http_index, fetch_concurrently, fetch_logs_by_hash_range, flush_failure_log_line,
        flush_timeout_label, flush_window_range, handle_empty_flush, live_head_gap_event,
        live_head_reorg_beyond_ring_event, resolve_http_provider, sort_ranges_ascending,
        walk_failure_event, walk_to_common_ancestor, yield_bounds,
    };
    use crate::rpc::LogRange;
    use crate::rpc::config::{CaptureTarget, HttpRpcSettings, ProviderSettings};
    use alloy::network::Ethereum;
    use alloy::primitives::B256;
    use alloy::providers::{DynProvider, Provider, ProviderBuilder};
    use alloy::rpc::types::{Header, 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`, own `hash`, and `parentHash`.
    ///
    /// Use this for tests that exercise `collect_window_hashes`, which trusts
    /// a response only when both `block.header.number == current_number` and
    /// `block.header.hash` match the hash actually queried at that walk step
    /// — every `push_success` site must supply the hash the walk queries, not
    /// a placeholder.
    fn block_with_parent_at(number: u64, hash: B256, parent: B256) -> serde_json::Value {
        let zero = hb(0).to_string();
        serde_json::json!({
            "hash":             hash.to_string(),
            "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 meaningful fields are its own `hash` and
    /// `parentHash`; the block number is not checked by `walk_to_common_ancestor`,
    /// so it is always 0. `hash` must be the hash the walk step actually
    /// queried — a mismatch is trusted by no one and falls back to the next
    /// provider. Use `block_with_parent_at` when `collect_window_hashes` is
    /// under test (it also cross-checks the block number).
    fn block_with_parent(hash: B256, parent: B256) -> serde_json::Value {
        block_with_parent_at(0, hash, parent)
    }

    /// `ProviderSettings` configured with `n` HTTP providers at unreachable,
    /// never-dialed hosts — every sequential-fallback test supplies its own
    /// `reused` client set (`Some`), so `resolve_http_provider` never falls
    /// through to `connect_http` against these URLs. `n == 0` builds an empty
    /// set, for the precondition-guard tests.
    fn n_http_provider_settings(n: usize) -> ProviderSettings {
        let http_providers = (0..n)
            .map(|i| HttpRpcSettings::new(format!("http://unused-{i}"), 1, 100))
            .collect();
        ProviderSettings::new(
            http_providers,
            vec!["ws://127.0.0.1:9".parse().expect("static url")],
        )
    }

    /// One mocked HTTP `DynProvider` backed by `asserter`, the same
    /// construction every existing mock-provider test already uses.
    fn mock_http_provider(asserter: Asserter) -> DynProvider<Ethereum> {
        ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased()
    }

    /// 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:?}"),
        }
    }

    /// The ring holds no entry at the parent's height, so `hash_at` returned
    /// `None` and nothing was compared. Absence of a record is not a divergence,
    /// however far below the floor the parent sits.
    #[test]
    fn head_linkage_parent_below_floor_has_no_recorded_parent() {
        // 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::NoRecordedParent
        );
    }

    // ── I.4f §6.1: `lowest_worth_recording` / `extend_downward`, pure ────────

    #[test]
    fn lowest_worth_recording_is_none_on_an_empty_ring() {
        let linkage = HeadLinkage::new(8);
        assert_eq!(linkage.lowest_worth_recording(), None);
    }

    #[test]
    fn lowest_worth_recording_tracks_latest_at_depth() {
        let linkage = seed_chain(8, 100, 105);
        // latest = 105, depth = 8 → 105 - 7 = 98.
        assert_eq!(linkage.lowest_worth_recording(), Some(98));

        let deep = seed_chain(1, 100, 105);
        // depth = 1 → the accessor equals latest() itself.
        assert_eq!(deep.lowest_worth_recording(), Some(105));
    }

    #[test]
    fn extend_downward_records_pairs_below_the_floor() {
        let mut linkage = seed_chain(8, 103, 105);
        assert_eq!(linkage.floor(), Some(103));

        linkage.extend_downward(&[(101, hb(101)), (102, hb(102))]);

        assert_eq!(linkage.floor(), Some(101));
        assert_eq!(linkage.hash_at(101), Some(hb(101)));
        assert_eq!(linkage.hash_at(102), Some(hb(102)));
        // Existing entries and the tip are untouched.
        assert_eq!(linkage.hash_at(103), Some(hb(103)));
        assert_eq!(linkage.latest(), Some((105, hb(105))));
    }

    #[test]
    fn extend_downward_ignores_pairs_at_or_above_the_floor() {
        let mut linkage = seed_chain(8, 103, 105);
        let before = linkage.clone();
        // 103 (the floor itself) and 104 (already recorded, wrong hash here)
        // must both be dropped without duplicating or overwriting anything.
        linkage.extend_downward(&[(103, hb(103)), (104, hb(200))]);

        assert_eq!(
            linkage.floor(),
            before.floor(),
            "no-op: nothing below the floor"
        );
        assert_eq!(
            linkage.hash_at(104),
            Some(hb(104)),
            "an at-or-above pair must never overwrite an existing entry"
        );
        assert_eq!(linkage.ring.len(), before.ring.len());
    }

    #[test]
    fn extend_downward_batch_not_contiguous_with_front_records_nothing() {
        let mut linkage = seed_chain(8, 103, 105);
        // 100 is below the floor but not adjacent to it (103's predecessor is
        // 102, not 100) — a gap the ring cannot link across.
        linkage.extend_downward(&[(100, hb(100))]);
        assert_eq!(
            linkage.floor(),
            Some(103),
            "a non-contiguous batch is discarded whole"
        );
        assert_eq!(linkage.hash_at(100), None);
    }

    #[test]
    fn extend_downward_internal_gap_records_nothing() {
        let mut linkage = seed_chain(8, 103, 105);
        // Adjacent to the front (102 == 103 - 1), but 100 and 102 are not
        // adjacent to each other — a gap inside the batch itself.
        linkage.extend_downward(&[(100, hb(100)), (102, hb(102))]);
        assert_eq!(
            linkage.floor(),
            Some(103),
            "an internal gap is discarded whole"
        );
        assert_eq!(linkage.hash_at(102), None);
    }

    #[test]
    fn extend_downward_never_exceeds_depth() {
        // Ring already spans the full depth (3 entries for depth 3).
        let mut linkage = seed_chain(3, 103, 105);
        linkage.extend_downward(&[(100, hb(100)), (101, hb(101)), (102, hb(102))]);
        assert_eq!(
            linkage.floor(),
            Some(103),
            "a full ring must not extend further"
        );
        assert_eq!(linkage.ring.len(), 3);
    }

    #[test]
    fn extend_downward_caps_at_depth_even_when_pairs_go_deeper() {
        // Ring holds 3 of a 5-deep budget — room for exactly 2 more entries.
        let mut linkage = seed_chain(5, 103, 105);
        linkage.extend_downward(&[(99, hb(99)), (100, hb(100)), (101, hb(101)), (102, hb(102))]);
        assert_eq!(
            linkage.floor(),
            Some(101),
            "must stop exactly at depth, not walk past it"
        );
        assert_eq!(
            linkage.hash_at(100),
            None,
            "beyond depth must not be recorded"
        );
        assert_eq!(linkage.ring.len(), 5);
    }

    #[test]
    fn extend_downward_and_lowest_worth_recording_are_inert_at_depth_zero() {
        let mut linkage = HeadLinkage::new(0);
        assert_eq!(linkage.observe(100, hb(100), hb(99)), LinkOutcome::Linked);
        assert_eq!(linkage.lowest_worth_recording(), None);
        linkage.extend_downward(&[(99, hb(99))]);
        assert_eq!(
            linkage.floor(),
            None,
            "disabled linkage records nothing, ever"
        );
    }

    #[tokio::test]
    async fn walk_to_common_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(203), hb(202)));
        asserter.push_success(&block_with_parent(hb(202), hb(101)));
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);

        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(203), 104, 8).await;
        assert!(matches!(err, Ok(101)), "got {err:?}");
    }

    /// A fetched hash reached the ring's floor and disagreed with the hash
    /// recorded there. That comparison is the one thing this walk can establish
    /// about the chain, so it is the one place `ReorgBeyondRing` is raised.
    #[tokio::test]
    async fn walk_to_common_ancestor_fetched_hash_at_floor_is_reorg_beyond_ring() {
        // 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(203), hb(202)));
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);

        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(203), 104, 2).await;
        assert!(matches!(err, Err(RpcError::ReorgBeyondRing)), "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_reports_the_break_in_our_own_view() {
        // `number > tip + 1`: heads skipped ahead (e.g. across a reconnect).
        // The gap is missed observation, so it names the ring tip it broke from
        // rather than routing to the reorg walk, and the ring is left untouched
        // — no clear, no re-anchor, nothing fetched.
        let mut linkage = seed_chain(8, 100, 103); // tip = 103
        assert_eq!(
            linkage.observe(107, hb(107), hb(106)),
            LinkOutcome::Gap { ring_tip: 103 }
        );
        // 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: walk_to_common_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 walk_to_common_ancestor_walk_depth_1() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(200), hb(109))); // 110'→canonical 109
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);
        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(200), 111, 15).await;
        assert!(matches!(err, Ok(109)), "got {err:?}");
    }

    #[tokio::test]
    async fn walk_to_common_ancestor_walk_depth_2() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(200), hb(201))); // 110'→109'
        asserter.push_success(&block_with_parent(hb(201), hb(108))); // 109'→canonical 108
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);
        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(200), 111, 15).await;
        assert!(matches!(err, Ok(108)), "got {err:?}");
    }

    #[tokio::test]
    async fn walk_to_common_ancestor_walk_depth_3() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(200), hb(201))); // 110'→109'
        asserter.push_success(&block_with_parent(hb(201), hb(202))); // 109'→108'
        asserter.push_success(&block_with_parent(hb(202), hb(107))); // 108'→canonical 107
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);
        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(200), 111, 15).await;
        assert!(matches!(err, Ok(107)), "got {err:?}");
    }

    #[tokio::test]
    async fn walk_to_common_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(200), hb(201)));
        asserter.push_success(&block_with_parent(hb(201), hb(202)));
        asserter.push_success(&block_with_parent(hb(202), hb(203)));
        asserter.push_success(&block_with_parent(hb(203), hb(204)));
        asserter.push_success(&block_with_parent(hb(204), hb(205)));
        asserter.push_success(&block_with_parent(hb(205), hb(206)));
        asserter.push_success(&block_with_parent(hb(206), hb(103))); // final→canonical 103
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);
        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(200), 111, 15).await;
        assert!(matches!(err, Ok(103)), "got {err:?}");
    }

    #[tokio::test]
    async fn walk_to_common_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(200), hb(201))); // 104'→103'
        asserter.push_success(&block_with_parent(hb(201), hb(202))); // 103'→102'
        asserter.push_success(&block_with_parent(hb(202), hb(203))); // 102'→101'
        asserter.push_success(&block_with_parent(hb(203), hb(100))); // 101'→canonical 100
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);
        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(200), 105, 5).await;
        assert!(matches!(err, Ok(100)), "got {err:?}");
    }

    // ── G.8 §6: a gap never reaches the walk ─────────────────────────────────

    /// The change this task turns on, asserted at the seam that exists.
    ///
    /// A forward gap leaves `observe` as `LinkOutcome::Gap`, which the live
    /// loop reports as `RpcError::Gap` — it no longer produces the `Detached`
    /// that routes to `walk_to_common_ancestor`. The arm keys on
    /// `number > tip_number + 1` alone, so a 1,300-block gap costs exactly
    /// what a 2-block gap costs: nothing. Reverting the arm to `Detached`
    /// fails this outright.
    #[test]
    fn forward_gap_never_reaches_the_walk_whatever_its_length() {
        for gap in [2u64, 1_300] {
            // Canonical ring [100, 103], depth 8; a head arrives `gap` above the tip.
            let mut linkage = seed_chain(8, 100, 103);
            let head = 103 + gap;
            assert_eq!(
                linkage.observe(head, hb(207), hb(206)),
                LinkOutcome::Gap { ring_tip: 103 },
                "a gap of {gap} must report the break in our own view, never route \
                 to the reorg walk"
            );
            assert_eq!(
                linkage.latest(),
                Some((103, hb(103))),
                "a gap of {gap} must not mutate the ring"
            );
        }
    }

    /// The other half of the same proof: the walk still costs what it always
    /// cost — the per-block fetches the gap path no longer pays. Asserted by
    /// exhausting an `Asserter` primed with exactly the walk's expected steps,
    /// so a walk that fetched fewer or more would fail.
    #[tokio::test]
    async fn walk_still_spends_a_fetch_per_step_when_a_divergence_reaches_it() {
        // Canonical ring [100, 103], depth 8. A differing parent inside the
        // ring — the one arm that still reaches the walk.
        let mut linkage = seed_chain(8, 100, 103);
        assert_eq!(
            linkage.observe(104, hb(204), hb(203)),
            LinkOutcome::Detached {
                parent_hash: hb(203),
                child_number: 104,
            }
        );

        // Three steps down the new chain: 103'→102'→101'→ canonical 100.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(203), hb(202)));
        asserter.push_success(&block_with_parent(hb(202), hb(201)));
        asserter.push_success(&block_with_parent(hb(201), hb(100)));
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);

        let ancestor =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(203), 104, 8).await;
        assert!(matches!(ancestor, Ok(100)), "got {ancestor:?}");
    }

    /// A head the ring already records, re-observed. Neither an error nor a
    /// chain claim: the ring holds this exact `(number, hash)`, so nothing
    /// changed. A one-entry ring is the shape a restart leaves behind, where
    /// `hash_at(number - 1)` is `None` and the duplicate is the only thing
    /// standing between an identical block and a fabricated verdict.
    #[test]
    fn re_observing_the_only_recorded_head_is_not_a_divergence() {
        let mut linkage = HeadLinkage::new(96);
        assert_eq!(linkage.observe(500, hb(50), hb(49)), LinkOutcome::Linked);
        assert_eq!(linkage.floor(), Some(500));
        assert_eq!(
            linkage.hash_at(499),
            None,
            "the parent's height is genuinely unrecorded — that is the shape \
             this test exists for"
        );

        assert_eq!(
            linkage.observe(500, hb(50), hb(49)),
            LinkOutcome::Linked,
            "the same block at the same height is not new information"
        );
        assert_eq!(linkage.latest(), Some((500, hb(50))));
        assert_eq!(linkage.ring.len(), 1, "a duplicate must not grow the ring");
    }

    /// A differing parent exactly at the floor of a one-entry ring — the
    /// bootstrap shape, where `floor == tip` — reaches the walk's floor check on
    /// iteration zero, having fetched nothing. `parent_hash` is still
    /// `detached_parent`, the value the push supplied, so the only disagreement
    /// is with a provider's own word: `Gap`, never a chain claim.
    #[tokio::test]
    async fn walk_at_a_one_entry_rings_floor_claims_nothing_before_a_fetch() {
        let mut linkage = HeadLinkage::new(96);
        assert_eq!(linkage.observe(500, hb(50), hb(49)), LinkOutcome::Linked);
        assert_eq!(linkage.floor(), linkage.latest().map(|(n, _)| n));

        // A head at 501 whose parent hash disagrees with the recorded 500.
        assert_eq!(
            linkage.observe(501, hb(151), hb(150)),
            LinkOutcome::Detached {
                parent_hash: hb(150),
                child_number: 501,
            }
        );

        // No pushes: reaching a fetch at all would fail here.
        let provider = mock_http_provider(Asserter::new());
        let settings = n_http_provider_settings(1);
        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(150), 501, 96).await;
        assert!(
            matches!(err, Err(RpcError::Gap)),
            "an unfetched disagreement at the floor must not claim a divergence; got {err:?}"
        );
    }

    /// The split the walk draws at its floor check, both halves against one
    /// ring. At the floor the ring holds a record, so a fetched hash that
    /// disagrees with it is a fact about the chain. Below the floor nothing was
    /// ever recorded, so there is nothing to disagree with and the same check
    /// may claim nothing.
    #[tokio::test]
    async fn walk_at_the_floor_diverges_but_below_it_only_discontinues() {
        // Ring [102, 103] (depth 2); floor = 102.
        let linkage = seed_chain(2, 100, 103);
        assert_eq!(linkage.floor(), Some(102));
        let settings = n_http_provider_settings(1);

        // At the floor: child 104 detaches at 103' (hb(203)); one confirmed step
        // puts a fetched hb(202) against the recorded canonical 102.
        let at_floor = Asserter::new();
        at_floor.push_success(&block_with_parent(hb(203), hb(202)));
        let err = walk_to_common_ancestor(
            &mock_http_provider(at_floor),
            &settings,
            None,
            &linkage,
            hb(203),
            104,
            8,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::ReorgBeyondRing)),
            "a fetched hash disagreeing with a recorded one at the floor is a \
             chain claim; got {err:?}"
        );

        // Below the floor: child 102 puts the first check at 101, a height the
        // ring never recorded. No pushes — reaching a fetch would fail here.
        let err = walk_to_common_ancestor(
            &mock_http_provider(Asserter::new()),
            &settings,
            None,
            &linkage,
            hb(202),
            102,
            8,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::Gap)),
            "a height the ring never recorded cannot diverge from it; got {err:?}"
        );
    }

    /// The budget ran out with the ring intact and no disagreement observed:
    /// a statement about how far we were willing to look, never about the chain.
    #[tokio::test]
    async fn walk_to_common_ancestor_loop_exhausted_is_a_gap() {
        // 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(200), hb(201)));
        asserter.push_success(&block_with_parent(hb(201), hb(202)));
        asserter.push_success(&block_with_parent(hb(202), hb(203))); // loop ends after this fetch
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);
        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(200), 111, 3).await;
        assert!(matches!(err, Err(RpcError::Gap)), "got {err:?}");
    }

    /// The floor check firing before any fetch: `detached_parent` is still the
    /// incoming push's own claim, so the disagreement is with a provider's word,
    /// not with a hash we confirmed. That is a gap, not a chain claim.
    #[tokio::test]
    async fn walk_to_common_ancestor_parent_at_floor_before_a_fetch_is_a_gap() {
        // `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 = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);
        // child=101 → parent_number=100 = floor; first iteration hits the floor check.
        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(200), 101, 10).await;
        assert!(matches!(err, Err(RpcError::Gap)), "got {err:?}");
    }

    /// A single configured provider whose primary AND fallback both answer
    /// `Ok(None)` has simply not answered — exhaustion is a view claim, never
    /// a chain claim, so this is `ConnectionError`, not `ReorgBeyondRing`.
    #[tokio::test]
    async fn walk_to_common_ancestor_single_provider_absent_is_unresolved() {
        let linkage = seed_chain(15, 100, 109);
        let ws_asserter = Asserter::new();
        ws_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(ws_asserter);
        let settings = n_http_provider_settings(1);
        let http_asserter = Asserter::new();
        http_asserter.push_success(&serde_json::json!(null));
        let reused = vec![mock_http_provider(http_asserter)];
        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(200),
            111,
            15,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::ConnectionError(_))),
            "got {err:?}"
        );
    }

    /// A mid-walk transport error is recoverable (the stream can restart and
    /// re-attempt) — same `ConnectionError` verdict as the null case above,
    /// mechanically unchanged by the trust rule.
    #[tokio::test]
    async fn walk_to_common_ancestor_transport_error_is_unresolved() {
        let linkage = seed_chain(15, 100, 109);
        let settings = n_http_provider_settings(1);
        let http_asserter = Asserter::new();
        http_asserter.push_failure_msg("transport blip");
        let reused = vec![mock_http_provider(http_asserter)];
        let asserter = Asserter::new();
        asserter.push_failure_msg("transport blip");
        let provider = mock_http_provider(asserter);
        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(200),
            111,
            15,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::ConnectionError(_))),
            "got {err:?}"
        );
    }

    // ── walk_to_common_ancestor: sequential fallback matrix ──────────────────
    // The WS primary contributes no answer of its own; a would-be-terminal
    // primary answer falls back to the configured HTTP providers, one at a
    // time, starting from the walk's own cursor (0 on the first step, then
    // wherever the previous step's fallback left off).

    /// THE REVERT-FAILS TEST FOR THE INCIDENT: a single provider's `Ok(None)`
    /// must not be conclusive when another configured HTTP provider serves
    /// the queried block. Fails against pre-trust-rule code that returns
    /// a walk verdict on the primary's first null without trying the other
    /// providers.
    #[tokio::test]
    async fn walk_to_common_ancestor_second_source_recovers_the_incident_scenario() {
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        ws_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(ws_asserter);
        let settings = n_http_provider_settings(3);

        let voter0 = Asserter::new();
        voter0.push_success(&serde_json::json!(null));
        let voter1 = Asserter::new();
        voter1.push_success(&block_with_parent(hb(203), hb(102))); // recovers → parent 102 ∈ ring
        let voter2 = Asserter::new();
        voter2.push_success(&serde_json::json!(null));
        let reused = vec![
            mock_http_provider(voter0),
            mock_http_provider(voter1),
            mock_http_provider(voter2),
        ];

        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(203),
            104,
            8,
        )
        .await;
        assert!(matches!(err, Ok(102)), "got {err:?}");
    }

    /// N=3, all three HTTP providers null: every provider tried, none
    /// answered — a view claim, `ConnectionError`, never `ReorgBeyondRing`.
    #[tokio::test]
    async fn walk_to_common_ancestor_all_sources_absent_is_unresolved() {
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        ws_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(ws_asserter);
        let settings = n_http_provider_settings(3);
        let reused: Vec<_> = (0..3)
            .map(|_| {
                let a = Asserter::new();
                a.push_success(&serde_json::json!(null));
                mock_http_provider(a)
            })
            .collect();

        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(203),
            104,
            8,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::ConnectionError(_))),
            "got {err:?}"
        );
    }

    /// N=3, 1 null + 2 transport-fail: the incident's actual weather — every
    /// provider still gets its own turn, and none answers.
    #[tokio::test]
    async fn walk_to_common_ancestor_lone_absent_answer_is_unresolved() {
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        ws_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(ws_asserter);
        let settings = n_http_provider_settings(3);
        let voter0 = Asserter::new();
        voter0.push_success(&serde_json::json!(null));
        let voter1 = Asserter::new();
        voter1.push_failure_msg("blip");
        let voter2 = Asserter::new();
        voter2.push_failure_msg("blip");
        let reused = vec![
            mock_http_provider(voter0),
            mock_http_provider(voter1),
            mock_http_provider(voter2),
        ];

        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(203),
            104,
            8,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::ConnectionError(_))),
            "got {err:?}"
        );
    }

    /// N=3, 2 null + 1 transport-fail — a mixed reason for not answering is
    /// still "did not answer"; the walk still tries all three and reports
    /// `ConnectionError`.
    #[tokio::test]
    async fn walk_to_common_ancestor_two_of_three_absent_is_unresolved() {
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        ws_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(ws_asserter);
        let settings = n_http_provider_settings(3);
        let voter0 = Asserter::new();
        voter0.push_success(&serde_json::json!(null));
        let voter1 = Asserter::new();
        voter1.push_success(&serde_json::json!(null));
        let voter2 = Asserter::new();
        voter2.push_failure_msg("blip");
        let reused = vec![
            mock_http_provider(voter0),
            mock_http_provider(voter1),
            mock_http_provider(voter2),
        ];

        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(203),
            104,
            8,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::ConnectionError(_))),
            "got {err:?}"
        );
    }

    /// N=2: 1 null + 1 transport-fail, and separately both null — either way
    /// every configured provider gets tried and none answers.
    #[tokio::test]
    async fn walk_to_common_ancestor_n_equals_2() {
        let linkage = seed_chain(8, 100, 103);
        let settings = n_http_provider_settings(2);

        {
            let ws_asserter = Asserter::new();
            ws_asserter.push_success(&serde_json::json!(null));
            let provider = mock_http_provider(ws_asserter);
            let voter0 = Asserter::new();
            voter0.push_success(&serde_json::json!(null));
            let voter1 = Asserter::new();
            voter1.push_failure_msg("blip");
            let reused = vec![mock_http_provider(voter0), mock_http_provider(voter1)];
            let err = walk_to_common_ancestor(
                &provider,
                &settings,
                Some(&reused),
                &linkage,
                hb(203),
                104,
                8,
            )
            .await;
            assert!(
                matches!(err, Err(RpcError::ConnectionError(_))),
                "got {err:?}"
            );
        }
        {
            let ws_asserter = Asserter::new();
            ws_asserter.push_success(&serde_json::json!(null));
            let provider = mock_http_provider(ws_asserter);
            let voter0 = Asserter::new();
            voter0.push_success(&serde_json::json!(null));
            let voter1 = Asserter::new();
            voter1.push_success(&serde_json::json!(null));
            let reused = vec![mock_http_provider(voter0), mock_http_provider(voter1)];
            let err = walk_to_common_ancestor(
                &provider,
                &settings,
                Some(&reused),
                &linkage,
                hb(203),
                104,
                8,
            )
            .await;
            assert!(
                matches!(err, Err(RpcError::ConnectionError(_))),
                "got {err:?}"
            );
        }
    }

    /// A primary transport failure is "did not answer," same as a null: one
    /// HTTP provider serving the block still recovers the walk; all
    /// providers transport-failing yields `ConnectionError`.
    #[tokio::test]
    async fn walk_to_common_ancestor_primary_transport_fail() {
        let linkage = seed_chain(8, 100, 103);
        let settings = n_http_provider_settings(3);

        {
            let ws_asserter = Asserter::new();
            ws_asserter.push_failure_msg("blip");
            let provider = mock_http_provider(ws_asserter);
            let voter0 = Asserter::new();
            voter0.push_success(&serde_json::json!(null));
            let voter1 = Asserter::new();
            voter1.push_success(&block_with_parent(hb(203), hb(102)));
            let voter2 = Asserter::new();
            voter2.push_success(&serde_json::json!(null));
            let reused = vec![
                mock_http_provider(voter0),
                mock_http_provider(voter1),
                mock_http_provider(voter2),
            ];
            let err = walk_to_common_ancestor(
                &provider,
                &settings,
                Some(&reused),
                &linkage,
                hb(203),
                104,
                8,
            )
            .await;
            assert!(matches!(err, Ok(102)), "got {err:?}");
        }
        {
            let ws_asserter = Asserter::new();
            ws_asserter.push_failure_msg("blip");
            let provider = mock_http_provider(ws_asserter);
            let reused: Vec<_> = (0..3)
                .map(|_| {
                    let a = Asserter::new();
                    a.push_failure_msg("blip");
                    mock_http_provider(a)
                })
                .collect();
            let err = walk_to_common_ancestor(
                &provider,
                &settings,
                Some(&reused),
                &linkage,
                hb(203),
                104,
                8,
            )
            .await;
            assert!(
                matches!(err, Err(RpcError::ConnectionError(_))),
                "got {err:?}"
            );
        }
    }

    /// Two recovered steps then a clean ring re-link, proving the walk's
    /// cursor carries forward instead of restarting at index 0 every step:
    /// step 1 starts at index 0, misses, and recovers at index 1; step 2
    /// starts from index 1 — not index 0 again — misses there, and recovers
    /// at index 2. Every `Asserter` is given EXACTLY the responses the
    /// sequential loop consumes.
    #[tokio::test]
    async fn walk_to_common_ancestor_two_recoveries_then_ring_relink() {
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        ws_asserter.push_success(&serde_json::json!(null)); // step 1 trigger
        ws_asserter.push_success(&serde_json::json!(null)); // step 2 trigger
        let provider = mock_http_provider(ws_asserter.clone());
        let settings = n_http_provider_settings(3);

        // Index 0 misses step 1; the cursor then skips it on step 2, so its
        // second queued response is never read — asserted below.
        let voter0 = Asserter::new();
        voter0.push_success(&serde_json::json!(null)); // step 1: miss
        voter0.push_success(&serde_json::json!(null)); // step 2: never read under the cursor
        // Index 1 recovers step 1 (hash hb(205)) with parent hb(204). The
        // cursor asks it first on step 2, where it misses.
        let voter1 = Asserter::new();
        voter1.push_success(&block_with_parent(hb(205), hb(204)));
        voter1.push_success(&serde_json::json!(null)); // step 2: miss
        // Index 2 is never reached in step 1 (index 1 already matched); it
        // recovers step 2 (hash hb(204)) with parent hb(102) ∈ ring.
        let voter2 = Asserter::new();
        voter2.push_success(&block_with_parent(hb(204), hb(102)));
        let reused = vec![
            mock_http_provider(voter0.clone()),
            mock_http_provider(voter1.clone()),
            mock_http_provider(voter2.clone()),
        ];

        // Detached at child=105 (parent_number starts at 104, one above the
        // linked ring): 104 → recovered to 103' (hb(204)) → recovered to 102 ∈ ring.
        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(205),
            105,
            8,
        )
        .await;
        assert!(matches!(err, Ok(102)), "got {err:?}");
        assert_eq!(
            ws_asserter.read_q().len(),
            0,
            "exactly 2 primary triggers, no third"
        );
        assert_eq!(
            voter0.read_q().len(),
            1,
            "cursor carries to index 1 on step 2; index 0's queued response is unread"
        );
        assert_eq!(voter1.read_q().len(), 0, "index 1 tried on both steps");
        assert_eq!(
            voter2.read_q().len(),
            0,
            "index 2 reached only once, on step 2"
        );
    }

    /// A step the WS `provider` answers directly, with no HTTP fallback call
    /// at all, must not reset the cursor a later step inherits. Step 1's
    /// fallback recovers at index 1; step 2 is answered by the WS provider
    /// directly; step 3's fallback must still start from index 1, not index
    /// 0, proving the WS-answered step in between left the cursor alone.
    #[tokio::test]
    async fn walk_to_common_ancestor_ws_answered_step_leaves_cursor_undisturbed() {
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        ws_asserter.push_success(&serde_json::json!(null)); // step 1 trigger
        ws_asserter.push_success(&block_with_parent(hb(205), hb(204))); // step 2: WS answers directly
        ws_asserter.push_success(&serde_json::json!(null)); // step 3 trigger
        let provider = mock_http_provider(ws_asserter.clone());
        let settings = n_http_provider_settings(3);

        // Index 0 misses step 1; if the cursor were reset instead of carried
        // past step 2, step 3 would ask it again — its second response would
        // then be consumed. Asserted unread below.
        let voter0 = Asserter::new();
        voter0.push_success(&serde_json::json!(null)); // step 1: miss
        voter0.push_success(&serde_json::json!(null)); // never read if the cursor survives
        // Index 1 recovers step 1 (hash hb(206)) with parent hb(205); step 2
        // never reaches it (WS answered directly). The cursor asks it first
        // on step 3, where it misses.
        let voter1 = Asserter::new();
        voter1.push_success(&block_with_parent(hb(206), hb(205)));
        voter1.push_success(&serde_json::json!(null)); // step 3: miss
        // Index 2 recovers step 3 (hash hb(204)) with parent hb(102) ∈ ring.
        let voter2 = Asserter::new();
        voter2.push_success(&block_with_parent(hb(204), hb(102)));
        let reused = vec![
            mock_http_provider(voter0.clone()),
            mock_http_provider(voter1.clone()),
            mock_http_provider(voter2.clone()),
        ];

        // Detached at child=106 (parent_number starts at 105): 105 →
        // recovered to 104' (hb(205)) → WS-answered to 103' (hb(204)) →
        // recovered to 102 ∈ ring.
        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(206),
            106,
            8,
        )
        .await;
        assert!(matches!(err, Ok(102)), "got {err:?}");
        assert_eq!(
            ws_asserter.read_q().len(),
            0,
            "exactly 3 primary triggers, no fourth"
        );
        assert_eq!(
            voter0.read_q().len(),
            1,
            "cursor still at index 1 on step 3; index 0's queued response is unread"
        );
        assert_eq!(voter1.read_q().len(), 0, "index 1 tried on steps 1 and 3");
        assert_eq!(
            voter2.read_q().len(),
            0,
            "index 2 reached only once, on step 3"
        );
    }

    /// A source answering `Ok(Some)` with the WRONG hash is neither a match
    /// nor evidence of anything about the chain — it is simply "did not
    /// answer," the same as a null or a transport error. This test asserts
    /// each in its own composition:
    ///
    /// - *the failure text proves every provider was tried* — 1 wrong hash +
    ///   1 null + 1 transport error, N=3: the returned error carries the
    ///   transport error's own text, which it can only do if the wrong-hash
    ///   answer was rejected and the walk moved on to ask the rest.
    /// - *not a match* — 1 wrong hash + 2 null. Accepting the wrong block
    ///   sends the walk down that block's parent pointer instead of trying
    ///   the other providers, so dropping the identity check inverts this
    ///   composition to `Ok(_)` instead of exhaustion.
    #[tokio::test]
    async fn walk_to_common_ancestor_wrong_block_is_not_a_recovery() {
        // Composition 1: wrong-identity must not short-circuit as a match.
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        ws_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(ws_asserter);
        let settings = n_http_provider_settings(3);
        let voter0 = Asserter::new();
        // Answers Ok(Some) but with a DIFFERENT hash than queried (hb(203)) —
        // must not be treated as a match.
        voter0.push_success(&block_with_parent(hb(9), hb(1)));
        let voter1 = Asserter::new();
        voter1.push_success(&serde_json::json!(null));
        let voter2 = Asserter::new();
        voter2.push_failure_msg("blip");
        let reused = vec![
            mock_http_provider(voter0),
            mock_http_provider(voter1),
            mock_http_provider(voter2),
        ];

        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(203),
            104,
            8,
        )
        .await;
        let Err(RpcError::ConnectionError(msg)) = err else {
            panic!("got {err:?}");
        };
        // The transport error's own text reaches the returned error — without it
        // a rate limit, a TLS reset, and a connect timeout are indistinguishable
        // to whoever reads this message.
        assert!(msg.contains("blip"), "provider error text dropped: {msg}");

        // Composition 2: wrong-identity must not be accepted as the block —
        // every provider tried, none answered, `ConnectionError`.
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        ws_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(ws_asserter);
        let voter0 = Asserter::new();
        // Its parent hb(102) sits in the ring — the bait for a dropped check:
        // accepting this block would send the walk to Ok(102) instead of
        // exhaustion.
        voter0.push_success(&block_with_parent(hb(77), hb(102)));
        let voter1 = Asserter::new();
        voter1.push_success(&serde_json::json!(null));
        let voter2 = Asserter::new();
        voter2.push_success(&serde_json::json!(null));
        let reused = vec![
            mock_http_provider(voter0),
            mock_http_provider(voter1),
            mock_http_provider(voter2),
        ];

        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(203),
            104,
            8,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::ConnectionError(_))),
            "got {err:?}"
        );
    }

    /// The PRIMARY's own hash-identity check, which nothing else exercises:
    /// every other fixture hands the primary an honest block. Hash identity is
    /// what makes it sound to take consecutive walk steps from different
    /// providers, so losing it silently is the worst failure this module has.
    ///
    /// The primary answers `Ok(Some)` with a block whose own hash is NOT the
    /// one queried, but whose `parent_hash` IS in the ring. Drop the check and
    /// the walk accepts the wrong block, follows its parent pointer, and
    /// reports `Ok(102)` — a fabricated rollback target the consumer would act
    /// on. Keep it and the step falls back to the one configured HTTP
    /// provider, which reports null, and the walk fails closed with
    /// `ConnectionError`.
    #[tokio::test]
    async fn walk_to_common_ancestor_primary_wrong_hash_does_not_advance_the_walk() {
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        // Queried for hb(203); answers with a block whose hash is hb(77).
        // Its parent hb(102) sits in the ring — the bait for a dropped check.
        ws_asserter.push_success(&block_with_parent(hb(77), hb(102)));
        let provider = mock_http_provider(ws_asserter);
        let settings = n_http_provider_settings(1);
        let source = Asserter::new();
        source.push_success(&serde_json::json!(null));
        let reused = vec![mock_http_provider(source)];

        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(203),
            104,
            8,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::ConnectionError(_))),
            "a wrong-hash primary block must never advance the walk; got {err:?}"
        );
    }

    /// A primary transport error is "did not answer," exactly like a null —
    /// exhaustion is a view claim, never a chain claim, so this stays
    /// `ConnectionError` even when every configured provider also misses.
    #[tokio::test]
    async fn walk_to_common_ancestor_primary_transport_error_with_absent_quorum_is_unresolved() {
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        ws_asserter.push_failure_msg("primary socket died");
        let provider = mock_http_provider(ws_asserter);
        let settings = n_http_provider_settings(3);
        let voter0 = Asserter::new();
        voter0.push_success(&serde_json::json!(null));
        let voter1 = Asserter::new();
        voter1.push_success(&serde_json::json!(null));
        let voter2 = Asserter::new();
        voter2.push_failure_msg("blip");
        let reused = vec![
            mock_http_provider(voter0),
            mock_http_provider(voter1),
            mock_http_provider(voter2),
        ];

        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(203),
            104,
            8,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::ConnectionError(_))),
            "got {err:?}"
        );
    }

    /// The `reused.len() != n` precondition, and why it is a correctness guard
    /// rather than tidiness: `resolve_http_provider` indexes the reused set
    /// modulo its length, so a set SHORTER than `n` silently aliases several
    /// ring positions onto one client — fewer independent fallback sources
    /// than configured, masquerading as `n` of them.
    ///
    /// The fixture makes that concrete: one client queued with 3 null
    /// answers against `n = 3`. The guard fails closed on the mismatch.
    /// Remove it and the sequential loop queries that one client under all
    /// three indices, silently substituting one host's opinion for three.
    #[tokio::test]
    async fn walk_to_common_ancestor_reused_length_mismatch_is_unresolved() {
        let linkage = seed_chain(8, 100, 103);
        let ws_asserter = Asserter::new();
        ws_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(ws_asserter);
        let settings = n_http_provider_settings(3);
        let lone = Asserter::new();
        lone.push_success(&serde_json::json!(null));
        lone.push_success(&serde_json::json!(null));
        lone.push_success(&serde_json::json!(null));
        let reused = vec![mock_http_provider(lone)]; // 1 client, n = 3

        let err = walk_to_common_ancestor(
            &provider,
            &settings,
            Some(&reused),
            &linkage,
            hb(203),
            104,
            8,
        )
        .await;
        assert!(
            matches!(err, Err(RpcError::ConnectionError(_))),
            "a short reused set must fail closed, never let one client vote \
             three times; got {err:?}"
        );
    }

    /// `N == 0` fails closed before any fetch, no panic (no HTTP providers
    /// configured is a configuration fault, not a walk outcome).
    #[tokio::test]
    async fn walk_to_common_ancestor_zero_http_providers_is_unresolved() {
        let linkage = seed_chain(8, 100, 103);
        let asserter = Asserter::new(); // no pushes needed — the guard fires first
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(0);

        let err =
            walk_to_common_ancestor(&provider, &settings, None, &linkage, hb(203), 104, 8).await;
        assert!(
            matches!(err, 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(103), hb(102))); // block 103 → parent 102
        walk_asserter.push_success(&block_with_parent_at(102, hb(102), hb(101))); // block 102 → parent 101
        walk_asserter.push_success(&block_with_parent_at(101, hb(101), hb(100))); // block 101 → parent 100
        let walk_provider = mock_http_provider(walk_asserter);
        let walk_settings = n_http_provider_settings(1);

        let pairs_desc =
            collect_window_hashes(&walk_provider, &walk_settings, None, 0, 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, range) in ranges.iter().enumerate() {
            let block_n = 100u64 + i as u64;
            assert_eq!(
                range.from_block, block_n,
                "from_block mismatch at index {i}"
            );
            assert_eq!(range.to_block, block_n, "to_block mismatch at index {i}");
            assert_eq!(range.logs.len(), 1, "expected one log at block {block_n}");
            assert_eq!(
                range.logs[0].block_number,
                Some(block_n),
                "log block number mismatch"
            );
            assert_eq!(
                range.tip_hash,
                Some(hb(block_n as u8)),
                "a by-hash range carries the hash it was fetched with"
            );
        }
    }

    /// Both by-hash producers stamp the range with the hash they fetched with.
    ///
    /// The live flush and the restart tail's `fetch_logs_by_hash_range` each
    /// yield one block per range, pinned by `at_block_hash`, so the canonical
    /// identity of the range's top block is known at construction. A production
    /// site reverted to `tip_hash: None` fails here. The `None` half of the rule
    /// lives with the by-number path that owns it, in `backfill`'s own tests.
    #[tokio::test]
    async fn by_hash_producers_stamp_the_range_with_its_canonical_tip_hash() {
        // The restart tail: one range per requested (number, hash) pair.
        let pairs = [(100u64, hb(100)), (101u64, hb(101))];
        let asserter = Asserter::new();
        for _ in 0..pairs.len() {
            asserter.push_success(&serde_json::json!([]));
        }
        let tail_settings = ProviderSettings::from_mock(mock_http_provider(asserter));
        let ranges = fetch_logs_by_hash_range(&tail_settings, &pairs, None)
            .await
            .expect("by-hash fetch should succeed");
        assert_eq!(ranges.len(), 2);
        for (range, &(number, hash)) in ranges.iter().zip(pairs.iter()) {
            assert_eq!((range.from_block, range.to_block), (number, number));
            assert_eq!(
                range.tip_hash,
                Some(hash),
                "the restart tail knows the hash it queried with"
            );
        }

        // The live flush: same rule, reached through the ring's own entry.
        let flush_asserter = Asserter::new();
        flush_asserter.push_success(&serde_json::json!([]));
        let flush_settings = ProviderSettings::from_mock(mock_http_provider(flush_asserter));
        // Depth 1 keeps the ring exactly full, so the top-up walk no-ops and
        // the single queued response is the flush's own.
        let mut linkage = HeadLinkage::new(1);
        assert_eq!(linkage.observe(300, hb(44), hb(43)), LinkOutcome::Linked);
        let mut coalescer = HeadCoalescer::new(Some(300), WINDOW);
        coalescer.observe_head(300, Instant::now());
        let ranges =
            flush_window_range(&mut linkage, &mut coalescer, &flush_settings, None, 0, None)
                .await
                .expect("flush should succeed");
        assert_eq!(ranges.len(), 1);
        assert_eq!(
            ranges[0].tip_hash,
            Some(hb(44)),
            "the live flush fetches by the ring's canonical hash and carries it"
        );
    }

    /// Ring bootstrapped at a head far above the flush cursor: the self-heal
    /// branch fires because the window's hashes are absent, and the walk it
    /// would need spans one block more than the bound allows. The bound is the
    /// constant plus the configured window, so the shortfall is measured
    /// against both — a flush must refuse before its first fetch, and the
    /// empty `Asserter` would error on any request that got through. Shared by
    /// the two self-heal-bound tests below.
    fn self_heal_bound_exceeded_fixture() -> (
        ProviderSettings,
        HeadLinkage,
        HeadCoalescer,
        Option<(u64, u64)>,
    ) {
        let settings = ProviderSettings::from_mock(mock_http_provider(Asserter::new()));

        let depth = 96u64;
        let head = 10_000_000u64;
        let cursor = head - (SELF_HEAL_WALK_MAX_BLOCKS + depth);
        let mut linkage = HeadLinkage::new(depth);
        assert_eq!(linkage.observe(head, hb(7), hb(6)), LinkOutcome::Linked);
        let mut coalescer = HeadCoalescer::new(Some(cursor), WINDOW);
        coalescer.observe_head(head, Instant::now());
        let pending_before = coalescer.pending_range();

        (settings, linkage, coalescer, pending_before)
    }

    /// G.17 — the self-heal walk stops at its length bound rather than walking a
    /// backlog that would park the live loop, and reports only what stopping
    /// establishes: we could not rebuild our own view.
    #[tokio::test]
    async fn flush_self_heal_walk_stops_at_its_length_bound() {
        let (settings, mut linkage, mut coalescer, pending_before) =
            self_heal_bound_exceeded_fixture();

        let err = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None)
            .await
            .expect_err("a walk past the bound must not run");
        assert!(matches!(err, RpcError::Gap), "got {err:?}");
        assert_eq!(
            coalescer.pending_range(),
            pending_before,
            "a refused flush leaves the cursor where it was — no silent gap"
        );
    }

    /// `bound_flush_time` passes the inner error through unchanged when
    /// `flush_max_total_secs == 0` — the default every chain runs — so the
    /// self-heal-bound error it raises reaches `classify_flush_failure` intact
    /// and classifies as `EndStream`, leaving the span a retry would re-run
    /// unchanged (asserted below).
    ///
    /// Does not pin the raise site to `Gap`: `classify_flush_failure`
    /// maps `Gap`, `ReorgBeyondRing` and `Reorg` all to
    /// `EndStream` (see `classify_flush_failure_rotates_only_on_a_transport_fault`),
    /// so retargeting the raise site to either of the other two leaves this
    /// test green.
    ///
    /// Residual: the live loop's dispatch arm has no hermetic seam — see the
    /// `Err(e)` arm in `stream_heads_with_logs_with_progress` (`TODO G.15`).
    #[tokio::test]
    async fn self_heal_bound_error_classifies_end_stream_and_moves_no_cursor() {
        let (settings, mut linkage, mut coalescer, pending_before) =
            self_heal_bound_exceeded_fixture();

        let err = bound_flush_time(
            settings.flush_max_total_secs,
            &flush_timeout_label(&settings, "ws=test-host"),
            flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None),
        )
        .await
        .expect_err("a walk past the bound must not run");

        assert_eq!(
            classify_flush_failure(&err),
            FlushFailure::EndStream,
            "a gap must reach the consumer, not another provider: {err:?}"
        );
        assert_eq!(
            coalescer.pending_range(),
            pending_before,
            "the span a retry would re-run is unchanged, so a retry cannot differ"
        );
    }

    /// `classify_flush_failure` over every `RpcError` variant: a transport
    /// fault, and the two self-clearing all-providers signals, keep the
    /// provider rotation; everything else ends the stream.
    ///
    /// The rotation half is what the branch must not break — a stalled or
    /// unconfigured provider is still another provider's to answer. This
    /// covers the classification alone; what the live loop then does with it
    /// has no hermetic seam (`TODO G.15`).
    #[test]
    fn classify_flush_failure_rotates_only_on_a_transport_fault() {
        for err in [
            RpcError::Gap,
            RpcError::ReorgBeyondRing,
            RpcError::Reorg {
                common_ancestor: 42,
            },
        ] {
            assert_eq!(
                classify_flush_failure(&err),
                FlushFailure::EndStream,
                "{err:?} states a conclusion no other provider can overturn"
            );
        }

        for err in [
            // What `bound_flush_time` raises on a stalled flush, and what
            // `flush_window_range` raises with no HTTP providers configured.
            RpcError::ConnectionError("stalled mid-flush".to_string()),
            RpcError::SubscriptionError("dropped".to_string()),
            RpcError::LogFetchError("eth_getLogs failed".to_string()),
            RpcError::TransportError("reset by peer".to_string()),
            RpcError::CursorPastTip {
                to_block: 10,
                reported_tip: 5,
            },
            RpcError::ProviderStalled {
                to_block: 10,
                reported_tip: 5,
                stalled_cycles: 3,
            },
            RpcError::Other("unclassified".to_string()),
        ] {
            assert_eq!(
                classify_flush_failure(&err),
                FlushFailure::SwitchProvider,
                "{err:?} is a transport-class failure another provider may answer"
            );
        }

        for err in [
            RpcError::AllProvidersSuspended("all down".to_string()),
            RpcError::AllProvidersIdle {
                providers_tried: 2,
                idle_secs: 30,
            },
        ] {
            assert_eq!(
                classify_flush_failure(&err),
                FlushFailure::SwitchProvider,
                "{err:?} is self-clearing, not a permanent claim about every provider"
            );
        }
    }

    /// The flush failure log line names which branch the dispatch took.
    ///
    /// The dispatch arm itself is untestable (see
    /// `self_heal_bound_error_classifies_end_stream_and_moves_no_cursor`'s
    /// residual), so this line is the only signal an operator has for which
    /// way an identical error was resolved. Pinning it keeps the two
    /// dispositions from collapsing into one indistinguishable message — and
    /// pins `SwitchProvider` against re-acquiring the "next provider" claim:
    /// the live loop's call site always retries `flush_window_range` from
    /// HTTP provider index 0, so no rotation happens on a retried flush
    /// regardless of configured provider count.
    #[test]
    fn flush_failure_log_line_states_the_disposition() {
        let err = RpcError::Gap;
        let ended = flush_failure_log_line(&err, FlushFailure::EndStream);
        let rotated = flush_failure_log_line(&err, FlushFailure::SwitchProvider);

        assert!(
            ended.contains("ending stream"),
            "an ended stream must say so: {ended}"
        );
        assert!(
            rotated.contains("retrying the same window"),
            "a retry must say so: {rotated}"
        );
        assert!(
            !rotated.contains("next provider"),
            "the retried flush's HTTP fetch never rotates off provider index \
             0 — the line must not claim a rotation this call site cannot \
             perform: {rotated}"
        );
        assert_ne!(
            ended, rotated,
            "one error text, two dispositions — the line must tell them apart"
        );
    }

    /// `LogRange` compares and hashes, as the tuple it replaced did for free.
    ///
    /// Compiling is most of the assertion — dropping a derive breaks this
    /// test's build. `tip_hash` carries the distinction that matters: two
    /// ranges over the same blocks on different chains are not the same range.
    #[test]
    fn log_ranges_compare_and_hash_across_every_field() {
        use std::collections::HashSet;

        let range = LogRange {
            from_block: 10,
            to_block: 12,
            logs: vec![],
            tip_hash: Some(hb(12)),
        };
        let same = range.clone();
        let competing_tip = LogRange {
            tip_hash: Some(hb(11)),
            ..range.clone()
        };

        assert_eq!(range, same);
        assert_ne!(range, competing_tip);

        let mut seen = HashSet::new();
        assert!(seen.insert(range));
        assert!(!seen.insert(same), "equal ranges must hash alike");
        assert!(seen.insert(competing_tip));
    }

    /// G.17 — the bound must clear the configured recovery window, never merely
    /// equal it. The self-heal branch's flush-failure trigger first fires with
    /// the cursor one block under a full ring's floor, so its first legitimate
    /// walk is `recovery_window_depth + 1` blocks long. A bound sized *at* the
    /// depth — what maximizing against the constant yields once a chain
    /// configures a window deeper than `SELF_HEAL_WALK_MAX_BLOCKS` — falls one
    /// block short of that walk and kills the stream during ordinary degraded
    /// operation. Adding the constant to the depth keeps its full headroom at
    /// every depth.
    ///
    /// Depth 2,000 is the only domain where the two formulas differ; at the
    /// depths in service today (96, and 0 on the L2s) the defect is inert, which
    /// is why `flush_self_heal_walk_stops_at_its_length_bound` cannot see it.
    #[tokio::test]
    async fn flush_self_heal_walk_succeeds_when_recovery_window_exceeds_the_constant_bound() {
        // One distinct hash per height: `hb` only spans 256 values, and the walk
        // rejects any block whose hash and number disagree with what it asked for.
        fn hash_at_height(number: u64) -> B256 {
            let mut bytes = [0u8; 32];
            bytes[24..].copy_from_slice(&number.to_be_bytes());
            B256::from(bytes)
        }

        let depth = 2_000u64;
        assert!(
            depth > SELF_HEAL_WALK_MAX_BLOCKS,
            "the defect is inert unless the configured window is the deeper of the two"
        );

        // A ring filled to its depth, with the flush cursor one block under its
        // floor — the shortfall a flush that kept failing while heads kept
        // arriving leaves behind the first time the self-heal branch fires.
        let tip = 10_000_000u64;
        let floor = tip - (depth - 1);
        let cursor = floor - 1;

        let mut linkage = HeadLinkage::new(depth);
        for number in floor..=tip {
            assert_eq!(
                linkage.observe(number, hash_at_height(number), hash_at_height(number - 1)),
                LinkOutcome::Linked
            );
        }

        // One `get_block_by_hash` per step from the ring anchor down to the
        // cursor, then the single pending block's logs.
        let asserter = Asserter::new();
        for number in (cursor + 1..=tip).rev() {
            asserter.push_success(&block_with_parent_at(
                number,
                hash_at_height(number),
                hash_at_height(number - 1),
            ));
        }
        asserter.push_success(&serde_json::json!([]));
        let settings = ProviderSettings::from_mock(mock_http_provider(asserter));

        let mut coalescer = HeadCoalescer::new(Some(cursor), WINDOW);
        coalescer.observe_head(cursor, Instant::now());

        let ranges = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None)
            .await
            .expect("the walk's first legitimate firing must not be refused by the bound");
        assert_eq!(ranges.len(), 1);
        assert_eq!(
            ranges[0].tip_hash,
            Some(hash_at_height(cursor)),
            "the healed window is fetched by the hash the walk established"
        );
    }

    /// A consumer's log-metric filter matches this exact string, and nothing
    /// else in the workspace reads it. Pinning it here is what makes a rename a
    /// test failure rather than a silently dead filter.
    #[test]
    fn live_head_gap_event_name_is_fixed() {
        assert_eq!(live_head_gap_event(), "live_head_gap");
    }

    /// Same contract as `live_head_gap_event_name_is_fixed`, for the event a
    /// future filter keys on to alarm on a chain claim specifically.
    #[test]
    fn live_head_reorg_beyond_ring_event_name_is_fixed() {
        assert_eq!(
            live_head_reorg_beyond_ring_event(),
            "live_head_reorg_beyond_ring"
        );
    }

    /// The walk's dispatch split, asserted directly since the live loop's own
    /// arm has no hermetic seam (`TODO G.15`). Only `ReorgBeyondRing` — the
    /// walk's one chain claim — earns the named event; every other exit,
    /// including a transport fault reaching the walk, must not, or a filter
    /// on the chain-claim event would count a stalled provider as one.
    #[test]
    fn walk_failure_event_names_only_the_chain_claim() {
        assert_eq!(
            walk_failure_event(&RpcError::ReorgBeyondRing),
            Some("live_head_reorg_beyond_ring"),
            "the chain claim must carry its own event"
        );

        for err in [
            RpcError::ConnectionError("stalled mid-walk".to_string()),
            RpcError::Gap,
            RpcError::Reorg {
                common_ancestor: 42,
            },
        ] {
            assert_eq!(
                walk_failure_event(&err),
                None,
                "{err:?} must keep the walk's unnamed line"
            );
        }
    }

    /// `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 for `get_logs`. The ring is at depth 8
    /// with a single entry, so the unconditional top-up (I.4f) also fires on this
    /// flush and drains the queue a second time via `get_block_by_hash`; finding it
    /// empty, the mock transport errors — not a panic — and the top-up's
    /// non-propagating error arm (settled decision 8) swallows it.
    #[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, for get_logs. The top-up's get_block_by_hash drains
        // the queue empty and errors — swallowed, not a panic (see the doc comment
        // above).
        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 range = &ranges[0];
        assert_eq!((range.from_block, range.to_block), (200, 200));
        assert_eq!(range.logs.len(), 1, "canonical log returned");
        assert_eq!(
            range.logs[0].block_hash,
            Some(hb(200)),
            "log must carry the canonical block hash, not an orphan's"
        );
        assert_eq!(
            range.tip_hash,
            Some(hb(200)),
            "the live flush fetches by hash and carries it"
        );
        // Cursor advanced past block 200.
        assert_eq!(coalescer.pending_range(), None);
    }

    /// A null block and a transport error are both "the primary did not
    /// answer" — with a single configured HTTP provider there is nowhere
    /// left to fall back to, so both yield `ConnectionError`, a view claim,
    /// never `ReorgBeyondRing`.
    #[tokio::test]
    async fn hash_mismatch_null_and_transport_are_unresolved() {
        {
            let asserter = Asserter::new();
            asserter.push_success(&serde_json::json!(null));
            let provider = mock_http_provider(asserter);
            let settings = n_http_provider_settings(1);
            // Walk from block 5 (hash hb(5)) down to block 1; the first fetch returns null.
            let err = collect_window_hashes(&provider, &settings, None, 0, hb(5), 5, 1)
                .await
                .expect_err("null block must be an error");
            assert!(
                matches!(err, RpcError::ConnectionError(_)),
                "null block must yield ConnectionError, got {err:?}"
            );
        }
        {
            let asserter = Asserter::new();
            asserter.push_failure_msg("connection reset");
            let provider = mock_http_provider(asserter);
            let settings = n_http_provider_settings(1);
            let err = collect_window_hashes(&provider, &settings, None, 0, 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: self-heal 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. That leaves the ring at depth
        // 8 with only 5 entries (101..=105), so the unconditional top-up
        // (I.4f) also fires: it issues one more get_block_by_hash, finds the
        // queue drained, and errors — swallowed by the top-up's
        // non-propagating error arm (settled decision 8), not affecting the
        // assertions below.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent_at(105, hb(105), hb(104)));
        asserter.push_success(&block_with_parent_at(104, hb(104), hb(103)));
        asserter.push_success(&block_with_parent_at(103, hb(103), hb(102)));
        asserter.push_success(&block_with_parent_at(102, hb(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, range) in ranges.iter().enumerate() {
            let n = 101u64 + i as u64;
            assert_eq!((range.from_block, range.to_block), (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 not an answer — the same bucket as a
    /// null or a wrong hash. With N=1 there is no other provider to fall
    /// back to, so this is `ConnectionError`, a view claim, never a verdict
    /// about the chain. The hash below is honestly the queried hash
    /// (hb(5)) precisely to prove that even a hash MATCH does not save a
    /// number mismatch.
    #[tokio::test]
    async fn collect_window_hashes_number_mismatch() {
        let asserter = Asserter::new();
        // Wrong number, but parent_hash (hb(4)) lands exactly on from_block:
        // if the number check is dropped, the walk accepts this block and
        // completes at Ok(vec![(5, hb(5)), (4, hb(4))]) instead of exhausting
        // — the sharpest signal a dropped guard can leave.
        asserter.push_success(&block_with_parent_at(99, hb(5), hb(4))); // wrong number
        let provider = mock_http_provider(asserter);
        let settings = n_http_provider_settings(1);

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

    /// `collect_window_hashes`'s own PRIMARY hash-identity check, which nothing
    /// else exercises: every other fixture hands the primary either an honest
    /// block or `null`. Hash identity is what makes it sound to collect a
    /// block from any provider into the window's `(number, hash)` pairs and
    /// later fetch logs by that hash, so losing it silently is the worst
    /// failure this function has.
    ///
    /// The primary answers `Ok(Some)` with a block whose own hash is NOT the
    /// one queried, but whose `number` IS correct — it must clear the
    /// (separately covered) number-integrity check and be caught by the
    /// hash-identity check instead. `parent_hash` sits one block short of
    /// `from_block`, so if the guard is dropped and the wrong block is
    /// accepted, the walk completes in a single further step instead of
    /// falling back to the other providers — `expect_err` below then fails
    /// outright rather than merely returning a different error, the
    /// sharpest signal a dropped guard can leave. Keep the guard and the
    /// primary's answer falls back to the other two providers, both null,
    /// and the walk fails closed with `ConnectionError`.
    #[tokio::test]
    async fn collect_window_hashes_primary_wrong_hash_does_not_answer_the_step() {
        let primary_asserter = Asserter::new();
        // Queried for hb(5); answers with a block of the correct number (5)
        // but hash hb(77) instead.
        primary_asserter.push_success(&block_with_parent_at(5, hb(77), hb(4)));
        let provider = mock_http_provider(primary_asserter);
        let settings = n_http_provider_settings(3);

        let corrob1 = Asserter::new();
        corrob1.push_success(&serde_json::json!(null));
        let corrob2 = Asserter::new();
        corrob2.push_success(&serde_json::json!(null));
        let reused = vec![
            mock_http_provider(Asserter::new()), // index 0: the primary, never dispatched
            mock_http_provider(corrob1),
            mock_http_provider(corrob2),
        ];

        let err = collect_window_hashes(&provider, &settings, Some(&reused), 0, hb(5), 5, 4)
            .await
            .expect_err("a wrong-hash primary block must never answer the step");
        assert!(matches!(err, RpcError::ConnectionError(_)), "got {err:?}");
    }

    // ── collect_window_hashes: sequential fallback matrix ────────────────────
    // The primary IS one of the N HTTP providers (`primary_index`); a miss
    // falls back to the other N-1, starting right after `primary_index`.

    /// N=3, primary null, one other provider serves the block (valid hash
    /// and number) → the walk continues; assert the full collected pair
    /// list, not just the outcome variant.
    #[tokio::test]
    async fn collect_window_hashes_second_source_continues_walk() {
        let primary_asserter = Asserter::new();
        primary_asserter.push_success(&serde_json::json!(null)); // step 1: triggers step resolution
        primary_asserter.push_success(&block_with_parent_at(4, hb(4), hb(3))); // step 2: direct match
        let provider = mock_http_provider(primary_asserter);
        let settings = n_http_provider_settings(3);

        let corrob1 = Asserter::new();
        corrob1.push_success(&block_with_parent_at(5, hb(5), hb(4))); // recovers step 1
        let corrob2 = Asserter::new();
        corrob2.push_success(&serde_json::json!(null));
        let reused = vec![
            mock_http_provider(Asserter::new()), // index 0: the primary, never dispatched
            mock_http_provider(corrob1),
            mock_http_provider(corrob2),
        ];

        let pairs_desc = collect_window_hashes(&provider, &settings, Some(&reused), 0, hb(5), 5, 3)
            .await
            .expect("one second source supplying the block must let the walk continue");
        assert_eq!(
            pairs_desc,
            vec![(5, hb(5)), (4, hb(4)), (3, hb(3))],
            "got {pairs_desc:?}"
        );
    }

    /// N=3, primary null, both other providers null: every provider tried,
    /// none answered — `ConnectionError`, a view claim, never a verdict about
    /// the chain.
    #[tokio::test]
    async fn collect_window_hashes_all_sources_absent_is_unresolved() {
        let primary_asserter = Asserter::new();
        primary_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(primary_asserter);
        let settings = n_http_provider_settings(3);
        let corrob1 = Asserter::new();
        corrob1.push_success(&serde_json::json!(null));
        let corrob2 = Asserter::new();
        corrob2.push_success(&serde_json::json!(null));
        let reused = vec![
            mock_http_provider(Asserter::new()),
            mock_http_provider(corrob1),
            mock_http_provider(corrob2),
        ];

        let err = collect_window_hashes(&provider, &settings, Some(&reused), 0, hb(5), 5, 1)
            .await
            .expect_err("exhausting every provider must be an error");
        assert!(matches!(err, RpcError::ConnectionError(_)), "got {err:?}");
    }

    /// N=3, primary null, both other providers transport-fail: same
    /// exhaustion, different reason — still `ConnectionError`.
    #[tokio::test]
    async fn collect_window_hashes_lone_absent_answer_is_unresolved() {
        let primary_asserter = Asserter::new();
        primary_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(primary_asserter);
        let settings = n_http_provider_settings(3);
        let corrob1 = Asserter::new();
        corrob1.push_failure_msg("blip");
        let corrob2 = Asserter::new();
        corrob2.push_failure_msg("blip");
        let reused = vec![
            mock_http_provider(Asserter::new()),
            mock_http_provider(corrob1),
            mock_http_provider(corrob2),
        ];

        let err = collect_window_hashes(&provider, &settings, Some(&reused), 0, hb(5), 5, 1)
            .await
            .expect_err("exhausting every provider must be an error");
        assert!(matches!(err, RpcError::ConnectionError(_)), "got {err:?}");
    }

    /// N=3, primary null, one other provider null + one transport-fail: the
    /// primary's own miss no longer counts toward anything — it is simply
    /// tried, then skipped, and the other two are tried and also miss.
    #[tokio::test]
    async fn collect_window_hashes_primary_miss_does_not_shorten_fallback() {
        let primary_asserter = Asserter::new();
        primary_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(primary_asserter);
        let settings = n_http_provider_settings(3);
        let corrob1 = Asserter::new();
        corrob1.push_success(&serde_json::json!(null));
        let corrob2 = Asserter::new();
        corrob2.push_failure_msg("blip");
        let reused = vec![
            mock_http_provider(Asserter::new()),
            mock_http_provider(corrob1),
            mock_http_provider(corrob2),
        ];

        let err = collect_window_hashes(&provider, &settings, Some(&reused), 0, hb(5), 5, 1)
            .await
            .expect_err("exhausting every provider must be an error");
        assert!(matches!(err, RpcError::ConnectionError(_)), "got {err:?}");
    }

    /// A second source's block failing the number-integrity check is not a
    /// match — it folds into the same "did not answer" bucket a transport
    /// error does.
    #[tokio::test]
    async fn collect_window_hashes_wrong_number_is_not_a_recovery() {
        let primary_asserter = Asserter::new();
        primary_asserter.push_success(&serde_json::json!(null));
        let provider = mock_http_provider(primary_asserter);
        let settings = n_http_provider_settings(3);
        let corrob1 = Asserter::new();
        // Right hash, WRONG number — fails the identity check, must not match.
        corrob1.push_success(&block_with_parent_at(99, hb(5), hb(4)));
        let corrob2 = Asserter::new();
        corrob2.push_failure_msg("blip");
        let reused = vec![
            mock_http_provider(Asserter::new()),
            mock_http_provider(corrob1),
            mock_http_provider(corrob2),
        ];

        let err = collect_window_hashes(&provider, &settings, Some(&reused), 0, hb(5), 5, 1)
            .await
            .expect_err("wrong-number second source must not match");
        assert!(matches!(err, RpcError::ConnectionError(_)), "got {err:?}");
    }

    /// `flush_window_range`: when `latest()` is None (ring completely empty),
    /// the function returns `Gap` rather than panicking — no anchor
    /// to walk from, and no record anywhere to disagree with.
    #[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::Gap), "got {err:?}");
    }

    /// The flush's own zero-provider guard: a configuration fault, resolved
    /// before any fetch and never a panic — the same posture
    /// `walk_to_common_ancestor_zero_http_providers_is_unresolved` holds for
    /// the walk. `http_settings`/`connect_http` index by modulo, so an empty
    /// set has no safe reader; only this guard keeps it out of them.
    #[tokio::test]
    async fn flush_window_range_zero_http_providers_is_fail_closed() {
        let settings = n_http_provider_settings(0);
        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());

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

    /// The label is built at the call site BEFORE `flush_window_range` can
    /// reach the guard above, so it must not be the thing that panics on the
    /// configuration this repo deliberately tolerates: `http_settings(0)`
    /// computes `0 % 0` on an empty set.
    #[test]
    fn flush_timeout_label_survives_an_empty_provider_set() {
        let ws_host = "ws-provider.example.invalid";
        let label = flush_timeout_label(&n_http_provider_settings(0), ws_host);
        assert!(
            label.contains(ws_host),
            "the label must still name the WS host with no HTTP provider \
             configured: {label}"
        );
    }

    /// With providers configured the label names the first one — the same
    /// index the flush passes to `flush_window_range`.
    #[test]
    fn flush_timeout_label_names_the_first_configured_provider() {
        let label = flush_timeout_label(&n_http_provider_settings(2), "ws-host.invalid");
        assert!(
            label.contains("ws-host.invalid") && label.contains("unused-0"),
            "the label must name the WS host and the first HTTP provider: {label}"
        );
    }

    /// 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() {
        // No queued responses: Part A's self-heal guard no-ops (top_n < from_block)
        // before any RPC. The depth-8 ring's top-up (I.4f) then fires too (floor 50
        // > target 43) and does issue one get_block_by_hash call, but the empty
        // asserter errors it immediately, and the top-up's non-propagating error
        // arm swallows that the same way.
        let asserter = Asserter::new();
        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(105), hb(104)));
        asserter.push_success(&block_with_parent_at(104, hb(104), hb(103)));
        asserter.push_success(&block_with_parent_at(103, hb(103), hb(102)));
        asserter.push_success(&block_with_parent_at(102, hb(102), hb(101)));
        for _ in 101u64..=105 {
            asserter.push_success(&serde_json::json!([]));
        }
        let reused = vec![mock_http_provider(asserter)];

        // `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, range) in ranges.iter().enumerate() {
            let n = 101u64 + i as u64;
            assert_eq!((range.from_block, range.to_block), (n, n));
        }
    }

    /// I.4f §6.2 — Part A alone: one flush from a ring bootstrapped one entry
    /// deep at the live head (105), cursor at 101 — the post-restart shape.
    /// The self-heal walk's already-fetched pairs land in the ring
    /// (`extend_downward`), and `depth` is sized so the ring is *exactly*
    /// full once they do (`depth == latest - from_block + 1 == 5`): the
    /// top-up then finds `floor() == lowest_worth_recording()` and skips,
    /// which is what isolates Part A from Part B's own deepening (settled
    /// decision 6 — Part B runs on every flush by design, so a looser
    /// `depth` would let it move the floor too, and the exact-cursor
    /// assertion below would no longer isolate anything).
    ///
    /// Reverting Part A (dropping the `extend_downward` call in the
    /// self-heal branch) leaves the ring at its pre-flush single entry —
    /// the floor stays at the live head (105), never reaching the cursor
    /// (101) — and this test fails.
    #[tokio::test]
    async fn flush_window_range_part_a_seeds_ring_down_to_the_flush_cursor() {
        let mut linkage = HeadLinkage::new(5);
        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)));

        // Self-heal walk: 105 -> 104 -> 103 -> 102 (4 calls; 101 == from_block,
        // no fetch for 101 itself), then 5 log fetches (101..=105). Exactly 9
        // responses queued — the ring lands exactly full, so the top-up finds
        // nothing left and issues no 10th call; a drained Asserter queried a
        // 10th time would surface as an error the flush's own `Result` masks,
        // so the real proof is the floor assertion below, not silence alone.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent_at(105, hb(105), hb(104)));
        asserter.push_success(&block_with_parent_at(104, hb(104), hb(103)));
        asserter.push_success(&block_with_parent_at(103, hb(103), hb(102)));
        asserter.push_success(&block_with_parent_at(102, hb(102), hb(101)));
        for _ in 101u64..=105 {
            asserter.push_success(&serde_json::json!([]));
        }
        let reused = vec![mock_http_provider(asserter)];
        let settings = n_http_provider_settings(1);

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

        assert_eq!(ranges.len(), 5, "one LogRange per block 101..=105");
        assert_eq!(
            linkage.floor(),
            Some(101),
            "Part A must drop the floor to the flush cursor (from_block)"
        );
        assert_eq!(
            linkage.lowest_worth_recording(),
            linkage.floor(),
            "depth was sized so the ring is already full — the top-up has \
             nothing left to walk"
        );
    }

    /// I.4f §6.3 — Part B alone, and it must not ride Part A's branch. The
    /// ring holds only the live head (200, depth 96 — production magnitude),
    /// and every flush in this test asks for exactly that one block, so the
    /// self-heal branch (`heads.rs:544`'s ring-miss predicate) can never
    /// fire: whatever moves the floor here is Part B.
    ///
    /// Reverting Part B leaves the floor pinned at 200 on every flush; this
    /// test fails on the very first iteration, while §6.2's test (which
    /// never touches Part B in a way this scenario could distinguish) does
    /// not — the separation the mutation check (§6.5) depends on.
    #[tokio::test]
    async fn flush_window_range_part_b_deepens_ring_without_a_self_heal_walk() {
        const DEPTH: u64 = 96;
        const TIP: u64 = 200;
        const TARGET: u64 = TIP - (DEPTH - 1); // 105 — lowest_worth_recording

        let mut linkage = HeadLinkage::new(DEPTH);
        assert_eq!(
            linkage.observe(TIP, hb(TIP as u8), hb((TIP - 1) as u8)),
            LinkOutcome::Linked
        );

        // Cap-sized steps down to TARGET: eleven steps of the full cap (8),
        // then one shorter step (7) landing exactly on TARGET — mirrors the
        // prompt's own derivation (a 96-deep ring converges in ~12 flushes
        // at 8 steps/flush).
        let mut steps = vec![REORG_RING_TOP_UP_STEPS; 11];
        steps.push(TIP - TARGET - 11 * REORG_RING_TOP_UP_STEPS); // 7
        assert_eq!(
            steps.iter().sum::<u64>(),
            TIP - TARGET,
            "sanity: steps span the full descent"
        );

        // Queue each flush's one `get_logs` response, then that flush's
        // block-fetch responses (descending, matching the walk's own order).
        let asserter = Asserter::new();
        let mut floor = TIP;
        for &step in &steps {
            asserter.push_success(&serde_json::json!([]));
            for n in ((floor - step + 1)..=floor).rev() {
                asserter.push_success(&block_with_parent_at(n, hb(n as u8), hb((n - 1) as u8)));
            }
            floor -= step;
        }
        assert_eq!(floor, TARGET);

        // Kept to queue one more `get_logs` response for the final,
        // already-full-ring flush below — `Asserter` shares its queue
        // through an inner `Arc`, so this handle and the provider's own
        // stay in sync.
        let asserter_handle = asserter.clone();
        let reused = vec![mock_http_provider(asserter)];
        let settings = n_http_provider_settings(1);

        let mut expected_floor = TIP;
        for (i, &step) in steps.iter().enumerate() {
            let mut coalescer = HeadCoalescer::new(Some(TIP), WINDOW);
            coalescer.observe_head(TIP, Instant::now());
            assert_eq!(coalescer.pending_range(), Some((TIP, TIP)));

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

            expected_floor -= step;
            assert_eq!(
                linkage.floor(),
                Some(expected_floor),
                "flush {i}: floor must descend by exactly this step"
            );
        }
        assert_eq!(
            steps[0], REORG_RING_TOP_UP_STEPS,
            "the first flush's descent must be the full cap"
        );
        assert_eq!(linkage.floor(), Some(TARGET));
        assert_eq!(linkage.lowest_worth_recording(), Some(TARGET));

        // Ring is now full: one more flush must not move the floor past
        // `lowest_worth_recording` — the top-up self-disables rather than
        // walking below it.
        let mut coalescer = HeadCoalescer::new(Some(TIP), WINDOW);
        coalescer.observe_head(TIP, Instant::now());
        asserter_handle.push_success(&serde_json::json!([])); // this flush's own get_logs(200)
        flush_window_range(
            &mut linkage,
            &mut coalescer,
            &settings,
            None,
            0,
            Some(&reused),
        )
        .await
        .expect("flush must succeed even once the ring is already full");
        assert_eq!(
            linkage.floor(),
            Some(TARGET),
            "floor must stop at the accessor, not below it"
        );
    }

    /// I.4f §6.4 — the incident, end to end (2026-08-14 01:17Z in miniature).
    /// A stream restart bootstraps the ring one entry deep at the live head.
    /// A reorg's common ancestor lies below that head but within `depth` —
    /// exactly the shape that produced a beyond-finality report against a ring
    /// that had never been filled.
    ///
    /// Two linkages, same detached-parent walk: `undeepened` never flushes
    /// (the ring stays exactly where a restart leaves it — "at `EXPECT_TIP`"
    /// in the goal's own words), `deepened` runs ordinary flushes first (Part
    /// B alone) so its floor passes the ancestor before the reorg arrives.
    /// `walk_to_common_ancestor` on `undeepened` reaches the floor with a
    /// fetched hash that disagrees with the one recorded there, so it reports
    /// `ReorgBeyondRing`; on `deepened` it resolves to the ancestor.
    #[tokio::test]
    async fn walk_to_common_ancestor_resolves_after_ring_deepens_past_the_ancestor() {
        const DEPTH: u64 = 96;
        const HEAD: u64 = 200;
        // Within DEPTH of HEAD (200 - 170 = 30 < 95), but below where the
        // ring's floor sits the instant a stream restarts (HEAD itself).
        const ANCESTOR: u64 = 170;
        const CHILD_NUMBER: u64 = HEAD + 2;

        // A byte range (50..=81) disjoint from the canonical chain's own
        // (105..=200), so no fork hash accidentally matches a ring entry.
        fn fork_hash(n: u64) -> B256 {
            hb((50 + (n - ANCESTOR)) as u8)
        }

        let detached_parent = fork_hash(CHILD_NUMBER - 1);

        // `undeepened`: the ring exactly as a restart leaves it — bootstrapped
        // one entry deep at HEAD, never flushed.
        let mut undeepened = HeadLinkage::new(DEPTH);
        assert_eq!(
            undeepened.observe(HEAD, hb(HEAD as u8), hb((HEAD - 1) as u8)),
            LinkOutcome::Linked
        );
        assert_eq!(undeepened.floor(), Some(HEAD));

        let undeepened_asserter = Asserter::new();
        // One step: the walk fetches CHILD_NUMBER - 1's block, finds its
        // parent (at HEAD) doesn't match the ring, and HEAD is the floor —
        // a confirmed disagreement without a second fetch.
        undeepened_asserter.push_success(&block_with_parent(detached_parent, fork_hash(HEAD)));
        let undeepened_provider = mock_http_provider(undeepened_asserter);
        let settings = n_http_provider_settings(1);

        let undeepened_result = walk_to_common_ancestor(
            &undeepened_provider,
            &settings,
            None,
            &undeepened,
            detached_parent,
            CHILD_NUMBER,
            DEPTH,
        )
        .await;
        assert!(
            matches!(undeepened_result, Err(RpcError::ReorgBeyondRing)),
            "got {undeepened_result:?}"
        );

        // `deepened`: the same bootstrap, then ordinary flushes (Part B
        // alone — the window is always covered by the ring's own tip, same
        // shape as §6.3's test) carry the floor down past ANCESTOR before
        // the reorg arrives.
        let mut deepened = HeadLinkage::new(DEPTH);
        assert_eq!(
            deepened.observe(HEAD, hb(HEAD as u8), hb((HEAD - 1) as u8)),
            LinkOutcome::Linked
        );
        // The top-up's own stop condition is `lowest_worth_recording`
        // (HEAD - (DEPTH - 1) = 105), not ANCESTOR — the fixtures below must
        // match what the real implementation actually walks toward, and the
        // outer loop below simply stops queuing once that real progression
        // has carried the floor at or below ANCESTOR.
        let lowest_worth_recording = HEAD - (DEPTH - 1);
        let flush_asserter = Asserter::new();
        let mut floor = HEAD;
        while floor > ANCESTOR {
            flush_asserter.push_success(&serde_json::json!([])); // this flush's get_logs(HEAD)
            let walk_from =
                lowest_worth_recording.max(floor.saturating_sub(REORG_RING_TOP_UP_STEPS));
            for n in ((walk_from + 1)..=floor).rev() {
                flush_asserter.push_success(&block_with_parent_at(
                    n,
                    hb(n as u8),
                    hb((n - 1) as u8),
                ));
            }
            floor = walk_from;
        }
        let flush_reused = vec![mock_http_provider(flush_asserter)];
        while deepened.floor().expect("bootstrapped") > ANCESTOR {
            let mut coalescer = HeadCoalescer::new(Some(HEAD), WINDOW);
            coalescer.observe_head(HEAD, Instant::now());
            flush_window_range(
                &mut deepened,
                &mut coalescer,
                &settings,
                None,
                0,
                Some(&flush_reused),
            )
            .await
            .expect("flush must succeed");
        }
        assert!(
            deepened.floor().expect("bootstrapped") <= ANCESTOR,
            "the ring must deepen past the ancestor before the reorg arrives"
        );

        // The reorg walk itself: CHILD_NUMBER - 1 down to ANCESTOR + 1, where
        // the fork's parent hash finally matches the ring's canonical entry.
        let walk_asserter = Asserter::new();
        for n in ((ANCESTOR + 1)..CHILD_NUMBER).rev() {
            let parent = if n == ANCESTOR + 1 {
                hb(ANCESTOR as u8)
            } else {
                fork_hash(n - 1)
            };
            walk_asserter.push_success(&block_with_parent(fork_hash(n), parent));
        }
        let walk_provider = mock_http_provider(walk_asserter);

        let deepened_result = walk_to_common_ancestor(
            &walk_provider,
            &settings,
            None,
            &deepened,
            detached_parent,
            CHILD_NUMBER,
            DEPTH,
        )
        .await;
        assert!(
            matches!(deepened_result, Ok(ANCESTOR)),
            "got {deepened_result:?}"
        );
    }

    /// 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
    /// `poll_providers_for_block_number`'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:?}");
    }

    /// `resolve_http_provider` is unreachable at `N == 0` in production (every
    /// caller fails closed first), but the arithmetic underneath it still
    /// divides by zero if that guard is ever removed. The guard's message must
    /// name the misconfiguration, not the arithmetic.
    #[test]
    #[should_panic(expected = "resolve_http_provider: no HTTP providers configured")]
    fn resolve_http_provider_panics_with_labeled_message_on_empty_http_providers() {
        let settings = ProviderSettings::new(
            vec![], // zero configured HTTP providers
            vec!["ws://127.0.0.1:9".parse().expect("static url")],
        );
        resolve_http_provider(&settings, None, 0);
    }

    /// 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, "unused at max_secs=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, "ws=test-host", 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:?}"
        );
    }

    /// Proves both G.5 checkboxes together: the timeout names the WS host and
    /// the fallback host it was given, ahead of the prose so identity survives
    /// even a tight cap, and the flush-failure log line carries that message
    /// complete through "switching provider" rather than cutting it mid-word.
    /// Reverting the label (checkbox 1) fails the host assertions below;
    /// reverting the log line's bound back to a 100-byte cap (checkbox 2)
    /// fails the tail assertion, since the labeled message is longer than 100
    /// bytes on its own.
    #[tokio::test(start_paused = true)]
    async fn bound_flush_time_timeout_names_both_providers_and_keeps_the_tail() {
        let ws_host = "ws-provider.example.invalid";
        let fallback_host = "fallback-provider.example.invalid";
        let label = format!("ws={ws_host} fallback={fallback_host}");

        let result: Result<u32, RpcError> = bound_flush_time(1, &label, async {
            tokio::time::sleep(Duration::from_millis(1200)).await;
            Ok(42)
        })
        .await;
        let err = result.expect_err("a future exceeding max_secs must time out");
        let display = err.to_string();

        let ws_pos = display
            .find(ws_host)
            .expect("timeout message must name the WS host");
        let fallback_pos = display
            .find(fallback_host)
            .expect("timeout message must name the fallback host");
        assert!(
            ws_pos < 100 && fallback_pos < 100,
            "both hosts must land before byte 100 of the error's Display output \
             so identity survives a tight cap, got ws@{ws_pos} fallback@{fallback_pos}: \
             {display}"
        );

        let log_line = flush_failure_log_line(&err, classify_flush_failure(&err));
        assert!(
            log_line.contains("stalled mid-flush; switching provider"),
            "the log line must carry the message complete through \
             'switching provider', not cut mid-word: {log_line}"
        );
    }

    /// SETTLING TEST for the top-up's placement invariant: `top_up_reorg_ring`'s
    /// call site inside `flush_window_range` must sit before
    /// `coalescer.advance_cursor`, never after. A `bound_flush_time` timeout
    /// firing mid-top-up must drop the top-up's own effect together with the
    /// already-fetched log ranges, never advance the cursor past blocks whose
    /// logs were not yielded. Moving the call after `advance_cursor` (the
    /// regression the placement comment on that call site warns against)
    /// makes this test fail: the cursor would advance to (last + 1) before the
    /// top-up's timeout is ever reached, so `pending_range()` would come back
    /// changed, not unchanged.
    ///
    /// `get_block_by_hash` returns a concrete builder type (`EthGetBlock`), not
    /// an `async fn` trait method, so the `Provider`-trait override this file
    /// otherwise uses for `get_logs` (see
    /// `flush_window_range_dispatches_concurrently_and_restores_ascending_order`)
    /// does not reach it. `DelayingTransport` intercepts one level lower, at the
    /// `tower::Service<RequestPacket>` the `RpcClient` dispatches through,
    /// delaying only `eth_getBlockByHash` requests so the flush's own `get_logs`
    /// call still resolves immediately.
    #[tokio::test(start_paused = true)]
    async fn flush_window_range_timeout_mid_top_up_drops_cursor_advance() {
        use alloy::rpc::client::RpcClient;
        use alloy::rpc::json_rpc::{RequestPacket, ResponsePacket};
        use alloy::transports::TransportError;
        use alloy::transports::TransportFut;
        use alloy::transports::mock::MockTransport;

        #[derive(Clone)]
        struct DelayingTransport {
            inner: MockTransport,
            delay: Duration,
        }

        impl tower::Service<RequestPacket> for DelayingTransport {
            type Response = ResponsePacket;
            type Error = TransportError;
            type Future = TransportFut<'static>;

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

            fn call(&mut self, req: RequestPacket) -> Self::Future {
                let mut inner = self.inner.clone();
                let delay_this = req.method_names().any(|m| m == "eth_getBlockByHash");
                let delay = self.delay;
                Box::pin(async move {
                    if delay_this {
                        tokio::time::sleep(delay).await;
                    }
                    tower::Service::call(&mut inner, req).await
                })
            }
        }

        // Depth 96 (BSC's finality depth), ring at a single entry (200): floor
        // (200) sits far above target (200 - 95 = 105), so the top-up has real
        // steps to walk. The flush's own in-window fetch (block 200) is already
        // covered by the ring, so the self-heal branch never fires — only the
        // top-up calls `get_block_by_hash`.
        let mut linkage = HeadLinkage::new(96);
        assert_eq!(linkage.observe(200, hb(200), hb(199)), LinkOutcome::Linked);
        let mut coalescer = HeadCoalescer::new(Some(200), WINDOW);
        coalescer.observe_head(200, Instant::now());
        let pending_before_flush = coalescer.pending_range();

        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!([])); // the one in-window get_logs call
        let transport = DelayingTransport {
            inner: MockTransport::new(asserter),
            delay: Duration::from_millis(1_200),
        };
        let provider: DynProvider<Ethereum> = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_client(RpcClient::new(transport, false))
            .erased();
        let settings = ProviderSettings::from_mock(provider);

        let result = bound_flush_time(
            1,
            "ws=test-host fallback=test-host",
            flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0, None),
        )
        .await;

        assert!(
            matches!(result, Err(RpcError::ConnectionError(_))),
            "a top-up stalled past the flush bound must surface as a timeout, got {result:?}"
        );
        assert_eq!(
            coalescer.pending_range(),
            pending_before_flush,
            "a timeout mid-top-up must leave the cursor exactly where the flush found it — \
             the fetched ranges and the top-up's own walk both dropped together"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn idle_head_await_returns_a_header_well_within_the_bound() {
        let mut s = futures_util::stream::iter(vec![Header::default()]);
        let outcome = await_head_within_idle_bound(&mut s).await;
        assert!(
            matches!(outcome, IdleHeadAwait::Header(_)),
            "a stream with a ready header must resolve to it, not expire"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn idle_head_await_distinguishes_stream_ended_from_expiry() {
        let mut s = futures_util::stream::empty::<Header>();
        let outcome = await_head_within_idle_bound(&mut s).await;
        assert!(
            matches!(outcome, IdleHeadAwait::StreamEnded),
            "an ended stream must resolve to StreamEnded, not Expired"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn idle_head_await_expires_on_a_silent_stream() {
        let mut s = futures_util::stream::pending::<Header>();
        // Race the helper against its own outer bound rather than awaiting it
        // directly: under paused time, a reverted helper (no internal
        // timeout) registers no timer at all and would hang the test
        // forever. The outer timeout always registers one, so it fires
        // regardless of whether the helper is bounded — turning "no expiry"
        // into a clean assertion failure instead of a deadlock.
        let outcome = tokio::time::timeout(
            Duration::from_secs(IDLE_HEAD_AWAIT_SECS + 60),
            await_head_within_idle_bound(&mut s),
        )
        .await
        .expect("the idle bound must expire on its own, not hang past it");
        assert!(
            matches!(outcome, IdleHeadAwait::Expired),
            "a silently pending stream must resolve to Expired once the idle bound elapses"
        );
    }

    #[test]
    fn idle_rotation_counter_continues_below_threshold() {
        let mut counter = IdleRotationCounter::new();
        assert!(matches!(counter.expire(3), IdleRotation::Continue));
        assert!(matches!(counter.expire(3), IdleRotation::Continue));
    }

    #[test]
    fn idle_rotation_counter_fires_at_exactly_the_configured_provider_count() {
        let mut counter = IdleRotationCounter::new();
        assert!(matches!(counter.expire(3), IdleRotation::Continue));
        assert!(matches!(counter.expire(3), IdleRotation::Continue));
        assert!(matches!(counter.expire(3), IdleRotation::AllIdle));
    }

    #[test]
    fn idle_rotation_counter_fires_at_one_for_a_single_provider() {
        let mut counter = IdleRotationCounter::new();
        assert!(matches!(counter.expire(1), IdleRotation::AllIdle));
    }

    #[test]
    fn idle_rotation_counter_header_resets_the_count() {
        let mut counter = IdleRotationCounter::new();
        assert!(matches!(counter.expire(3), IdleRotation::Continue));
        assert!(matches!(counter.expire(3), IdleRotation::Continue));
        counter.reset();
        // Without the reset, this third expiry would have hit the threshold.
        assert!(matches!(counter.expire(3), IdleRotation::Continue));
        assert!(matches!(counter.expire(3), IdleRotation::Continue));
        assert!(matches!(counter.expire(3), IdleRotation::AllIdle));
    }

    /// 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(105), hb(104)));
        asserter.push_success(&block_with_parent_at(104, hb(104), hb(103)));
        asserter.push_success(&block_with_parent_at(103, hb(103), hb(102)));
        asserter.push_success(&block_with_parent_at(102, hb(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, range) in ranges.iter().enumerate() {
            let n = 101u64 + i as u64;
            assert_eq!(
                (range.from_block, range.to_block),
                (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<LogRange> = [105u64, 100, 103]
            .into_iter()
            .map(|n| LogRange {
                from_block: n,
                to_block: n,
                logs: vec![],
                tip_hash: None,
            })
            .collect();

        let sorted = sort_ranges_ascending(unordered);

        let numbers: Vec<u64> = sorted.iter().map(|range| range.from_block).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. This test's ring covers
        /// the whole flushed range, so `flush_window_range` never takes the
        /// self-heal `get_block_by_hash` walk — but the ring is short of `depth`
        /// (4 of 8), so the unconditional top-up (I.4f) does call `get_block_by_hash`
        /// through this stub's `root()`, drains its empty `Asserter`, and errors,
        /// which the top-up logs and swallows (settled decision 8) rather than
        /// failing the flush. `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(); // queried by the top-up walk; drained, so it errors and is swallowed
        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(|range| range.from_block).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));
        // The ring (6 of depth 8) is short of full depth, so the unconditional
        // top-up (I.4f) queries this stub's `root()` via `get_block_by_hash`;
        // its `Asserter` is empty, so the walk errors and the top-up logs and
        // swallows it (settled decision 8) rather than failing the flush.
        let unused_asserter = Asserter::new();
        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);
    }
}