cachekit-rs 0.13.1

//! Reliability tier: backpressure (bounded backend concurrency), retry with
//! exponential backoff + jitter, and a closed/open/half-open circuit breaker
//! around backend operations.
//!
//! All three are composed by a private `ReliableBackend` decorator around any
//! [`crate::backend::Backend`], applied by the builder when a [`ReliabilityConfig`] is set
//! (see [`crate::CacheKitBuilder::reliability`]). The intent presets
//! `production`, `secure`, and `io` enable it by default; `minimal` does
//! not — mirroring the TypeScript SDK's preset posture.
//!
//! Composition order is `backpressure(breaker(retry(op)))`:
//!
//! - The retry loop is *inside* the breaker (matching the TypeScript SDK's
//!   `ReliabilityExecutor`), so one exhausted retry sequence counts as a
//!   single breaker failure, and a fast-failing open breaker never spends
//!   time retrying.
//! - The concurrency limiter is *outermost*: one permit per logical cache
//!   operation, held across the entire breaker/retry sequence. That bounds
//!   in-flight work including retry amplification (K callers mid-backoff are
//!   still K permits — new work queues behind them instead of piling onto a
//!   struggling backend), and a shed call never touches breaker counters or
//!   half-open probe slots, so the breaker keeps measuring backend health,
//!   not caller-side overload. A permit holder never re-enters the limiter
//!   (backend ops don't nest), so holding permits across retry backoff
//!   cannot deadlock.
//!
//! Unlike the TypeScript breaker (which counts every error), only errors
//! classified retryable by [`crate::error::BackendErrorKind::is_retryable`] (`Transient`,
//! `Timeout`) count toward opening the circuit: they are the backend-health
//! signals. `Permanent` / `Authentication` errors are request-specific — five
//! malformed requests must not cut off healthy traffic.
//!
//! Requires a tokio runtime for backoff timers (`redis` and `cachekitio`
//! backends already do). Not available on wasm32 targets.

use std::future::Future;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, Instant};

use async_trait::async_trait;

use crate::backend::{Backend, Freshness, HealthStatus, LockableBackend};
use crate::client::SharedBackend;
use crate::error::BackendError;
use crate::random_unit;

// ── Configuration ────────────────────────────────────────────────────────────

/// Retry policy configuration (truncated exponential backoff with jitter).
///
/// Under [`CacheKit::io`](crate::CacheKit::io), all attempts of one operation
/// also share one deadline: 5 s for a read (`get`, `exists`) and 10 s for a
/// write (`set`, `delete`), matching the cachekit.io backend's per-attempt
/// timeouts. An attempt still running at the deadline is cancelled and the
/// operation fails with a `Timeout` error, and no backoff sleep starts that
/// would end past it. A cachekit.io request that stops answering therefore
/// costs one attempt, not `max_attempts` of them, while fast failures (a
/// `503`, a dropped connection) still get every attempt that fits. Other
/// backends, and a backend set on the builder after `io`, get no deadline:
/// their operations run to their own timeouts.
#[derive(Debug, Clone, PartialEq)]
pub struct RetryConfig {
    /// Total attempts, including the first (default: 3). `0` behaves as `1`.
    pub max_attempts: u32,
    /// Backoff base delay; attempt *n* waits `base_delay * 2^n` (default: 100 ms).
    pub base_delay: Duration,
    /// Backoff ceiling (default: 5 s).
    pub max_delay: Duration,
    /// Multiply each delay by a random factor in `[0.5, 1.5)` (default: true).
    pub jitter: bool,
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_attempts: 3,
            base_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(5),
            jitter: true,
        }
    }
}

/// Circuit breaker configuration.
///
/// Defaults mirror the TypeScript SDK's production preset
/// (`PRODUCTION_RELIABILITY` in `cachekit-ts/src/intents.ts`).
#[derive(Debug, Clone, PartialEq)]
pub struct CircuitBreakerConfig {
    /// Retryable failures within [`Self::rolling_window`] before the circuit
    /// opens (default: 5).
    pub failure_threshold: u32,
    /// Successes in half-open state required to close the circuit (default: 3).
    pub success_threshold: u32,
    /// How long the circuit stays open before allowing half-open probes
    /// (default: 5 s).
    pub open_timeout: Duration,
    /// Maximum concurrent probe calls in half-open state (default: 3).
    pub half_open_max_calls: u32,
    /// Rolling window for failure counting (default: 60 s).
    pub rolling_window: Duration,
}

impl Default for CircuitBreakerConfig {
    fn default() -> Self {
        Self {
            failure_threshold: 5,
            success_threshold: 3,
            open_timeout: Duration::from_secs(5),
            half_open_max_calls: 3,
            rolling_window: Duration::from_secs(60),
        }
    }
}

/// Backpressure configuration: bound how many backend data operations may be
/// in flight at once, so a slow or failing backend cannot exhaust the
/// caller's connection pool or memory.
///
/// Defaults mirror the Python SDK's `BackpressureConfig`
/// (`max_concurrent_requests: 100`, `queue_size: 1000`, `timeout: 0.1s`).
///
/// Over-limit calls first join a bounded waiting queue; a caller that finds
/// the queue full, or that waits longer than [`Self::acquire_timeout`]
/// without a permit freeing up, is shed with a
/// [`crate::error::BackendErrorKind::Backpressure`] error — never queued
/// unboundedly. Shed calls do not reach the backend and do not count toward
/// opening the circuit breaker.
///
/// On the `#[cachekit]` macro's plain path a shed is outage-class — exactly
/// like `CircuitOpen`, the wrapped function runs uncached (fail-open);
/// `secure` paths fail closed on a shed like on every other backend error.
#[derive(Debug, Clone, PartialEq)]
pub struct BackpressureConfig {
    /// Maximum backend data operations in flight at once (default: 100).
    /// `0` behaves as `1`; values above tokio's `Semaphore::MAX_PERMITS`
    /// (`usize::MAX >> 3`) are clamped to it, so `usize::MAX` reads as
    /// "effectively unbounded" rather than panicking the builder.
    pub max_concurrent: usize,
    /// Maximum callers waiting for a permit before further calls are shed
    /// immediately (default: 1000). `0` disables waiting entirely: a call
    /// that cannot take a permit on the spot is shed.
    pub max_queue: usize,
    /// How long a queued caller waits for a permit before it is shed
    /// (default: 100 ms).
    pub acquire_timeout: Duration,
}

