anodizer-core 0.16.1

//! Uniform retry-with-exponential-backoff primitives.
//!
//! Replaces six open-coded retry loops in `stage-docker` (3×) and
//! `stage-release` (3×) that had diverged on backoff formulas —
//! `2^(n-2)`, `2^(n-1)`, and `500 << (attempt-1)` all coexisted.
//!
//! The canonical policy is exponential backoff with multiplier 2 starting at
//! `base_delay` and capped at `max_delay`:
//!
//! ```text
//! attempt 1:  f() executes immediately
//! attempt 2:  sleep base_delay
//! attempt 3:  sleep base_delay * 2
//! attempt N:  sleep min(base_delay * 2^(N-2), max_delay)
//! ```
//!
//! `ControlFlow<Break, Continue>` lets the operation decide retry policy per
//! failure (e.g. 4xx → Break, 5xx → Continue) without the helper encoding
//! protocol-specific predicates.
//!
//! Both a sync (`retry_sync`) and async (`retry_async`) variant are provided so
//! that sites can adopt without crossing a sync/async boundary.

use std::error::Error as StdError;
use std::fmt;
use std::io;
use std::ops::ControlFlow;
use std::time::Duration;

use crate::log::StageLogger;

/// Names the operation a retry engine is driving and carries the logger that
/// surfaces per-attempt failures.
///
/// A required parameter on every retry engine — not an optional builder — so
/// a silent retry is unrepresentable: a backoff ladder can sleep for many
/// minutes (10 attempts × 5m cap), and an operator watching a run must be
/// able to tell "waiting on a transient failure" from "hung".
#[derive(Clone, Copy)]
pub struct RetryLog<'a> {
    desc: &'a str,
    log: &'a StageLogger,
}

impl<'a> RetryLog<'a> {
    /// `desc` is a short human description of the operation being retried
    /// (e.g. `"chocolatey push"`, `"mastodon announce"`); it prefixes every
    /// per-attempt warn line.
    pub fn new(desc: &'a str, log: &'a StageLogger) -> Self {
        Self { desc, log }
    }

    /// The operation description supplied at construction.
    pub fn desc(&self) -> &str {
        self.desc
    }

    fn warn_retry(&self, attempt: u32, max: u32, cause: &dyn fmt::Display, delay: Duration) {
        // Spelled through the tool's one duration format (`45s`, `2m15s`) so a
        // retry line and an adjacent heartbeat line read the same way.
        self.log.warn(&format!(
            "{} attempt {}/{} failed ({}); retrying in {}",
            self.desc,
            attempt,
            max,
            cause,
            crate::progress::format_elapsed(delay)
        ));
    }

    /// Warn that the ladder exhausted its attempts (or wall-clock budget) and is
    /// giving up after `attempts` tries. Paired with the error the engine then
    /// returns: the error names *what* failed, this line records that the
    /// retries themselves are spent so a watcher does not wait for more.
    fn warn_giving_up(&self, attempts: u32) {
        self.log.warn(&format!(
            "{} failed after {} attempt(s), giving up",
            self.desc, attempts
        ));
    }

    /// Note (default-visible) that the operation recovered after `attempts`
    /// tries — the transient failure cleared. Only emitted once at least one
    /// retry has happened, so a clean first attempt stays silent.
    fn note_succeeded(&self, attempts: u32) {
        // status, not warn: a recovered transient is a positive per-operation
        // result an operator wants at default verbosity, mirroring the
        // rollback/dry-run default events — not a command echo.
        self.log.status(&format!(
            "{} succeeded after {} attempt(s)",
            self.desc, attempts
        )); // status-ok: recovered-after-retry is a per-operation result event
    }
}

/// Retry policy used by `retry_sync` / `retry_async`.
#[derive(Debug, Clone, Copy)]
pub struct RetryPolicy {
    /// Total attempts, including the first.
    ///
    /// Invariant: must be `>= 1`. The clamp is enforced at two layers so
    /// every construction path is safe:
    ///
    /// 1. [`crate::config::RetryConfig::to_policy`] clamps user YAML
    ///    (`attempts: 0` -> `1`) at the config-surface boundary.
    /// 2. [`retry_sync`] / [`retry_async`] clamp again at the loop boundary
    ///    to protect direct `RetryPolicy { max_attempts: 0, .. }`
    ///    constructions (e.g. test fixtures).
    ///
    /// Callers therefore do NOT need to clamp `max_attempts` again at the
    /// call site.
    pub max_attempts: u32,
    /// Delay before the second attempt (no wait before the first).
    pub base_delay: Duration,
    /// Upper bound on any individual sleep between attempts.
    pub max_delay: Duration,
}

impl RetryPolicy {
    /// Canonical upload policy: 10 attempts, 50ms
    /// base, 30s cap.
    pub const UPLOAD: RetryPolicy = RetryPolicy {
        max_attempts: 10,
        base_delay: Duration::from_millis(50),
        max_delay: Duration::from_secs(30),
    };

    /// Shallow policy for best-effort pre-publish probes: 3 attempts, 200ms
    /// base, 1s cap.
    ///
    /// Pre-publish probes (token `whoami`, registry index GET, GitHub repo
    /// scope, npm duplicate-version) are an advisory warning gate, not a
    /// write that must land. They run sequentially across every configured
    /// publisher, so the production write-ladder (10 attempts / 10s base /
    /// 5m cap) would let a single wedged endpoint stall the gate for tens of
    /// minutes. A shallow bound keeps the probe responsive while still
    /// absorbing a transient blip; the per-request HTTP timeout still bounds
    /// each individual attempt.
    pub const PREFLIGHT: RetryPolicy = RetryPolicy {
        max_attempts: 3,
        base_delay: Duration::from_millis(200),
        max_delay: Duration::from_secs(1),
    };

    /// Shallow policy for burn-detection guard probes (published-state
    /// registry lookups made before a destructive rollback): 3 attempts, 1s
    /// base, 30s cap.
    ///
    /// A guard consults one registry endpoint per crate/package, and a
    /// multi-crate workspace probes many of them in one pass, so the
    /// production write-ladder (up to ~25 minutes of backoff per operation)
    /// would let a registry outage stall the guard for hours before it can
    /// classify the outcome. A shallow, capped ladder keeps the whole probe
    /// pass bounded while still absorbing a transient blip; the guard's own
    /// fail-closed / fail-open classification handles genuine outages.
    pub const GUARD_PROBE: RetryPolicy = RetryPolicy {
        max_attempts: 3,
        base_delay: Duration::from_secs(1),
        max_delay: Duration::from_secs(30),
    };

    pub fn delay_for(&self, next_attempt: u32) -> Duration {
        // `next_attempt` is the attempt we're about to run (≥2). The wait
        // before attempt 2 uses base_delay; before attempt 3 uses base_delay*2;
        // i.e. multiplier = 2^(next_attempt - 2).
        let exp = next_attempt.saturating_sub(2);
        let mult = 1u64.checked_shl(exp).unwrap_or(u64::MAX);
        let ms = (self.base_delay.as_millis() as u64).saturating_mul(mult);
        std::cmp::min(Duration::from_millis(ms), self.max_delay)
    }

    /// Raise this policy's `max_attempts` to at least [`IDEMPOTENT_PUT_ATTEMPTS`]
    /// without disturbing its backoff shape, returning the adjusted policy.
    ///
    /// An idempotent PUT/POST to a fixed target (an Artifactory/generic upload,
    /// a GemFury push, a Snap Store upload, a bucket blob PUT, a GitHub asset
    /// upload) lands the same bytes at the same path on every re-issue, so a
    /// transient 5xx/429 or dropped connection must retry a bounded number of
    /// times even when a stateful mode (`--publish-only`) resolves the
    /// configured policy down to `attempts: 1`. The floor is a `max()` — it
    /// only widens the bound for the retriable classes and never lowers an
    /// operator-set higher value. 4xx responses still fast-fail inside the
    /// per-attempt classifier regardless of this floor.
    pub fn with_idempotent_floor(self) -> RetryPolicy {
        self.with_floor(IDEMPOTENT_PUT_ATTEMPTS)
    }

    /// Raise this policy's `max_attempts` to at least `min`, leaving the backoff
    /// shape untouched. A `max()` floor, never a clamp that lowers a higher
    /// operator-set value.
    pub fn with_floor(self, min: u32) -> RetryPolicy {
        RetryPolicy {
            max_attempts: self.max_attempts.max(min),
            ..self
        }
    }

    /// Whether the wait before `next_attempt` would carry total wall-time past
    /// `deadline`. Checked before each backoff so a long registry storm exits
    /// cleanly (the last error is returned, and an idempotent write recovers on
    /// re-run) instead of being SIGKILLed mid-publish by the outer job timeout.
    /// Shared by the sync and async ladders so both bound identically.
    ///
    /// A saturating check: an uncapped policy (`max_delay: Duration::MAX`) can
    /// project a backoff so large that `Instant::now() + delay` would overflow;
    /// an overflowing projection is treated as past any real deadline (the ladder
    /// stops) rather than panicking.
    pub fn budget_exhausted(&self, next_attempt: u32, deadline: std::time::Instant) -> bool {
        match std::time::Instant::now().checked_add(self.delay_for(next_attempt)) {
            Some(projected) => projected > deadline,
            None => true,
        }
    }
}

/// Total attempt floor for an idempotent PUT/POST, single-sourcing the
/// "3 total attempts" guarantee shared by every idempotent-upload publisher
/// (HTTP upload, GemFury, Snapcraft, GitHub asset, blob). Applied via
/// [`RetryPolicy::with_idempotent_floor`] as a `max()` so a stateful mode
/// (`--publish-only`) that resolves `attempts: 1` still keeps a bounded
/// transient retry, while an operator-set higher cap is preserved.
pub const IDEMPOTENT_PUT_ATTEMPTS: u32 = 3;

/// Default wall-clock budget for a retry ladder when `retry.max_elapsed` is not
/// set. Resolved into an absolute deadline by [`crate::Context::retry_deadline`]
/// and threaded into the engine by publishers, so a ladder bounded only by
/// attempt count cannot run unbounded on a slow-but-not-failing endpoint. It is
/// a *default*, not a hard ceiling: an operator raises (or lowers) it with
/// `retry.max_elapsed`, and a caller that threads `None` is still unbounded.
pub const DEFAULT_MAX_ELAPSED: Duration = Duration::from_secs(15 * 60);

/// Wall-clock time slept in retry backoff so far this run, in milliseconds.
///
/// A release runs as one process, so a single process-global accumulator
/// captures every stage's backoff without threading a handle through the many
/// independent per-stage retry loops — several of which sleep via an injected
/// callback that has no path to carry a handle. Parallel upload workers add
/// concurrently through the atomic. Read once at summary time via
/// [`total_retry_backoff`]; the run surfaces it as a `retry_backoff_secs`
/// field and an operator status line.
static RETRY_BACKOFF_MILLIS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Per-scope retry tally (backoff sleeps + summed wait), keyed by the label of
/// the enclosing [`RetryScope`]. Backoff recorded with no active scope lands
/// under [`UNATTRIBUTED_SCOPE`] so the per-scope rows always sum to the global
/// total. A `Mutex` (not a lock-free map) is ample: retry sleeps are seconds
/// apart, so contention among the parallel upload workers is negligible.
static PER_SCOPE_RETRY: std::sync::Mutex<std::collections::BTreeMap<String, ScopeRetry>> =
    std::sync::Mutex::new(std::collections::BTreeMap::new());

/// The label backoff is attributed to while a [`RetryScope`] is active. Stages
/// run serially and each installs one scope, so a single global cell suffices:
/// the release stage's parallel upload tasks all read the same constant value
/// ("release") for the stage's duration, and the serial publish loop swaps it
/// per publisher. No task-local is needed because the value never differs
/// between two concurrently-running sleeps.
static CURRENT_SCOPE: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);

/// Bucket key for backoff recorded outside any [`RetryScope`].
const UNATTRIBUTED_SCOPE: &str = "(unattributed)";

#[derive(Clone, Copy, Default)]
struct ScopeRetry {
    /// Number of backoff sleeps (i.e. retries) recorded against this scope.
    retries: u32,
    /// Summed backoff wait for this scope, in milliseconds.
    backoff_ms: u64,
}

/// RAII scope that attributes every backoff sleep recorded during its lifetime
/// to `name` (a publisher or stage label). Restores the previous scope on drop,
/// so nested/sequential scopes compose. Install one around each publisher's
/// `run` and around a stage's whole retrying section.
#[must_use = "the scope only applies while the guard is alive"]
pub struct RetryScope {
    prev: Option<String>,
}

