alloy-transport 2.3.0

use alloy_json_rpc::{ErrorPayload, Id, RpcError, RpcResult};
use serde::Deserialize;
use serde_json::value::RawValue;
use std::{error::Error as StdError, fmt::Debug, time::Duration};
use thiserror::Error;

/// A transport error is an [`RpcError`] containing a [`TransportErrorKind`].
pub type TransportError<ErrResp = Box<RawValue>> = RpcError<TransportErrorKind, ErrResp>;

/// A transport result is a [`Result`] containing a [`TransportError`].
pub type TransportResult<T, ErrResp = Box<RawValue>> = RpcResult<T, TransportErrorKind, ErrResp>;

/// Transport error.
///
/// All transport errors are wrapped in this enum.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum TransportErrorKind {
    /// Missing batch response.
    ///
    /// This error is returned when a batch request is sent and the response
    /// does not contain a response for a request. For convenience the ID is
    /// specified.
    #[error("missing response for request with ID {0}")]
    MissingBatchResponse(Id),

    /// Backend connection task has stopped.
    #[error("backend connection task has stopped")]
    BackendGone,

    /// Pubsub service is not available for the current provider.
    #[error("subscriptions are not available on this provider")]
    PubsubUnavailable,

    /// HTTP Error with code and body
    #[error("{0}")]
    HttpError(#[from] HttpError),

    /// HTTP error with a server-provided retry delay.
    ///
    /// Error statuses (4xx/5xx) are always considered retryable since the server explicitly
    /// requested a retry. Redirects are not: `Retry-After` on a 3xx delays the redirected
    /// request, not a replay of the original one.
    #[error("{error}")]
    HttpErrorWithRetryAfter {
        /// The HTTP error.
        #[source]
        error: HttpError,
        /// The delay requested by the server.
        retry_after: Duration,
    },

    /// Custom error.
    #[error("{0}")]
    Custom(#[source] Box<dyn StdError + Send + Sync + 'static>),

    /// Deterministic, terminal failure that retry loops should not retry.
    ///
    /// Use this when the operation is guaranteed to fail again on retry —
    /// for example a malformed request, a permission error, or any other
    /// protocol-level failure that is not a transient socket / IO problem.
    #[error("{0}")]
    NonRetryable(#[source] Box<dyn StdError + Send + Sync + 'static>),
}

impl TransportErrorKind {
    /// Returns `true` if the error is potentially recoverable.
    /// This is a naive heuristic and should be used with caution.
    pub const fn recoverable(&self) -> bool {
        matches!(self, Self::MissingBatchResponse(_))
    }

    /// Instantiate a new `TransportError` from a custom error.
    pub fn custom_str(err: &str) -> TransportError {
        RpcError::Transport(Self::Custom(err.into()))
    }

    /// Instantiate a new `TransportError` from a custom error.
    pub fn custom(err: impl StdError + Send + Sync + 'static) -> TransportError {
        RpcError::Transport(Self::Custom(Box::new(err)))
    }

    /// Instantiate a new non-retryable `TransportError` from a string.
    pub fn non_retryable_str(err: &str) -> TransportError {
        RpcError::Transport(Self::NonRetryable(err.into()))
    }

    /// Instantiate a new non-retryable `TransportError` from a custom error.
    pub fn non_retryable(err: impl StdError + Send + Sync + 'static) -> TransportError {
        RpcError::Transport(Self::NonRetryable(Box::new(err)))
    }

    /// Returns true if this is [`TransportErrorKind::NonRetryable`].
    pub const fn is_non_retryable(&self) -> bool {
        matches!(self, Self::NonRetryable(_))
    }

    /// Instantiate a new `TransportError` from a missing ID.
    pub const fn missing_batch_response(id: Id) -> TransportError {
        RpcError::Transport(Self::MissingBatchResponse(id))
    }

    /// Instantiate a new `TransportError::BackendGone`.
    pub const fn backend_gone() -> TransportError {
        RpcError::Transport(Self::BackendGone)
    }

    /// Instantiate a new `TransportError::PubsubUnavailable`.
    pub const fn pubsub_unavailable() -> TransportError {
        RpcError::Transport(Self::PubsubUnavailable)
    }

    /// Instantiate a new `TransportError::HttpError`.
    pub const fn http_error(status: u16, body: String) -> TransportError {
        RpcError::Transport(Self::HttpError(HttpError { status, body }))
    }

    /// Instantiate a new HTTP error that optionally carries the retry delay requested by the
    /// server via a `Retry-After` header.
    pub const fn http_error_with_retry_after(
        status: u16,
        body: String,
        retry_after: Option<Duration>,
    ) -> TransportError {
        match retry_after {
            Some(retry_after) => RpcError::Transport(Self::HttpErrorWithRetryAfter {
                error: HttpError { status, body },
                retry_after,
            }),
            None => Self::http_error(status, body),
        }
    }

    /// Returns true if this is [`TransportErrorKind::PubsubUnavailable`].
    pub const fn is_pubsub_unavailable(&self) -> bool {
        matches!(self, Self::PubsubUnavailable)
    }

    /// Returns true if this is [`TransportErrorKind::BackendGone`].
    pub const fn is_backend_gone(&self) -> bool {
        matches!(self, Self::BackendGone)
    }

    /// Returns true if this is an HTTP error.
    pub const fn is_http_error(&self) -> bool {
        matches!(self, Self::HttpError(_) | Self::HttpErrorWithRetryAfter { .. })
    }

    /// Returns the [`HttpError`] if this is an HTTP error.
    pub const fn as_http_error(&self) -> Option<&HttpError> {
        match self {
            Self::HttpError(err) => Some(err),
            Self::HttpErrorWithRetryAfter { error, .. } => Some(error),
            _ => None,
        }
    }

    /// Returns the server-provided retry delay, if present.
    pub const fn retry_after(&self) -> Option<Duration> {
        match self {
            Self::HttpErrorWithRetryAfter { retry_after, .. } => Some(*retry_after),
            _ => None,
        }
    }

    /// Returns the custom error if this is [`TransportErrorKind::Custom`].
    pub const fn as_custom(&self) -> Option<&(dyn StdError + Send + Sync + 'static)> {
        match self {
            Self::Custom(err) => Some(&**err),
            _ => None,
        }
    }

    /// Analyzes the [TransportErrorKind] and decides if the request should be retried based on the
    /// variant.
    pub fn is_retry_err(&self) -> bool {
        match self {
            // Missing batch response errors can be retried.
            Self::MissingBatchResponse(_) => true,
            Self::HttpError(http_err) => {
                http_err.is_rate_limit_err() || http_err.is_temporarily_unavailable()
            }
            // The server explicitly requested a retry with a delay. Redirects are excluded
            // because their `Retry-After` delays the redirected request, not a replay.
            Self::HttpErrorWithRetryAfter { error, .. } => error.status >= 400,
            Self::Custom(err) => {
                let msg = err.to_string();
                msg.contains("429 Too Many Requests")
            }
            _ => false,
        }
    }
}

/// Type for holding HTTP errors such as 429 rate limit error.
#[derive(Debug, thiserror::Error)]
#[error(
    "HTTP error {status} with {}",
    if body.is_empty() { "empty body".to_string() } else { format!("body: {body}") }
)]
pub struct HttpError {
    /// The HTTP status code.
    pub status: u16,
    /// The HTTP response body.
    pub body: String,
}