impl Default for BackpressureConfig {
    fn default() -> Self {
        Self {
            max_concurrent: 100,
            max_queue: 1000,
            acquire_timeout: Duration::from_millis(100),
        }
    }
}

/// Reliability stack configuration: which layers to apply around backend ops.
///
/// The `Default` enables all layers with production defaults. Disable a
/// layer by setting its field to `None`:
///
/// ```
/// use cachekit::reliability::ReliabilityConfig;
///
/// let retry_only = ReliabilityConfig {
///     circuit_breaker: None,
///     backpressure: None,
///     ..ReliabilityConfig::default()
/// };
/// assert!(retry_only.retry.is_some());
/// ```
#[derive(Debug, Clone, PartialEq)]
pub struct ReliabilityConfig {
    /// Retry policy, or `None` to propagate every error on first failure.
    pub retry: Option<RetryConfig>,
    /// Circuit breaker, or `None` to never fail fast.
    pub circuit_breaker: Option<CircuitBreakerConfig>,
    /// Concurrency limiter, or `None` for unbounded backend concurrency.
    pub backpressure: Option<BackpressureConfig>,
}

impl Default for ReliabilityConfig {
    fn default() -> Self {
        Self {
            retry: Some(RetryConfig::default()),
            circuit_breaker: Some(CircuitBreakerConfig::default()),
            backpressure: Some(BackpressureConfig::default()),
        }
    }
}

impl ReliabilityConfig {
    /// A config with every layer off — the documented preset opt-out.
    ///
    /// Prefer this over spelling out a struct literal with all-`None`
    /// fields: a literal breaks downstream code every time the stack gains
    /// a layer (it has, twice).
    ///
    /// ```
    /// use cachekit::reliability::ReliabilityConfig;
    ///
    /// assert!(ReliabilityConfig::disabled().is_disabled());
    /// assert!(!ReliabilityConfig::default().is_disabled());
    /// ```
    #[must_use]
    pub fn disabled() -> Self {
        Self {
            retry: None,
            circuit_breaker: None,
            backpressure: None,
        }
    }

    /// `true` when no layer is enabled — the builder skips the (no-op)
    /// `ReliableBackend` decorator entirely. Lives here, next to the fields,
    /// so adding a layer cannot silently miss the builder gate again.
    #[must_use]
    pub fn is_disabled(&self) -> bool {
        self.retry.is_none() && self.circuit_breaker.is_none() && self.backpressure.is_none()
    }
}

// ── RetryPolicy ──────────────────────────────────────────────────────────────

/// One deadline shared by every attempt of an operation (see [`RetryConfig`]).
/// Crate-private and set only by [`CacheKit::io`](crate::CacheKit::io): the
/// budgets match the cachekit.io client's per-attempt timeouts, and a
/// backend with slower healthy operations must not be cut short by them.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct RetryDeadlines {
    /// For `get`, `get_with_freshness` and `exists`.
    pub(crate) read: Duration,
    /// For `set` and `delete`.
    pub(crate) write: Duration,
}

/// The cachekit.io deadlines: one per-attempt timeout each.
#[cfg(feature = "cachekitio")]
pub(crate) const CACHEKITIO_DEADLINES: RetryDeadlines = RetryDeadlines {
    read: Duration::from_secs(5),
    write: Duration::from_secs(10),
};

/// Retries an operation on errors where [`crate::error::BackendErrorKind::is_retryable`] is
/// true, sleeping a truncated exponential backoff (with jitter) between
/// attempts, all inside one deadline. `Permanent` and `Authentication` errors,
/// and a quota or balance deny, propagate immediately.
#[derive(Debug)]
pub(crate) struct RetryPolicy {
    config: RetryConfig,
}

impl RetryPolicy {
    pub(crate) fn new(config: RetryConfig) -> Self {
        Self { config }
    }

    fn delay(&self, attempt: u32) -> Duration {
        let exp = self
            .config
            .base_delay
            .saturating_mul(2u32.saturating_pow(attempt));
        let capped = exp.min(self.config.max_delay);
        if self.config.jitter {
            capped.mul_f64(0.5 + random_unit())
        } else {
            capped
        }
    }

    /// Run `f` until it succeeds, fails for good, or `budget` (if any) runs out.
    ///
    /// The tokio clock measures the budget, so a paused test runtime controls it.
    pub(crate) async fn execute<T, F, Fut>(
        &self,
        budget: Option<Duration>,
        f: F,
    ) -> Result<T, BackendError>
    where
        F: Fn() -> Fut,
        Fut: Future<Output = Result<T, BackendError>>,
    {
        let deadline = budget.map(|budget| tokio::time::Instant::now() + budget);
        let mut attempt: u32 = 0;
        loop {
            let result = match deadline {
                Some(deadline) => match tokio::time::timeout_at(deadline, f()).await {
                    Ok(result) => result,
                    Err(_elapsed) => {
                        return Err(BackendError::timeout(format!(
                            "operation did not complete within its {:?} deadline ({} attempt(s))",
                            budget.unwrap_or_default(),
                            attempt + 1
                        )))
                    }
                },
                None => f().await,
            };
            match result {
                Ok(v) => return Ok(v),
                Err(e)
                    if e.kind.is_retryable()
                        && !e.is_quota_denied()
                        && attempt + 1 < self.config.max_attempts =>
                {
                    let delay = self.delay(attempt);
                    // A retry that cannot start before the deadline would only
                    // turn this error into a deadline timeout: return it now.
                    if deadline.is_some_and(|d| tokio::time::Instant::now() + delay >= d) {
                        return Err(e);
                    }
                    tokio::time::sleep(delay).await;
                    attempt += 1;
                }
                Err(e) => return Err(e),
            }
        }
    }
}

// ── CircuitBreaker ───────────────────────────────────────────────────────────

/// Circuit breaker state, as reported by
/// [`CacheKit::circuit_state`](crate::CacheKit::circuit_state) and carried by
/// the `tracing` transition events (`from` / `to` fields).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CircuitState {
    /// Normal operation; calls pass through.
    Closed,
    /// Failing fast; calls return a [`crate::error::BackendErrorKind::CircuitOpen`] error
    /// without reaching the backend.
    Open,
    /// Probing recovery with a bounded number of trial calls.
    HalfOpen,
}

#[derive(Debug)]
enum State {
    Closed,
    Open { since: Instant },
    HalfOpen,
}

impl From<&State> for CircuitState {
    fn from(state: &State) -> Self {
        match state {
            State::Closed => Self::Closed,
            State::Open { .. } => Self::Open,
            State::HalfOpen => Self::HalfOpen,
        }
    }
}