impl RetryScope {
    /// Enter a retry-attribution scope named `name`.
    pub fn enter(name: impl Into<String>) -> Self {
        let mut cur = CURRENT_SCOPE.lock().unwrap_or_else(|e| e.into_inner());
        let prev = cur.replace(name.into());
        RetryScope { prev }
    }
}

impl Drop for RetryScope {
    fn drop(&mut self) {
        *CURRENT_SCOPE.lock().unwrap_or_else(|e| e.into_inner()) = self.prev.take();
    }
}

/// Record a backoff sleep of `d` against this run's total and the active scope.
/// Callers that sleep for retry should prefer [`sleep_backoff_blocking`] /
/// [`sleep_backoff_async`] (which record and sleep together); use this directly
/// only when the sleep is performed elsewhere (e.g. an injected `sleep`
/// callback in stage-sign).
pub fn record_retry_backoff(d: Duration) {
    let ms = u64::try_from(d.as_millis()).unwrap_or(u64::MAX);
    RETRY_BACKOFF_MILLIS.fetch_add(ms, std::sync::atomic::Ordering::Relaxed);

    let key = CURRENT_SCOPE
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .clone()
        .unwrap_or_else(|| UNATTRIBUTED_SCOPE.to_string());
    let mut map = PER_SCOPE_RETRY.lock().unwrap_or_else(|e| e.into_inner());
    let entry = map.entry(key).or_default();
    entry.retries = entry.retries.saturating_add(1);
    entry.backoff_ms = entry.backoff_ms.saturating_add(ms);
}

/// Sleep `d` (blocking) and record it as retry backoff.
///
/// A zero `d` is a no-op: it neither sleeps nor records. A caller that owns its
/// own wait (a rate-limit reset probe that already blocked until quota returned)
/// passes `Duration::ZERO` to re-attempt immediately, and such a re-attempt must
/// not inflate the per-scope "backoff sleeps" counter with a sleep that never
/// happened.
pub fn sleep_backoff_blocking(d: Duration) {
    if d.is_zero() {
        return;
    }
    record_retry_backoff(d);
    std::thread::sleep(d);
}

/// Sleep `d` (async) and record it as retry backoff. A zero `d` is a no-op —
/// see [`sleep_backoff_blocking`].
pub async fn sleep_backoff_async(d: Duration) {
    if d.is_zero() {
        return;
    }
    record_retry_backoff(d);
    tokio::time::sleep(d).await;
}

/// Total wall-clock time slept in retry backoff so far this run.
pub fn total_retry_backoff() -> Duration {
    Duration::from_millis(RETRY_BACKOFF_MILLIS.load(std::sync::atomic::Ordering::Relaxed))
}

/// Per-scope retry breakdown so far this run: `(scope, retries, backoff)` per
/// publisher/stage that backed off, sorted by backoff descending (biggest
/// offender first). Their backoff sums to [`total_retry_backoff`].
pub fn retry_scope_breakdown() -> Vec<(String, u32, Duration)> {
    let map = PER_SCOPE_RETRY.lock().unwrap_or_else(|e| e.into_inner());
    let mut rows: Vec<(String, u32, Duration)> = map
        .iter()
        .map(|(k, v)| (k.clone(), v.retries, Duration::from_millis(v.backoff_ms)))
        .collect();
    // Descending by backoff, then name for a stable tie-break.
    rows.sort_by(|a, b| b.2.cmp(&a.2).then_with(|| a.0.cmp(&b.0)));
    rows
}

/// Retry a synchronous operation according to `policy`.
///
/// `op` returns:
/// - `Ok(T)` on success (no retry).
/// - `Err(ControlFlow::Continue(e))` to retry if attempts remain.
/// - `Err(ControlFlow::Break(e))` to stop immediately (4xx-style fast-fail).
///
/// Returns the last error if all attempts are exhausted.
///
/// This variant is attempt-count-bounded only; a caller that wants a wall-clock
/// budget (a shorter or operator-raised [`DEFAULT_MAX_ELAPSED`]) uses
/// [`retry_sync_deadline`] with the deadline from
/// [`crate::Context::retry_deadline`].
///
/// Every failed attempt that will be retried emits a default-visible warn
/// (`<desc> attempt n/max failed (<cause>); retrying in <delay>`) via `rlog`
/// before the backoff sleep, so a multi-minute ladder is never silent.
pub fn retry_sync<T, E, F>(rlog: RetryLog<'_>, policy: &RetryPolicy, op: F) -> Result<T, E>
where
    E: fmt::Display,
    F: FnMut(u32) -> Result<T, ControlFlow<E, E>>,
{
    retry_sync_deadline(rlog, policy, None, op)
}

/// Like [`retry_sync`], but stops retrying once the next backoff sleep would
/// push total wall-time past `deadline`. On budget exhaustion it returns the
/// last error observed before the budget was hit, so a caller whose write is
/// idempotent recovers on re-run instead of being killed mid-attempt by an
/// outer timeout. `deadline: None` is byte-for-byte the attempt-count-only
/// behavior of [`retry_sync`].
pub fn retry_sync_deadline<T, E, F>(
    rlog: RetryLog<'_>,
    policy: &RetryPolicy,
    deadline: Option<std::time::Instant>,
    mut op: F,
) -> Result<T, E>
where
    E: fmt::Display,
    F: FnMut(u32) -> Result<T, ControlFlow<E, E>>,
{
    retry_steps_sync(rlog, policy.max_attempts, deadline, |attempt| {
        controlflow_to_step(policy, attempt, op(attempt))
    })
}

/// Adapt a [`ControlFlow`]-classified result into a [`RetryStep`], using
/// `policy` for the backoff shape: `Ok` → `Done`, `Break` → `Fail` (fast-fail),
/// `Continue(e)` → `Retry` sleeping `policy.delay_for(attempt + 1)` with `e`'s
/// `Display` as the per-attempt cause. Single-sources the mapping the sync and
/// async [`ControlFlow`] adapters share.
fn controlflow_to_step<T, E: fmt::Display>(
    policy: &RetryPolicy,
    attempt: u32,
    result: Result<T, ControlFlow<E, E>>,
) -> RetryStep<T, E> {
    match result {
        Ok(v) => RetryStep::Done(v),
        Err(ControlFlow::Break(e)) => RetryStep::Fail(e),
        Err(ControlFlow::Continue(e)) => {
            let cause = e.to_string();
            RetryStep::Retry {
                error: e,
                delay: policy.delay_for(attempt + 1),
                cause,
            }
        }
    }
}

/// Retry an asynchronous operation according to `policy`.
///
/// Same semantics as `retry_sync` but awaits `op` and uses `tokio::time::sleep`.
pub async fn retry_async<T, E, F, Fut>(
    rlog: RetryLog<'_>,
    policy: &RetryPolicy,
    op: F,
) -> Result<T, E>
where
    E: fmt::Display,
    F: FnMut(u32) -> Fut,
    Fut: std::future::Future<Output = Result<T, ControlFlow<E, E>>>,
{
    retry_async_deadline(rlog, policy, None, op).await
}

/// Like [`retry_async`], but stops once the next backoff would push total
/// wall-time past `deadline` — the async counterpart of [`retry_sync_deadline`],
/// so async publishers (release-asset uploads, GitLab/Gitea API calls layered on
/// [`retry_http_async`]) can honor the same [`crate::Context::retry_deadline`]
/// budget. `deadline: None` is byte-for-byte the attempt-count-only behavior.
pub async fn retry_async_deadline<T, E, F, Fut>(
    rlog: RetryLog<'_>,
    policy: &RetryPolicy,
    deadline: Option<std::time::Instant>,
    mut op: F,
) -> Result<T, E>
where
    E: fmt::Display,
    F: FnMut(u32) -> Fut,
    Fut: std::future::Future<Output = Result<T, ControlFlow<E, E>>>,
{
    retry_steps_async(rlog, policy.max_attempts, deadline, |attempt| {
        let fut = op(attempt);
        async move { controlflow_to_step(policy, attempt, fut.await) }
    })
    .await
}

/// One attempt's outcome for the step-based retry engines
/// ([`retry_steps_sync`] / [`retry_steps_async`]).
///
/// The operation closure owns *both* classification and the backoff duration;
/// the engine owns everything a hand-rolled loop repeatedly gets wrong — the
/// attempt cap, the wall-clock deadline, backoff accounting, and the full
/// warn / giving-up / succeeded log lifecycle. This is the single primitive
/// every retry ladder in the tree routes through: a publisher that needs a
/// bespoke delay (unjittered exponential, a linear `5·attempt` ladder, a
/// rate-limit reset window) or a bespoke classifier (transient-output markers,
/// index-propagation lag, a partial-upload probe) expresses it in the closure
/// instead of re-implementing the loop and drifting from the others.
///
/// The [`ControlFlow`]-based [`retry_sync`] / [`retry_async`] adapters and the
/// HTTP wrappers are themselves thin layers over these engines, so a fixed
/// [`RetryPolicy`] and a caller-owned delay share one loop, one deadline check,
/// and one set of log lines.
pub enum RetryStep<T, E> {
    /// Stop and succeed with this value. When at least one retry preceded it,
    /// the engine emits the recovery ("succeeded after N attempt(s)") line —
    /// so reserve `Done` for a *clean* success the operator wants confirmed.
    Done(T),
    /// Stop and succeed with this value, but suppress the recovery line. For a
    /// terminal outcome that is success-valued yet carries its own narrative —
    /// an idempotent skip, a tolerated degraded disposition (a kept-stale
    /// asset) — where a "succeeded after N attempt(s)" note would contradict
    /// the closure's own log line rather than confirm a recovery.
    DoneQuiet(T),
    /// Stop and fail with this non-retriable error (a 4xx-style fast-fail).
    /// The engine emits no giving-up line: the operation already classified
    /// this as terminal and knows why.
    Fail(E),
    /// A retriable failure. If an attempt and the wall-clock budget both
    /// remain, the engine sleeps `delay` (recorded as run backoff) and re-runs
    /// the closure; otherwise it stops and returns `error`. `cause` is the
    /// compact, human-readable reason rendered in the per-attempt warn line
    /// (e.g. `"status=503"`, `"sparse-index propagation lag"`).
    Retry {
        error: E,
        delay: Duration,
        cause: String,
    },
}

/// Retry a synchronous operation whose closure owns classification and backoff.
///
/// `max_attempts` bounds the attempt count (clamped to ≥1). `deadline`
/// optionally bounds wall-clock time: before each backoff sleep the engine
/// checks whether `now + delay` would pass it and, if so, stops with the last
/// error (an idempotent write then recovers on re-run instead of being killed
/// mid-attempt by an outer timeout). Every retriable failure emits a
/// default-visible warn before its sleep, an exhausted ladder emits a
/// giving-up warn, and a recovery after ≥1 retry emits a succeeded line — the
/// one retry-log lifecycle shared by every ladder.
pub fn retry_steps_sync<T, E, F>(
    rlog: RetryLog<'_>,
    max_attempts: u32,
    deadline: Option<std::time::Instant>,
    mut op: F,
) -> Result<T, E>
where
    F: FnMut(u32) -> RetryStep<T, E>,
{
    let max = max_attempts.max(1);
    let mut attempt: u32 = 1;
    loop {
        match op(attempt) {
            RetryStep::Done(v) => {
                if attempt > 1 {
                    rlog.note_succeeded(attempt);
                }
                return Ok(v);
            }
            RetryStep::DoneQuiet(v) => return Ok(v),
            RetryStep::Fail(e) => return Err(e),
            RetryStep::Retry {
                error,
                delay,
                cause,
            } => {
                if attempt >= max || deadline_exhausted(deadline, delay) {
                    rlog.warn_giving_up(attempt);
                    return Err(error);
                }
                rlog.warn_retry(attempt, max, &cause, delay);
                sleep_backoff_blocking(delay);
            }
        }
        attempt += 1;
    }
}

/// Async counterpart of [`retry_steps_sync`]; sleeps via [`sleep_backoff_async`]
/// so async ladders honor the same deadline and backoff accounting.
pub async fn retry_steps_async<T, E, F, Fut>(
    rlog: RetryLog<'_>,
    max_attempts: u32,
    deadline: Option<std::time::Instant>,
    mut op: F,
) -> Result<T, E>
where
    F: FnMut(u32) -> Fut,
    Fut: std::future::Future<Output = RetryStep<T, E>>,
{
    let max = max_attempts.max(1);
    let mut attempt: u32 = 1;
    loop {
        match op(attempt).await {
            RetryStep::Done(v) => {
                if attempt > 1 {
                    rlog.note_succeeded(attempt);
                }
                return Ok(v);
            }
            RetryStep::DoneQuiet(v) => return Ok(v),
            RetryStep::Fail(e) => return Err(e),
            RetryStep::Retry {
                error,
                delay,
                cause,
            } => {
                if attempt >= max || deadline_exhausted(deadline, delay) {
                    rlog.warn_giving_up(attempt);
                    return Err(error);
                }
                rlog.warn_retry(attempt, max, &cause, delay);
                sleep_backoff_async(delay).await;
            }
        }
        attempt += 1;
    }
}

/// Whether sleeping `delay` now would carry total wall-time past `deadline`.
/// A saturating check: a projection that overflows `Instant` is treated as
/// past any real deadline (stop) rather than panicking. `None` deadline is
/// never exhausted (attempt-count-only bound).
fn deadline_exhausted(deadline: Option<std::time::Instant>, delay: Duration) -> bool {
    deadline.is_some_and(|d| {
        std::time::Instant::now()
            .checked_add(delay)
            .is_none_or(|projected| projected > d)
    })
}

/// Whether to consider 3xx redirects a success outcome (most upload-style
/// publishers do, since the underlying client follows redirects under the
/// hood; some callers explicitly want only 2xx).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SuccessClass {
    /// 2xx only. Any 3xx is treated as a non-success status (eligible for
    /// retry / fast-fail per `is_retriable`).
    Strict,
    /// 2xx OR 3xx. Used by upload publishers whose servers may emit a
    /// 301/302/307 in the success path (artifactory does this for some
    /// virtual repo configurations).
    AllowRedirects,
}