impl HttpError {
    /// Checks the `status` to determine whether the request should be retried.
    pub const fn is_rate_limit_err(&self) -> bool {
        self.status == 429
    }

    /// Checks the `status` to determine whether the service was temporarily unavailable and should
    /// be retried.
    pub const fn is_temporarily_unavailable(&self) -> bool {
        self.status == 503
    }
}

/// Extension trait to implement methods for [`RpcError<TransportErrorKind, E>`].
pub(crate) trait RpcErrorExt {
    /// Analyzes whether to retry the request depending on the error.
    fn is_retryable(&self) -> bool;

    /// Fetches the backoff hint from the error message if present
    fn backoff_hint(&self) -> Option<std::time::Duration>;
}

impl RpcErrorExt for RpcError<TransportErrorKind> {
    fn is_retryable(&self) -> bool {
        match self {
            // There was a transport-level error. This is either a non-retryable error,
            // or a server error that should be retried.
            Self::Transport(err) => err.is_retry_err(),
            // The transport could not serialize the error itself. The request was malformed from
            // the start.
            Self::SerError(_) => false,
            Self::DeserError { text, .. } => {
                if let Ok(resp) = serde_json::from_str::<ErrorPayload>(text) {
                    return resp.is_retry_err();
                }

                // some providers send invalid JSON RPC in the error case (no `id:u64`), but the
                // text should be a `JsonRpcError`
                #[derive(Deserialize)]
                struct Resp {
                    error: ErrorPayload,
                }

                if let Ok(resp) = serde_json::from_str::<Resp>(text) {
                    return resp.error.is_retry_err();
                }

                false
            }
            Self::ErrorResp(err) => err.is_retry_err(),
            Self::NullResp => true,
            _ => false,
        }
    }

    fn backoff_hint(&self) -> Option<std::time::Duration> {
        // Server-provided hints are clamped so that a misbehaving server cannot stall retries
        // with an absurd delay.
        raw_backoff_hint(self).map(|hint| hint.min(MAX_BACKOFF_HINT))
    }
}

/// Maximum server-provided backoff honored, whether from a `Retry-After` header or an error
/// payload.
const MAX_BACKOFF_HINT: Duration = Duration::from_secs(5 * 60);