#[derive(Debug)]
struct BreakerInner {
    state: State,
    /// Timestamps of counted failures inside the rolling window.
    failures: Vec<Instant>,
    half_open_successes: u32,
    half_open_calls: u32,
    /// Number of state transitions so far; stamps each [`Transition`] with
    /// its `seq` under the lock.
    transitions: u64,
}

/// How a completed call is reported back to the breaker.
enum Outcome {
    Success,
    /// A retryable-kind failure — a backend-health signal.
    Failure,
    /// A non-retryable failure (permanent/auth) — request-specific, does not
    /// count toward opening the circuit but must release its half-open slot,
    /// or a burst of permanent errors would wedge the breaker half-open.
    Neutral,
}

/// State machine: closed → (failures ≥ threshold in window) → open →
/// (open_timeout elapsed) → half-open → (successes ≥ threshold) → closed,
/// or (any counted failure) → open.
#[derive(Debug)]
pub(crate) struct CircuitBreaker {
    /// Process-unique id, stable for the breaker's lifetime; the `breaker`
    /// field on transition events. Several clients under one subscriber each
    /// restart `seq` at 1, so `(breaker, seq)` is the ordering key.
    id: u64,
    config: CircuitBreakerConfig,
    inner: Mutex<BreakerInner>,
}

/// Source of [`CircuitBreaker::id`]s. Per process: fleet-wide correlation
/// combines the field with the host/process context the subscriber adds.
static BREAKER_IDS: AtomicU64 = AtomicU64::new(0);

impl CircuitBreaker {
    pub(crate) fn new(config: CircuitBreakerConfig) -> Self {
        Self {
            id: BREAKER_IDS.fetch_add(1, Ordering::Relaxed) + 1,
            config,
            inner: Mutex::new(BreakerInner {
                state: State::Closed,
                failures: Vec::new(),
                half_open_successes: 0,
                half_open_calls: 0,
                transitions: 0,
            }),
        }
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, BreakerInner> {
        // A poisoned lock means a panic mid-update; breaker state is advisory,
        // so recovering the guard is strictly better than propagating panics.
        self.inner.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// Current state (transitions open → half-open lazily on inspection).
    pub(crate) fn state(&self) -> CircuitState {
        let mut inner = self.lock();
        let change = self.maybe_half_open(&mut inner);
        let state = CircuitState::from(&inner.state);
        drop(inner);
        emit(change);
        state
    }

    fn maybe_half_open(&self, inner: &mut BreakerInner) -> Option<Transition> {
        let State::Open { since } = inner.state else {
            return None;
        };
        if since.elapsed() < self.config.open_timeout {
            return None;
        }
        let change = transition(self.id, inner, State::HalfOpen);
        inner.half_open_successes = 0;
        inner.half_open_calls = 0;
        Some(change)
    }

    /// Admit a call, or fail fast with a circuit-open error.
    ///
    /// Returns an RAII [`ProbePermit`]: if the guarded future is cancelled
    /// (caller timeout/`select!`) or panics before an outcome is recorded,
    /// the permit's `Drop` releases any half-open probe slot it took —
    /// otherwise `half_open_max_calls` cancelled probes would wedge the
    /// breaker half-open forever, fast-failing every call even against a
    /// recovered backend.
    fn try_acquire(&self) -> Result<ProbePermit<'_>, BackendError> {
        let mut inner = self.lock();
        let change = self.maybe_half_open(&mut inner);
        let admitted = match inner.state {
            State::Closed => Ok(ProbePermit {
                breaker: self,
                took_slot: false,
            }),
            State::Open { .. } => Err(BackendError::circuit_open(
                "circuit breaker is open: backend calls are failing fast",
            )),
            State::HalfOpen => {
                if inner.half_open_calls >= self.config.half_open_max_calls {
                    Err(BackendError::circuit_open(
                        "circuit breaker is half-open and the probe limit is reached",
                    ))
                } else {
                    inner.half_open_calls += 1;
                    Ok(ProbePermit {
                        breaker: self,
                        took_slot: true,
                    })
                }
            }
        };
        drop(inner);
        emit(change);
        admitted
    }

    fn record(&self, outcome: &Outcome) {
        let mut inner = self.lock();
        let mut change = None;
        match outcome {
            Outcome::Success => {
                if matches!(inner.state, State::HalfOpen) {
                    inner.half_open_successes += 1;
                    if inner.half_open_successes >= self.config.success_threshold {
                        change = Some(transition(self.id, &mut inner, State::Closed));
                        inner.failures.clear();
                        inner.half_open_successes = 0;
                        inner.half_open_calls = 0;
                    } else {
                        // Release this probe's slot. `half_open_calls` caps the
                        // number of *in-flight* probes, so a success that does
                        // not yet close the breaker must free its slot (exactly
                        // as `Neutral` does). Without this, a config with
                        // success_threshold > half_open_max_calls wedges the
                        // breaker half-open forever: the slots fill, successes
                        // stall below the threshold, and every subsequent call
                        // fails fast with CircuitOpen even against a healthy
                        // backend.
                        inner.half_open_calls = inner.half_open_calls.saturating_sub(1);
                    }
                }
            }
            Outcome::Failure => match inner.state {
                State::HalfOpen => {
                    let now = Instant::now();
                    change = Some(transition(self.id, &mut inner, State::Open { since: now }));
                    inner.half_open_successes = 0;
                    inner.half_open_calls = 0;
                }
                State::Closed => {
                    let now = Instant::now();
                    inner.failures.push(now);
                    let window = self.config.rolling_window;
                    inner.failures.retain(|t| now.duration_since(*t) <= window);
                    if inner.failures.len() >= self.config.failure_threshold as usize {
                        change = Some(transition(self.id, &mut inner, State::Open { since: now }));
                        inner.failures.clear();
                    }
                }
                // Open without an admitted call cannot report a failure;
                // ignore rather than extend the open window.
                State::Open { .. } => {}
            },
            Outcome::Neutral => {
                if matches!(inner.state, State::HalfOpen) {
                    inner.half_open_calls = inner.half_open_calls.saturating_sub(1);
                }
            }
        }
        drop(inner);
        emit(change);
    }
}

/// A breaker state change handed back by [`transition`] so the caller can
/// report it once the breaker lock is released.
///
/// `(breaker, seq)` is the ordering key. `seq` is assigned under the lock and
/// counts this breaker's transitions; `breaker` says which breaker, since
/// several clients under one subscriber each restart `seq` at 1. Emission
/// happens after unlock, so two transitions can reach a subscriber in the
/// wrong wall-clock order (`Closed → Open` committed, unlock, a zero
/// `open_timeout` lets another caller commit and emit `Open → HalfOpen`
/// first); a consumer that groups by `breaker` and orders by `seq`
/// reconstructs each true timeline.
#[cfg_attr(not(feature = "tracing"), allow(dead_code))]
#[derive(Debug, Clone, Copy)]
struct Transition {
    breaker: u64,
    seq: u64,
    from: CircuitState,
    to: CircuitState,
}

/// Apply a breaker state change under the lock and return it for [`emit`].
fn transition(breaker: u64, inner: &mut BreakerInner, to: State) -> Transition {
    inner.transitions += 1;
    let change = Transition {
        breaker,
        seq: inner.transitions,
        from: CircuitState::from(&inner.state),
        to: CircuitState::from(&to),
    };
    inner.state = to;
    change
}

/// Report a transition on the `tracing` target (feature `tracing`): `warn`
/// when the circuit opens — traffic is now failing fast — `info` for half-open
/// and closed. Fields: `breaker`, `seq`, `from`, `to`.
///
/// Must be called with the breaker lock **released**: a subscriber may react
/// to the event by calling
/// [`CacheKit::circuit_state`](crate::CacheKit::circuit_state), which takes
/// the same lock, and a synchronous writer must not stall every other
/// worker's `try_acquire` for the duration of the write. The price is that
/// emission order is not transition order under contention — see
/// [`Transition`]; `(breaker, seq)` is what a consumer orders by.
fn emit(change: Option<Transition>) {
    #[cfg(feature = "tracing")]
    if let Some(Transition {
        breaker,
        seq,
        from,
        to,
    }) = change
    {
        if to == CircuitState::Open {
            tracing::warn!(breaker, seq, from = ?from, to = ?to, "circuit breaker opened");
        } else {
            tracing::info!(breaker, seq, from = ?from, to = ?to, "circuit breaker transitioned");
        }
    }
    #[cfg(not(feature = "tracing"))]
    let _ = change;
}

// ── ProbePermit ──────────────────────────────────────────────────────────────

/// RAII token for a breaker-admitted call.
///
/// Slot accounting lives in exactly one of two places: [`Self::complete`]
/// (normal return — the outcome arms of `record` own the bookkeeping from
/// there) or `Drop` (cancel/panic — release the slot like `Neutral`, no
/// transition). Manual increment/decrement pairs leaked twice before this
/// guard existed; do not reintroduce them.
#[derive(Debug)]
struct ProbePermit<'a> {
    breaker: &'a CircuitBreaker,
    /// Whether this admission consumed a half-open probe slot.
    took_slot: bool,
}

