zeph-channels 0.22.4

// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

//! Shared HTTP 429 retry-with-backoff helper for channel REST clients.
//!
//! Discord, Telegram, and Slack all rate-limit via HTTP 429 with a
//! `Retry-After` header (or, for Discord and some Telegram responses, a JSON
//! body field of the same name). This module centralises the retry loop so
//! each channel's REST client only has to build the request.

use std::time::Duration;

use serde::Deserialize;

/// Upper bound applied to any `Retry-After` value, whether from the header or body.
const MAX_RETRY_SECS: f64 = 60.0;
/// Maximum number of retry attempts before giving up and surfacing the error.
const MAX_RETRIES: u32 = 3;

/// Guards against `MAX_RETRY_SECS` ever being edited into a negative,
/// non-finite, or overflow-inducing value: `delay_secs.min(MAX_RETRY_SECS)`
/// would silently let such a value flow into [`Duration::from_secs_f64`],
/// which panics on all three cases (negative/NaN, ±infinity, and magnitudes
/// beyond `Duration::MAX`) — the same panic fixed for parsed `Retry-After`
/// values by [`valid_retry_secs`]. The 3600.0 (one hour) ceiling bounds how
/// large a single backoff step can grow if `MAX_RETRY_SECS` is ever edited;
/// the cumulative worst case across all retries (`MAX_RETRIES *
/// MAX_RETRY_SECS`) is documented in the `# Timing` section of
/// [`send_with_retry`] for whatever value is configured.
const _: () = assert!(
    MAX_RETRY_SECS.is_finite() && MAX_RETRY_SECS >= 0.0 && MAX_RETRY_SECS <= 3600.0,
    "MAX_RETRY_SECS must be finite, non-negative, and at most 3600.0 (1 hour): \
     out-of-range values reach Duration::from_secs_f64, which panics on negative, \
     non-finite, or overflowing input"
);

/// Body shape used by rate-limit responses that carry `retry_after` as JSON.
#[derive(Deserialize, Default)]
struct RateLimitBody {
    retry_after: Option<f64>,
}

/// Discards a `Retry-After` value that is negative or non-finite (`NaN`/`±inf`),
/// treating it the same as a missing or unparseable value so it cannot reach
/// [`Duration::from_secs_f64`], which panics on such inputs.
fn valid_retry_secs(secs: Option<f64>) -> Option<f64> {
    secs.filter(|v| v.is_finite() && *v >= 0.0)
}

/// Executes a request with automatic 429 retry-after backoff.
///
/// Builds a fresh request each attempt via `make_req`. On HTTP 429 the function
/// reads the `Retry-After` header (falling back to the JSON body's `retry_after`
/// field, then a 1 s default), discarding negative or non-finite values from
/// either source as if they were unparseable, clamps to [`MAX_RETRY_SECS`],
/// sleeps, and retries up to [`MAX_RETRIES`] times. When retries are exhausted
/// a final request is issued to obtain a `reqwest::Error` with the original
/// HTTP status — `reqwest::Error` cannot be constructed directly.
///
/// `context` is a short label (e.g. `"discord"`, `"telegram"`, `"slack"`) attached
/// to the warning logs so retries can be attributed to the originating channel.
///
/// # Timing
///
/// Under sustained 429s, the worst-case wall-clock across all attempts is
/// approximately `(MAX_RETRIES + 2) * <per-request timeout>` (each attempt's
/// own `send()`, including the final un-retried attempt) plus
/// `MAX_RETRIES * MAX_RETRY_SECS` (the backoff sleeps between them) — with the
/// current constants, on the order of minutes, not seconds. Callers that await
/// this on a hot path without an outer timeout should account for this.
///
/// # Errors
///
/// Returns a [`reqwest::Error`] when all retries are exhausted, a non-429 HTTP
/// error is received, or the request's own timeout is exceeded.
pub(crate) async fn send_with_retry<F>(
    context: &str,
    make_req: F,
) -> Result<reqwest::Response, reqwest::Error>
where
    F: Fn() -> reqwest::RequestBuilder,
{
    let mut attempts = 0u32;
    loop {
        let resp = make_req().send().await?;

        if resp.status() != reqwest::StatusCode::TOO_MANY_REQUESTS {
            return resp.error_for_status();
        }

        // Parse retry delay: header wins, then body field, then default 1 s.
        // Negative/NaN/infinite values are treated as absent (see `valid_retry_secs`)
        // so they can't reach `Duration::from_secs_f64`, which panics on them.
        let header_secs = valid_retry_secs(
            resp.headers()
                .get(reqwest::header::RETRY_AFTER)
                .and_then(|v| v.to_str().ok())
                .and_then(|s| s.parse::<f64>().ok()),
        );

        let body_secs = valid_retry_secs(
            resp.json::<RateLimitBody>()
                .await
                .unwrap_or_default()
                .retry_after,
        );

        let delay_secs = header_secs.or(body_secs).unwrap_or(1.0).min(MAX_RETRY_SECS);

        attempts += 1;
        if attempts > MAX_RETRIES {
            tracing::warn!(
                context,
                delay_secs,
                attempts,
                "rate-limited and retries exhausted"
            );
            // Surface as an error by issuing the request once more without retry.
            return make_req().send().await?.error_for_status();
        }

        tracing::warn!(
            context,
            delay_secs,
            attempt = attempts,
            max = MAX_RETRIES,
            "rate-limited (429), backing off"
        );
        tokio::time::sleep(Duration::from_secs_f64(delay_secs)).await;
    }
}

