babelforce-manager-sdk 0.46.0

//! Automatic request retries, implemented as transport-layer middleware — the Rust analogue of
//! the TS SDK's `fetch` wrapper (`retry.ts`) and the Go SDK's `RoundTripper` (`retry.go`).
//!
//! Transient failures — network errors and a configurable set of "try again" status codes
//! (429/502/503/504 by default) — are retried with exponential backoff and full jitter, honouring
//! a `Retry-After` header (delta-seconds or HTTP-date, capped at [`RetryPolicy::max_delay`]) when
//! present. Idempotency derives from the HTTP method: non-idempotent methods (POST/PATCH) are
//! only retried on `429`, where the server explicitly rejected the request without acting on it.

use std::collections::HashSet;
use std::time::Duration;

use rand::Rng;
use reqwest::{Method, Request, Response};
use reqwest_middleware::Next;

/// Automatic-retry tuning. See the module docs for the retry semantics; disable with
/// [`RetryPolicy::disabled`] or tune via [`crate::ManagerClientBuilder::retry`].
#[derive(Clone, Debug)]
pub struct RetryPolicy {
    /// Retry attempts after the initial request. `0` disables retries.
    pub max_retries: u32,
    /// Base backoff; grows exponentially per attempt (with full jitter).
    pub base_delay: Duration,
    /// Upper bound on any single backoff. Also caps a server-sent `Retry-After`.
    pub max_delay: Duration,
    /// Response status codes that trigger a retry. Default: 429, 502, 503, 504.
    pub retry_status: Vec<u16>,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            max_retries: 2,
            base_delay: Duration::from_millis(250),
            max_delay: Duration::from_secs(10),
            retry_status: vec![429, 502, 503, 504],
        }
    }
}

impl RetryPolicy {
    /// A policy that performs no retries.
    pub fn disabled() -> Self {
        Self {
            max_retries: 0,
            ..Self::default()
        }
    }
}

/// Methods safe to replay after a network error or retryable 5xx.
fn idempotent(method: &Method) -> bool {
    matches!(
        *method,
        Method::GET | Method::HEAD | Method::PUT | Method::DELETE | Method::OPTIONS
    )
}

/// Parse a `Retry-After` header (delta-seconds or HTTP-date) into a wait duration.
fn retry_after(resp: &Response) -> Option<Duration> {
    let header = resp
        .headers()
        .get(reqwest::header::RETRY_AFTER)?
        .to_str()
        .ok()?;
    if let Ok(secs) = header.trim().parse::<u64>() {
        return Some(Duration::from_secs(secs));
    }
    let when = httpdate::parse_http_date(header.trim()).ok()?;
    Some(
        when.duration_since(std::time::SystemTime::now())
            .unwrap_or(Duration::ZERO),
    )
}

/// The reqwest middleware that performs the retries. Installed once on the shared HTTP client by
/// `ManagerClientBuilder::connect`, so every request — generated-client calls, the facade's raw
/// JSON paths and the `/oauth/token` grants — gets the same retry semantics.
pub(crate) struct RetryMiddleware {
    policy: RetryPolicy,
    codes: HashSet<u16>,
}

impl RetryMiddleware {
    pub(crate) fn new(policy: RetryPolicy) -> Self {
        let codes = policy.retry_status.iter().copied().collect();
        RetryMiddleware { policy, codes }
    }

    /// Exponential backoff with full jitter, capped at `max_delay`.
    fn backoff(&self, attempt: u32) -> Duration {
        let factor = 1u32.checked_shl(attempt.min(16)).unwrap_or(u32::MAX);
        let ceiling = self
            .policy
            .base_delay
            .saturating_mul(factor)
            .min(self.policy.max_delay);
        let ceiling_ms = u64::try_from(ceiling.as_millis()).unwrap_or(u64::MAX);
        Duration::from_millis(rand::rng().random_range(0..=ceiling_ms))
    }

    /// Decide whether to retry and, if so, how long to wait first.
    fn retry_wait(
        &self,
        attempt: u32,
        idempotent: bool,
        result: &reqwest_middleware::Result<Response>,
    ) -> Option<Duration> {
        if attempt >= self.policy.max_retries {
            return None;
        }
        match result {
            // Network-level failure: only safe to replay idempotent requests.
            Err(_) => idempotent.then(|| self.backoff(attempt)),
            Ok(resp) => {
                let status = resp.status().as_u16();
                if !self.codes.contains(&status) {
                    return None;
                }
                // Non-idempotent methods only retry on 429 (rejected without acting on it).
                if !idempotent && status != 429 {
                    return None;
                }
                Some(
                    retry_after(resp)
                        .map(|d| d.min(self.policy.max_delay))
                        .unwrap_or_else(|| self.backoff(attempt)),
                )
            }
        }
    }
}

#[async_trait::async_trait]
impl reqwest_middleware::Middleware for RetryMiddleware {
    async fn handle(
        &self,
        req: Request,
        extensions: &mut http::Extensions,
        next: Next<'_>,
    ) -> reqwest_middleware::Result<Response> {
        if self.policy.max_retries == 0 {
            return next.run(req, extensions).await;
        }
        let idempotent = idempotent(req.method());
        let mut attempt: u32 = 0;
        loop {
            // Streaming bodies (e.g. multipart file uploads) can't be replayed — single attempt.
            let Some(attempt_req) = req.try_clone() else {
                return next.run(req, extensions).await;
            };
            let result = next.clone().run(attempt_req, extensions).await;
            let Some(wait) = self.retry_wait(attempt, idempotent, &result) else {
                return result;
            };
            // Drop the abandoned response before sleeping so its connection is released.
            drop(result);
            tokio::time::sleep(wait).await;
            attempt += 1;
        }
    }
}

/// Invoke `f` — retained as the facade's uniform call seam. Retries themselves happen in
/// [`RetryMiddleware`] on the shared HTTP client, where response headers (`Retry-After`) are
/// visible and idempotency derives from the HTTP method — matching the TS/Go SDKs. The
/// `_policy`/`_idempotent` arguments document call-site intent and keep every resource's call
/// shape stable.
pub(crate) async fn with_retry<F, Fut, R, E>(
    _policy: &RetryPolicy,
    _idempotent: bool,
    mut f: F,
) -> Result<R, E>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<R, E>>,
{
    f().await
}