impl ProbePermit<'_> {
    /// Report the call's outcome and disarm the drop-release.
    fn complete(mut self, outcome: &Outcome) {
        self.took_slot = false;
        self.breaker.record(outcome);
    }
}

impl Drop for ProbePermit<'_> {
    fn drop(&mut self) {
        if !self.took_slot {
            return;
        }
        // No outcome was recorded: the guarded future was cancelled mid-await
        // or panicked. Free the probe slot so the half-open window can keep
        // probing; if the breaker transitioned meanwhile (counters reset),
        // the saturating decrement is a no-op.
        let mut inner = self.breaker.lock();
        if matches!(inner.state, State::HalfOpen) {
            inner.half_open_calls = inner.half_open_calls.saturating_sub(1);
        }
    }
}

// ── ConcurrencyLimiter ───────────────────────────────────────────────────────

/// Bounds concurrent backend data operations with a semaphore and a bounded
/// waiting queue (two-phase, like the Python SDK's `BackpressureController`):
/// a saturated limiter admits up to `max_queue` waiters for at most
/// `acquire_timeout` each; everyone else is shed with a
/// [`crate::error::BackendErrorKind::Backpressure`] error.
#[derive(Debug)]
pub(crate) struct ConcurrencyLimiter {
    semaphore: tokio::sync::Semaphore,
    /// Callers currently waiting for a permit (phase-2 queue depth).
    waiting: std::sync::atomic::AtomicUsize,
    config: BackpressureConfig,
}

/// RAII guard for a slot in the waiting queue: decrements `waiting` on every
/// exit path, including cancellation mid-`acquire` (same lesson as
/// [`ProbePermit`] — manual increment/decrement pairs leak on cancel).
struct QueueSlot<'a> {
    waiting: &'a std::sync::atomic::AtomicUsize,
}

impl Drop for QueueSlot<'_> {
    fn drop(&mut self) {
        self.waiting
            .fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
    }
}

impl ConcurrencyLimiter {
    pub(crate) fn new(config: BackpressureConfig) -> Self {
        Self {
            // `Semaphore::new(0)` would shed every call after acquire_timeout
            // with nothing ever admitted — clamp like RetryConfig's "0
            // behaves as 1". The upper clamp matters too: `Semaphore::new`
            // PANICS above `MAX_PERMITS` (usize::MAX >> 3), and usize::MAX
            // is the natural "effectively unbounded" sentinel a caller will
            // reach for — a config value must never panic the builder.
            semaphore: tokio::sync::Semaphore::new(
                config
                    .max_concurrent
                    .clamp(1, tokio::sync::Semaphore::MAX_PERMITS),
            ),
            waiting: std::sync::atomic::AtomicUsize::new(0),
            config,
        }
    }

    /// Take a permit, or shed the call.
    ///
    /// Phase 1: a free permit is taken on the spot — no queue accounting.
    /// Phase 2 (saturated): join the bounded waiting queue and wait up to
    /// `acquire_timeout` for a permit; queue-full and wait-timeout both shed
    /// with a `Backpressure` error. The returned permit releases on drop, so
    /// a cancelled or panicking caller can never leak capacity.
    async fn acquire(&self) -> Result<tokio::sync::SemaphorePermit<'_>, BackendError> {
        use std::sync::atomic::Ordering;