/// Extracts the backoff hint from the error, unbounded.
fn raw_backoff_hint(err: &RpcError<TransportErrorKind>) -> Option<Duration> {
    if let RpcError::Transport(TransportErrorKind::HttpErrorWithRetryAfter {
        retry_after, ..
    }) = err
    {
        return Some(*retry_after);
    }

    if let RpcError::ErrorResp(resp) = err {
        // try to extract backoff from the error data (infura-style)
        let data = resp.try_data_as::<serde_json::Value>();
        if let Some(Ok(data)) = data {
            // if daily rate limit exceeded, infura returns the requested backoff in the error
            // response
            let backoff_seconds = &data["rate"]["backoff_seconds"];
            // infura rate limit error
            if let Some(seconds) = backoff_seconds.as_u64() {
                return Some(Duration::from_secs(seconds));
            }
            if let Some(seconds) = backoff_seconds.as_f64() {
                return Some(Duration::from_secs((seconds as u64).saturating_add(1)));
            }
        }

        // try to extract backoff from the error message, e.g. "try again in 4ms"
        if let Some(duration) = parse_retry_after(&resp.message) {
            return Some(duration);
        }
    }
    None
}

/// Parses a duration from messages like "try again in 4ms", "try again in 1s".
fn parse_retry_after(message: &str) -> Option<std::time::Duration> {
    let after = message.split_once("try again in ")?.1.trim_start();

    let digits_len = after.as_bytes().iter().take_while(|b| b.is_ascii_digit()).count();
    let (digits, rest) = after.split_at(digits_len);
    let value: u64 = digits.parse().ok()?;

    let unit = rest.trim().trim_end_matches(|c: char| c.is_ascii_punctuation());
    match unit {
        "ms" => Some(std::time::Duration::from_millis(value)),
        "s" => Some(std::time::Duration::from_secs(value)),
        _ => None,
    }
}

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

    #[test]
    fn test_retry_error() {
        let err = "{\"code\":-32007,\"message\":\"100/second request limit reached - reduce calls per second or upgrade your account at quicknode.com\"}";
        let err = serde_json::from_str::<ErrorPayload>(err).unwrap();
        assert!(TransportError::ErrorResp(err).is_retryable());
    }

    #[test]
    fn test_retry_error_rate_limited() {
        let err = r#"{"code":-32005,"message":"rate limited, try again in 4ms","data":null}"#;
        let err = serde_json::from_str::<ErrorPayload>(err).unwrap();
        let err = TransportError::ErrorResp(err);
        assert!(err.is_retryable());
        assert_eq!(err.backoff_hint(), Some(std::time::Duration::from_millis(4)));
    }

    #[test]
    fn parse_retry_after_millis() {
        assert_eq!(
            parse_retry_after("try again in 4ms"),
            Some(std::time::Duration::from_millis(4))
        );
        assert_eq!(
            parse_retry_after("rate limited, try again in 100ms"),
            Some(std::time::Duration::from_millis(100))
        );
    }

    #[test]
    fn parse_retry_after_seconds() {
        assert_eq!(parse_retry_after("try again in 2s"), Some(std::time::Duration::from_secs(2)));
    }

    #[test]
    fn parse_retry_after_none() {
        assert_eq!(parse_retry_after("some other error"), None);
        assert_eq!(parse_retry_after("try again in"), None);
        assert_eq!(parse_retry_after("try again in ms"), None);
        assert_eq!(parse_retry_after("try again in 4us"), None);
    }

    #[test]
    fn test_retry_error_429() {
        let err = r#"{"code":429,"event":-33200,"message":"Too Many Requests","details":"You have surpassed your allowed throughput limit. Reduce the amount of requests per second or upgrade for more capacity."}"#;
        let err = serde_json::from_str::<ErrorPayload>(err).unwrap();
        assert!(TransportError::ErrorResp(err).is_retryable());
    }

    #[test]
    fn http_retry_after_retryable_for_error_statuses() {
        let err = TransportErrorKind::http_error_with_retry_after(
            500,
            String::new(),
            Some(Duration::from_secs(1)),
        );
        assert!(err.is_retryable());
        assert_eq!(err.backoff_hint(), Some(Duration::from_secs(1)));

        // without the header a plain 500 stays non-retryable
        assert!(!TransportErrorKind::http_error(500, String::new()).is_retryable());

        // redirects are not replayed even with a Retry-After
        let err = TransportErrorKind::http_error_with_retry_after(
            301,
            String::new(),
            Some(Duration::from_secs(1)),
        );
        assert!(!err.is_retryable());
    }

    #[test]
    fn backoff_hint_clamped() {
        let err = TransportErrorKind::http_error_with_retry_after(
            429,
            String::new(),
            Some(Duration::from_secs(u64::MAX)),
        );
        assert_eq!(err.backoff_hint(), Some(MAX_BACKOFF_HINT));
    }

    #[test]
    fn http_retry_after_preserves_http_error() {
        let err = TransportErrorKind::http_error_with_retry_after(
            429,
            "Too Many Requests".to_owned(),
            Some(Duration::from_secs(52)),
        );

        assert!(err.is_retryable());
        assert_eq!(err.backoff_hint(), Some(Duration::from_secs(52)));

        let TransportError::Transport(kind) = err else { panic!("expected transport error") };
        assert!(kind.is_http_error());
        assert_eq!(kind.retry_after(), Some(Duration::from_secs(52)));
        assert_eq!(kind.as_http_error().unwrap().status, 429);
    }
}