ethl 0.1.42

use std::cmp::{max, min};
use std::time::{Duration, Instant};

use alloy::{
    network::Ethereum,
    providers::{DynProvider, Provider},
    rpc::types::{Filter, Log},
};
use async_stream::stream;
use futures_util::{Stream, future::join_all};
use tokio::time::sleep;
use tracing::{debug, info, trace, 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,
};
use crate::rpc::{
    LogRange, RpcError,
    config::{BASE_BACKOFF_SECS, ProviderSettings},
};

const BACKFILL_PROGRESS_LOG_INTERVAL: Duration = Duration::from_secs(120);

mod sizing {
    use std::cmp::{max, min};

    use crate::rpc::config::HttpRpcSettings;

    /// Fraction of `max_logs_per_request` the sizer aims to approach from below.
    const TARGET_FRACTION_OF_MAX_LOGS: f64 = 0.9;
    /// Shared shrink step for both "the success response was too big" and "the
    /// provider said the range itself is too wide" — same underlying signal (we
    /// asked for too much), same remedy, one policy instead of two.
    const SHRINK_FACTOR: f64 = 0.85;
    /// Growth step on the success path when the response was comfortably small.
    const GROWTH_FACTOR: f64 = 1.02;
    /// Shared floor for both shrink paths. The two previously had different, unexplained
    /// floors (10 vs 5) - unified, and set to the true physically-irreducible minimum: you
    /// cannot query less than one block's logs, matching `get_logs_bisecting`'s own stopping
    /// point elsewhere in this file. Safe because `record_success` can re-expand afterward -
    /// a temporary dip to 1 doesn't strand the sizer there.
    const MIN_BATCH_SIZE: u32 = 1;

    /// Result of asking the sizer to react to a range-too-wide error.
    pub(super) enum ShrinkOutcome {
        /// Batch size was reduced to this new value.
        Shrunk(u32),
        /// Already at the floor - shrinking further wouldn't change anything.
        AtFloor,
    }