#[cfg(test)]
mod tests {
    use wiremock::matchers::{method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    use super::*;

    #[tokio::test]
    async fn send_with_retry_succeeds_on_200() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"id": "1"})))
            .mount(&server)
            .await;

        let client = reqwest::Client::new();
        let url = format!("{}/channels/ch1/messages", server.uri());
        let resp = send_with_retry("test", || client.post(&url)).await.unwrap();
        assert_eq!(resp.status(), 200);
    }

    #[tokio::test]
    async fn send_with_retry_retries_on_429_then_succeeds() {
        let server = MockServer::start().await;

        // First call → 429 with Retry-After header.
        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(
                ResponseTemplate::new(429)
                    .append_header("Retry-After", "0")
                    .set_body_json(serde_json::json!({"retry_after": 0.0})),
            )
            .up_to_n_times(1)
            .mount(&server)
            .await;

        // Second call → 200.
        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"id": "2"})))
            .mount(&server)
            .await;

        let client = reqwest::Client::new();
        let url = format!("{}/channels/ch1/messages", server.uri());
        let resp = send_with_retry("test", || client.post(&url)).await.unwrap();
        assert_eq!(resp.status(), 200);
    }

    #[tokio::test]
    async fn send_with_retry_uses_body_retry_after_when_no_header() {
        let server = MockServer::start().await;

        // Three 429 responses without Retry-After header but with body field.
        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(
                ResponseTemplate::new(429).set_body_json(serde_json::json!({"retry_after": 0.0})),
            )
            .up_to_n_times(3)
            .mount(&server)
            .await;

        // Fourth → 200.
        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"id": "3"})))
            .mount(&server)
            .await;

        let client = reqwest::Client::new();
        let url = format!("{}/channels/ch1/messages", server.uri());
        let resp = send_with_retry("test", || client.post(&url)).await.unwrap();
        assert_eq!(resp.status(), 200);
    }

    // #6737: negative/invalid Retry-After falls back to a real 1s default sleep in
    // send_with_retry; start_paused fast-forwards it. Safe specifically because `client`
    // below is `reqwest::Client::new()`, which arms no connect/request timeout — there is no
    // competing timer for tokio's paused-clock auto-advance to race against while this test
    // waits on the real (loopback) wiremock response. This does NOT generalize to any client
    // configured with a real timeout (see chat_429_rate_limit_propagates in zeph-llm's
    // openai/tests.rs, which arms connect_timeout(30s)/timeout(600s) and is exposed to exactly
    // that race).
    #[tokio::test(start_paused = true)]
    async fn send_with_retry_ignores_negative_retry_after_header() {
        let server = MockServer::start().await;

        // 429 with a negative Retry-After header must not panic when converted
        // to a Duration — it should fall back to the 1 s default instead.
        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(ResponseTemplate::new(429).append_header("Retry-After", "-5"))
            .up_to_n_times(1)
            .mount(&server)
            .await;

        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"id": "4"})))
            .mount(&server)
            .await;

        let client = reqwest::Client::new();
        let url = format!("{}/channels/ch1/messages", server.uri());
        let resp = send_with_retry("test", || client.post(&url)).await.unwrap();
        assert_eq!(resp.status(), 200);
    }

    // #6737: see send_with_retry_ignores_negative_retry_after_header for why start_paused
    // is safe here (no-timeout client, not a general wiremock property).
    #[tokio::test(start_paused = true)]
    async fn send_with_retry_ignores_negative_retry_after_body() {
        let server = MockServer::start().await;

        // 429 with no header but a negative retry_after body field must also
        // fall back to the default instead of reaching Duration::from_secs_f64.
        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(
                ResponseTemplate::new(429).set_body_json(serde_json::json!({"retry_after": -1.0})),
            )
            .up_to_n_times(1)
            .mount(&server)
            .await;

        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"id": "5"})))
            .mount(&server)
            .await;

        let client = reqwest::Client::new();
        let url = format!("{}/channels/ch1/messages", server.uri());
        let resp = send_with_retry("test", || client.post(&url)).await.unwrap();
        assert_eq!(resp.status(), 200);
    }

    #[test]
    fn valid_retry_secs_rejects_negative_and_non_finite() {
        assert_eq!(valid_retry_secs(Some(-5.0)), None);
        assert_eq!(valid_retry_secs(Some(f64::NAN)), None);
        assert_eq!(valid_retry_secs(Some(f64::INFINITY)), None);
        assert_eq!(valid_retry_secs(Some(f64::NEG_INFINITY)), None);
        assert_eq!(valid_retry_secs(None), None);
        assert_eq!(valid_retry_secs(Some(0.0)), Some(0.0));
        assert_eq!(valid_retry_secs(Some(2.5)), Some(2.5));
    }

    #[tokio::test]
    async fn send_with_retry_propagates_non_429_errors() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(ResponseTemplate::new(403))
            .mount(&server)
            .await;

        let client = reqwest::Client::new();
        let url = format!("{}/channels/ch1/messages", server.uri());
        let result = send_with_retry("test", || client.post(&url)).await;
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.status(), Some(reqwest::StatusCode::FORBIDDEN));
    }

    #[tokio::test]
    async fn send_with_retry_errors_when_retries_exhausted() {
        let server = MockServer::start().await;

        // Return 429 for all requests — exhausts MAX_RETRIES (3) then the final attempt.
        Mock::given(method("POST"))
            .and(path("/channels/ch1/messages"))
            .respond_with(
                ResponseTemplate::new(429)
                    .append_header("Retry-After", "0")
                    .set_body_json(serde_json::json!({"retry_after": 0.0})),
            )
            .mount(&server)
            .await;

        let client = reqwest::Client::new();
        let url = format!("{}/channels/ch1/messages", server.uri());
        let result = send_with_retry("test", || client.post(&url)).await;
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.status(), Some(reqwest::StatusCode::TOO_MANY_REQUESTS));
    }

    #[test]
    fn rate_limit_body_defaults_to_none() {
        let body: RateLimitBody = serde_json::from_str("{}").unwrap();
        assert!(body.retry_after.is_none());
    }

    #[test]
    fn rate_limit_body_parses_float() {
        let body: RateLimitBody = serde_json::from_str(r#"{"retry_after": 1.5}"#).unwrap();
        assert!((body.retry_after.unwrap() - 1.5).abs() < f64::EPSILON);
    }

    #[test]
    fn max_retry_secs_clamps() {
        let unclamped: f64 = 120.0;
        assert_eq!(
            unclamped.min(MAX_RETRY_SECS).to_bits(),
            MAX_RETRY_SECS.to_bits()
        );
    }
}