        if let Ok(permit) = self.semaphore.try_acquire() {
            return Ok(permit);
        }
        if self.waiting.fetch_add(1, Ordering::AcqRel) >= self.config.max_queue {
            self.waiting.fetch_sub(1, Ordering::AcqRel);
            return Err(BackendError::backpressure(format!(
                "backpressure: waiting queue is full (max_queue={}), call shed without reaching the backend",
                self.config.max_queue
            )));
        }
        let _slot = QueueSlot {
            waiting: &self.waiting,
        };
        match tokio::time::timeout(self.config.acquire_timeout, self.semaphore.acquire()).await {
            Ok(Ok(permit)) => Ok(permit),
            // The semaphore is never closed; treat a close defensively as shed.
            Ok(Err(_closed)) => Err(BackendError::backpressure(
                "backpressure: limiter unavailable, call shed without reaching the backend",
            )),
            Err(_elapsed) => Err(BackendError::backpressure(format!(
                "backpressure: timed out waiting for a permit after {:?}, call shed without reaching the backend",
                self.config.acquire_timeout
            ))),
        }
    }
}

// ── ReliableBackend ──────────────────────────────────────────────────────────

/// Decorator that applies the reliability stack to every cache operation of
/// an inner [`Backend`]: `backpressure(breaker(retry(op)))`.
///
/// - `get`/`set`/`delete`/`exists` take a concurrency-limiter permit, are
///   retried on retryable errors, and gated by the circuit breaker.
/// - `health` passes through unguarded — it is a diagnostic and must keep
///   reporting truthfully while the breaker fails data calls fast (or the
///   limiter sheds them).
/// - [`Backend::as_lockable`] forwards to the inner backend so distributed
///   fill locks bypass the stack (locks are best-effort advisory).
pub(crate) struct ReliableBackend {
    inner: SharedBackend,
    retry: Option<RetryPolicy>,
    /// Shared with the owning `CacheKit` so
    /// [`CacheKit::circuit_state`](crate::CacheKit::circuit_state) can read
    /// it through the type-erased backend.
    breaker: Option<Arc<CircuitBreaker>>,
    limiter: Option<ConcurrencyLimiter>,
    deadlines: Option<RetryDeadlines>,
}

impl ReliableBackend {
    pub(crate) fn new(
        inner: SharedBackend,
        config: ReliabilityConfig,
        deadlines: Option<RetryDeadlines>,
    ) -> Self {
        Self {
            inner,
            deadlines,
            retry: config.retry.map(RetryPolicy::new),
            breaker: config
                .circuit_breaker
                .map(|config| Arc::new(CircuitBreaker::new(config))),
            limiter: config.backpressure.map(ConcurrencyLimiter::new),
        }
    }

    /// Run `f` through the stack; `budget` is the retry deadline for this op, if any.
    async fn guarded<T, F, Fut>(&self, budget: Option<Duration>, f: F) -> Result<T, BackendError>
    where
        F: Fn() -> Fut,
        Fut: Future<Output = Result<T, BackendError>>,
    {
        // Outermost layer: one permit per logical operation, held across the
        // whole breaker/retry sequence (see the module docs for why). A shed
        // call returns here — before touching breaker state.
        let _permit = match &self.limiter {
            Some(limiter) => Some(limiter.acquire().await?),
            None => None,
        };
        let permit = match &self.breaker {
            Some(cb) => Some(cb.try_acquire()?),
            None => None,
        };
        let result = match &self.retry {
            Some(retry) => retry.execute(budget, f).await,
            None => f().await,
        };
        if let Some(permit) = permit {
            let outcome = match &result {
                Ok(_) => Outcome::Success,
                Err(e) if e.kind.is_retryable() => Outcome::Failure,
                Err(_) => Outcome::Neutral,
            };
            permit.complete(&outcome);
        }
        result
    }
}

#[cfg_attr(not(feature = "unsync"), async_trait)]
#[cfg_attr(feature = "unsync", async_trait(?Send))]
impl Backend for ReliableBackend {
    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, BackendError> {
        self.guarded(self.deadlines.map(|d| d.read), || self.inner.get(key))
            .await
    }

    // Forwarded, not defaulted: the default would call `self.get` and drop
    // the inner backend's freshness, unbounding the L1 backfill.
    async fn get_with_freshness(
        &self,
        key: &str,
    ) -> Result<Option<(Vec<u8>, Freshness)>, BackendError> {
        self.guarded(self.deadlines.map(|d| d.read), || {
            self.inner.get_with_freshness(key)
        })
        .await
    }

    async fn set(
        &self,
        key: &str,
        value: Vec<u8>,
        ttl: Option<Duration>,
    ) -> Result<(), BackendError> {
        // Clone per attempt: the inner call consumes the buffer.
        self.guarded(self.deadlines.map(|d| d.write), || {
            self.inner.set(key, value.clone(), ttl)
        })
        .await
    }

    async fn delete(&self, key: &str) -> Result<bool, BackendError> {
        self.guarded(self.deadlines.map(|d| d.write), || self.inner.delete(key))
            .await
    }

    async fn exists(&self, key: &str) -> Result<bool, BackendError> {
        self.guarded(self.deadlines.map(|d| d.read), || self.inner.exists(key))
            .await
    }

    async fn health(&self) -> Result<HealthStatus, BackendError> {
        self.inner.health().await
    }

    fn as_lockable(&self) -> Option<&dyn LockableBackend> {
        self.inner.as_lockable()
    }
}

/// Wrap `inner` in a [`ReliableBackend`] and re-share it, handing back the
/// breaker (if configured) for the client's state accessor.
pub(crate) fn wrap_reliable(
    inner: SharedBackend,
    config: ReliabilityConfig,
    deadlines: Option<RetryDeadlines>,
) -> (SharedBackend, Option<Arc<CircuitBreaker>>) {
    let reliable = ReliableBackend::new(inner, config, deadlines);
    let breaker = reliable.breaker.clone();
    #[cfg(not(feature = "unsync"))]
    let shared: SharedBackend = std::sync::Arc::new(reliable);
    #[cfg(feature = "unsync")]
    let shared: SharedBackend = std::rc::Rc::new(reliable);
    (shared, breaker)
}

// ── Unit tests ───────────────────────────────────────────────────────────────

#[cfg(test)]
#[allow(clippy::expect_used)] // test-only: failed acquire/probe should panic loudly
mod tests {
    use std::pin::Pin;
    use std::task::{Context, Poll, Waker};

    use super::*;
    use crate::error::BackendErrorKind;

    fn breaker(failure_threshold: u32, open_timeout: Duration) -> CircuitBreaker {
        CircuitBreaker::new(CircuitBreakerConfig {
            failure_threshold,
            success_threshold: 2,
            open_timeout,
            half_open_max_calls: 2,
            rolling_window: Duration::from_secs(60),
        })
    }

    /// Admit a call and immediately report its outcome.
    fn admit_and(cb: &CircuitBreaker, outcome: &Outcome) {
        let permit = cb.try_acquire().expect("breaker admits the call");
        permit.complete(outcome);
    }