    /// Tracks and adjusts the block-range width used for a single provider's fetches.
    /// Owns both directions of adjustment - success-driven growth/shrink via
    /// `record_success`, and error-driven shrink via `record_range_too_wide` - so
    /// `batch_size` is never written from outside this module.
    pub(super) struct DynamicBatchSizer<'a> {
        settings: &'a HttpRpcSettings,
        target: usize,
        batch_size: u32,
    }

    impl<'a> DynamicBatchSizer<'a> {
        pub(super) fn new(settings: &'a HttpRpcSettings) -> Self {
            Self {
                settings,
                target: (settings.max_logs_per_request as f64 * TARGET_FRACTION_OF_MAX_LOGS)
                    as usize,
                batch_size: settings.init_batch_size,
            }
        }

        pub(super) fn batch_size(&self) -> u32 {
            self.batch_size
        }

        /// React to a successful fetch: grow toward the target, or back off if the
        /// response came back bigger than we wanted.
        pub(super) fn record_success(&mut self, log_count: usize) {
            if log_count > self.target {
                self.batch_size = max(
                    MIN_BATCH_SIZE,
                    (self.batch_size as f64 * SHRINK_FACTOR) as u32,
                );
            } else if log_count < self.target / 2 {
                self.batch_size = min(
                    self.settings.max_batch_size,
                    (self.batch_size as f64 * GROWTH_FACTOR) as u32,
                );
            }
        }

        /// React to a range-too-wide error: shrink toward a safer width, or report
        /// that we're already at the floor and this isn't a sizing problem anymore.
        pub(super) fn record_range_too_wide(&mut self) -> ShrinkOutcome {
            let new_batch = max(
                MIN_BATCH_SIZE,
                (self.batch_size as f64 * SHRINK_FACTOR) as u32,
            );
            if new_batch >= self.batch_size {
                return ShrinkOutcome::AtFloor;
            }
            self.batch_size = new_batch;
            ShrinkOutcome::Shrunk(new_batch)
        }

        pub(super) fn stride(&self) -> u64 {
            (self.batch_size * self.settings.max_concurrency) as u64
        }
    }

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

        fn test_settings() -> HttpRpcSettings {
            HttpRpcSettings {
                url: "http://test".parse().unwrap(),
                max_concurrency: 8,
                max_batch_size: 5000,
                init_batch_size: 1000,
                max_logs_per_request: 10_000,
            }
        }

        #[test]
        fn test_dynamic_batch_sizer_new() {
            let settings = test_settings();
            let sizer = DynamicBatchSizer::new(&settings);

            assert_eq!(sizer.batch_size(), 1000); // init_batch_size
            assert_eq!(sizer.target, 9000); // 90% of max_logs_per_request
        }

        #[test]
        fn test_dynamic_batch_sizer_decreases_on_high_count() {
            let settings = test_settings();
            let mut sizer = DynamicBatchSizer::new(&settings);

            let initial = sizer.batch_size();
            sizer.record_success(10_000); // Above target (9000)

            assert!(
                sizer.batch_size() < initial,
                "Batch size should decrease when count exceeds target"
            );
        }

        #[test]
        fn test_dynamic_batch_sizer_increases_on_low_count() {
            let settings = test_settings();
            let mut sizer = DynamicBatchSizer::new(&settings);

            let initial = sizer.batch_size();
            sizer.record_success(1000); // Below target (9000)

            assert!(
                sizer.batch_size() > initial,
                "Batch size should increase when count is below target"
            );
        }

        #[test]
        fn test_dynamic_batch_sizer_respects_max() {
            let settings = test_settings();
            let mut sizer = DynamicBatchSizer::new(&settings);
            sizer.batch_size = 4900; // Near max - same-module test access to the private field

            sizer.record_success(100); // Low count should try to increase

            assert!(
                sizer.batch_size() <= settings.max_batch_size,
                "Batch size should not exceed max"
            );
        }

        #[test]
        fn test_dynamic_batch_sizer_stride() {
            let settings = test_settings();
            let sizer = DynamicBatchSizer::new(&settings);

            // stride = batch_size * max_concurrency = 1000 * 8 = 8000
            assert_eq!(sizer.stride(), 8000);
        }

        #[test]
        fn record_range_too_wide_shrinks_gently_not_a_halving() {
            // 15% step (reusing the existing success-path shrink factor), not 50% - a flat
            // halving overshoots straight past the safe boundary (Alchemy suggested ~400
            // blocks against a 500-wide failing request in production); a gentler step
            // converges closer to it over a few corrections instead of one big overshoot.
            let settings = test_settings();
            let mut sizer = DynamicBatchSizer::new(&settings);
            sizer.batch_size = 500;

            match sizer.record_range_too_wide() {
                ShrinkOutcome::Shrunk(n) => assert_eq!(n, 425),
                ShrinkOutcome::AtFloor => panic!("expected a shrink, not AtFloor"),
            }
            assert_eq!(sizer.batch_size(), 425);

            match sizer.record_range_too_wide() {
                ShrinkOutcome::Shrunk(n) => assert_eq!(n, 361),
                ShrinkOutcome::AtFloor => panic!("expected a shrink, not AtFloor"),
            }
        }

        #[test]
        fn record_range_too_wide_reports_at_floor_when_it_cannot_shrink_further() {
            let settings = test_settings();
            let mut sizer = DynamicBatchSizer::new(&settings);
            sizer.batch_size = 1; // already at MIN_BATCH_SIZE

            match sizer.record_range_too_wide() {
                ShrinkOutcome::AtFloor => {}
                ShrinkOutcome::Shrunk(n) => panic!("expected AtFloor, got Shrunk({n})"),
            }
            // AtFloor must not have mutated batch_size
            assert_eq!(sizer.batch_size(), 1);
        }
    }
}
use sizing::{DynamicBatchSizer, ShrinkOutcome};

