ethl 0.1.34

//! Common error tracking utilities for RPC providers.
//!
//! This module provides shared error tracking infrastructure used by both
//! the backfill (HTTP) and heads (WebSocket) streaming modules.

use std::time::{Duration, Instant};
use tokio::time::sleep;
use tracing::{info, warn};

use crate::rpc::config;

/// Maximum length for error messages before truncation
pub const MAX_ERROR_MSG_LEN: usize = 512;

/// Maximum errors allowed within the sliding window before suspension
pub const MAX_ERRORS_PER_WINDOW: usize = config::MAX_RETRIES as usize;

/// Duration of the sliding window for error tracking (10 minutes)
pub const ERROR_WINDOW_SECONDS: u64 = 600;

/// Truncates error messages that are too long for logging
pub fn truncate_error(err: impl std::fmt::Display) -> String {
    let msg = err.to_string();
    if msg.len() <= MAX_ERROR_MSG_LEN {
        msg
    } else {
        format!(
            "{}... [truncated {} chars]",
            &msg[..MAX_ERROR_MSG_LEN],
            msg.len() - MAX_ERROR_MSG_LEN
        )
    }
}

/// Truncates error message to a specified length for inline logging
pub fn truncate_error_short(err: impl std::fmt::Display, max_len: usize) -> String {
    let msg = err.to_string();
    if msg.len() <= max_len {
        msg
    } else {
        msg[..max_len].to_string()
    }
}

/// Categorizes RPC errors for appropriate handling
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ErrorCategory {
    /// Query range too wide: either the payload exceeded a size limit, or the
    /// provider timed out scanning the range and told us to shrink it. Both are
    /// the same underlying signal - reduce batch size and retry the same provider.
    RangeTooWide,
    /// Rate limit hit - apply backoff and retry
    RateLimit,
    /// Connection error - switch provider
    Connection,
    /// Other/unknown error - count towards suspension
    Other,
}

impl ErrorCategory {
    /// Categorize an error message into the appropriate category
    pub fn from_error_msg(error_msg: &str) -> Self {
        // Range-too-wide errors: an outright size limit, or a provider-side
        // timeout that names the range as the cause (checked ahead of the
        // generic "timeout" match in the Connection block below, since these
        // messages often also contain the word "timeout").
        if error_msg.contains("32081")
            || error_msg.contains("too large")
            || error_msg.contains("max results")
            || error_msg.contains("deserialization")
            || error_msg.contains("EOF")
            || error_msg.contains("truncated")
            || error_msg.contains("decoding response")
            || error_msg.contains("block range")
            || error_msg.contains("response size")
            || (error_msg.contains("reduc") && error_msg.contains("range"))
        {
            return Self::RangeTooWide;
        }

        // Rate limit errors
        if error_msg.contains("429")
            || error_msg.contains("Too Many Requests")
            || error_msg.contains("rate limit")
            || error_msg.contains("Rate limit")
        {
            return Self::RateLimit;
        }

        // Connection errors
        if error_msg.contains("Connection reset")
            || error_msg.contains("connection")
            || error_msg.contains("timeout")
            || error_msg.contains("Timeout")
        {
            return Self::Connection;
        }

        Self::Other
    }
}

/// Tracks error timestamps for a single provider using a sliding window.
///
/// This struct maintains a record of recent errors and can suspend a provider
/// if too many errors occur within the sliding window period.
pub struct ProviderErrorTracker {
    /// Timestamps of recent errors within the sliding window
    error_timestamps: Vec<Instant>,
    /// Whether this provider is currently suspended
    suspended: bool,
    /// Provider identifier for logging (hostname or endpoint name)
    identifier: String,
}

impl ProviderErrorTracker {
    /// Creates a new error tracker for a provider
    pub fn new(identifier: impl Into<String>) -> Self {
        Self {
            error_timestamps: Vec::new(),
            suspended: false,
            identifier: identifier.into(),
        }
    }