/// Drive a single HTTP call to completion, retrying transient failures via
/// the shared [`retry_sync`] machinery.
///
/// On every attempt, `send` is invoked to construct + dispatch a fresh
/// request. The closure must rebuild the request from scratch (multipart
/// `Form`, streamed body, etc. are move-only). The helper:
///
/// 1. On `Err` (transport-level): wrap in [`HttpError::from_response`] +
///    a `<label>: <stage> transport error` context, classify with
///    [`is_retriable`] (so EOF / connection-reset retry, plain "dial
///    failed" fast-fails), and dispatch `Continue`/`Break`.
/// 2. On non-success status: drain the body, format the outer message via
///    `error_msg`, wrap in [`HttpError::new`] with the upstream status, and
///    classify (5xx/429 → `Continue`, 4xx → `Break`).
/// 3. On success status: return `(status, body)`.
///
/// The `error_msg` closure receives the response status and body so callers
/// can format publisher-specific envelopes (e.g. artifactory's
/// `{"errors":[...]}` JSON).
///
/// Replaces three nearly-identical retry loops:
/// - `stage-publish/cloudsmith.rs::retry_request`
/// - `stage-publish/artifactory.rs::upload_single_artifact` (inline)
/// - `stage-announce/helpers.rs::retry_http` (now wraps this helper; see
///   announce/helpers.rs for the thin adapter that returns the body string
///   instead of `(StatusCode, String)`).
pub fn retry_http_blocking<F, M>(
    rlog: RetryLog<'_>,
    policy: &RetryPolicy,
    success_class: SuccessClass,
    send: F,
    error_msg: M,
) -> anyhow::Result<(reqwest::StatusCode, String)>
where
    F: FnMut(u32) -> Result<reqwest::blocking::Response, reqwest::Error>,
    M: Fn(reqwest::StatusCode, &str) -> String,
{
    retry_http_blocking_deadline(rlog, policy, None, success_class, send, error_msg)
}

/// Like [`retry_http_blocking`], but stops once the next backoff would push
/// total wall-time past `deadline` (from [`crate::Context::retry_deadline`]), so
/// a long upload storm exits resumable before the outer job timeout instead of
/// running the full attempt ladder. `deadline: None` is the unbounded form.
pub fn retry_http_blocking_deadline<F, M>(
    rlog: RetryLog<'_>,
    policy: &RetryPolicy,
    deadline: Option<std::time::Instant>,
    success_class: SuccessClass,
    mut send: F,
    error_msg: M,
) -> anyhow::Result<(reqwest::StatusCode, String)>
where
    F: FnMut(u32) -> Result<reqwest::blocking::Response, reqwest::Error>,
    M: Fn(reqwest::StatusCode, &str) -> String,
{
    use anyhow::Context as _;
    retry_sync_deadline(rlog, policy, deadline, |attempt| {
        match send(attempt) {
            Ok(resp) => {
                let status = resp.status();
                let succeeded = match success_class {
                    SuccessClass::Strict => status.is_success(),
                    SuccessClass::AllowRedirects => status.is_success() || status.is_redirection(),
                };
                let body = resp
                    .text()
                    .unwrap_or_else(|e| format!("<failed to read body: {e}>"));
                if succeeded {
                    Ok((status, body))
                } else {
                    let msg = error_msg(status, &body);
                    let inner = anyhow::anyhow!("{msg}");
                    let wrapped = anyhow::Error::new(HttpError::new(
                        std::io::Error::other(inner.to_string()),
                        status.as_u16(),
                    ))
                    .context(inner);
                    // `as_ref()` is the head of the chain; `is_retriable` walks
                    // `.source()` to reach `HttpError`. `root_cause()` would
                    // unwrap past `HttpError` to the io::Error leaf and miss
                    // the status. Pinned by
                    // `classifier_5xx_via_anyhow_chain_uses_as_ref`.
                    if is_retriable(wrapped.as_ref()) {
                        Err(ControlFlow::Continue(wrapped))
                    } else {
                        Err(ControlFlow::Break(wrapped))
                    }
                }
            }
            Err(e) => {
                // Transport-layer failure: always wrap in HttpError(status=0)
                // so the chain-walking classifier can see network-error
                // substrings via the inner io::Error message.
                let err = anyhow::Error::new(HttpError::from_response(e, None))
                    .context(format!("{}: HTTP transport error", rlog.desc()));
                if is_retriable(err.as_ref()) {
                    Err(ControlFlow::Continue(err))
                } else {
                    Err(ControlFlow::Break(err))
                }
            }
        }
    })
    .with_context(|| format!("{}: exhausted retry attempts", rlog.desc()))
}

/// Binary-body sibling of [`retry_http_blocking`] for endpoints whose success
/// payload is not valid UTF-8 (e.g. a gzip-compressed `.crate` tarball).
///
/// `resp.text()` runs a lossy UTF-8 conversion that silently rewrites
/// non-UTF-8 byte sequences to U+FFFD, corrupting a binary payload with no
/// error raised — a caller hashing the "recovered" bytes would never match
/// the original digest. This variant reads the success body via
/// `resp.bytes()` instead, keeping every other behavior (retry classification,
/// `HttpError` wrapping, `success_class`) identical to the text variant. The
/// error-path body is still decoded lossily into text purely for the
/// `error_msg` formatter — error responses are conventionally textual/JSON,
/// and a few replacement characters in an already-failing message are
/// harmless.
pub fn retry_http_blocking_bytes<F, M>(
    rlog: RetryLog<'_>,
    policy: &RetryPolicy,
    success_class: SuccessClass,
    send: F,
    error_msg: M,
) -> anyhow::Result<(reqwest::StatusCode, Vec<u8>)>
where
    F: FnMut(u32) -> Result<reqwest::blocking::Response, reqwest::Error>,
    M: Fn(reqwest::StatusCode, &str) -> String,
{
    retry_http_blocking_bytes_deadline(rlog, policy, None, success_class, send, error_msg)
}

/// Deadline-bounded sibling of [`retry_http_blocking_bytes`], mirroring
/// [`retry_http_blocking_deadline`] for binary success bodies. `deadline: None`
/// is the unbounded form.
pub fn retry_http_blocking_bytes_deadline<F, M>(
    rlog: RetryLog<'_>,
    policy: &RetryPolicy,
    deadline: Option<std::time::Instant>,
    success_class: SuccessClass,
    mut send: F,
    error_msg: M,
) -> anyhow::Result<(reqwest::StatusCode, Vec<u8>)>
where
    F: FnMut(u32) -> Result<reqwest::blocking::Response, reqwest::Error>,
    M: Fn(reqwest::StatusCode, &str) -> String,
{
    use anyhow::Context as _;
    retry_sync_deadline(rlog, policy, deadline, |attempt| match send(attempt) {
        Ok(resp) => {
            let status = resp.status();
            let succeeded = match success_class {
                SuccessClass::Strict => status.is_success(),
                SuccessClass::AllowRedirects => status.is_success() || status.is_redirection(),
            };
            let bytes = resp
                .bytes()
                .map(|b| b.to_vec())
                .unwrap_or_else(|e| format!("<failed to read body: {e}>").into_bytes());
            if succeeded {
                Ok((status, bytes))
            } else {
                let body_text = String::from_utf8_lossy(&bytes).into_owned();
                let msg = error_msg(status, &body_text);
                let inner = anyhow::anyhow!("{msg}");
                let wrapped = anyhow::Error::new(HttpError::new(
                    std::io::Error::other(inner.to_string()),
                    status.as_u16(),
                ))
                .context(inner);
                if is_retriable(wrapped.as_ref()) {
                    Err(ControlFlow::Continue(wrapped))
                } else {
                    Err(ControlFlow::Break(wrapped))
                }
            }
        }
        Err(e) => {
            let err = anyhow::Error::new(HttpError::from_response(e, None))
                .context(format!("{}: HTTP transport error", rlog.desc()));
            if is_retriable(err.as_ref()) {
                Err(ControlFlow::Continue(err))
            } else {
                Err(ControlFlow::Break(err))
            }
        }
    })
    .with_context(|| format!("{}: exhausted retry attempts", rlog.desc()))
}

/// Async sibling of [`retry_http_blocking`] for `reqwest::Client` (non-blocking)
/// call sites such as the GitLab and Gitea release publishers.
///
/// Each attempt invokes `send` (a fresh future) and:
///
/// 1. On `Err` (transport-level): wraps in [`HttpError::from_response`] +
///    a `<label>: HTTP transport error` context, classifies via
///    [`is_retriable`] (network-substring + EOF chain match), and dispatches
///    `Continue`/`Break`.
/// 2. On non-success status: drains the body via `Response::text().await`,
///    formats the outer message via `error_msg`, wraps in [`HttpError::new`]
///    with the upstream status, and classifies (5xx/429 → `Continue`, 4xx →
///    `Break`).
/// 3. On success status: returns the raw [`reqwest::Response`] for the
///    caller to consume (e.g. `.json()`, `.text()`, header inspection).
///
/// `success_class` mirrors the blocking variant: `Strict` rejects 3xx,
/// `AllowRedirects` accepts them. Most async API clients want `Strict`
/// (their reqwest::Client follows redirects by default, so a surfaced 3xx
/// is itself an error).
pub async fn retry_http_async<F, Fut, M>(
    rlog: RetryLog<'_>,
    policy: &RetryPolicy,
    success_class: SuccessClass,
    send: F,
    error_msg: M,
) -> anyhow::Result<reqwest::Response>
where
    F: FnMut(u32) -> Fut,
    Fut: std::future::Future<Output = Result<reqwest::Response, reqwest::Error>>,
    M: Fn(reqwest::StatusCode, &str) -> String,
{
    retry_http_async_deadline(rlog, policy, None, success_class, send, error_msg).await
}