/// Backfills logs for a given block range using multiple HTTP providers.
/// This function will cycle through the provided HTTP providers to fetch logs in batches as failures occur.
/// It will yield a stream of ranged log tuples for each successfully fetched batch.
/// Ordering of logs is guaranteed across different providers.
/// Providers are round robin, starting with the first one in the list.
/// The amount of concurrency and batch size is determined by the provider settings and ongoing event output tracking.
///
/// # Error Handling
/// - Each provider tracks errors independently using a sliding window
/// - A provider is suspended after MAX_ERRORS_PER_WINDOW errors within ERROR_WINDOW_SECONDS
/// - Suspended providers are automatically un-suspended when their errors age out
/// - If all providers become suspended, yields an error and terminates
/// - Blocks are never skipped - any unrecoverable error yields Err and stops
///
/// A `from_block` of 0 is clamped to 1: archives and snapshots are 1-based, and
/// block 0 is genesis, which carries no logs on any indexed chain.
///
/// The returned stream is `Send` and may be spawned onto an executor (e.g. `tokio::spawn`).
///
/// Example usage:
/// ```no_run
/// use ethl::rpc::{backfill::{backfill}, config::ProviderSettings, RpcError, LogRange};
/// use futures_util::{pin_mut, StreamExt};
/// use alloy_primitives::Address;
/// use alloy::rpc::types::Filter;
/// use futures_util::Stream;
/// use anyhow::Result;
///
///
/// #[tokio::main]
/// async fn main() -> Result<()> {
///     let settings = ProviderSettings::default();
///     let filter = Filter::new()
///         .from_block(10000000u64);
///     let log_stream = backfill(&settings, &filter).await;
///     pin_mut!(log_stream);
///     while let Some(result) = log_stream.next().await {
///         match result {
///             Ok((from, to, logs)) => println!("Fetched {} logs from blocks {} to {}", logs.len(), from, to),
///             Err(e) => return Err(e.into()),
///         }
///     }
///    Ok(())
/// }
/// ```
pub async fn backfill(
    providers: &ProviderSettings,
    filter: &Filter,
) -> impl Stream<Item = Result<LogRange, RpcError>> + Send {
    let init_from_block = filter.get_from_block().unwrap_or(0);
    let init_to_block = filter
        .get_to_block()
        .expect("Filter::to_block must be set for backfill");

    let num_providers = providers.http_providers.len();

    stream! {
        // Initialize per-provider error tracking
        let mut provider_trackers: Vec<ProviderErrorTracker> = (0..num_providers)
            .map(|i| ProviderErrorTracker::new(providers.http_settings(i).host()))
            .collect();

        let mut provider_index = 0;
        // Closed range [from, to]; next_from is the first block not yet fetched. The
        // loop guard below is the emptiness test on the remaining interval, and every
        // iteration strictly advances next_from past the block(s) it just fetched.
        //
        // Archives and snapshots are 1-based - block 0 is genesis and carries no logs
        // on any indexed chain - so the cursor never starts below block 1. `.max(1)`
        // states that policy in one place rather than reintroducing a branch.
        let mut next_from: u64 = init_from_block.max(1);
        let progress_start_block = next_from;
        let total_blocks = if init_to_block >= progress_start_block {
            init_to_block - progress_start_block + 1
        } else {
            0
        };
        let started_at = Instant::now();
        let mut next_progress_log = started_at + BACKFILL_PROGRESS_LOG_INTERVAL;

        let mut gate = TipGate::new();
        while next_from <= init_to_block {
            // Find the next non-suspended provider
            let active_provider = find_active_provider(&mut provider_trackers, provider_index);
            let Some(idx) = active_provider else {
                yield Err(RpcError::AllProvidersSuspended(format!(
                    "All {} providers suspended at block {}. Unable to continue backfill.",
                    num_providers,
                    next_from
                )));
                return;
            };
            provider_index = idx;

            let provider = providers.connect_http(provider_index);
            let settings = providers.http_settings(provider_index);
            let mut sizer = DynamicBatchSizer::new(settings);

            'fetch: loop {
                if next_from > init_to_block {
                    break;
                }

                let from_block = next_from;
                let to_block = min(from_block + sizer.stride() - 1, init_to_block);
                let concurrency = max(1, ((to_block - from_block) + 1) / sizer.batch_size() as u64);

                let batch_filter = filter.clone().from_block(from_block).to_block(to_block);

                debug!(
                    "Fetching logs from block {} to {} using provider {} (c={}, b={})",
                    from_block, to_block, settings.host(), concurrency, sizer.batch_size()
                );

                let log_vector = get_logs_concurrently(&provider, batch_filter, concurrency as usize).await;
                match log_vector {
                    Ok((max_count, logs)) => {
                        debug!("Captured {} logs from blocks {} to {}", logs.len(), from_block, to_block);
                        sizer.record_success(max_count);

                        // 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 provider's reported tip before
                        // advancing the cursor.
                        if logs.is_empty() {
                            match provider.get_block_number().await {
                                Ok(reported_tip) => {
                                    if next_cursor_after_batch(to_block, true, reported_tip).is_err() {
                                        match wait_for_tip(
                                            to_block,
                                            reported_tip,
                                            &mut gate,
                                            providers,
                                            &settings.host(),
                                            || provider.get_block_number(),
                                        )
                                        .await
                                        {
                                            TipWaitOutcome::CaughtUp => continue 'fetch,
                                            TipWaitOutcome::Terminal {
                                                to_block,
                                                reported_tip,
                                                stalled_cycles,
                                            } => {
                                                yield Err(RpcError::ProviderStalled {
                                                    to_block,
                                                    reported_tip,
                                                    stalled_cycles,
                                                });
                                                return;
                                            }
                                            TipWaitOutcome::RepollFailed(err) => {
                                                provider_trackers[provider_index]
                                                    .record_and_backoff(
                                                        "tip re-poll failed during tip-wait",
                                                        from_block,
                                                        to_block,
                                                        &err,
                                                    )
                                                    .await;
                                                break 'fetch;
                                            }
                                        }
                                    }
                                }
                                Err(e) => {
                                    provider_trackers[provider_index]
                                        .record_and_backoff(
                                            "tip fetch failed during empty-batch guard",
                                            from_block,
                                            to_block,
                                            &e,
                                        )
                                        .await;
                                    break;
                                }
                            }
                        }

                        next_from = to_block + 1;
                        let now = Instant::now();
                        if total_blocks > 0 && now >= next_progress_log {
                            let completed_blocks =
                                next_from.saturating_sub(progress_start_block).min(total_blocks);
                            let progress_pct = completed_blocks as f64 / total_blocks as f64 * 100.0;
                            info!(
                                "Backfill progress: {} / {} blocks ({:.1}%), up to block {} in {}s",
                                completed_blocks,
                                total_blocks,
                                progress_pct,
                                to_block,
                                started_at.elapsed().as_secs()
                            );
                            next_progress_log = now + BACKFILL_PROGRESS_LOG_INTERVAL;
                        }
                        provider_trackers[provider_index].record_success();
                        yield Ok((from_block, to_block, logs));
                        gate.reset();
                    }
                    Err(e) => {
                        let error_msg = e.to_string();
                        let error_category = ErrorCategory::from_error_msg(&error_msg);

                        match error_category {
                            ErrorCategory::RangeTooWide => {
                                let previous_batch = sizer.batch_size();
                                match sizer.record_range_too_wide() {
                                    ShrinkOutcome::Shrunk(new_batch) => {
                                        warn!(
                                            "Range too wide for blocks {}-{} from {}, reducing batch size from {} to {} (error: {})",
                                            from_block, to_block, settings.host(), previous_batch, new_batch,
                                            &error_msg[..std::cmp::min(100, error_msg.len())]
                                        );
                                        sleep(Duration::from_secs(BASE_BACKOFF_SECS)).await;
                                        continue;
                                    }
                                    ShrinkOutcome::AtFloor => {
                                        // Can't shrink further - this is no longer a sizing
                                        // problem, it's the same "this provider isn't
                                        // working" signal as RateLimit/Connection/Other.
                                        // Fall through to the same tracked suspend-and-switch
                                        // handling below instead of a special-cased hard stop.
                                        provider_trackers[provider_index]
                                            .record_and_backoff(
                                                "range too wide, batch size at floor",
                                                from_block,
                                                to_block,
                                                &e,
                                            )
                                            .await;
                                    }
                                }
                            }
                            ErrorCategory::RateLimit | ErrorCategory::Connection | ErrorCategory::Other => {
                                provider_trackers[provider_index]
                                    .record_and_backoff("error fetching logs", from_block, to_block, &e)
                                    .await;
                            }
                        }

                        // Switch to next provider. RangeTooWide's Shrunk case already
                        // `continue`d above and never reaches this line.
                        break;
                    }
                }
            }

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

/// Fetch logs for a given filter using the specified provider with concurrency.
async fn get_logs_concurrently(
    provider: &DynProvider<Ethereum>,
    filter: Filter,
    concurrency: usize,
) -> Result<(usize, Vec<Log>), RpcError> {
    let chunks = chunk_filter(&filter, concurrency);
    let futures = chunks
        .into_iter()
        .map(|chunk| get_logs_bisecting(provider, chunk));

    let resolved = join_all(futures).await;

    let mut max_logs = 0;
    let mut result = vec![];
    for item in resolved {
        match item {
            Ok(logs) => {
                max_logs = max(max_logs, logs.len());
                result.extend(logs);
            }
            Err(e) => {
                return Err(e);
            }
        }
    }

    trace!(
        "Range {:?}->{:?} logs={:?}",
        &filter.get_from_block().unwrap(),
        &filter.get_to_block().unwrap(),
        result.len()
    );

    Ok((max_logs, result))
}

async fn get_logs_bisecting(
    provider: &DynProvider<Ethereum>,
    filter: Filter,
) -> Result<Vec<Log>, RpcError> {
    let logs = provider.get_logs(&filter).await;

    match logs {
        Ok(logs) => Ok(logs),
        Err(e) => {
            let from_block = filter.get_from_block().unwrap();
            let to_block = filter.get_to_block().unwrap();
            if e.to_string().contains("max") && to_block != from_block {
                let mid_block = (from_block + to_block) / 2;
                let left_query = filter.clone().from_block(from_block).to_block(mid_block);
                let right_query = filter.clone().from_block(mid_block + 1).to_block(to_block);

                let left_logs = Box::pin(get_logs_bisecting(provider, left_query)).await;
                let right_logs = Box::pin(get_logs_bisecting(provider, right_query)).await;

                match (left_logs, right_logs) {
                    (Ok(left), Ok(right)) => {
                        let mut merged_logs = left;
                        merged_logs.extend(right);
                        Ok(merged_logs)
                    }
                    (Err(e), _) => Err(e),
                    (_, Err(e)) => Err(e),
                }
            } else {
                Err(RpcError::LogFetchError(truncate_error(e)))
            }
        }
    }
}

fn chunk_filter(filter: &Filter, num_chunks: usize) -> Vec<Filter> {
    let from_block = filter.get_from_block().unwrap();
    let to_block = filter.get_to_block().unwrap();
    let span = (to_block - from_block + 1) as f64;
    let chunk_size = (span / num_chunks as f64).ceil() as usize;
    if chunk_size <= 1 {
        return vec![filter.clone()];
    }

    let mut filters = Vec::with_capacity(num_chunks);
    let mut current_from_block = from_block;
    while current_from_block <= to_block {
        let current_to_block = min(current_from_block + chunk_size as u64 - 1, to_block);
        filters.push(
            filter
                .clone()
                .from_block(current_from_block)
                .to_block(current_to_block),
        );
        current_from_block = current_to_block + 1;
    }

    filters
}

#[cfg(test)]
mod test {
    use super::*;
    use alloy::rpc::types::Filter;

    // ==================== chunk_filter tests ====================

    #[test]
    fn test_chunk_filter() {
        let filter = Filter::new().from_block(0).to_block(99);
        let chunks = chunk_filter(&filter, 10);
        assert_eq!(chunks.len(), 10);
        assert_eq!(chunks[0].get_from_block().unwrap(), 0);
        assert_eq!(chunks[0].get_to_block().unwrap(), 9);
        assert_eq!(chunks[9].get_from_block().unwrap(), 90);
        assert_eq!(chunks[9].get_to_block().unwrap(), 99);
    }

    #[test]
    fn test_chunk_filter_iter() {
        let filter = Filter::new().from_block(0).to_block(100);
        for i in 1..101 {
            let chunks = chunk_filter(&filter, i);
            assert_eq!(chunks.last().unwrap().get_to_block().unwrap(), 100);
            assert!(chunks.len() <= 101_usize);
        }
    }

    #[test]
    fn test_chunk_filter_single_block() {
        let filter = Filter::new().from_block(100).to_block(100);
        let chunks = chunk_filter(&filter, 10);
        assert_eq!(chunks.len(), 1);
        assert_eq!(chunks[0].get_from_block().unwrap(), 100);
        assert_eq!(chunks[0].get_to_block().unwrap(), 100);
    }

    #[test]
    fn chunk_filter_handles_num_chunks_far_beyond_u16_range() {
        // u16::MAX is 65_535 - proof the old u16 parameter was a real ceiling, not just a
        // latent risk. Reverting num_chunks back to u16 would make this fail to COMPILE
        // (100_000 doesn't fit in a u16 literal), which is a legitimate - if unusual - way
        // for "would fail if reverted" to hold: the regression is caught at the type level,
        // not by an assertion. span=1_000_000 with num_chunks=100_000 keeps chunk_size at
        // 10 (comfortably above 1), so this exercises the real splitting loop rather than
        // the chunk_size<=1 early-return path (see Boundaries in the task header - that
        // early-return is a separate, pre-existing quirk, not part of this fix).
        let num_chunks = 100_000usize;
        let filter = Filter::new().from_block(0).to_block(999_999u64);
        let chunks = chunk_filter(&filter, num_chunks);
        assert_eq!(chunks.len(), num_chunks);
        assert_eq!(chunks.first().unwrap().get_from_block().unwrap(), 0);
        assert_eq!(chunks.last().unwrap().get_to_block().unwrap(), 999_999);
    }

    // ==================== tip-wait integration tests ====================

    /// Provider tip starts below to_block, then catches up after one re-poll.
    /// The stream must eventually yield the range with no data loss and no ProviderStalled.
    /// Response queue (in FIFO order):
    ///   1. get_logs → []             (first fetch, empty, past tip)
    ///   2. get_block_number → 99     (initial tip check; to_block=100 > 99 → past tip)
    ///   3. get_block_number → 100    (re-poll inside tip-wait; tip caught up)
    ///   4. get_logs → []             (re-fetch same range; logs genuinely empty)
    ///   5. get_block_number → 100    (tip check after re-fetch; within tip, fall through)
    #[tokio::test]
    async fn backfill_tip_wait_advancing_tip_yields_range() {
        use alloy::providers::ProviderBuilder;
        use alloy::transports::mock::Asserter;
        use futures_util::StreamExt;
        use futures_util::pin_mut;

        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!([])); // get_logs → empty
        asserter.push_success(&serde_json::json!("0x63")); // get_block_number → 99
        asserter.push_success(&serde_json::json!("0x64")); // get_block_number → 100 (caught up)
        asserter.push_success(&serde_json::json!([])); // get_logs → empty (within tip)
        asserter.push_success(&serde_json::json!("0x64")); // get_block_number → 100 (within tip)

        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let settings = ProviderSettings::from_mock(provider); // tip_wait_backoff_secs=0

        let filter = Filter::new().from_block(100u64).to_block(100u64);
        let stream = backfill(&settings, &filter).await;
        pin_mut!(stream);

        let result = stream.next().await.expect("expected one range");
        let (from, to, logs) = result.expect("range should not error");
        assert_eq!(from, 100);
        assert_eq!(to, 100);
        assert!(logs.is_empty()); // genuinely empty within tip

        assert!(stream.next().await.is_none(), "stream should be exhausted");
    }

    /// Provider tip advances every cycle but never reaches to_block — stream must yield
    /// `ProviderStalled` due to the wall-clock budget, NOT the flat-tip gate.
    /// This is the 0.1.22 blind spot: an advancing tip never triggered `TipWait::Stall`.
    ///
    /// Uses `start_paused = true`: each `sleep(1s)` call auto-advances virtual time by 1s
    /// when no other tasks are runnable, so the test is deterministic and completes instantly.
    /// `tip_wait_backoff_secs = 1` is required (zero backoff never suspends, so time never
    /// auto-advances).
    ///
    /// Response queue (in FIFO order):
    ///   1. get_logs → []             (first fetch, empty, past tip)
    ///   2. get_block_number → 95     (initial tip check; to_block=100 > 95 → CursorPastTip)
    ///   3. get_block_number → 96     (re-poll cycle 1; tip advanced 95→96, still < 100)
    ///   4. get_block_number → 97     (re-poll cycle 2; tip advanced 96→97, still < 100)
    ///
    /// Flow: T=0 wait_started; sleep×2 auto-advances T to 2s; budget check (elapsed=2s≥2s) fires
    /// at start of cycle 3, yielding ProviderStalled even though the tip advanced every cycle.
    #[tokio::test(start_paused = true)]
    async fn backfill_tip_wait_wall_clock_budget_exceeded_yields_provider_stalled() {
        use alloy::providers::ProviderBuilder;
        use alloy::transports::mock::Asserter;
        use futures_util::StreamExt;
        use futures_util::pin_mut;

        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!([])); // get_logs → empty
        asserter.push_success(&serde_json::json!("0x5f")); // get_block_number → 95 (seed)
        asserter.push_success(&serde_json::json!("0x60")); // re-poll cycle 1 → 96 (advancing)
        asserter.push_success(&serde_json::json!("0x61")); // re-poll cycle 2 → 97 (advancing)

        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        // backoff=1s so sleep() suspends and virtual time auto-advances per cycle;
        // budget=2s fires after 2 sleeps (T=2s); stall_cycles=10 prevents flat-tip gate.
        let settings = ProviderSettings::from_mock(provider)
            .with_tip_wait_backoff_secs(1)
            .with_tip_wait_max_total_secs(2)
            .with_tip_wait_max_stall_cycles(10);

        let filter = Filter::new().from_block(100u64).to_block(100u64);
        let stream = backfill(&settings, &filter).await;
        pin_mut!(stream);

        let result = stream.next().await.expect("expected error result");
        // Tip advanced 95→96→97 (gate always returned Wait) but budget fires at cycle 3 (T=2s).
        // stalled_cycles=3 = local wait_cycle on the wall-clock path.
        assert!(
            matches!(
                result,
                Err(RpcError::ProviderStalled {
                    to_block: 100,
                    reported_tip: 97,
                    stalled_cycles: 3,
                })
            ),
            "expected ProviderStalled from wall-clock budget, got {:?}",
            result
        );
    }

    /// Provider tip stays flat for max_stall_cycles — stream must yield ProviderStalled.
    /// tip_wait_max_stall_cycles=2 → stalls after 2 flat re-polls following the seed.
    #[tokio::test]
    async fn backfill_tip_wait_stalled_tip_yields_provider_stalled() {
        use alloy::providers::ProviderBuilder;
        use alloy::transports::mock::Asserter;
        use futures_util::StreamExt;
        use futures_util::pin_mut;

        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!([])); // get_logs → empty
        asserter.push_success(&serde_json::json!("0x63")); // initial get_block_number → 99
        // Iteration 1: observe(99,2) → Wait (seeds last_tip); re-poll → 99
        asserter.push_success(&serde_json::json!("0x63"));
        // Iteration 2: observe(99,2) → stalled_cycles=1, Wait; re-poll → 99
        asserter.push_success(&serde_json::json!("0x63"));
        // Iteration 3: observe(99,2) → stalled_cycles=2 >= 2 → Stall(2)

        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let settings = ProviderSettings::from_mock(provider).with_tip_wait_max_stall_cycles(2); // hit stall faster

        let filter = Filter::new().from_block(100u64).to_block(100u64);
        let stream = backfill(&settings, &filter).await;
        pin_mut!(stream);

        let result = stream.next().await.expect("expected error result");
        assert!(
            matches!(
                result,
                Err(RpcError::ProviderStalled {
                    to_block: 100,
                    reported_tip: 99,
                    stalled_cycles: 2,
                })
            ),
            "expected ProviderStalled, got {:?}",
            result
        );
    }
}