    #[test]
    fn breaker_opens_after_threshold_and_fails_fast() {
        let cb = breaker(3, Duration::from_secs(60));
        for _ in 0..3 {
            admit_and(&cb, &Outcome::Failure);
        }
        assert_eq!(cb.state(), CircuitState::Open);
        let err = cb.try_acquire().expect_err("open breaker fails fast");
        assert_eq!(err.kind, BackendErrorKind::CircuitOpen);
        assert!(!err.kind.is_retryable());
    }

    #[test]
    fn breaker_ignores_permanent_errors() {
        let cb = breaker(2, Duration::from_secs(60));
        for _ in 0..10 {
            admit_and(&cb, &Outcome::Neutral);
        }
        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[test]
    fn breaker_half_open_recovers_on_successes() {
        let cb = breaker(1, Duration::from_millis(0));
        admit_and(&cb, &Outcome::Failure);
        // open_timeout of zero → immediately half-open on next inspection
        assert_eq!(cb.state(), CircuitState::HalfOpen);
        for _ in 0..2 {
            admit_and(&cb, &Outcome::Success);
        }
        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[test]
    fn breaker_half_open_reopens_on_failure() {
        let cb = breaker(1, Duration::from_millis(0));
        admit_and(&cb, &Outcome::Failure);
        assert_eq!(cb.state(), CircuitState::HalfOpen);
        admit_and(&cb, &Outcome::Failure);
        // Freshly re-opened with a zero timeout flips half-open again on
        // inspection, so assert via the internal state before inspecting.
        assert!(matches!(cb.lock().state, State::Open { .. }));
    }

    #[test]
    fn breaker_half_open_slot_released_by_neutral_outcome() {
        let cb = breaker(1, Duration::from_millis(0));
        admit_and(&cb, &Outcome::Failure);
        assert_eq!(cb.state(), CircuitState::HalfOpen);
        // Exhaust both probe slots with permanent errors...
        admit_and(&cb, &Outcome::Neutral);
        admit_and(&cb, &Outcome::Neutral);
        // ...and the breaker still admits probes instead of wedging.
        let permit = cb
            .try_acquire()
            .expect("neutral outcomes release their probe slots");
        permit.complete(&Outcome::Neutral);
    }

    #[test]
    fn breaker_half_open_closes_when_success_threshold_exceeds_probe_cap() {
        // success_threshold (3) deliberately exceeds half_open_max_calls (1):
        // with a single in-flight probe slot, the breaker can only ever reach
        // three successes if each non-closing success RELEASES its slot. Before
        // the fix this wedged half-open forever — the slot filled after the
        // first success (which stalled at 1 < 3), so no further probe was
        // admitted and the breaker never re-closed.
        let cb = CircuitBreaker::new(CircuitBreakerConfig {
            failure_threshold: 1,
            success_threshold: 3,
            open_timeout: Duration::from_millis(0),
            half_open_max_calls: 1,
            rolling_window: Duration::from_secs(60),
        });
        admit_and(&cb, &Outcome::Failure);
        assert_eq!(cb.state(), CircuitState::HalfOpen);
        for _ in 0..3 {
            let permit = cb
                .try_acquire()
                .expect("a non-closing success must release its probe slot");
            permit.complete(&Outcome::Success);
        }
        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[test]
    fn breaker_dropped_permit_releases_probe_slot() {
        // A probe future cancelled (caller timeout / select!) or panicked
        // before recording an outcome must not consume its slot forever:
        // exhaust every half-open slot with plain drops and the breaker must
        // still admit probes instead of wedging half-open until restart.
        let cb = breaker(1, Duration::from_millis(0));
        admit_and(&cb, &Outcome::Failure);
        assert_eq!(cb.state(), CircuitState::HalfOpen);
        for _ in 0..2 {
            let permit = cb.try_acquire().expect("half-open admits a probe");
            drop(permit); // cancelled before any outcome
        }
        let permit = cb
            .try_acquire()
            .expect("dropped permits release their probe slots");
        permit.complete(&Outcome::Success);
    }

    #[test]
    fn breaker_closed_permit_drop_does_not_touch_half_open_accounting() {
        // A call admitted while CLOSED holds no probe slot; cancelling it
        // must not free (or corrupt) slots in a half-open window that opened
        // after its admission.
        let cb = breaker(1, Duration::from_millis(0));
        let closed_permit = cb.try_acquire().expect("closed breaker admits calls");
        // Another call's failure opens the breaker, then zero timeout flips
        // it half-open with a fresh probe window.
        admit_and(&cb, &Outcome::Failure);
        assert_eq!(cb.state(), CircuitState::HalfOpen);
        let p1 = cb.try_acquire().expect("probe slot 1");
        let p2 = cb.try_acquire().expect("probe slot 2");
        drop(closed_permit); // must be a no-op: it never took a slot
        assert!(
            cb.try_acquire().is_err(),
            "probe cap must still be enforced after a closed-state permit drops"
        );
        p1.complete(&Outcome::Success);
        p2.complete(&Outcome::Success);
        assert_eq!(cb.state(), CircuitState::Closed);
    }

    #[test]
    fn reliability_default_enables_backpressure_with_python_parity_defaults() {
        let config = ReliabilityConfig::default();
        let bp = config.backpressure.expect("backpressure is on by default");
        assert_eq!(bp.max_concurrent, 100);
        assert_eq!(bp.max_queue, 1000);
        assert_eq!(bp.acquire_timeout, Duration::from_millis(100));
    }

    #[tokio::test]
    async fn limiter_clamps_zero_max_concurrent_to_one() {
        let limiter = ConcurrencyLimiter::new(BackpressureConfig {
            max_concurrent: 0,
            max_queue: 0,
            acquire_timeout: Duration::from_millis(10),
        });
        let permit = limiter
            .acquire()
            .await
            .expect("0 behaves as 1 — one permit exists");
        drop(permit);
    }

    #[tokio::test]
    async fn limiter_clamps_huge_max_concurrent_instead_of_panicking() {
        // usize::MAX is the natural "unbounded" sentinel; Semaphore::new
        // panics above MAX_PERMITS, so the constructor must clamp.
        let limiter = ConcurrencyLimiter::new(BackpressureConfig {
            max_concurrent: usize::MAX,
            max_queue: 0,
            acquire_timeout: Duration::from_millis(10),
        });
        let permit = limiter.acquire().await.expect("clamped limiter admits");
        drop(permit);
    }

    #[tokio::test]
    async fn limiter_sheds_immediately_when_queue_disabled() {
        let limiter = ConcurrencyLimiter::new(BackpressureConfig {
            max_concurrent: 1,
            max_queue: 0,
            acquire_timeout: Duration::from_secs(5),
        });
        let _held = limiter.acquire().await.expect("first permit");
        let start = Instant::now();
        let err = limiter
            .acquire()
            .await
            .expect_err("saturated with no waiting queue");
        assert_eq!(err.kind, BackendErrorKind::Backpressure);
        assert!(!err.kind.is_retryable());
        assert!(
            start.elapsed() < Duration::from_millis(500),
            "queue-full sheds immediately, not after acquire_timeout"
        );
    }

    /// Polls `fut` exactly once, with a waker that does nothing. On a
    /// saturated limiter, a caller that joins the waiting queue is `Pending`
    /// after this poll and a queue-full shed is already `Ready`, so neither
    /// outcome depends on a timer or on how the scheduler runs the test.
    fn poll_once<F: Future>(fut: Pin<&mut F>) -> Poll<F::Output> {
        fut.poll(&mut Context::from_waker(Waker::noop()))
    }

    #[tokio::test]
    async fn limiter_waiting_slot_released_on_cancelled_wait() {
        // A waiter cancelled mid-acquire (caller timeout / select!) must free
        // its queue slot via the QueueSlot drop guard. With max_queue: 1, a
        // leaked slot would shed the next waiter as queue-full on its first
        // poll; parking in the queue instead proves the slot was released.
        let limiter = ConcurrencyLimiter::new(BackpressureConfig {
            max_concurrent: 1,
            max_queue: 1,
            acquire_timeout: Duration::from_millis(100),
        });
        let _held = limiter.acquire().await.expect("first permit");
        let mut cancelled = Box::pin(limiter.acquire());
        assert!(
            poll_once(cancelled.as_mut()).is_pending(),
            "waiter parks in the queue"
        );
        drop(cancelled); // the caller gives up mid-wait

        let mut next = Box::pin(limiter.acquire());
        assert!(
            poll_once(next.as_mut()).is_pending(),
            "next waiter must park in the queue — a queue-full shed means \
             the cancelled waiter leaked its slot"
        );
        let err = next
            .await
            .expect_err("permit never frees, waiter times out");
        assert_eq!(err.kind, BackendErrorKind::Backpressure);
    }

    #[tokio::test]
    async fn limiter_sheds_queue_full_at_nonzero_boundary() {
        // cap 1, queue 1: with the permit held and one waiter parked, a
        // third caller must shed instantly as queue-full — pinning the
        // fetch_add boundary arithmetic at a nonzero max_queue.
        let limiter = ConcurrencyLimiter::new(BackpressureConfig {
            max_concurrent: 1,
            max_queue: 1,
            acquire_timeout: Duration::from_millis(200),
        });
        let _held = limiter.acquire().await.expect("first permit");
        let mut waiter = Box::pin(limiter.acquire());
        assert!(
            poll_once(waiter.as_mut()).is_pending(),
            "waiter parks in the queue"
        );
        // Ready on the first poll is the queue-full shed: every other shed
        // path is only reachable after parking.
        let third = poll_once(Box::pin(limiter.acquire()).as_mut());
        assert!(
            matches!(&third, Poll::Ready(Err(err)) if err.kind == BackendErrorKind::Backpressure),
            "queue of 1 is full, so the third caller sheds on its first poll: {third:?}"
        );
        waiter
            .await
            .expect_err("the parked waiter itself times out");
    }

    #[test]
    fn retry_delay_is_capped_and_jittered() {
        let policy = RetryPolicy::new(RetryConfig {
            max_attempts: 5,
            base_delay: Duration::from_millis(100),
            max_delay: Duration::from_millis(300),
            jitter: true,
        });
        for attempt in 0..10 {
            let d = policy.delay(attempt);
            // cap 300ms × jitter [0.5, 1.5) → strictly under 450ms
            assert!(d < Duration::from_millis(450), "attempt {attempt}: {d:?}");
        }
        let no_jitter = RetryPolicy::new(RetryConfig {
            jitter: false,
            ..RetryConfig::default()
        });
        assert_eq!(no_jitter.delay(0), Duration::from_millis(100));
        assert_eq!(no_jitter.delay(1), Duration::from_millis(200));
        assert_eq!(no_jitter.delay(20), Duration::from_secs(5));
    }

    #[tokio::test]
    async fn a_quota_deny_is_not_retried() {
        let calls = AtomicU64::new(0);
        let policy = RetryPolicy::new(RetryConfig::default());

        let err = policy
            .execute(None, || async {
                calls.fetch_add(1, Ordering::SeqCst);
                Err::<(), _>(BackendError::from_http_status(429, b"").with_quota_denied())
            })
            .await
            .expect_err("denied");

        assert!(err.is_quota_denied());
        assert_eq!(calls.load(Ordering::SeqCst), 1);
    }
}

// ── Retry deadline (CacheKit::io) ────────────────────────────────────────────

/// The deadline is crate-private, so these run in-crate rather than in
/// `tests/reliability_tests.rs`.
#[cfg(all(test, feature = "cachekitio"))]
#[allow(clippy::expect_used)] // test-only
mod deadline_tests {
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::sync::Arc;
    use std::time::Duration;

    use async_trait::async_trait;

    use super::{CircuitState, ReliabilityConfig, RetryConfig, CACHEKITIO_DEADLINES};
    use crate::backend::{Backend, HealthStatus};
    use crate::client::SharedBackend;
    use crate::error::{BackendError, BackendErrorKind};
    use crate::{CacheKit, CachekitError};

    /// Every data op sleeps `latency` on the tokio clock, then fails with `kind`;
    /// `None` never completes. Counts calls.
    #[derive(Debug, Clone)]
    struct SlowBackend {
        latency: Option<Duration>,
        kind: BackendErrorKind,
        calls: Arc<AtomicU32>,
    }

    impl SlowBackend {
        fn new_with_handle(
            latency: Option<Duration>,
            kind: BackendErrorKind,
        ) -> (SharedBackend, Self) {
            let backend = Self {
                latency,
                kind,
                calls: Arc::new(AtomicU32::new(0)),
            };
            let handle = backend.clone();
            #[cfg(not(feature = "unsync"))]
            let shared: SharedBackend = Arc::new(backend);
            #[cfg(feature = "unsync")]
            let shared: SharedBackend = std::rc::Rc::new(backend);
            (shared, handle)
        }

        fn calls(&self) -> u32 {
            self.calls.load(Ordering::SeqCst)
        }

        async fn op<T>(&self) -> Result<T, BackendError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            match self.latency {
                Some(latency) => tokio::time::sleep(latency).await,
                None => std::future::pending().await,
            }
            Err(BackendError {
                kind: self.kind.clone(),
                message: "scripted failure".to_owned(),
                source: None,
            })
        }
    }

    #[cfg_attr(not(feature = "unsync"), async_trait)]
    #[cfg_attr(feature = "unsync", async_trait(?Send))]
    impl Backend for SlowBackend {
        async fn get(&self, _key: &str) -> Result<Option<Vec<u8>>, BackendError> {
            self.op().await
        }

        async fn set(
            &self,
            _key: &str,
            _value: Vec<u8>,
            _ttl: Option<Duration>,
        ) -> Result<(), BackendError> {
            self.op().await
        }

        async fn delete(&self, _key: &str) -> Result<bool, BackendError> {
            self.op().await
        }

        async fn exists(&self, _key: &str) -> Result<bool, BackendError> {
            self.op().await
        }

        async fn health(&self) -> Result<HealthStatus, BackendError> {
            Ok(HealthStatus {
                is_healthy: true,
                latency_ms: 0.0,
                backend_type: "slow".to_owned(),
                details: HashMap::new(),
            })
        }
    }

    /// The `CacheKit::io` stack (default layers plus the cachekit.io deadlines)
    /// around `backend`, with jitter off so virtual time is exact.
    fn default_stack(backend: SharedBackend) -> CacheKit {
        let mut builder = CacheKit::builder()
            .backend(backend)
            .reliability(ReliabilityConfig {
                retry: Some(RetryConfig {
                    jitter: false,
                    ..RetryConfig::default()
                }),
                ..ReliabilityConfig::default()
            })
            .no_l1();
        builder.retry_deadlines = Some(CACHEKITIO_DEADLINES);
        builder.build().expect("client builds")
    }

    fn backend_kind(err: &CachekitError) -> Option<&BackendErrorKind> {
        match err {
            CachekitError::Backend(b) => Some(&b.kind),
            _ => None,
        }
    }

    /// Run `op` and return its result with the virtual time it took.
    async fn timed<T>(op: impl std::future::Future<Output = T>) -> (T, Duration) {
        let start = tokio::time::Instant::now();
        let out = op.await;
        (out, start.elapsed())
    }

    #[tokio::test(start_paused = true)]
    async fn a_stalled_get_fails_at_the_5_s_read_deadline() {
        let (backend, handle) = SlowBackend::new_with_handle(None, BackendErrorKind::Timeout);
        let client = default_stack(backend);

        let (result, elapsed) = timed(client.get::<u32>("k")).await;

        let err = result.expect_err("stalled");
        assert_eq!(backend_kind(&err), Some(&BackendErrorKind::Timeout));
        assert_eq!(elapsed, Duration::from_secs(5));
        assert_eq!(handle.calls(), 1);
    }

    #[tokio::test(start_paused = true)]
    async fn a_stalled_set_fails_at_the_10_s_write_deadline() {
        let (backend, handle) = SlowBackend::new_with_handle(None, BackendErrorKind::Timeout);
        let client = default_stack(backend);

        let (result, elapsed) = timed(client.set("k", &7u32)).await;

        let err = result.expect_err("stalled");
        assert_eq!(backend_kind(&err), Some(&BackendErrorKind::Timeout));
        assert_eq!(elapsed, Duration::from_secs(10));
        assert_eq!(handle.calls(), 1);
    }

    /// The old cachekit.io shape: each attempt times out after 30 s. Without the
    /// deadline this took 3 x 30 s + 0.3 s backoff = 90.3 s.
    #[tokio::test(start_paused = true)]
    async fn a_30_s_attempt_timeout_is_cut_to_the_deadline() {
        let (backend, handle) =
            SlowBackend::new_with_handle(Some(Duration::from_secs(30)), BackendErrorKind::Timeout);
        let client = default_stack(backend);

        let (result, elapsed) = timed(client.get::<u32>("k")).await;

        result.expect_err("timed out");
        assert_eq!(elapsed, Duration::from_secs(5));
        assert_eq!(handle.calls(), 1);
    }

    #[tokio::test(start_paused = true)]
    async fn fast_failures_still_get_every_attempt() {
        let (backend, handle) =
            SlowBackend::new_with_handle(Some(Duration::ZERO), BackendErrorKind::Transient);
        let client = default_stack(backend);

        let (result, elapsed) = timed(client.get::<u32>("k")).await;

        let err = result.expect_err("always fails");
        assert_eq!(backend_kind(&err), Some(&BackendErrorKind::Transient));
        assert_eq!(handle.calls(), 3);
        assert_eq!(elapsed, Duration::from_millis(300), "100 + 200 ms backoff");
    }

    #[tokio::test(start_paused = true)]
    async fn no_backoff_starts_past_the_deadline() {
        // 4.95 s attempt + 100 ms backoff would start attempt 2 past 5 s: return
        // the attempt's own error rather than a deadline timeout.
        let (backend, handle) = SlowBackend::new_with_handle(
            Some(Duration::from_millis(4950)),
            BackendErrorKind::Transient,
        );
        let client = default_stack(backend);

        let (result, elapsed) = timed(client.get::<u32>("k")).await;

        let err = result.expect_err("fails");
        assert_eq!(backend_kind(&err), Some(&BackendErrorKind::Transient));
        assert_eq!(handle.calls(), 1);
        assert_eq!(elapsed, Duration::from_millis(4950));
    }

    #[tokio::test(start_paused = true)]
    async fn serial_stalled_gets_open_the_breaker_inside_its_window() {
        let (backend, handle) = SlowBackend::new_with_handle(None, BackendErrorKind::Timeout);
        let client = default_stack(backend);

        // Five 5 s failures fit the 60 s rolling window; at 90 s each they never did.
        for _ in 0..5 {
            client.get::<u32>("k").await.expect_err("stalled");
        }
        assert_eq!(client.circuit_state(), Some(CircuitState::Open));

        let calls = handle.calls();
        let err = client.get::<u32>("k").await.expect_err("open");
        assert_eq!(backend_kind(&err), Some(&BackendErrorKind::CircuitOpen));
        assert_eq!(handle.calls(), calls, "an open breaker sends nothing");
    }
}