/// Deadline-bounded sibling of [`retry_http_async`], the async counterpart of
/// [`retry_http_blocking_deadline`], so async upload publishers (GitLab/Gitea
/// release-asset uploads) honor the [`crate::Context::retry_deadline`] budget.
/// `deadline: None` is the unbounded form.
pub async fn retry_http_async_deadline<F, Fut, M>(
    rlog: RetryLog<'_>,
    policy: &RetryPolicy,
    deadline: Option<std::time::Instant>,
    success_class: SuccessClass,
    mut send: F,
    error_msg: M,
) -> anyhow::Result<reqwest::Response>
where
    F: FnMut(u32) -> Fut,
    Fut: std::future::Future<Output = Result<reqwest::Response, reqwest::Error>>,
    M: Fn(reqwest::StatusCode, &str) -> String,
{
    use anyhow::Context as _;
    retry_async_deadline(rlog, policy, deadline, |attempt| {
        let fut = send(attempt);
        let error_msg = &error_msg;
        async move {
            match fut.await {
                Ok(resp) => {
                    let status = resp.status();
                    let succeeded = match success_class {
                        SuccessClass::Strict => status.is_success(),
                        SuccessClass::AllowRedirects => {
                            status.is_success() || status.is_redirection()
                        }
                    };
                    if succeeded {
                        Ok(resp)
                    } else {
                        let body = resp
                            .text()
                            .await
                            .unwrap_or_else(|e| format!("<failed to read body: {e}>"));
                        let msg = error_msg(status, &body);
                        let inner = anyhow::anyhow!("{msg}");
                        let wrapped = anyhow::Error::new(HttpError::new(
                            std::io::Error::other(inner.to_string()),
                            status.as_u16(),
                        ))
                        .context(inner);
                        // `as_ref()` is the head of the chain; `is_retriable`
                        // walks `.source()` to reach `HttpError`. `root_cause()`
                        // would unwrap past `HttpError` to the io::Error leaf
                        // and miss the status. Pinned by
                        // `classifier_5xx_via_anyhow_chain_uses_as_ref`.
                        if is_retriable(wrapped.as_ref()) {
                            Err(ControlFlow::Continue(wrapped))
                        } else {
                            Err(ControlFlow::Break(wrapped))
                        }
                    }
                }
                Err(e) => {
                    // Transport-layer failure: wrap in HttpError(status=0) so
                    // the chain-walking classifier can see network-error
                    // substrings via the inner io::Error message.
                    let err = anyhow::Error::new(HttpError::from_response(e, None))
                        .context(format!("{}: HTTP transport error", rlog.desc()));
                    if is_retriable(err.as_ref()) {
                        Err(ControlFlow::Continue(err))
                    } else {
                        Err(ControlFlow::Break(err))
                    }
                }
            }
        }
    })
    .await
    .with_context(|| format!("{}: exhausted retry attempts", rlog.desc()))
}

/// Classify a `reqwest::Result<reqwest::blocking::Response>` into the
/// `ControlFlow` shape expected by `retry_sync` for a typical HTTP call:
/// 5xx + transport errors retry, 4xx fast-fails, 2xx/3xx returns Ok. The
/// returned response (Ok branch) is the caller's to consume.
///
/// This is the convention shared by every HTTP-uploading publisher; see audit
/// A7 dedup S5.
pub fn classify_http_sync(
    result: reqwest::Result<reqwest::blocking::Response>,
) -> Result<reqwest::blocking::Response, ControlFlow<anyhow::Error, anyhow::Error>> {
    use anyhow::anyhow;
    match result {
        Ok(resp) => {
            let status = resp.status();
            if status.is_success() || status.is_redirection() {
                Ok(resp)
            } else if status.is_server_error() {
                Err(ControlFlow::Continue(anyhow!(
                    "HTTP {} {}",
                    status.as_u16(),
                    status.canonical_reason().unwrap_or("server error")
                )))
            } else {
                // 4xx (and any other non-success/redirect/5xx): fast-fail
                Err(ControlFlow::Break(anyhow!(
                    "HTTP {} {}",
                    status.as_u16(),
                    status.canonical_reason().unwrap_or("client error")
                )))
            }
        }
        // Transport-layer failure (DNS, connect, TLS, timeout): retry.
        Err(e) => Err(ControlFlow::Continue(anyhow!(e))),
    }
}

// ---------------------------------------------------------------------------
// Retriable-error classification
// ---------------------------------------------------------------------------

/// Carries an HTTP status code alongside the original error so
/// [`is_retriable`] can route 5xx / 429 to retry and 4xx to fast-fail.
///
/// HTTP error carrying status + message. Construct via [`HttpError::new`]
/// (status-only) or wrap an existing `reqwest::Response` via
/// [`HttpError::from_response`].
///
/// A `status` of `0` denotes a network-level failure where no response was
/// ever received (the no-response branch). Network-level failures
/// are still classified via the inner error's message, so wrapping them in
/// `HttpError { status: 0, .. }` does not lose retriability information.
#[derive(Debug)]
pub struct HttpError {
    /// The wrapped error (transport, decode, or status-derived message).
    /// Reachable via the [`StdError::source`] trait method (not directly).
    source: Box<dyn StdError + Send + Sync + 'static>,
    /// HTTP status code; `0` for transport-level failures.
    pub status: u16,
}

impl HttpError {
    /// Wrap an error with a status code. `0` denotes a network-level failure
    /// (no response received).
    pub fn new<E>(source: E, status: u16) -> Self
    where
        E: StdError + Send + Sync + 'static,
    {
        Self {
            source: Box::new(source),
            status,
        }
    }

    /// Wrap a transport-layer error with the status code from the (possibly
    /// missing) response.
    /// `None` resp yields status `0` (network-level failure).
    pub fn from_response<E>(err: E, resp: Option<&reqwest::Response>) -> Self
    where
        E: StdError + Send + Sync + 'static,
    {
        Self::new(err, resp.map(|r| r.status().as_u16()).unwrap_or(0))
    }
}

/// Extract the upstream HTTP status from an [`anyhow::Error`] chain produced by
/// [`retry_http_blocking`] / [`retry_http_async`].
///
/// Returns `0` when no [`HttpError`] is present in the chain — a transport-level
/// failure that never received a response, or a non-HTTP error.
pub fn http_status(err: &anyhow::Error) -> u16 {
    err.chain()
        .find_map(|e| e.downcast_ref::<HttpError>().map(|h| h.status))
        .unwrap_or(0)
}

impl fmt::Display for HttpError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        // Defer to the inner error so messages stay focused on the cause.
        // Delegate to the inner error message.
        fmt::Display::fmt(&self.source, f)
    }
}

impl StdError for HttpError {
    fn source(&self) -> Option<&(dyn StdError + 'static)> {
        Some(&*self.source)
    }
}

/// Marker error wrapping any inner error so [`is_retriable`] returns `true`
/// regardless of class — useful when a
/// caller knows the failure is transient (e.g. an idempotent registry write
/// returning 422 because of a transient race condition) and wants the retry
/// loop to ignore the usual 4xx fast-fail.
#[derive(Debug)]
pub struct Retriable(Box<dyn StdError + Send + Sync + 'static>);

impl Retriable {
    /// Wrap any error so [`is_retriable`] returns `true` regardless of class.
    /// Use this when a caller knows a 4xx is transient (e.g. a 422 from an
    /// idempotent registry write losing a race) and wants to override the
    /// usual fast-fail. For `Option<E>` inputs, see [`is_retriable_opt`] —
    /// this constructor itself is non-nullable.
    pub fn new<E>(source: E) -> Self
    where
        E: StdError + Send + Sync + 'static,
    {
        Self(Box::new(source))
    }
}

impl fmt::Display for Retriable {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Display::fmt(&self.0, f)
    }
}

impl StdError for Retriable {
    fn source(&self) -> Option<&(dyn StdError + 'static)> {
        Some(&*self.0)
    }
}

