use std::collections::VecDeque;
use std::time::Duration;

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

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

/// Maximum block range per WS request to prevent "max results" errors
const MAX_BLOCKS_PER_WS_REQUEST: u64 = 300;
/// Coalesce rapid head updates into a bounded polling cadence.
const LIVE_HEAD_COALESCE_WINDOW: Duration = Duration::from_millis(500);

#[derive(Debug, Clone)]
struct HeadCoalescer {
    block_cursor: Option<u64>,
    latest_seen_head: Option<u64>,
    flush_deadline: Option<Instant>,
}

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

    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 + LIVE_HEAD_COALESCE_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);

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

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

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

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

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

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

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

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

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

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

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

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

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

        // A forward gap (heads skipped ahead with no verifiable reorg signal —
        // e.g. across a reconnect): linkage cannot be asserted over the gap, so
        // re-anchor here rather than risk a false reorg.
        if number > tip_number + 1 {
            self.ring.clear();
            self.push(number, hash);
            return LinkOutcome::Linked;
        }

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

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

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

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

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

async fn get_logs_with_http_fallback_retry(
    providers: &ProviderSettings,
    query_filter: &Filter,
    start_index: usize,
    from_block: u64,
    to_block: u64,
) -> 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 = providers.connect_http(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,
    ))
}

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

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

pub async fn stream_heads_with_logs(
    providers: &ProviderSettings,
    filter: Option<&Filter>,
) -> impl Stream<Item = Result<LogRange, RpcError>> {
    let num_providers = providers.wss_endpoints.len();

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

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

        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;
                }
            };

            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.
                    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);

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

                                // 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. Verify against the WS
                                // provider's reported tip before advancing the cursor.
                                if logs.is_empty() {
                                    match provider.get_block_number().await {
                                        Ok(reported_tip) => {
                                            if let Err(e) = next_cursor_after_batch(to_block, true, reported_tip) {
                                                yield Err(e);
                                                return;
                                            }
                                        }
                                        Err(e) => {
                                            let err_str = e.to_string();
                                            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;
                                        }
                                    }
                                }

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

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

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

                                                // 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.
                                                if logs.is_empty() {
                                                    let http_provider = providers.connect_http(http_provider_idx);
                                                    let http_host = providers.http_settings(http_provider_idx).host();
                                                    match http_provider.get_block_number().await {
                                                        Ok(reported_tip) => {
                                                            if let Err(e) = next_cursor_after_batch(to_block, true, reported_tip) {
                                                                yield Err(e);
                                                                return;
                                                            }
                                                        }
                                                        Err(e) => {
                                                            warn!(
                                                                "HTTP fallback {} tip fetch failed during empty-batch guard for blocks {}-{}: {}",
                                                                http_host, from_block, to_block,
                                                                truncate_error_short(e.to_string(), 100)
                                                            );
                                                            should_switch_provider = true;
                                                            break;
                                                        }
                                                    }
                                                }

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

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

                    if should_switch_provider {
                        break;
                    }

                    continue;
                }

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

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

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

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

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

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

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

/// Resolve the block number to use as the capture target based on `settings.capture_target`.
///
/// - `Latest` → current chain tip (existing behavior).
/// - `Lag(n)` → tip saturating_sub n.
/// - `Finalized` → the `finalized` tagged block; returns `Err` if the provider returns none.
///
/// Emits a one-time `warn!` when a deep-lag policy is active so operators notice
/// the intentional delay in their logs.
pub async fn resolve_capture_target(settings: &ProviderSettings) -> Result<u64, RpcError> {
    match settings.capture_target {
        CaptureTarget::Latest => last_block(settings).await,
        CaptureTarget::Lag(n) => {
            let tip = last_block(settings).await?;
            warn!(
                "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,
                tip.saturating_sub(n),
                n,
            );
            Ok(tip.saturating_sub(n))
        }
        CaptureTarget::Finalized => {
            let provider = settings.connect_http(0);
            let result = provider
                .get_block_by_number(BlockNumberOrTag::Finalized)
                .await
                .map_err(|e| {
                    RpcError::ConnectionError(format!("finalized block fetch failed: {e}"))
                })?;
            match result {
                Some(block) => {
                    let finalized = block.header.number;
                    warn!(
                        "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.",
                        finalized
                    );
                    Ok(finalized)
                }
                None => Err(RpcError::Other(
                    "Provider returned no finalized block; cannot determine safe capture target. \
                     Ensure the provider supports the `finalized` tag (mainnet only)."
                        .to_string(),
                )),
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        HeadCoalescer, HeadLinkage, LIVE_HEAD_COALESCE_WINDOW, LinkOutcome,
        MAX_BLOCKS_PER_WS_REQUEST, RpcError, fallback_http_exhausted_error, fallback_http_index,
        resolve_reorg_ancestor,
    };
    use alloy::primitives::B256;
    use alloy::providers::{Provider, ProviderBuilder};
    use alloy::transports::mock::Asserter;
    use std::time::Duration;
    use tokio::time::Instant;

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

        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));

        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 + LIVE_HEAD_COALESCE_WINDOW - Duration::from_millis(1))
        );
        assert!(coalescer.should_flush(now + LIVE_HEAD_COALESCE_WINDOW));
    }

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

        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));

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

    #[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));

        // 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 + LIVE_HEAD_COALESCE_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 + LIVE_HEAD_COALESCE_WINDOW));
    }

    /// Synthetic block whose only meaningful field is `parentHash` (the walk
    /// reads nothing else). `eth_getBlockByHash` response for the RPC fallback.
    fn block_with_parent(parent: B256) -> serde_json::Value {
        let zero = hb(0).to_string();
        serde_json::json!({
            "hash":             zero,
            "parentHash":       parent.to_string(),
            "sha3Uncles":       zero,
            "miner":            "0x0000000000000000000000000000000000000000",
            "stateRoot":        zero,
            "transactionsRoot": zero,
            "receiptsRoot":     zero,
            "logsBloom":        format!("0x{}", "0".repeat(512)),
            "difficulty":       "0x0",
            "number":           "0x0",
            "gasLimit":         "0x0",
            "gasUsed":          "0x0",
            "timestamp":        "0x0",
            "extraData":        "0x",
            "mixHash":          zero,
            "nonce":            "0x0000000000000000",
            "transactions":     [],
            "uncles":           []
        })
    }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    #[test]
    fn head_linkage_forward_gap_reanchors_without_false_reorg() {
        // `number > tip + 1`: heads skipped ahead (e.g. across a reconnect).
        // The distinct arm clears the ring + re-anchors at the new head — no reorg.
        let mut linkage = seed_chain(8, 100, 103); // tip = 103
        assert_eq!(linkage.observe(107, hb(107), hb(106)), LinkOutcome::Linked);
        // Old entries gone; ring re-anchored at 107 only.
        assert_eq!(linkage.floor(), Some(107));
        assert_eq!(linkage.ring.len(), 1);
        assert_eq!(linkage.hash_at(103), None);
        // Clean extension from the new tip continues normally.
        assert_eq!(linkage.observe(108, hb(108), hb(107)), LinkOutcome::Linked);
        assert_eq!(linkage.floor(), Some(107));
        assert_eq!(linkage.ring.len(), 2);
    }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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