    /// Records an error and returns true if the provider should be suspended
    pub fn record_error(&mut self) -> bool {
        let now = Instant::now();
        self.cleanup_old_errors(now);
        self.error_timestamps.push(now);

        if self.error_timestamps.len() >= MAX_ERRORS_PER_WINDOW {
            self.suspended = true;
            warn!(
                "Provider {} suspended: {} errors in {} seconds",
                self.identifier,
                self.error_timestamps.len(),
                ERROR_WINDOW_SECONDS
            );
        }

        self.suspended
    }

    /// Clears error history on successful request
    pub fn record_success(&mut self) {
        self.error_timestamps.clear();
    }

    /// Removes errors older than the sliding window
    fn cleanup_old_errors(&mut self, now: Instant) {
        let window = Duration::from_secs(ERROR_WINDOW_SECONDS);
        self.error_timestamps
            .retain(|&ts| now.duration_since(ts) < window);

        // Un-suspend if errors have aged out
        if self.suspended && self.error_timestamps.len() < MAX_ERRORS_PER_WINDOW {
            info!("Provider {} un-suspended: errors aged out", self.identifier);
            self.suspended = false;
        }
    }

    /// Returns true if the provider is currently suspended
    pub fn is_suspended(&mut self) -> bool {
        self.cleanup_old_errors(Instant::now());
        self.suspended
    }

    /// Returns the current error count within the sliding window
    pub fn error_count(&mut self) -> usize {
        self.cleanup_old_errors(Instant::now());
        self.error_timestamps.len()
    }

    /// Returns the provider identifier
    pub fn identifier(&self) -> &str {
        &self.identifier
    }

    /// Calculates the backoff duration based on current error count
    pub fn backoff_duration(&mut self) -> Duration {
        let error_count = self.error_count();
        Duration::from_secs(config::BASE_BACKOFF_SECS.pow(error_count as u32))
    }

    /// Records a fetch failure against this provider's sliding window, logs one of two
    /// messages depending on whether this trips suspension, and sleeps the resulting
    /// backoff before returning - no sleep if the provider just got suspended, since
    /// there's no point backing off; the caller should move on immediately. Always
    /// truncates the error before logging (some call sites previously didn't, risking
    /// an unbounded error body landing in the log).
    pub async fn record_and_backoff(
        &mut self,
        context: &str,
        from_block: u64,
        to_block: u64,
        err: impl std::fmt::Display,
    ) {
        let suspended = self.record_error();
        let error_count = self.error_count();
        if suspended {
            warn!(
                "Provider {} suspended after {} for blocks {}-{}: {}",
                self.identifier,
                context,
                from_block,
                to_block,
                truncate_error(&err)
            );
        } else {
            let backoff = self.backoff_duration();
            warn!(
                "{} for blocks {}-{} from {}: {} (error {}/{}, backoff {:?})",
                context,
                from_block,
                to_block,
                self.identifier,
                truncate_error(&err),
                error_count,
                MAX_ERRORS_PER_WINDOW,
                backoff
            );
            sleep(backoff).await;
        }
    }
}