/// Returns `true` if the message looks like a transient network-layer failure.
///
/// Network-error classification, extended for Rust /
/// Windows. Each link in the error chain is checked two ways:
///
/// 1a. **Structural [`io::ErrorKind`] check** via `downcast_ref::<io::Error>()`.
///     Treats `UnexpectedEof`, `TimedOut`, `ConnectionRefused`,
///     `ConnectionReset`, `ConnectionAborted`, and `BrokenPipe` as transient.
///     The OS-classified `ErrorKind` is robust where Display text is not:
///     Linux's connect-refused says `"Connection refused"` but Windows
///     surfaces a transient connect failure as
///     `io::Error { kind: TimedOut, message: "operation timed out" }`, and
///     a Windows-reset reads `"An existing connection was forcibly closed"`.
///     Matching `kind()` catches all of them regardless of phrasing. Also
///     recognises any `io::Error` whose Display form is `"EOF"` /
///     `"unexpected eof"` (rustls / hyper convention; Rust has no
///     equivalent of Go's `io.EOF` sentinel).
///
/// 1b. **Substring match on the lowercased Display form** against
///     [`NETWORK_ERROR_NEEDLES`]. Covers the canonical surface plus the
///     Windows / Rust-stdlib phrasings that bypass the kind check when an
///     error has been wrapped (e.g. reqwest coercing the inner kind to
///     `Other` while preserving the OS message text).
///
/// Walks `.source()` for both branches — Rust's `Display` impls do NOT
/// inherit the wrapped error's text the way Go's `err.Error()` does, so a
/// reqwest "Connection refused" message buried under an anyhow context would
/// otherwise be invisible to the head-only string.
pub fn is_network_error(err: &(dyn StdError + 'static)) -> bool {
    let mut cur: Option<&(dyn StdError + 'static)> = Some(err);
    while let Some(e) = cur {
        // 1a. Structural ErrorKind check — robust to platform Display drift
        //     (Windows's "operation timed out" vs Linux's "Connection refused").
        if let Some(io_err) = e.downcast_ref::<io::Error>() {
            match io_err.kind() {
                io::ErrorKind::UnexpectedEof
                | io::ErrorKind::TimedOut
                | io::ErrorKind::ConnectionRefused
                | io::ErrorKind::ConnectionReset
                | io::ErrorKind::ConnectionAborted
                | io::ErrorKind::BrokenPipe => return true,
                _ => {}
            }
            let m = io_err.to_string().to_lowercase();
            if m == "eof" || m == "unexpected eof" {
                return true;
            }
        }

        // 1b. Substring match on each link's own Display (NOT the full
        //     chain "{e:#}" form, which would double-count the same text on
        //     deeper links). Lowercased once per link.
        let s = e.to_string().to_lowercase();
        if NETWORK_ERROR_NEEDLES.iter().any(|n| s.contains(n)) {
            return true;
        }

        cur = e.source();
    }
    false
}

/// The set of substrings classified as transient.
///
/// The first nine entries are the canonical network-error needles
/// (matching is case-insensitive). The remaining entries cover Windows and
/// Rust-stdlib phrasings of transient transport failures that surface when
/// an `io::Error` has been wrapped by a higher layer (reqwest, hyper,
/// anyhow), losing the original `ErrorKind` classification but preserving
/// the OS message text. Without these, every publisher running on Windows
/// fast-failed on the first transient connect blip instead of retrying.
const NETWORK_ERROR_NEEDLES: &[&str] = &[
    "connection reset",
    "network is unreachable",
    "connection closed",
    "connection refused",
    "tls handshake timeout",
    "i/o timeout",
    "broken pipe",
    "timeout awaiting response headers",
    "context deadline exceeded",
    // Windows + macOS phrasing of ErrorKind::TimedOut after wrapping.
    "operation timed out",
    // Windows ErrorKind::ConnectionAborted phrasing.
    "the network connection was aborted",
    // Windows ErrorKind::ConnectionReset phrasing.
    "an existing connection was forcibly closed",
    // hyper-util / reqwest DNS-resolution failures wrapped through the
    // connector. Surfaces as `client error (Connect): dns error: ...` with
    // a platform-specific resolver tail ("Name or service not known" on
    // Linux/glibc, "nodename nor servname provided, or not known" on macOS,
    // "No such host is known" on Windows). The leading "dns error" prefix
    // is the cross-platform constant.
    "dns error",
    // GAI (getaddrinfo) wording across resolvers; covers the Linux
    // resolver tail above and BSD/macOS phrasing.
    "failed to lookup address",
    // Windows resolver tail when DNS-resolution fails.
    "no such host is known",
];

/// Classify an error as retriable.
///
/// Returns `true` for:
/// - any [`is_network_error`] match (substring + EOF / UnexpectedEof in the
///   `source()` chain)
/// - any error whose chain contains a [`Retriable`] wrapper
/// - any error whose chain contains an [`HttpError`] with status `>= 500`
///   or status `429` (Too Many Requests)
///
/// Returns `false` for plain errors and 4xx HTTP errors (other than 429) —
/// those are fast-failed by the retry loop.
pub fn is_retriable(err: &(dyn StdError + 'static)) -> bool {
    // 1. Any link in the chain is an explicit Retriable marker.
    let mut cur: Option<&(dyn StdError + 'static)> = Some(err);
    while let Some(e) = cur {
        if e.is::<Retriable>() {
            return true;
        }
        if let Some(http) = e.downcast_ref::<HttpError>()
            && status_is_retriable(http.status)
        {
            return true;
        }
        cur = e.source();
    }

    // 2. Network-error substring / EOF chain match.
    is_network_error(err)
}

/// The canonical retriable-HTTP-status rule: server errors (`>= 500`) and
/// `429 Too Many Requests`. Everything else — notably the remaining 4xx
/// range — is fast-failed.
///
/// [`is_retriable`]'s [`HttpError`] arm delegates here, and raw-status
/// classifiers that cannot route through [`HttpError`] (the gemfury and
/// chocolatey multipart push loops, whose conflict-as-success / hard-fail
/// cases need bespoke `ControlFlow` handling) call it directly, so the
/// fast-fail/retry split for a bare status code has exactly one
/// definition. Extending the rule (408/425, `Retry-After` awareness)
/// updates every consumer at once — including the one-way-door publishers
/// where a mis-fast-failed transient burns an unrecoverable publish
/// attempt.
pub fn status_is_retriable(status: u16) -> bool {
    status >= 500 || status == 429
}

/// Convenience: `None` passes through as `false`. The
/// `IsRetriable(nil) -> false` semantics.
pub fn is_retriable_opt(err: Option<&(dyn StdError + 'static)>) -> bool {
    err.is_some_and(is_retriable)
}

/// Apply ±20 % pseudo-jitter to `base` using a cheap subsecond-nanos modulo.
///
/// Returns a value in `[base * 0.8, base * 1.2)`. No `rand` crate dependency:
/// `SystemTime::now().subsec_nanos()` provides ~nanosecond entropy that is
/// sufficient for retry jitter (the goal is spreading out concurrent retriers,
/// not cryptographic unpredictability).
///
/// The ±20 % window is a widely-adopted convention (AWS SDK, GCP client libs).
/// Jitter only ever widens the sleep by up to 20 %; it never shortens it below
/// 80 % of the nominal delay, so `Retry-After` honoring is conservative.
pub fn jitter_duration(base: Duration) -> Duration {
    let nanos = base.as_nanos() as u64;
    // 20 % of the nominal duration.
    let window = nanos / 5;
    if window == 0 {
        return base;
    }
    // Cheap pseudo-random offset in [0, window * 2) centred on window,
    // giving a net range of [base - window, base + window). The wall-clock
    // seed is XORed with a process-local Weyl sequence (odd-constant atomic
    // counter, so consecutive draws stay well spread) because under
    // SOURCE_DATE_EPOCH a pinned clock would collapse jitter to a constant
    // and re-synchronize concurrent retriers on every round — recreating
    // the exact collision jitter exists to break. SOURCE_DATE_EPOCH pins
    // BUILD OUTPUT bytes; a retry sleep duration never reaches an artifact,
    // so varying it is determinism-safe (which is also why this reads the
    // real clock instead of `sde::resolve_now()`).
    static JITTER_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let clock = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.subsec_nanos() as u64)
        .unwrap_or(0);
    let seq = JITTER_SEQ.fetch_add(0x9E37_79B9_7F4A_7C15, std::sync::atomic::Ordering::Relaxed);
    let seed = clock ^ seq;
    let offset = seed % (window * 2);
    // Saturating arithmetic so we never panic on extreme values.
    let jittered = nanos.saturating_sub(window).saturating_add(offset);
    Duration::from_nanos(jittered)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU32, Ordering};

    use crate::test_helpers::{test_logger, test_retry_log as tlog};

    #[test]
    fn backoff_accumulator_is_monotonic_and_sleep_helper_records() {
        // The accumulator is process-global and other retry tests run
        // concurrently against it, so assert on the DELTA (never smaller than
        // this test's own contribution) rather than an absolute total — a reset
        // would race those tests. `record_retry_backoff` adds without sleeping;
        // `sleep_backoff_blocking` both sleeps the duration and records it.
        let before = total_retry_backoff();
        record_retry_backoff(Duration::from_millis(250));
        assert!(
            total_retry_backoff().saturating_sub(before) >= Duration::from_millis(250),
            "record_retry_backoff must add at least its duration"
        );

        let before_sleep = total_retry_backoff();
        let start = std::time::Instant::now();
        sleep_backoff_blocking(Duration::from_millis(30));
        assert!(
            start.elapsed() >= Duration::from_millis(30),
            "helper must sleep"
        );
        assert!(
            total_retry_backoff().saturating_sub(before_sleep) >= Duration::from_millis(30),
            "sleep_backoff_blocking must record its sleep"
        );
    }

    #[test]
    fn retry_scope_attributes_backoff_to_its_label() {
        // Isolation rests on the unique scope name plus `>=` delta assertions,
        // not on serialization: no other test in this crate enters a
        // `RetryScope`, so nothing swaps `CURRENT_SCOPE` away between the two
        // records here, and a uniquely-named key can only grow inside this
        // test's guarded block.
        let scope_name = "test-scope-attributes-2f9c";
        let read = |name: &str| -> (u32, Duration) {
            retry_scope_breakdown()
                .into_iter()
                .find(|(k, _, _)| k == name)
                .map(|(_, r, d)| (r, d))
                .unwrap_or((0, Duration::ZERO))
        };

        let (r0, d0) = read(scope_name);
        {
            let _scope = RetryScope::enter(scope_name);
            record_retry_backoff(Duration::from_millis(40));
            record_retry_backoff(Duration::from_millis(60));
        }
        let (r1, d1) = read(scope_name);
        assert!(r1 >= r0 + 2, "two records must add at least two retries");
        assert!(
            d1.saturating_sub(d0) >= Duration::from_millis(100),
            "scope backoff must sum the recorded sleeps"
        );

        // After the guard drops, backoff falls back to the unattributed bucket,
        // not this scope — so a later record does not grow this scope's tally.
        record_retry_backoff(Duration::from_millis(10));
        assert_eq!(
            read(scope_name).0,
            r1,
            "records outside the scope must not attribute to it"
        );
    }

    fn fast_policy() -> RetryPolicy {
        RetryPolicy {
            max_attempts: 4,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(5),
        }
    }

    /// Locks the shallow shape of the best-effort pre-publish probe policy so a
    /// future edit cannot silently re-point preflight probes at the production
    /// write-ladder (10 attempts / 10s base / 5m cap), which would let one
    /// wedged endpoint stall the gate for tens of minutes.
    #[test]
    fn preflight_policy_is_shallow() {
        let p = RetryPolicy::PREFLIGHT;
        assert_eq!(p.max_attempts, 3);
        assert_eq!(p.base_delay, Duration::from_millis(200));
        assert_eq!(p.max_delay, Duration::from_secs(1));
        // Sub-second base + low cap: the whole probe ladder must stay well
        // under a second of sleeps even when every attempt is exhausted.
        let total_sleep: Duration = (2..=p.max_attempts).map(|n| p.delay_for(n)).sum();
        assert!(
            total_sleep < Duration::from_secs(1),
            "preflight backoff sleeps must stay sub-second, got {total_sleep:?}"
        );
    }

    /// Locks the shallow shape of the burn-detection guard probe policy so a
    /// future edit cannot silently re-point the published-state guards at the
    /// production write-ladder, which would let a registry outage stall a
    /// multi-crate probe pass for hours before it can fail closed.
    #[test]
    fn guard_probe_policy_is_shallow_and_capped() {
        let p = RetryPolicy::GUARD_PROBE;
        assert_eq!(p.max_attempts, 3);
        assert_eq!(p.base_delay, Duration::from_secs(1));
        assert_eq!(p.max_delay, Duration::from_secs(30));
        // Every individual sleep must respect the 30s cap, and the whole
        // ladder must stay bounded (worst case: 1s + 2s of backoff).
        for n in 2..=p.max_attempts {
            assert!(p.delay_for(n) <= Duration::from_secs(30));
        }
        let total_sleep: Duration = (2..=p.max_attempts).map(|n| p.delay_for(n)).sum();
        assert!(
            total_sleep <= Duration::from_secs(3),
            "guard probe backoff must stay in seconds, got {total_sleep:?}"
        );
    }

    #[test]
    fn http_status_extracts_status_from_chain() {
        let wrapped = anyhow::Error::new(HttpError::new(std::io::Error::other("boom"), 429))
            .context("outer context");
        assert_eq!(http_status(&wrapped), 429);
    }

    #[test]
    fn http_status_is_zero_without_http_error() {
        let plain = anyhow::anyhow!("not an http error");
        assert_eq!(http_status(&plain), 0);
    }

    /// The idempotent floor raises a sub-floor cap to [`IDEMPOTENT_PUT_ATTEMPTS`]
    /// but never lowers an operator-set higher cap. Fails if the floor constant
    /// is reverted to 1 (or the `max()` semantics flip to a clamp).
    #[test]
    fn idempotent_floor_raises_low_cap_and_preserves_high_cap() {
        let raised = RetryPolicy {
            max_attempts: 1,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(5),
        }
        .with_idempotent_floor();
        assert_eq!(
            raised.max_attempts, IDEMPOTENT_PUT_ATTEMPTS,
            "a single-attempt cap must be raised to the idempotent floor"
        );

        let preserved = RetryPolicy {
            max_attempts: 7,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(5),
        }
        .with_idempotent_floor();
        assert_eq!(
            preserved.max_attempts, 7,
            "an operator-set cap above the floor must be preserved, not lowered"
        );
    }

    #[test]
    fn jitter_returns_base_when_window_rounds_to_zero() {
        // For any duration under 5ns the ±20 % window (`nanos / 5`) floors to
        // 0, so jitter is a no-op and the base is returned unchanged — the
        // early-return guard that avoids a `% 0` panic on tiny delays.
        for n in 0..5u64 {
            let base = Duration::from_nanos(n);
            assert_eq!(
                jitter_duration(base),
                base,
                "sub-5ns base {n} must pass through unjittered"
            );
        }
    }

    #[test]
    fn jitter_stays_within_plus_minus_twenty_percent() {
        // The jittered value never leaves [base*0.8, base*1.2) — the documented
        // window. Uses a duration large enough that `nanos / 5 > 0`.
        let base = Duration::from_millis(100);
        let jittered = jitter_duration(base);
        let lo = base.mul_f64(0.8);
        let hi = base.mul_f64(1.2);
        assert!(
            jittered >= lo && jittered < hi,
            "jittered {jittered:?} outside [{lo:?}, {hi:?})"
        );
    }

    #[test]
    fn jitter_spreads_consecutive_draws_even_with_a_pinned_clock() {
        // The Weyl-sequence XOR guarantees consecutive draws differ even if
        // the wall clock were frozen (the SOURCE_DATE_EPOCH-style failure
        // mode where a constant seed re-synchronizes concurrent retriers).
        // The clock here is real, but the sequence term alone already forces
        // distinct offsets, so all-equal draws would mean the mixing broke.
        let base = Duration::from_millis(100);
        let draws: Vec<Duration> = (0..8).map(|_| jitter_duration(base)).collect();
        assert!(
            draws.windows(2).any(|w| w[0] != w[1]),
            "8 consecutive jitter draws were all identical: {draws:?}"
        );
    }

    #[test]
    fn delay_progression_caps_at_max() {
        let p = RetryPolicy {
            max_attempts: 10,
            base_delay: Duration::from_millis(100),
            max_delay: Duration::from_millis(500),
        };
        assert_eq!(p.delay_for(2), Duration::from_millis(100));
        assert_eq!(p.delay_for(3), Duration::from_millis(200));
        assert_eq!(p.delay_for(4), Duration::from_millis(400));
        assert_eq!(p.delay_for(5), Duration::from_millis(500)); // capped
        assert_eq!(p.delay_for(8), Duration::from_millis(500)); // capped
    }

    #[test]
    fn sync_succeeds_on_first_attempt() {
        let calls = AtomicU32::new(0);
        let result: Result<&str, &str> = retry_sync(tlog(), &fast_policy(), |_| {
            calls.fetch_add(1, Ordering::SeqCst);
            Ok("ok")
        });
        assert_eq!(result, Ok("ok"));
        assert_eq!(calls.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn sync_retries_until_success() {
        let calls = AtomicU32::new(0);
        let result: Result<u32, &str> = retry_sync(tlog(), &fast_policy(), |attempt| {
            calls.fetch_add(1, Ordering::SeqCst);
            if attempt < 3 {
                Err(ControlFlow::Continue("transient"))
            } else {
                Ok(attempt)
            }
        });
        assert_eq!(result, Ok(3));
        assert_eq!(calls.load(Ordering::SeqCst), 3);
    }

    #[test]
    fn sync_break_stops_immediately() {
        let calls = AtomicU32::new(0);
        let result: Result<(), &str> = retry_sync(tlog(), &fast_policy(), |_| {
            calls.fetch_add(1, Ordering::SeqCst);
            Err(ControlFlow::Break("fatal"))
        });
        assert_eq!(result, Err("fatal"));
        assert_eq!(calls.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn sync_returns_last_error_after_exhaustion() {
        let calls = AtomicU32::new(0);
        let result: Result<(), String> = retry_sync(tlog(), &fast_policy(), |attempt| {
            calls.fetch_add(1, Ordering::SeqCst);
            Err(ControlFlow::Continue(format!("fail {attempt}")))
        });
        assert_eq!(result, Err("fail 4".to_string()));
        assert_eq!(calls.load(Ordering::SeqCst), 4);
    }

    /// Build a captured logger + a `RetryLog` borrowing it, so the lifecycle
    /// tests can assert on the exact warn / status lines the engine emits.
    fn captured() -> (StageLogger, crate::log::LogCapture) {
        StageLogger::with_capture("test", crate::log::Verbosity::Normal)
    }

    const TINY: Duration = Duration::from_millis(1);

    #[test]
    fn steps_sync_first_try_done_is_silent() {
        let (log, cap) = captured();
        let out: Result<u32, &str> =
            retry_steps_sync(RetryLog::new("op", &log), 4, None, |_| RetryStep::Done(7));
        assert_eq!(out, Ok(7));
        assert_eq!(cap.total_count(), 0, "a clean first attempt must not log");
    }

    #[test]
    fn steps_sync_retry_then_done_emits_succeeded() {
        let (log, cap) = captured();
        let out: Result<u32, &str> =
            retry_steps_sync(RetryLog::new("op", &log), 5, None, |attempt| {
                if attempt < 3 {
                    RetryStep::Retry {
                        error: "transient",
                        delay: TINY,
                        cause: format!("blip {attempt}"),
                    }
                } else {
                    RetryStep::Done(attempt)
                }
            });
        assert_eq!(out, Ok(3));
        assert_eq!(cap.warn_count(), 2, "one warn per retried attempt");
        assert!(
            cap.all_messages()
                .iter()
                .any(|(lvl, m)| *lvl == crate::log::LogLevel::Status
                    && m.contains("op succeeded after 3 attempt(s)")),
            "recovery after retries must emit a succeeded status line: {:?}",
            cap.all_messages()
        );
    }

    #[test]
    fn steps_sync_done_quiet_recovers_without_succeeded_line() {
        // DoneQuiet returns the value like Done, but a recovery after retries
        // must NOT emit the "succeeded after N" note — the closure owns its own
        // resolution narrative (a tolerated skip / degraded disposition).
        let (log, cap) = captured();
        let out: Result<u32, &str> =
            retry_steps_sync(RetryLog::new("op", &log), 5, None, |attempt| {
                if attempt < 3 {
                    RetryStep::Retry {
                        error: "transient",
                        delay: TINY,
                        cause: "blip".into(),
                    }
                } else {
                    RetryStep::DoneQuiet(attempt)
                }
            });
        assert_eq!(out, Ok(3));
        assert_eq!(cap.warn_count(), 2, "per-attempt warns still fire");
        assert!(
            !cap.all_messages()
                .iter()
                .any(|(_, m)| m.contains("succeeded after")),
            "DoneQuiet must suppress the recovery line: {:?}",
            cap.all_messages()
        );
    }

    #[test]
    fn zero_delay_retry_is_not_counted_as_a_backoff_sleep() {
        // A caller that owns its own wait (a rate-limit reset probe) passes a
        // zero delay to re-attempt immediately; that must not inflate the
        // per-scope backoff-sleep count with a sleep that never happened.
        let (log, _cap) = captured();
        let scope = "zero-delay-accounting-probe";
        let _guard = RetryScope::enter(scope);
        let out: Result<u32, &str> =
            retry_steps_sync(RetryLog::new("op", &log), 5, None, |attempt| {
                if attempt < 3 {
                    RetryStep::Retry {
                        error: "transient",
                        delay: Duration::ZERO,
                        cause: "inline wait already served".into(),
                    }
                } else {
                    RetryStep::Done(attempt)
                }
            });
        assert_eq!(out, Ok(3));
        let recorded = retry_scope_breakdown()
            .into_iter()
            .find(|(name, _, _)| name == scope);
        assert!(
            recorded.is_none(),
            "two zero-delay retries must record no backoff sleeps: {recorded:?}"
        );
    }

    #[test]
    fn steps_sync_fail_fast_is_terminal_and_quiet() {
        let (log, cap) = captured();
        let calls = AtomicU32::new(0);
        let out: Result<(), &str> = retry_steps_sync(RetryLog::new("op", &log), 5, None, |_| {
            calls.fetch_add(1, Ordering::SeqCst);
            RetryStep::Fail("fatal")
        });
        assert_eq!(out, Err("fatal"));
        assert_eq!(calls.load(Ordering::SeqCst), 1, "Fail must not retry");
        assert_eq!(
            cap.warn_count(),
            0,
            "a fast-fail owns its own reason; the engine emits no giving-up line"
        );
    }

    #[test]
    fn steps_sync_exhaustion_emits_giving_up() {
        let (log, cap) = captured();
        let calls = AtomicU32::new(0);
        let out: Result<(), String> =
            retry_steps_sync(RetryLog::new("op", &log), 3, None, |attempt| {
                calls.fetch_add(1, Ordering::SeqCst);
                RetryStep::Retry {
                    error: format!("fail {attempt}"),
                    delay: TINY,
                    cause: "blip".into(),
                }
            });
        assert_eq!(out, Err("fail 3".to_string()));
        assert_eq!(calls.load(Ordering::SeqCst), 3);
        assert!(
            cap.warn_messages()
                .iter()
                .any(|m| m.contains("op failed after 3 attempt(s), giving up")),
            "exhausting the ladder must emit a giving-up warn: {:?}",
            cap.warn_messages()
        );
    }

    #[test]
    fn steps_sync_caller_delay_honors_deadline() {
        let (log, _cap) = captured();
        let calls = AtomicU32::new(0);
        // Deadline already elapsed: the caller-owned delay pushes `now + delay`
        // past it on the first classification, so the ladder stops after one op.
        let deadline = std::time::Instant::now();
        let out: Result<(), &str> =
            retry_steps_sync(RetryLog::new("op", &log), 10, Some(deadline), |_| {
                calls.fetch_add(1, Ordering::SeqCst);
                RetryStep::Retry {
                    error: "transient",
                    delay: Duration::from_secs(10),
                    cause: "blip".into(),
                }
            });
        assert_eq!(out, Err("transient"));
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "a delay that overshoots the deadline stops after one attempt"
        );
    }

    #[tokio::test]
    async fn steps_async_retry_then_done_emits_succeeded() {
        let (log, cap) = captured();
        let out: Result<u32, &str> =
            retry_steps_async(RetryLog::new("op", &log), 5, None, |attempt| async move {
                if attempt < 2 {
                    RetryStep::Retry {
                        error: "transient",
                        delay: TINY,
                        cause: "blip".into(),
                    }
                } else {
                    RetryStep::Done(attempt)
                }
            })
            .await;
        assert_eq!(out, Ok(2));
        assert_eq!(cap.warn_count(), 1);
        assert!(
            cap.all_messages()
                .iter()
                .any(|(lvl, m)| *lvl == crate::log::LogLevel::Status
                    && m.contains("op succeeded after 2 attempt(s)"))
        );
    }

    #[test]
    fn deadline_already_elapsed_stops_after_one_attempt_without_sleeping() {
        // A large base_delay proves the pre-attempt sleep is SKIPPED: with a
        // deadline already in the past, the budget check must fire after the
        // first Continue and return before any 10s sleep runs.
        let policy = RetryPolicy {
            max_attempts: 10,
            base_delay: Duration::from_secs(10),
            max_delay: Duration::from_secs(300),
        };
        let deadline = std::time::Instant::now();
        let calls = AtomicU32::new(0);
        let start = std::time::Instant::now();
        let result: Result<(), &str> = retry_sync_deadline(tlog(), &policy, Some(deadline), |_| {
            calls.fetch_add(1, Ordering::SeqCst);
            Err(ControlFlow::Continue("transient"))
        });
        assert_eq!(result, Err("transient"));
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "budget-exhausted retry must call op exactly once"
        );
        assert!(
            start.elapsed() < Duration::from_secs(1),
            "deadline check must skip the 10s backoff sleep, took {:?}",
            start.elapsed()
        );
    }

    #[test]
    fn deadline_none_matches_retry_sync_on_success() {
        let calls = AtomicU32::new(0);
        let result: Result<u32, &str> =
            retry_sync_deadline(tlog(), &fast_policy(), None, |attempt| {
                calls.fetch_add(1, Ordering::SeqCst);
                if attempt < 2 {
                    Err(ControlFlow::Continue("transient"))
                } else {
                    Ok(attempt)
                }
            });
        assert_eq!(result, Ok(2));
        assert_eq!(calls.load(Ordering::SeqCst), 2);

        let sync_calls = AtomicU32::new(0);
        let sync_result: Result<u32, &str> = retry_sync(tlog(), &fast_policy(), |attempt| {
            sync_calls.fetch_add(1, Ordering::SeqCst);
            if attempt < 2 {
                Err(ControlFlow::Continue("transient"))
            } else {
                Ok(attempt)
            }
        });
        assert_eq!(sync_result, result);
        assert_eq!(sync_calls.load(Ordering::SeqCst), 2);
    }

    #[test]
    fn deadline_far_in_future_does_not_change_behavior() {
        let deadline = std::time::Instant::now() + Duration::from_secs(3600);
        let calls = AtomicU32::new(0);
        let result: Result<u32, &str> =
            retry_sync_deadline(tlog(), &fast_policy(), Some(deadline), |attempt| {
                calls.fetch_add(1, Ordering::SeqCst);
                if attempt < 3 {
                    Err(ControlFlow::Continue("transient"))
                } else {
                    Ok(attempt)
                }
            });
        assert_eq!(result, Ok(3));
        assert_eq!(calls.load(Ordering::SeqCst), 3);
    }

    #[test]
    fn budget_exhausted_fires_on_a_past_deadline_and_not_a_future_one() {
        let policy = RetryPolicy {
            max_attempts: 10,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(1),
        };
        let now = std::time::Instant::now();
        assert!(policy.budget_exhausted(2, now - Duration::from_secs(1)));
        assert!(!policy.budget_exhausted(2, now + Duration::from_secs(3600)));
    }

    #[test]
    fn budget_exhausted_saturates_instead_of_panicking_on_uncapped_backoff() {
        // An uncapped policy projects a backoff near Duration::MAX; the check must
        // treat the (overflowing) projection as past the deadline, never panic on
        // `Instant + Duration` overflow (the docker/podman `max_delay: MAX` path).
        let policy = RetryPolicy {
            max_attempts: 100,
            base_delay: Duration::from_secs(30),
            max_delay: Duration::MAX,
        };
        let now = std::time::Instant::now();
        assert!(policy.budget_exhausted(64, now + Duration::from_secs(3600)));
    }

    #[tokio::test]
    async fn async_deadline_none_is_unbounded_and_exhausts_by_count() {
        // retry_async keeps the attempt-count-only contract: a None deadline runs
        // every configured attempt regardless of wall-time.
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(1),
        };
        let calls = std::sync::Arc::new(AtomicU32::new(0));
        let calls_inner = calls.clone();
        let result: Result<(), &str> = retry_async(tlog(), &policy, move |_| {
            let c = calls_inner.clone();
            async move {
                c.fetch_add(1, Ordering::SeqCst);
                Err(ControlFlow::Continue("transient"))
            }
        })
        .await;
        assert_eq!(result, Err("transient"));
        assert_eq!(calls.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn async_deadline_already_elapsed_stops_after_one_attempt() {
        // The async budget check mirrors the sync one: a past deadline stops the
        // ladder after the first Continue without sleeping the 10s backoff.
        let policy = RetryPolicy {
            max_attempts: 10,
            base_delay: Duration::from_secs(10),
            max_delay: Duration::from_secs(300),
        };
        let deadline = std::time::Instant::now();
        let calls = std::sync::Arc::new(AtomicU32::new(0));
        let calls_inner = calls.clone();
        let start = std::time::Instant::now();
        let result: Result<(), &str> =
            retry_async_deadline(tlog(), &policy, Some(deadline), move |_| {
                let c = calls_inner.clone();
                async move {
                    c.fetch_add(1, Ordering::SeqCst);
                    Err(ControlFlow::Continue("transient"))
                }
            })
            .await;
        assert_eq!(result, Err("transient"));
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        assert!(start.elapsed() < Duration::from_secs(1));
    }

    #[tokio::test]
    async fn async_retries_until_success() {
        let calls = std::sync::Arc::new(AtomicU32::new(0));
        let calls_inner = calls.clone();
        let result: Result<u32, &str> = retry_async(tlog(), &fast_policy(), move |attempt| {
            let c = calls_inner.clone();
            async move {
                c.fetch_add(1, Ordering::SeqCst);
                if attempt < 2 {
                    Err(ControlFlow::Continue("transient"))
                } else {
                    Ok(attempt)
                }
            }
        })
        .await;
        assert_eq!(result, Ok(2));
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

    // -----------------------------------------------------------------------
    // is_network_error / is_retriable / HttpError / Retriable
    //
    // Network-error classification test cases.
    // -----------------------------------------------------------------------

    /// Plain string error wrapper used in classification tests.
    #[derive(Debug)]
    struct StrErr(&'static str);
    impl fmt::Display for StrErr {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.write_str(self.0)
        }
    }
    impl StdError for StrErr {}

    #[derive(Debug)]
    struct OwnedErr(String);
    impl fmt::Display for OwnedErr {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.write_str(&self.0)
        }
    }
    impl StdError for OwnedErr {}

    #[test]
    fn network_error_substrings_match() {
        for s in [
            "connection reset by peer",
            "network is unreachable",
            "connection closed unexpectedly",
            "connection refused",
            "tls handshake timeout",
            "i/o timeout",
            "CONNECTION RESET",
            "TLS Handshake Timeout",
            "write: broken pipe",
            "net/http: timeout awaiting response headers",
            "context deadline exceeded",
            // DNS-resolution failures across platforms (hyper-util connector
            // surfaces these via reqwest as `client error (Connect): dns
            // error: <platform tail>`). Pin every tail we know about so a
            // cross-platform CI failure cannot reintroduce the gap.
            "client error (Connect): dns error: failed to lookup address information: Name or service not known",
            "dns error: nodename nor servname provided, or not known",
            "dns error: No such host is known. (os error 11001)",
        ] {
            let e = OwnedErr(s.to_string());
            assert!(is_network_error(&e), "expected network error: {s:?}");
        }
    }

    #[test]
    fn network_error_io_eof_kinds() {
        let e = io::Error::from(io::ErrorKind::UnexpectedEof);
        assert!(is_network_error(&e));

        // A custom-kind io::Error whose Display is "EOF" (rustls / hyper convention).
        let e2 = io::Error::other("EOF");
        assert!(is_network_error(&e2));
    }

    // Windows-CI regression: connect() on Windows surfaces transient failures
    // as io::Error { kind: TimedOut, message: "operation timed out" }, neither
    // of which matched the original EOF-only kind check or the
    // needle list. Same shape for the connection-* kinds across platforms —
    // pin each branch.

    #[test]
    fn is_network_error_classifies_io_timedout() {
        let e = io::Error::from(io::ErrorKind::TimedOut);
        assert!(is_network_error(&e));
        assert!(is_retriable(&e));
    }

    #[test]
    fn is_network_error_classifies_io_connection_refused() {
        let e = io::Error::from(io::ErrorKind::ConnectionRefused);
        assert!(is_network_error(&e));
        assert!(is_retriable(&e));
    }

    #[test]
    fn is_network_error_classifies_io_connection_reset() {
        let e = io::Error::from(io::ErrorKind::ConnectionReset);
        assert!(is_network_error(&e));
        assert!(is_retriable(&e));
    }

    #[test]
    fn is_network_error_classifies_io_connection_aborted() {
        let e = io::Error::from(io::ErrorKind::ConnectionAborted);
        assert!(is_network_error(&e));
        assert!(is_retriable(&e));
    }

    #[test]
    fn is_network_error_classifies_io_broken_pipe() {
        let e = io::Error::from(io::ErrorKind::BrokenPipe);
        assert!(is_network_error(&e));
        assert!(is_retriable(&e));
    }

    #[test]
    fn is_network_error_classifies_operation_timed_out_substring() {
        // Simulate a reqwest- or hyper-wrapped error whose io::ErrorKind has
        // been coerced to Other but whose Display still carries the Windows /
        // macOS TimedOut phrasing. Both the substring path and the
        // ErrorKind path must classify this independently.
        let other_kind = io::Error::other("operation timed out");
        assert!(is_network_error(&other_kind));
        assert!(is_retriable(&other_kind));

        let kind_only = io::Error::from(io::ErrorKind::TimedOut);
        assert!(is_network_error(&kind_only));
        assert!(is_retriable(&kind_only));
    }

    #[test]
    fn network_error_wrapped_unexpected_eof() {
        // Wrap an UnexpectedEof in an outer error so chain-walking is exercised.
        #[derive(Debug)]
        struct Wrap(io::Error);
        impl fmt::Display for Wrap {
            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
                write!(f, "read failed")
            }
        }
        impl StdError for Wrap {
            fn source(&self) -> Option<&(dyn StdError + 'static)> {
                Some(&self.0)
            }
        }
        let inner = io::Error::from(io::ErrorKind::UnexpectedEof);
        let outer = Wrap(inner);
        assert!(is_network_error(&outer));
    }

    #[test]
    fn network_error_non_network_strings_reject() {
        for s in [
            "file not found",
            "permission denied",
            "dial tcp: lookup example.com: no such host",
            "",
        ] {
            let e = OwnedErr(s.to_string());
            assert!(!is_network_error(&e), "expected NOT network error: {s:?}");
        }
    }

    #[test]
    fn retriable_opt_nil_passthrough() {
        assert!(!is_retriable_opt(None));
    }

    #[test]
    fn http_error_500_retriable() {
        let e = HttpError::new(StrErr("internal server error"), 500);
        assert!(is_retriable(&e));
    }

    #[test]
    fn http_error_502_503_retriable() {
        for s in [502u16, 503] {
            let e = HttpError::new(StrErr("bad gateway"), s);
            assert!(is_retriable(&e), "status {s} should be retriable");
        }
    }

    #[test]
    fn http_error_429_retriable() {
        let e = HttpError::new(StrErr("rate limited"), 429);
        assert!(is_retriable(&e));
    }

    #[test]
    fn http_error_4xx_not_retriable() {
        for s in [400u16, 401, 403, 404, 422] {
            let e = HttpError::new(StrErr("client err"), s);
            assert!(!is_retriable(&e), "status {s} should NOT be retriable");
        }
    }

    #[test]
    fn http_error_zero_status_routes_via_message() {
        // Status 0 == network-level failure with no response. Retriability
        // falls back to the network-error substring matcher on the inner.
        let net = HttpError::new(StrErr("connection reset"), 0);
        assert!(is_retriable(&net));

        let non_net = HttpError::new(StrErr("dial failed"), 0);
        assert!(!is_retriable(&non_net));
    }

    #[test]
    fn http_error_unwrap_chain_visible() {
        let inner = StrErr("inner");
        let e = HttpError::new(inner, 503);
        assert!(e.source().is_some());
    }

    #[test]
    fn from_response_nil_resp_yields_status_zero() {
        // No response means status 0.
        // Use a concrete `io::Error` since `reqwest::Error` cannot be
        // synthesised in tests; the API accepts any `E: StdError + Send + Sync`.
        let inner = io::Error::other("connect: dial tcp");
        let e = HttpError::from_response(inner, None);
        assert_eq!(e.status, 0);
    }

    #[test]
    fn from_response_unwrap_chain_visible() {
        // The inner error must remain reachable via the StdError chain so
        // is_retriable's network-error matcher can still see the cause.
        let inner = io::Error::other("connection reset by peer");
        let e = HttpError::from_response(inner, None);
        assert!(
            e.source().is_some(),
            "inner error must be reachable via source()"
        );
        // And classification must walk through to the network-error matcher.
        assert!(is_retriable(&e));
    }

    #[test]
    fn retriable_wrapper_is_retriable() {
        let e = Retriable::new(StrErr("retry me"));
        assert!(is_retriable(&e));
    }

    #[test]
    fn retriable_wrapper_overrides_4xx() {
        // A 422 wrapped in Retriable is still retriable.
        let inner = HttpError::new(StrErr("exists"), 422);
        let outer = Retriable::new(inner);
        assert!(is_retriable(&outer));
    }

    #[test]
    fn retriable_wrapper_unwrap_chain_visible() {
        let inner = StrErr("inner");
        let e = Retriable::new(inner);
        assert!(e.source().is_some());
    }

    #[test]
    fn plain_error_not_retriable() {
        let e = StrErr("something");
        assert!(!is_retriable(&e));
    }

    #[test]
    fn anyhow_error_threadable() {
        // Ensure is_retriable works through anyhow::Error's deref-to-dyn path
        // (which is the canonical caller form across the codebase).
        let e: anyhow::Error = anyhow::anyhow!("connection refused");
        assert!(is_retriable(e.as_ref()));

        let e2: anyhow::Error = anyhow::anyhow!("permission denied");
        assert!(!is_retriable(e2.as_ref()));
    }

    #[test]
    fn is_retriable_chain_walks_to_http_error() {
        // An anyhow::Error wrapping a concrete HttpError must be classified
        // by walking source(), not by Display alone — the message "outer"
        // gives no hint, the 503 status does.
        let inner = HttpError::new(StrErr("bad gateway"), 503);
        let wrapped: anyhow::Error = anyhow::Error::new(inner).context("publish failed");
        assert!(is_retriable(wrapped.as_ref()));
    }

    // ----- as_ref vs root_cause drift guard ---------------------------------
    //
    // Every consumer of `retry_http_blocking` (artifactory, cloudsmith, the
    // future stage-blob upload paths) classifies via `is_retriable(err.as_ref())`.
    // A subtle but catastrophic regression is to "simplify" that to
    // `is_retriable(err.root_cause())`, which walks past the HttpError wrapper
    // to the leaf io::Error — at which point 5xx misclassifies as fast-fail
    // (the leaf has no status code), and the entire retry policy becomes a
    // no-op. These tests pin the distinction once at the helper's home.

    #[test]
    fn classifier_5xx_via_anyhow_chain_uses_as_ref() {
        let wrapped: anyhow::Error =
            anyhow::Error::new(HttpError::new(std::io::Error::other("503"), 503))
                .context("publish");
        assert!(
            is_retriable(wrapped.as_ref()),
            "5xx HttpError reached via as_ref() must classify retriable"
        );
    }

    #[test]
    fn classifier_root_cause_walks_past_http_error_drift_guard() {
        // Drift guard: root_cause() unwraps to the leaf io::Error, which
        // has no status. If a future caller ever swaps as_ref → root_cause
        // they'll regress 5xx retry handling. This assertion locks the
        // distinction.
        let wrapped: anyhow::Error =
            anyhow::Error::new(HttpError::new(std::io::Error::other("503"), 503))
                .context("publish");
        assert!(
            !is_retriable(wrapped.root_cause()),
            "root_cause() walks past HttpError; 5xx must NOT be detected via the leaf"
        );
    }

    #[test]
    fn classifier_429_via_anyhow_chain_uses_as_ref() {
        // Symmetry with the 5xx case: 429 is the other retriable status
        // class and must also stay reachable via as_ref().
        let wrapped: anyhow::Error =
            anyhow::Error::new(HttpError::new(std::io::Error::other("429"), 429))
                .context("publish");
        assert!(is_retriable(wrapped.as_ref()));
        assert!(!is_retriable(wrapped.root_cause()));
    }

    // ----- retry_http_blocking behavioural tests ---------------------------
    //
    // `reqwest::Error` has no public constructor, so the transport-error
    // branch is exercised indirectly via per-publisher integration tests
    // (which mock at the network layer). The unit tests here drive a tiny
    // hand-rolled TCP server so we can exercise the success / non-success
    // status branches with a real reqwest::blocking::Client end-to-end.

    use crate::test_helpers::responder::spawn_oneshot_http_responder;

    #[test]
    fn retry_http_blocking_success_returns_first_attempt() {
        let (addr, calls) =
            spawn_oneshot_http_responder(vec!["HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"]);
        let client = reqwest::blocking::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_blocking(
            RetryLog::new("test", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |_, _| String::from("should not be called on success"),
        );
        let (status, body) = result.expect("success");
        assert_eq!(status.as_u16(), 200);
        assert_eq!(body, "ok");
        assert_eq!(calls.load(Ordering::SeqCst), 1, "single attempt");
    }

    #[test]
    fn retry_http_blocking_retries_5xx_then_succeeds() {
        let (addr, calls) = spawn_oneshot_http_responder(vec![
            "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n",
            "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok",
        ]);
        let client = reqwest::blocking::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_blocking(
            RetryLog::new("test", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |status, body| format!("{status}: {body}"),
        );
        let (status, _) = result.expect("eventually succeeds");
        assert_eq!(status.as_u16(), 200);
        assert_eq!(calls.load(Ordering::SeqCst), 2, "one retry then success");
    }

    #[test]
    fn retry_http_blocking_deadline_past_stops_after_one_attempt() {
        let (addr, calls) = spawn_oneshot_http_responder(vec![
            "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n",
            "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok",
        ]);
        let client = reqwest::blocking::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_secs(10),
            max_delay: Duration::from_secs(300),
        };
        let deadline = std::time::Instant::now();
        let result = retry_http_blocking_deadline(
            RetryLog::new("test", test_logger()),
            &policy,
            Some(deadline),
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |status, body| format!("{status}: {body}"),
        );
        assert!(result.is_err(), "past deadline must fail on the 503");
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "past deadline stops before the second attempt"
        );
    }

    #[test]
    fn retry_http_blocking_4xx_fast_fails_no_retry() {
        let (addr, calls) = spawn_oneshot_http_responder(vec![
            "HTTP/1.1 404 Not Found\r\nContent-Length: 9\r\n\r\nnot found",
        ]);
        let client = reqwest::blocking::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 5,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_blocking(
            RetryLog::new("myscope", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |status, body| format!("custom error: {status} body={body}"),
        );
        let err = result.expect_err("4xx must fast-fail");
        let chain = format!("{err:#}");
        assert!(
            chain.contains("custom error"),
            "error formatter must be invoked on non-success; got: {chain}"
        );
        assert!(chain.contains("404"), "status must be in chain: {chain}");
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "4xx must NOT retry (only one connection accepted)"
        );
    }

    #[test]
    fn retry_http_blocking_redirect_class_alters_success_predicate() {
        let (addr, _calls) = spawn_oneshot_http_responder(vec![
            "HTTP/1.1 307 Temporary Redirect\r\nLocation: /next\r\nContent-Length: 0\r\n\r\n",
        ]);
        let client = reqwest::blocking::Client::builder()
            .timeout(Duration::from_secs(2))
            // Disable redirect-following so the 307 surfaces to our helper.
            .redirect(reqwest::redirect::Policy::none())
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_blocking(
            RetryLog::new("test", test_logger()),
            &policy,
            SuccessClass::AllowRedirects,
            |_| client.get(format!("http://{addr}/")).send(),
            |_, _| String::from("should not be called on 3xx with AllowRedirects"),
        );
        let (status, _) = result.expect("3xx is success under AllowRedirects");
        assert_eq!(status.as_u16(), 307);
    }

    // ----- retry_http_blocking_bytes behavioural tests ---------------------

    #[test]
    fn retry_http_blocking_bytes_preserves_non_utf8_body() {
        // A body with invalid-UTF-8 byte sequences (gzip magic + a bare
        // continuation byte) proves the bytes variant does not run a lossy
        // UTF-8 pass over the success payload — `resp.text()` would silently
        // rewrite these to U+FFFD, corrupting the digest of whatever the
        // caller hashes.
        let body: Vec<u8> = vec![0x1f, 0x8b, 0x08, 0x00, 0x80, 0xff, 0xfe, 0x00];
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
        let addr = listener.local_addr().expect("local_addr");
        let body_for_thread = body.clone();
        std::thread::spawn(move || {
            use std::io::{Read, Write};
            if let Ok((mut stream, _)) = listener.accept() {
                let mut buf = [0u8; 1024];
                let _ = stream.read(&mut buf);
                let header = format!(
                    "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                    body_for_thread.len()
                );
                let _ = stream.write_all(header.as_bytes());
                let _ = stream.write_all(&body_for_thread);
                let _ = stream.flush();
                let _ = stream.shutdown(std::net::Shutdown::Both);
            }
        });
        let client = reqwest::blocking::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 1,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_blocking_bytes(
            RetryLog::new("test", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |_, _| String::from("should not be called on success"),
        );
        let (status, bytes) = result.expect("success");
        assert_eq!(status.as_u16(), 200);
        assert_eq!(bytes, body, "binary body must round-trip byte-for-byte");
    }

    #[test]
    fn retry_http_blocking_bytes_4xx_fast_fails_no_retry() {
        let (addr, calls) = spawn_oneshot_http_responder(vec![
            "HTTP/1.1 404 Not Found\r\nContent-Length: 9\r\n\r\nnot found",
        ]);
        let client = reqwest::blocking::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 5,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_blocking_bytes(
            RetryLog::new("myscope", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |status, body| format!("custom error: {status} body={body}"),
        );
        let err = result.expect_err("4xx must fast-fail");
        let chain = format!("{err:#}");
        assert!(
            chain.contains("custom error") && chain.contains("not found"),
            "error formatter must see the (lossily-decoded) error body: {chain}"
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "4xx must NOT retry (only one connection accepted)"
        );
    }

    #[test]
    fn retry_http_blocking_bytes_retries_5xx_then_succeeds() {
        let (addr, calls) = spawn_oneshot_http_responder(vec![
            "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n",
            "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok",
        ]);
        let client = reqwest::blocking::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_blocking_bytes(
            RetryLog::new("test", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |status, body| format!("{status}: {body}"),
        );
        let (status, bytes) = result.expect("eventually succeeds");
        assert_eq!(status.as_u16(), 200);
        assert_eq!(bytes, b"ok");
        assert_eq!(calls.load(Ordering::SeqCst), 2, "one retry then success");
    }

    // ----- retry_http_async behavioural tests ------------------------------
    //
    // Mirrors the blocking suite but drives an async reqwest::Client against
    // the same hand-rolled TCP responder (running on a worker thread, so the
    // tokio reactor is free to drive the client futures). The transport-error
    // arm (Err(reqwest::Error)) is exercised by
    // `retry_http_{async,blocking}_transport_error_retries_then_fails` below,
    // which bind an ephemeral port, drop the listener, then point the client
    // at the now-defunct address.

    #[tokio::test]
    async fn retry_http_async_success_returns_first_attempt() {
        let (addr, calls) =
            spawn_oneshot_http_responder(vec!["HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"]);
        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_async(
            RetryLog::new("test", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |_, _| String::from("should not be called on success"),
        )
        .await;
        let resp = result.expect("success");
        assert_eq!(resp.status().as_u16(), 200);
        let body = resp.text().await.expect("body");
        assert_eq!(body, "ok");
        assert_eq!(calls.load(Ordering::SeqCst), 1, "single attempt");
    }

    #[tokio::test]
    async fn retry_http_async_retries_5xx_then_succeeds() {
        let (addr, calls) = spawn_oneshot_http_responder(vec![
            "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n",
            "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok",
        ]);
        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_async(
            RetryLog::new("test", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |status, body| format!("{status}: {body}"),
        )
        .await;
        let resp = result.expect("eventually succeeds");
        assert_eq!(resp.status().as_u16(), 200);
        assert_eq!(calls.load(Ordering::SeqCst), 2, "one retry then success");
    }

    #[tokio::test]
    async fn retry_http_async_4xx_fast_fails_no_retry() {
        let (addr, calls) = spawn_oneshot_http_responder(vec![
            "HTTP/1.1 404 Not Found\r\nContent-Length: 9\r\n\r\nnot found",
        ]);
        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 5,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_async(
            RetryLog::new("myscope", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |status, body| format!("custom error: {status} body={body}"),
        )
        .await;
        let err = result.expect_err("4xx must fast-fail");
        let chain = format!("{err:#}");
        assert!(
            chain.contains("custom error"),
            "error formatter must be invoked on non-success; got: {chain}"
        );
        assert!(chain.contains("404"), "status must be in chain: {chain}");
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "4xx must NOT retry (only one connection accepted)"
        );
    }

    #[tokio::test]
    async fn retry_http_async_429_retries_then_succeeds() {
        // 429 (Too Many Requests) is the second retriable class alongside
        // 5xx. Ensures the helper doesn't accidentally fast-fail on rate
        // limits — a regression here would defeat the whole point of
        // wiring retry into release publishers.
        let (addr, calls) = spawn_oneshot_http_responder(vec![
            "HTTP/1.1 429 Too Many Requests\r\nContent-Length: 0\r\n\r\n",
            "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok",
        ]);
        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(2))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_async(
            RetryLog::new("test", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| client.get(format!("http://{addr}/")).send(),
            |status, body| format!("{status}: {body}"),
        )
        .await;
        let resp = result.expect("429 retried then success");
        assert_eq!(resp.status().as_u16(), 200);
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

    // ----- transport-error behavioural tests -------------------------------
    //
    // The transport-error arm (Err(reqwest::Error): DNS failure, connection
    // refused, EOF, TLS handshake failure, etc.) is the single most
    // reviewer-load-bearing path: it is the one the helper claims to retry
    // and that publishers rely on for resilience against transient network
    // blips. The pattern below dials the RFC 2606-reserved `.invalid` TLD,
    // which is guaranteed never to resolve, so every attempt fails at the
    // DNS-resolution stage in a few milliseconds on Linux, macOS, and
    // Windows alike.
    //
    // We verify:
    //   1. the helper retries (attempt counter > 1)
    //   2. eventually surfaces an Err with the configured label in the chain
    // The outer attempt counter is incremented inside the closure, so it
    // sees one bump per attempt regardless of the underlying transport
    // outcome.
    //
    // RFC 2606 (https://datatracker.ietf.org/doc/html/rfc2606) reserves the
    // `.invalid` TLD precisely for this purpose; using it removes any
    // dependence on OS-level TCP semantics (Windows' kernel can retransmit
    // SYN against an unbound loopback port until the connect timeout fires
    // rather than refusing synchronously like Linux + macOS do).
    const TRANSPORT_FAIL_URL: &str = "http://nonexistent.invalid/";

    #[test]
    fn retry_http_blocking_transport_error_retries_then_fails() {
        let attempts = std::sync::Arc::new(AtomicU32::new(0));
        let attempts_inner = attempts.clone();
        let client = reqwest::blocking::Client::builder()
            .timeout(Duration::from_millis(500))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_blocking(
            RetryLog::new("test-transport", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| {
                attempts_inner.fetch_add(1, Ordering::SeqCst);
                client.get(TRANSPORT_FAIL_URL).send()
            },
            |_, _| String::from("non-success branch should not be reached"),
        );
        let err = result.expect_err("transport error must surface as Err");
        let chain = format!("{err:#}");
        assert!(
            attempts.load(Ordering::SeqCst) > 1,
            "transport error must be retried; got {} attempts; chain={chain}",
            attempts.load(Ordering::SeqCst)
        );
        assert!(
            chain.contains("test-transport"),
            "label must surface in error chain; got: {chain}"
        );
    }

    #[tokio::test]
    async fn retry_http_async_transport_error_retries_then_fails() {
        let attempts = std::sync::Arc::new(AtomicU32::new(0));
        let attempts_inner = attempts.clone();
        let client = reqwest::Client::builder()
            .timeout(Duration::from_millis(500))
            .build()
            .expect("client");
        let policy = RetryPolicy {
            max_attempts: 3,
            base_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(2),
        };
        let result = retry_http_async(
            RetryLog::new("test-transport-async", test_logger()),
            &policy,
            SuccessClass::Strict,
            |_| {
                attempts_inner.fetch_add(1, Ordering::SeqCst);
                client.get(TRANSPORT_FAIL_URL).send()
            },
            |_, _| String::from("non-success branch should not be reached"),
        )
        .await;
        let err = result.expect_err("transport error must surface as Err");
        assert!(
            attempts.load(Ordering::SeqCst) > 1,
            "transport error must be retried; got {} attempts",
            attempts.load(Ordering::SeqCst)
        );
        let chain = format!("{err:#}");
        assert!(
            chain.contains("test-transport-async"),
            "label must surface in error chain; got: {chain}"
        );
    }
}