/// Finds the next active (non-suspended) provider starting from the given index.
/// Returns None if all providers are suspended.
pub fn find_active_provider(
    trackers: &mut [ProviderErrorTracker],
    start_index: usize,
) -> Option<usize> {
    let num_providers = trackers.len();
    for i in 0..num_providers {
        let idx = (start_index + i) % num_providers;
        if !trackers[idx].is_suspended() {
            return Some(idx);
        }
    }
    None
}

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

    #[test]
    fn test_error_category_size_errors() {
        assert_eq!(
            ErrorCategory::from_error_msg("Response 32081 too large"),
            ErrorCategory::RangeTooWide
        );
        assert_eq!(
            ErrorCategory::from_error_msg("max results exceeded"),
            ErrorCategory::RangeTooWide
        );
        assert_eq!(
            ErrorCategory::from_error_msg("deserialization failed"),
            ErrorCategory::RangeTooWide
        );
        assert_eq!(
            ErrorCategory::from_error_msg("error decoding response body"),
            ErrorCategory::RangeTooWide
        );
    }

    /// One case per literal substring `from_error_msg` checks, plus boundary cases
    /// that must NOT match despite superficial overlap (e.g. "range" alone). Each
    /// message is built to trip exactly one signal so deleting or typo-ing any
    /// single branch in `from_error_msg` fails exactly one case here. Messages are
    /// representative shapes, not verified vendor quotes, except the Alchemy one,
    /// which is verbatim from a production eth_getLogs timeout (robinhood-mainnet,
    /// captured 2026-07-15).
    #[test]
    fn error_category_covers_expected_message_classes() {
        let cases: &[(&str, ErrorCategory)] = &[
            // --- RangeTooWide: outright size-limit errors ---
            (
                "json-rpc error -32081: request rejected",
                ErrorCategory::RangeTooWide,
            ),
            (
                "the response payload was too large to return",
                ErrorCategory::RangeTooWide,
            ),
            (
                "query exceeded max results for this endpoint",
                ErrorCategory::RangeTooWide,
            ),
            (
                "deserialization error while parsing the RPC response",
                ErrorCategory::RangeTooWide,
            ),
            (
                "unexpected EOF while reading response body",
                ErrorCategory::RangeTooWide,
            ),
            (
                "response body was truncated before parsing completed",
                ErrorCategory::RangeTooWide,
            ),
            (
                "error decoding response: invalid UTF-8",
                ErrorCategory::RangeTooWide,
            ),
            // --- RangeTooWide: scan-timeout errors that name the range, not the payload ---
            (
                "the requested block range exceeds this node's limit",
                ErrorCategory::RangeTooWide,
            ),
            (
                "exceeded the response size limit for this request",
                ErrorCategory::RangeTooWide,
            ),
            (
                "please reduce the width of your query range and try again",
                ErrorCategory::RangeTooWide,
            ),
            (
                // Verbatim Alchemy eth_getLogs timeout body. Contains "timeout" but must
                // classify as RangeTooWide (checked ahead of the Connection block), or it
                // gets treated as a generic connection failure: counted against the
                // suspension budget and round-robined to the next provider instead of
                // shrinking the batch.
                "Query timeout exceeded. Consider reducing your block range. Based on your \
                 parameters and the response size limit, this block range should work: \
                 [0x7b3958, 0x7b3ae8]",
                ErrorCategory::RangeTooWide,
            ),
            // --- RateLimit ---
            (
                "HTTP error 429 returned by upstream",
                ErrorCategory::RateLimit,
            ),
            ("HTTP status: Too Many Requests", ErrorCategory::RateLimit),
            (
                "you have hit the rate limit for this endpoint",
                ErrorCategory::RateLimit,
            ),
            (
                "Rate limit exceeded, retry after 5s",
                ErrorCategory::RateLimit,
            ),
            // --- Connection ---
            (
                "Connection reset by peer while streaming response",
                ErrorCategory::Connection,
            ),
            (
                "could not establish connection to host",
                ErrorCategory::Connection,
            ),
            (
                "connection attempt timed out, retrying",
                ErrorCategory::Connection,
            ),
            (
                "Timeout waiting for upstream response",
                ErrorCategory::Connection,
            ),
            // --- Other: genuine, unrelated JSON-RPC errors must not be swept into
            //     RangeTooWide, RateLimit, or Connection ---
            (
                "execution reverted: insufficient balance for transfer",
                ErrorCategory::Other,
            ),
            (
                r#"{"code":-32602,"message":"invalid argument 0: json: cannot unmarshal hex string of odd length"}"#,
                ErrorCategory::Other,
            ),
            // --- Boundary: contains "range" but not "block range" or "reduc"+"range" -
            //     must not false-positive into RangeTooWide just because the word
            //     "range" appears in an unrelated context ---
            (
                "value out of range for uint256 conversion",
                ErrorCategory::Other,
            ),
            ("index out of range: 5 >= len 3", ErrorCategory::Other),
        ];

        for (msg, expected) in cases {
            assert_eq!(
                ErrorCategory::from_error_msg(msg),
                *expected,
                "message {:?} should classify as {:?}",
                msg,
                expected
            );
        }
    }

    #[test]
    fn test_error_category_rate_limit() {
        assert_eq!(
            ErrorCategory::from_error_msg("HTTP error: 429 Too Many Requests"),
            ErrorCategory::RateLimit
        );
        assert_eq!(
            ErrorCategory::from_error_msg("rate limit exceeded"),
            ErrorCategory::RateLimit
        );
    }

    #[test]
    fn test_error_category_connection() {
        assert_eq!(
            ErrorCategory::from_error_msg("Connection reset without closing handshake"),
            ErrorCategory::Connection
        );
        assert_eq!(
            ErrorCategory::from_error_msg("request timeout"),
            ErrorCategory::Connection
        );
    }

    #[test]
    fn test_provider_error_tracker() {
        let mut tracker = ProviderErrorTracker::new("test-provider");
        assert!(!tracker.is_suspended());
        assert_eq!(tracker.error_count(), 0);

        // Record some errors but not enough to suspend
        for _ in 0..(MAX_ERRORS_PER_WINDOW - 1) {
            assert!(!tracker.record_error());
        }
        assert!(!tracker.is_suspended());
        assert_eq!(tracker.error_count(), MAX_ERRORS_PER_WINDOW - 1);

        // One more error should suspend
        assert!(tracker.record_error());
        assert!(tracker.is_suspended());

        // Success should clear errors
        tracker.record_success();
        assert_eq!(tracker.error_count(), 0);
        // Note: suspended state is only cleared by cleanup_old_errors, not record_success
    }

    #[test]
    fn test_truncate_error() {
        let short_msg = "short error";
        assert_eq!(truncate_error(short_msg), short_msg);

        let long_msg = "x".repeat(1000);
        let truncated = truncate_error(&long_msg);
        assert!(truncated.len() < long_msg.len());
        assert!(truncated.contains("[truncated"));
    }

    #[test]
    fn test_find_active_provider() {
        let mut trackers = vec![
            ProviderErrorTracker::new("provider-0"),
            ProviderErrorTracker::new("provider-1"),
            ProviderErrorTracker::new("provider-2"),
        ];

        // All active - should return start index
        assert_eq!(find_active_provider(&mut trackers, 0), Some(0));
        assert_eq!(find_active_provider(&mut trackers, 1), Some(1));

        // Suspend first provider
        for _ in 0..MAX_ERRORS_PER_WINDOW {
            trackers[0].record_error();
        }
        assert_eq!(find_active_provider(&mut trackers, 0), Some(1));

        // Suspend all providers
        for tracker in &mut trackers {
            for _ in 0..MAX_ERRORS_PER_WINDOW {
                tracker.record_error();
            }
        }
        assert_eq!(find_active_provider(&mut trackers, 0), None);
    }

    #[tokio::test(start_paused = true)]
    async fn record_and_backoff_sleeps_until_suspended_then_stops_sleeping() {
        // Under MAX_ERRORS_PER_WINDOW calls: each one should have recorded an error
        // (error_count increments) - can't easily assert the sleep happened without
        // timing the test, so assert the state transition instead, which is what
        // actually matters: suspension flips exactly at the configured threshold.
        let mut tracker = ProviderErrorTracker::new("test-provider");
        for _ in 0..(MAX_ERRORS_PER_WINDOW - 1) {
            tracker.record_and_backoff("test error", 1, 2, "boom").await;
            assert!(!tracker.is_suspended());
        }
        tracker.record_and_backoff("test error", 1, 2, "boom").await;
        assert!(tracker.is_suspended());
    }
}