autumn-web 0.9.0

// autumn-determinism-gate: production code in this module must read time and
// mint identifiers through the framework's injected seams (ClockSource /
// Entropy), never `Instant::now()` / `Utc::now()` / `SystemTime::now()` /
// `Uuid::new_v4()` directly. See CONTRIBUTING.md "Determinism seam gate"
// (issue #1797). Justify exceptions with
// #[allow(clippy::disallowed_methods, reason = "…")] at the narrowest scope.
#![cfg_attr(not(test), deny(clippy::disallowed_methods))]
// autumn-panic-gate: request-path module — production code path must be panic-free.
// See CONTRIBUTING.md "Request-path panic gate". Justify exceptions with
// #[allow(clippy::<lint>, reason = "…")] at the narrowest scope.
#![cfg_attr(
    not(test),
    deny(
        clippy::unwrap_used,
        clippy::expect_used,
        clippy::panic,
        clippy::unreachable,
        clippy::todo,
        clippy::unimplemented,
        clippy::indexing_slicing,
        clippy::string_slice,
        clippy::arithmetic_side_effects,
    )
)]

use bytes::Bytes;
use futures::StreamExt as FuturesStreamExt;

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, PoisonError, RwLock};
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

use axum::body::Body;
use axum::http::{HeaderMap, Method, Request, Response, StatusCode, request::Parts};
use axum::response::IntoResponse;
use sha2::Digest as _;
use tower::{Layer, Service};

static IDEMPOTENCY_KEY_HEADER: &str = "idempotency-key";
static X_IDEMPOTENT_REPLAYED: &str = "x-idempotent-replayed";

/// Maximum response body size stored in the idempotency cache. Responses
/// larger than this are returned to the client as-is but not cached, so a
/// subsequent retry with the same key will re-execute the handler.
const MAX_CACHEABLE_RESPONSE_BODY: usize = 10 * 1024 * 1024; // 10 MiB

/// Fallback request body read limit used when the upload middleware extension
/// is absent (e.g. when `IdempotencyLayer` is used directly without the full
/// framework stack). Matches the framework default for
/// `security.upload.max_request_size_bytes`.
const DEFAULT_REQUEST_BODY_LIMIT: usize = 32 * 1024 * 1024; // 32 MiB

/// Default in-flight lock TTL. It applies only after a crash, a cancelled
/// request, or a failed record write. It is separate from the response TTL.
pub const DEFAULT_IN_FLIGHT_TTL: Duration = Duration::from_secs(60);

const fn is_mutating_method(method: &Method) -> bool {
    matches!(
        *method,
        Method::POST | Method::PUT | Method::PATCH | Method::DELETE
    )
}

fn compute_body_hash(bytes: &[u8], content_type: Option<&[u8]>) -> Vec<u8> {
    let mut hasher = sha2::Sha256::new();
    hasher.update(b"content-type:");
    if let Some(content_type) = content_type {
        hasher.update(content_type);
    }
    hasher.update(b"\nbody:");
    hasher.update(bytes);
    hasher.finalize().to_vec()
}

fn principal_scope_digest(session_id: Option<&str>) -> String {
    let mut hasher = sha2::Sha256::new();
    hasher.update(b"authorization:");
    hasher.update(b"\nsession:");
    if let Some(session_id) = session_id {
        hasher.update(session_id.as_bytes());
    }
    hex::encode(hasher.finalize())
}

fn push_storage_key_component(hasher: &mut sha2::Sha256, label: &str, value: &[u8]) {
    hasher.update(label.as_bytes());
    hasher.update(b":");
    hasher.update((value.len() as u64).to_be_bytes());
    hasher.update(b":");
    hasher.update(value);
    hasher.update(b";");
}

/// The tenant Autumn's own tenancy middleware resolved for the in-flight
/// request, if any.
///
/// Read from the [`CURRENT_TENANT`](crate::tenancy::CURRENT_TENANT) task-local
/// rather than from a request header, which is what makes it safe to key on:
/// the value is the framework's *resolution* under `[tenancy] source`
/// (`header`, `subdomain`, `session`, `jwt`), not a string the client can
/// restate at will. A legitimate retry from the same tenant therefore
/// reproduces the same component and still finds its cached response, while a
/// request that resolved to a different tenant lands in a different slot
/// instead of replaying that tenant's stored mutation.
///
/// `None` — tenancy disabled, or a `[tenancy] public_paths` route the
/// middleware exempts before it scopes anything — pushes no component at all,
/// so every storage key an app without tenancy computes is byte-identical to
/// the one it computed before this existed.
fn current_tenant_scope() -> Option<String> {
    crate::tenancy::CURRENT_TENANT
        .try_with(std::clone::Clone::clone)
        .ok()
        .flatten()
}

/// Namespace the cache key by method, path, the resolved tenant, a stable
/// principal digest, and the client-supplied idempotency key.
///
/// Namespacing by method+path prevents cross-endpoint cache collisions (P2).
/// Namespacing by session scope prevents cross-principal collisions (P1) for
/// cookie-backed authenticated sessions, and namespacing by the
/// framework-resolved tenant prevents them across tenants — for `header`,
/// `subdomain` and `jwt` tenancy the two requests are otherwise
/// indistinguishable to this key (same method, same target, and for a
/// token-authenticated API the same empty session scope), so tenant B replaying
/// tenant A's key would be served tenant A's stored response with the handler —
/// and every `tenant_scoped` repository predicate inside it — never running.
///
/// Bearer clients are scoped too (see [`BearerScope`]). A token client has no
/// cookie-backed session, so without this every token client of one route
/// shared a single slot: client B sending client A's key and body was served
/// A's stored response, and the handler — with whatever principal check it
/// does — never ran. The verified principal from
/// [`RequireApiToken`](crate::auth::RequireApiToken) is preferred, because it
/// survives token rotation; a request with no verified principal and no
/// session falls back to a digest of its raw `Authorization` header, which
/// covers routes that check the credential inside the handler.
///
/// The configured tenant *header* is intentionally excluded: the tenant is
/// taken from the middleware's task-local resolution rather than from the
/// wire. Opaque route layers that resolve their own tenants or policy state
/// must still use the fail-closed replay path instead of storage-key
/// partitioning. Each component is length-delimited inside a SHA-256 digest so
/// raw `:` bytes in paths or client-controlled keys cannot synthesize another
/// storage key.
#[derive(Clone)]
struct StorageKeyContext {
    idempotency_key: String,
    method: Method,
    target: String,
    /// Captured once, on the request path, while the tenancy middleware's
    /// task-local scope is still established — the alias keys computed later
    /// (see [`DeferredIdempotencyCommit::add_session_alias`]) must land in the
    /// same tenant's namespace as the primary key.
    tenant: Option<String>,
    /// Captured once from the original request, so alias keys computed later
    /// carry the same bearer scope as the primary key.
    bearer: Option<BearerScope>,
}

/// The bearer credential an idempotency storage key is scoped by.
#[derive(Clone, Debug, PartialEq, Eq)]
enum BearerScope {
    /// The principal [`RequireApiToken`](crate::auth::RequireApiToken)
    /// verified before this layer ran. A client cannot forge it, and it stays
    /// the same when the client rotates its token, so a retry with a fresh
    /// token still replays. Scopes the key whether or not a session exists.
    Verified(String),
    /// A SHA-256 digest of the raw `Authorization` header, for a request with
    /// no verified principal. Scopes the key only when the request has no
    /// cookie-backed session, so session clients keep the keys they had.
    ///
    /// A client that changes its header between attempts — a refreshed token,
    /// or none at all — misses and runs the handler again. That is the safe
    /// direction: the alternative is serving one client's stored response to
    /// another.
    Header(String),
}

/// The bearer scope for a request, read before the handler runs.
fn bearer_scope_for_parts(parts: &axum::http::request::Parts) -> Option<BearerScope> {
    if let Some(principal) = parts.extensions.get::<crate::auth::ApiTokenPrincipal>() {
        return Some(BearerScope::Verified(principal.0.clone()));
    }
    let header = parts.headers.get(axum::http::header::AUTHORIZATION)?;
    Some(BearerScope::Header(hex::encode(sha2::Sha256::digest(
        header.as_bytes(),
    ))))
}

impl StorageKeyContext {
    fn from_parts(idempotency_key: String, parts: &axum::http::request::Parts) -> Self {
        let target = parts
            .uri
            .path_and_query()
            .map_or_else(|| parts.uri.path().to_owned(), |pq| pq.as_str().to_owned());
        Self {
            idempotency_key,
            method: parts.method.clone(),
            target,
            tenant: current_tenant_scope(),
            bearer: bearer_scope_for_parts(parts),
        }
    }

    fn storage_key(&self, session_id: Option<&str>, tenant_override: Option<&str>) -> String {
        build_storage_key(
            &self.idempotency_key,
            self.method.as_str(),
            &self.target,
            session_id,
            tenant_override.or(self.tenant.as_deref()),
            self.bearer.as_ref(),
        )
    }
}

fn build_storage_key(
    idempotency_key: &str,
    method: &str,
    target: &str,
    session_id: Option<&str>,
    tenant: Option<&str>,
    bearer: Option<&BearerScope>,
) -> String {
    let principal = principal_scope_digest(session_id);
    let mut hasher = sha2::Sha256::new();
    push_storage_key_component(&mut hasher, "method", method.as_bytes());
    push_storage_key_component(&mut hasher, "target", target.as_bytes());
    push_storage_key_component(&mut hasher, "scope-header-count", b"0");
    push_storage_key_component(&mut hasher, "principal", principal.as_bytes());
    // Pushed only when the request carried a bearer credential, so a
    // session-only app keeps the exact keys it had before, like the tenant
    // component below.
    match bearer {
        Some(BearerScope::Verified(principal_id)) => {
            push_storage_key_component(&mut hasher, "bearer-principal", principal_id.as_bytes());
        }
        Some(BearerScope::Header(digest)) if session_id.is_none() => {
            push_storage_key_component(&mut hasher, "authorization", digest.as_bytes());
        }
        Some(BearerScope::Header(_)) | None => {}
    }
    // Pushed only when a tenant was resolved, so an app that does not use
    // tenancy keeps the exact keys it had before — no cache-wide miss, and no
    // duplicate execution of a mutation retried across an upgrade.
    if let Some(tenant) = tenant {
        push_storage_key_component(&mut hasher, "tenant", tenant.as_bytes());
    }
    push_storage_key_component(&mut hasher, "idempotency-key", idempotency_key.as_bytes());
    format!("v2:{}", hex::encode(hasher.finalize()))
}

async fn storage_session_id_for_parts(parts: &axum::http::request::Parts) -> Option<String> {
    let session_scope = parts
        .extensions
        .get::<IdempotencySessionScope>()
        .and_then(|scope| scope.session_id.as_deref().map(str::to_owned));
    let session = parts.extensions.get::<crate::session::Session>().cloned();
    if session_scope.is_some() {
        session_scope
    } else if let Some(session) = session
        && session.is_cookie_backed().await
    {
        Some(session.id().await)
    } else {
        None
    }
}

fn stale_cookie_session_id_for_parts(parts: &axum::http::request::Parts) -> Option<String> {
    parts
        .extensions
        .get::<IdempotencySessionScope>()
        .and_then(|scope| scope.stale_cookie_session_id.as_deref().map(str::to_owned))
}

fn extract_replay_headers(headers: &HeaderMap) -> Vec<(String, Vec<u8>)> {
    extract_replay_headers_with_policy(headers, false)
}

fn extract_finalized_session_replay_headers(headers: &HeaderMap) -> Vec<(String, Vec<u8>)> {
    extract_replay_headers_with_policy(headers, true)
}

fn extract_replay_headers_with_policy(
    headers: &HeaderMap,
    include_set_cookie: bool,
) -> Vec<(String, Vec<u8>)> {
    // Headers that must not be cached or replayed.
    // `set-cookie` is replayed only for finalized session-mutating responses
    // so lost successful mutations can deliver the session state they created.
    const SKIP: &[&str] = &[
        "connection",
        "transfer-encoding",
        "keep-alive",
        "upgrade",
        "proxy-authenticate",
        "proxy-authorization",
        "te",
        "trailer",
        "x-idempotent-replayed",
    ];
    headers
        .iter()
        .filter(|(name, _)| {
            !SKIP.contains(&name.as_str()) && (include_set_cookie || name.as_str() != "set-cookie")
        })
        .map(|(name, value)| (name.to_string(), value.as_bytes().to_vec()))
        .collect()
}

// ── Public types ──────────────────────────────────────────────────────────────

/// Stored response associated with an idempotency key.
#[derive(Clone)]
pub struct IdempotencyRecord {
    pub status: u16,
    pub headers: Vec<(String, Vec<u8>)>,
    pub body: Vec<u8>,
    pub metadata: Vec<(String, Vec<u8>)>,
}

const FINALIZED_SESSION_SCOPE_METADATA: &str = "__autumn.idempotency.finalized-session-scope";
const FINALIZED_SESSION_OLD_SCOPE: &[u8] = b"old-session-scope";
const FINALIZED_SESSION_CURRENT_SCOPE: &[u8] = b"current-session-scope";

fn finalized_session_record(
    mut record: IdempotencyRecord,
    scope: &'static [u8],
) -> IdempotencyRecord {
    record
        .metadata
        .retain(|(name, _)| name != FINALIZED_SESSION_SCOPE_METADATA);
    record
        .metadata
        .push((FINALIZED_SESSION_SCOPE_METADATA.to_owned(), scope.to_vec()));
    record
}

#[doc(hidden)]
#[derive(Clone, Copy, Debug)]
pub struct IdempotencyCacheCommittedErrorResponse;

#[derive(Clone, Debug)]
pub(crate) struct IdempotencySessionScope {
    session_id: Option<String>,
    stale_cookie_session_id: Option<String>,
}

impl IdempotencySessionScope {
    #[must_use]
    pub(crate) const fn new(
        session_id: Option<String>,
        stale_cookie_session_id: Option<String>,
    ) -> Self {
        Self {
            session_id,
            stale_cookie_session_id,
        }
    }
}

#[doc(hidden)]
#[derive(Clone, Debug, Default)]
pub struct IdempotencyReplayMetadata {
    entries: Vec<(String, Vec<u8>)>,
}

impl IdempotencyReplayMetadata {
    #[must_use]
    pub const fn new(entries: Vec<(String, Vec<u8>)>) -> Self {
        Self { entries }
    }

    fn into_entries(self) -> Vec<(String, Vec<u8>)> {
        self.entries
    }
}

/// Request-scoped idempotency metadata made available to inner handlers.
///
/// The raw `Idempotency-Key` header is available via [`Self::key`]. The
/// scoped key is the framework's collision-safe, principal-scoped storage key
/// and is the safer value to reuse for durable side-effect deduplication.
#[derive(Clone, Debug)]
pub struct IdempotencyContext {
    key: String,
    scoped_key: String,
    mutation_sequence: Arc<AtomicU64>,
}

impl IdempotencyContext {
    #[must_use]
    pub(crate) fn new(key: String, scoped_key: String) -> Self {
        Self {
            key,
            scoped_key,
            mutation_sequence: Arc::new(AtomicU64::new(0)),
        }
    }

    #[must_use]
    pub fn key(&self) -> &str {
        &self.key
    }

    #[must_use]
    pub fn scoped_key(&self) -> &str {
        &self.scoped_key
    }

    /// Return the next request-local mutation discriminator.
    ///
    /// Generated repository code uses this to distinguish multiple durable
    /// side effects produced by one idempotent request while keeping the same
    /// mutation slot stable across duplicate request attempts.
    #[must_use]
    pub fn next_mutation_discriminator(&self) -> String {
        self.mutation_sequence
            .fetch_add(1, Ordering::Relaxed)
            .to_string()
    }
}

impl PartialEq for IdempotencyContext {
    fn eq(&self, other: &Self) -> bool {
        self.key == other.key && self.scoped_key == other.scoped_key
    }
}

impl Eq for IdempotencyContext {}

/// Cache entry wrapping a record with expiry and request body fingerprint.
#[derive(Clone)]
pub struct IdempotencyEntry {
    pub record: IdempotencyRecord,
    pub body_hash: Vec<u8>,
    pub expires_at: Instant,
}

// ── Store trait ───────────────────────────────────────────────────────────────

/// Error returned when an idempotency backend fails to persist a successful
/// mutation response.
#[derive(Debug, Clone, thiserror::Error)]
#[error("{message}")]
pub struct IdempotencyStoreError {
    message: String,
}

impl IdempotencyStoreError {
    #[must_use]
    pub fn backend(message: impl Into<String>) -> Self {
        Self {
            message: message.into(),
        }
    }
}

/// Future returned by every [`IdempotencyStore`] operation.
pub type IdempotencyFuture<'a, T> =
    Pin<Box<dyn Future<Output = Result<T, IdempotencyStoreError>> + Send + 'a>>;

/// Pluggable storage backend for idempotency entries.
///
/// All operations are async and return [`IdempotencyFuture`]. Do not block a
/// runtime thread in an implementation: the middleware also runs on a
/// current-thread runtime.
///
/// Every error fails closed. A `get` or `set` error gives `503`. A `try_lock`
/// or `renew_lock` error gives `409`. An `unlock` error is logged; the lock
/// then expires by its TTL.
pub trait IdempotencyStore: std::any::Any + Send + Sync {
    /// Return the entry for `key`, or `None` if there is none or it expired.
    fn get<'a>(&'a self, key: &'a str) -> IdempotencyFuture<'a, Option<IdempotencyEntry>>;

    /// Store the response for `key` for `ttl`, as the lock `owner`.
    ///
    /// Do nothing while another owner holds a live lock on `key`, or has
    /// stored a response for it that has not expired: an attempt that
    /// outlived its lock must not replace the newer request's response. An
    /// owner may rewrite its own response.
    ///
    /// Returns `true` when the response was stored, `false` when another
    /// owner fenced it out.
    fn set<'a>(
        &'a self,
        key: &'a str,
        owner: &'a str,
        record: IdempotencyRecord,
        body_hash: Vec<u8>,
        ttl: Duration,
    ) -> IdempotencyFuture<'a, bool>;

    /// Acquire the in-flight lock for `key` as `owner` for `lock_ttl`.
    ///
    /// Returns `true` when acquired, `false` when another owner holds it.
    fn try_lock<'a>(
        &'a self,
        key: &'a str,
        owner: &'a str,
        lock_ttl: Duration,
    ) -> IdempotencyFuture<'a, bool>;

    /// Release the in-flight lock for `key` only if `owner` still holds it.
    fn unlock<'a>(&'a self, key: &'a str, owner: &'a str) -> IdempotencyFuture<'a, ()>;

    /// Run `owner`'s in-flight lock on `key` for `lock_ttl` from now, only if
    /// `owner` still holds it: `true` when renewed, `false` when another
    /// owner took the key or nobody holds it.
    ///
    /// The middleware and submit tokens call this when the work between
    /// taking the lock and running the handler used more than a tenth of the
    /// TTL, so the handler starts with a full lock. The default returns
    /// `false`: a store that does not implement it answers such a request
    /// `409` instead of running the handler on a lock that may have lapsed.
    fn renew_lock<'a>(
        &'a self,
        key: &'a str,
        owner: &'a str,
        lock_ttl: Duration,
    ) -> IdempotencyFuture<'a, bool> {
        let _ = (key, owner, lock_ttl);
        ready(false)
    }

    /// The response retention for [`IdempotencyLayer::new`]. Default: 24 hours.
    fn default_ttl(&self) -> Duration {
        Duration::from_secs(86_400)
    }
}

/// Wrap an already-known value as an [`IdempotencyFuture`].
fn ready<'a, T: Send + 'a>(value: T) -> IdempotencyFuture<'a, T> {
    Box::pin(std::future::ready(Ok(value)))
}

// ── Memory store ──────────────────────────────────────────────────────────────

/// In-memory idempotency store backed by a `RwLock<HashMap>`.
///
/// Evicts expired entries lazily on `get`. In-flight markers remain held until
/// `unlock` or until their configured in-flight lock TTL expires.
///
/// Suitable for single-process deployments and integration tests. For
/// multi-replica deployments configure `backend = "redis"` in `autumn.toml`.
pub struct MemoryIdempotencyStore {
    entries: RwLock<HashMap<String, MemoryEntry>>,
    in_flight: RwLock<HashMap<String, MemoryInFlightLock>>,
    /// Counts `set` calls to trigger periodic expired-entry eviction.
    write_count: AtomicU64,
    default_ttl: Duration,
}

struct MemoryInFlightLock {
    owner: String,
    expires_at: Instant,
}

/// A stored entry and the lock owner that wrote it.
struct MemoryEntry {
    entry: IdempotencyEntry,
    owner: String,
}

/// Compute an expiry `Instant` for `ttl`, saturating instead of panicking on
/// overflow.
///
/// `Instant::now() + ttl` panics when the sum is not representable by the
/// platform clock — a pathological `ttl` such as `Duration::MAX` or
/// `Duration::from_secs(u64::MAX)` (which is entirely attacker-influenceable
/// via configured TTLs) triggers this. Instead of panicking we clamp the
/// deadline to ~10 years out (far enough that the entry is effectively
/// non-expiring). See [`crate::time_math::saturating_deadline`], which the
/// job and job-tracking modules share.
fn saturating_deadline(ttl: Duration) -> Instant {
    crate::time_math::saturating_deadline(crate::time::ambient_instant(), ttl)
}

impl MemoryIdempotencyStore {
    #[must_use]
    pub fn new(default_ttl: Duration) -> Self {
        Self {
            entries: RwLock::new(HashMap::new()),
            in_flight: RwLock::new(HashMap::new()),
            write_count: AtomicU64::new(0),
            default_ttl,
        }
    }
}

impl MemoryIdempotencyStore {
    fn get_now(&self, key: &str) -> Option<IdempotencyEntry> {
        // Release the read lock immediately after cloning.
        let entry = self
            .entries
            .read()
            .unwrap_or_else(PoisonError::into_inner)
            .get(key)
            .map(|stored| stored.entry.clone());
        entry.filter(|e| e.expires_at > crate::time::ambient_instant())
    }

    fn set_now(
        &self,
        key: &str,
        owner: &str,
        record: IdempotencyRecord,
        body_hash: Vec<u8>,
        ttl: Duration,
    ) -> bool {
        // Write nothing when another owner holds a live lock, or stored an
        // entry that has not expired: a request that outlived its lock never
        // replaces the newer response. Keep the guard until the entry is in,
        // so no `try_lock` can take the key in between.
        let now = crate::time::ambient_instant();
        let in_flight = self
            .in_flight
            .read()
            .unwrap_or_else(PoisonError::into_inner);
        let held_by_another = in_flight
            .get(key)
            .is_some_and(|lock| lock.owner != owner && lock.expires_at > now);
        if held_by_another {
            return false;
        }
        let mut entries = self.entries.write().unwrap_or_else(PoisonError::into_inner);
        let stored_by_another = entries
            .get(key)
            .is_some_and(|stored| stored.owner != owner && stored.entry.expires_at > now);
        if stored_by_another {
            return false;
        }
        let entry = IdempotencyEntry {
            record,
            body_hash,
            expires_at: saturating_deadline(ttl),
        };
        entries.insert(
            key.to_owned(),
            MemoryEntry {
                entry,
                owner: owner.to_owned(),
            },
        );
        drop(in_flight);
        // Periodically evict expired entries to bound memory growth for
        // long-running processes. O(N) scan is amortised over every 128 writes.
        let n = self.write_count.fetch_add(1, Ordering::Relaxed);
        if n.is_multiple_of(128) {
            entries.retain(|_, v| v.entry.expires_at > now);
        }
        true
    }

    fn try_lock_now(&self, key: &str, owner: &str, lock_ttl: Duration) -> bool {
        let now = crate::time::ambient_instant();
        let mut in_flight = self
            .in_flight
            .write()
            .unwrap_or_else(PoisonError::into_inner);
        // Check only the requested key's active in-flight marker.
        if let Some(lock) = in_flight.get(key)
            && lock.expires_at > now
        {
            return false; // still in flight
        }
        // Not locked: acquire until the handler finishes, unlocks, or the
        // safety TTL expires after cancellation.
        let ttl = if lock_ttl.is_zero() {
            Duration::from_secs(1)
        } else {
            lock_ttl
        };
        in_flight.insert(
            key.to_owned(),
            MemoryInFlightLock {
                owner: owner.to_owned(),
                expires_at: saturating_deadline(ttl),
            },
        );
        true
    }

    fn renew_lock_now(&self, key: &str, owner: &str, lock_ttl: Duration) -> bool {
        let ttl = if lock_ttl.is_zero() {
            Duration::from_secs(1)
        } else {
            lock_ttl
        };
        let mut in_flight = self
            .in_flight
            .write()
            .unwrap_or_else(PoisonError::into_inner);
        let renewed = match in_flight.get_mut(key) {
            Some(lock) if lock.owner == owner => {
                lock.expires_at = saturating_deadline(ttl);
                true
            }
            _ => false,
        };
        drop(in_flight);
        renewed
    }

    fn unlock_now(&self, key: &str, owner: &str) {
        let mut in_flight = self
            .in_flight
            .write()
            .unwrap_or_else(PoisonError::into_inner);
        if in_flight
            .get(key)
            .is_some_and(|lock| lock.owner.as_str() == owner)
        {
            in_flight.remove(key);
        }
    }
}

impl IdempotencyStore for MemoryIdempotencyStore {
    fn get<'a>(&'a self, key: &'a str) -> IdempotencyFuture<'a, Option<IdempotencyEntry>> {
        ready(self.get_now(key))
    }

    fn set<'a>(
        &'a self,
        key: &'a str,
        owner: &'a str,
        record: IdempotencyRecord,
        body_hash: Vec<u8>,
        ttl: Duration,
    ) -> IdempotencyFuture<'a, bool> {
        ready(self.set_now(key, owner, record, body_hash, ttl))
    }

    fn try_lock<'a>(
        &'a self,
        key: &'a str,
        owner: &'a str,
        lock_ttl: Duration,
    ) -> IdempotencyFuture<'a, bool> {
        ready(self.try_lock_now(key, owner, lock_ttl))
    }

    fn unlock<'a>(&'a self, key: &'a str, owner: &'a str) -> IdempotencyFuture<'a, ()> {
        self.unlock_now(key, owner);
        ready(())
    }

    fn renew_lock<'a>(
        &'a self,
        key: &'a str,
        owner: &'a str,
        lock_ttl: Duration,
    ) -> IdempotencyFuture<'a, bool> {
        ready(self.renew_lock_now(key, owner, lock_ttl))
    }

    fn default_ttl(&self) -> Duration {
        self.default_ttl
    }
}

// ── Shared encoding ───────────────────────────────────────────────────────────

/// Encoded entry. The Redis and database stores keep entries in this form.
#[cfg(any(feature = "redis", feature = "db"))]
#[derive(serde::Serialize, serde::Deserialize)]
struct StoredEntry {
    status: u16,
    headers: Vec<(String, Vec<u8>)>,
    body: Vec<u8>,
    #[serde(default)]
    metadata: Vec<(String, Vec<u8>)>,
    body_hash: Vec<u8>,
}

#[cfg(any(feature = "redis", feature = "db"))]
impl StoredEntry {
    fn encode(
        record: IdempotencyRecord,
        body_hash: Vec<u8>,
    ) -> Result<Vec<u8>, IdempotencyStoreError> {
        let entry = Self {
            status: record.status,
            headers: record.headers,
            body: record.body,
            metadata: record.metadata,
            body_hash,
        };
        serde_json::to_vec(&entry).map_err(|e| {
            IdempotencyStoreError::backend(format!("failed to serialize idempotency entry: {e}"))
        })
    }

    /// Decode an entry. The backend owns the expiry, so the in-process expiry
    /// is set 24 h out and never fires first.
    fn decode(bytes: &[u8]) -> Result<IdempotencyEntry, IdempotencyStoreError> {
        let entry: Self = serde_json::from_slice(bytes).map_err(|e| {
            IdempotencyStoreError::backend(format!("failed to deserialize idempotency entry: {e}"))
        })?;
        Ok(IdempotencyEntry {
            record: IdempotencyRecord {
                status: entry.status,
                headers: entry.headers,
                body: entry.body,
                metadata: entry.metadata,
            },
            body_hash: entry.body_hash,
            expires_at: saturating_deadline(Duration::from_secs(86_400)),
        })
    }
}

// ── Redis store ───────────────────────────────────────────────────────────────

#[cfg(feature = "redis")]
mod redis_store {
    use super::{
        IdempotencyEntry, IdempotencyFuture, IdempotencyRecord, IdempotencyStore,
        IdempotencyStoreError, StoredEntry,
    };
    use redis::{AsyncCommands, aio::ConnectionManager, aio::ConnectionManagerConfig};
    use std::time::Duration;

    /// Redis-backed idempotency store for multi-replica deployments.
    ///
    /// Configured via `[idempotency.redis]` in `autumn.toml` or
    /// `AUTUMN_IDEMPOTENCY__REDIS__URL` env var.
    ///
    /// The response is written after the handler returns, not in the handler's
    /// database transaction. Use [`DbIdempotencyStore`](super::DbIdempotencyStore)
    /// when the record must commit with the mutation.
    pub struct RedisIdempotencyStore {
        connection: ConnectionManager,
        key_prefix: String,
    }

    impl RedisIdempotencyStore {
        /// Creates a [`RedisIdempotencyStore`] from the application idempotency config.
        ///
        /// # Errors
        /// Returns an error string if no Redis URL is configured or if the Redis
        /// client cannot be opened.
        pub fn from_config(config: &crate::config::IdempotencyConfig) -> Result<Self, String> {
            let url = config
                .redis
                .url
                .as_deref()
                .filter(|u| !u.trim().is_empty())
                .ok_or_else(|| {
                    "Redis idempotency backend requires a URL. \
                     Set AUTUMN_IDEMPOTENCY__REDIS__URL or \
                     [idempotency.redis] url in autumn.toml."
                        .to_owned()
                })?;
            let client = crate::redis_tls::open_client(url).map_err(|e| e.to_string())?;
            let connection =
                ConnectionManager::new_lazy_with_config(client, ConnectionManagerConfig::new())
                    .map_err(|e| e.to_string())?;
            Ok(Self {
                connection,
                key_prefix: config.redis.key_prefix.clone(),
            })
        }

        fn entry_key(&self, key: &str) -> String {
            format!("{}:entry:{}", self.key_prefix, key)
        }

        fn lock_key(&self, key: &str) -> String {
            format!("{}:lock:{}", self.key_prefix, key)
        }

        /// The lock owner that wrote the entry; it expires with the entry.
        fn owner_key(&self, key: &str) -> String {
            format!("{}:owner:{}", self.key_prefix, key)
        }
    }

    fn backend_error(action: &str, error: &redis::RedisError) -> IdempotencyStoreError {
        IdempotencyStoreError::backend(format!("failed to {action} in Redis: {error}"))
    }

    impl IdempotencyStore for RedisIdempotencyStore {
        fn get<'a>(&'a self, key: &'a str) -> IdempotencyFuture<'a, Option<IdempotencyEntry>> {
            let redis_key = self.entry_key(key);
            let mut conn = self.connection.clone();
            Box::pin(async move {
                let data: Option<Vec<u8>> = conn
                    .get(&redis_key)
                    .await
                    .map_err(|e| backend_error("read idempotency entry", &e))?;
                data.map(|bytes| StoredEntry::decode(&bytes)).transpose()
            })
        }

        fn set<'a>(
            &'a self,
            key: &'a str,
            owner: &'a str,
            record: IdempotencyRecord,
            body_hash: Vec<u8>,
            ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            let lock_key = self.lock_key(key);
            let redis_key = self.entry_key(key);
            let owner_key = self.owner_key(key);
            let mut conn = self.connection.clone();
            Box::pin(async move {
                let bytes = StoredEntry::encode(record, body_hash)?;
                // One script: write nothing while another owner holds the
                // lock, or wrote an entry that has not expired.
                let written: redis::RedisResult<i32> = redis::Script::new(
                    "local holder = redis.call('GET', KEYS[1]) \
                     if holder and holder ~= ARGV[1] then return 0 end \
                     local writer = redis.call('GET', KEYS[3]) \
                     if writer and writer ~= ARGV[1] \
                        and redis.call('EXISTS', KEYS[2]) == 1 then return 0 end \
                     redis.call('SET', KEYS[2], ARGV[2], 'EX', ARGV[3]) \
                     redis.call('SET', KEYS[3], ARGV[1], 'EX', ARGV[3]) \
                     return 1",
                )
                .key(&lock_key)
                .key(&redis_key)
                .key(&owner_key)
                .arg(owner)
                .arg(bytes)
                .arg(ttl.as_secs().max(1))
                .invoke_async(&mut conn)
                .await;
                written
                    .map(|written| written == 1)
                    .map_err(|e| backend_error("persist idempotency entry", &e))
            })
        }

        fn try_lock<'a>(
            &'a self,
            key: &'a str,
            owner: &'a str,
            lock_ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            let lock_key = self.lock_key(key);
            let mut conn = self.connection.clone();
            Box::pin(async move {
                let acquired: Option<String> = redis::cmd("SET")
                    .arg(&lock_key)
                    .arg(owner)
                    .arg("NX")
                    .arg("EX")
                    .arg(lock_ttl.as_secs().max(1))
                    .query_async(&mut conn)
                    .await
                    .map_err(|e| backend_error("acquire idempotency lock", &e))?;
                // `Some("OK")` = acquired; `None` = already held.
                Ok(acquired.is_some())
            })
        }

        fn unlock<'a>(&'a self, key: &'a str, owner: &'a str) -> IdempotencyFuture<'a, ()> {
            let lock_key = self.lock_key(key);
            let mut conn = self.connection.clone();
            Box::pin(async move {
                let released: redis::RedisResult<i32> = redis::Script::new(
                    "if redis.call('GET', KEYS[1]) == ARGV[1] then \
                     return redis.call('DEL', KEYS[1]) else return 0 end",
                )
                .key(&lock_key)
                .arg(owner)
                .invoke_async(&mut conn)
                .await;
                released
                    .map(drop)
                    .map_err(|e| backend_error("release idempotency lock", &e))
            })
        }

        fn renew_lock<'a>(
            &'a self,
            key: &'a str,
            owner: &'a str,
            lock_ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            let lock_key = self.lock_key(key);
            let mut conn = self.connection.clone();
            Box::pin(async move {
                let renewed: redis::RedisResult<i32> = redis::Script::new(
                    "if redis.call('GET', KEYS[1]) == ARGV[1] then \
                     return redis.call('EXPIRE', KEYS[1], ARGV[2]) else return 0 end",
                )
                .key(&lock_key)
                .arg(owner)
                .arg(lock_ttl.as_secs().max(1))
                .invoke_async(&mut conn)
                .await;
                renewed
                    .map(|renewed| renewed == 1)
                    .map_err(|e| backend_error("renew idempotency lock", &e))
            })
        }
    }
}

#[cfg(feature = "redis")]
pub use redis_store::RedisIdempotencyStore;

#[cfg(feature = "db")]
mod db_store;
#[cfg(feature = "db")]
pub use db_store::{DbIdempotencyStore, IdempotencyTx};

#[doc(hidden)]
#[derive(Clone)]
pub struct IdempotencyReplayResponse {
    record: IdempotencyRecord,
}

impl IdempotencyReplayResponse {
    fn into_response(self) -> Response<Body> {
        response_from_record(self.record)
    }

    #[must_use]
    pub fn metadata(&self, key: &str) -> Option<&[u8]> {
        self.record
            .metadata
            .iter()
            .find(|(name, _)| name == key)
            .map(|(_, value)| value.as_slice())
    }
}

#[doc(hidden)]
#[must_use]
pub fn __replay_response(
    replay: &Option<axum::extract::Extension<IdempotencyReplayResponse>>,
) -> Option<Response<Body>> {
    replay
        .as_ref()
        .map(|axum::extract::Extension(replay)| replay.clone().into_response())
}

#[doc(hidden)]
#[must_use]
pub fn __replay_finalized_session_response(
    replay: &Option<axum::extract::Extension<IdempotencyReplayResponse>>,
) -> Option<Response<Body>> {
    replay
        .as_ref()
        .and_then(|axum::extract::Extension(replay)| {
            let has_finalized_session_cookie = replay
                .record
                .headers
                .iter()
                .any(|(name, _)| name.eq_ignore_ascii_case("set-cookie"));
            let is_old_session_scope = replay.metadata(FINALIZED_SESSION_SCOPE_METADATA)
                == Some(FINALIZED_SESSION_OLD_SCOPE);
            (has_finalized_session_cookie && is_old_session_scope)
                .then(|| replay.clone().into_response())
        })
}

#[doc(hidden)]
pub async fn __replay_finalized_session_response_for_anonymous(
    session: &crate::session::Session,
    auth_session_key: &str,
    replay: &Option<axum::extract::Extension<IdempotencyReplayResponse>>,
) -> Option<Response<Body>> {
    if session.get(auth_session_key).await.is_some() {
        return None;
    }
    __replay_finalized_session_response(replay)
}

#[doc(hidden)]
#[must_use]
pub const fn __cache_committed_error_response(error: crate::AutumnError) -> crate::AutumnError {
    error.cache_idempotency_response()
}

#[doc(hidden)]
#[must_use]
pub fn __replay_metadata(
    replay: &Option<axum::extract::Extension<IdempotencyReplayResponse>>,
    key: &str,
) -> Option<Vec<u8>> {
    replay
        .as_ref()
        .and_then(|axum::extract::Extension(replay)| replay.metadata(key).map(<[u8]>::to_vec))
}

#[doc(hidden)]
pub enum IdempotencyReplayOr<T> {
    Replay(Response<Body>),
    Inner(T),
    InnerWithReplayMetadata(T, Vec<(String, Vec<u8>)>),
}

impl<T> IntoResponse for IdempotencyReplayOr<T>
where
    T: IntoResponse,
{
    fn into_response(self) -> Response<Body> {
        match self {
            Self::Replay(response) => response,
            Self::Inner(inner) => inner.into_response(),
            Self::InnerWithReplayMetadata(inner, metadata) => {
                let mut response = inner.into_response();
                response
                    .extensions_mut()
                    .insert(IdempotencyReplayMetadata::new(metadata));
                response
            }
        }
    }
}

/// Inner route layer used by Autumn-generated handlers to stop before the
/// mutating handler when an outer idempotency layer has already found a replay.
#[derive(Clone, Copy, Debug, Default)]
pub struct IdempotencyReplayLayer;

#[derive(Clone)]
pub struct IdempotencyReplayService<S> {
    inner: S,
}

impl<S> Layer<S> for IdempotencyReplayLayer {
    type Service = IdempotencyReplayService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        IdempotencyReplayService { inner }
    }
}

impl<S> Service<Request<Body>> for IdempotencyReplayService<S>
where
    S: Service<Request<Body>, Response = Response<Body>, Error = std::convert::Infallible>
        + Clone
        + Send
        + 'static,
    S::Future: Send + 'static,
{
    type Response = Response<Body>;
    type Error = std::convert::Infallible;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut req: Request<Body>) -> Self::Future {
        if let Some(replay) = req.extensions_mut().remove::<IdempotencyReplayResponse>() {
            return Box::pin(async move { Ok(replay.into_response()) });
        }

        // The overwhelming majority of requests carry no replay marker (only
        // the capsule replay driver ever sets one), so this runs for nearly
        // every request through this layer. Nothing here needs `self.inner`
        // cloned into an owned value first: `self.inner.call(req)` can be
        // boxed directly instead of cloning `self.inner` (a
        // `BoxCloneSyncService` at this point in the stack, whose `Clone`
        // impl allocates a fresh box) purely to move the clone into an
        // `async move` block that immediately `.await`s it and does nothing
        // else.
        Box::pin(self.inner.call(req))
    }
}

// ── Layer ─────────────────────────────────────────────────────────────────────

/// Tower [`Layer`] that enforces HTTP idempotency semantics per IETF
/// `draft-ietf-httpapi-idempotency-key-header`.
///
/// Applies only to mutating HTTP methods (POST, PUT, PATCH, DELETE).
/// Requests without an `Idempotency-Key` header are passed through unchanged.
///
/// - **Cache hit, same body**: replays the stored response with
///   `X-Idempotent-Replayed: true` and skips the handler.
/// - **Cache hit, different body**: returns `422 Unprocessable Entity`.
/// - **Concurrent duplicate** (same key, already in flight): returns
///   `409 Conflict` with `Retry-After: 1`.
/// - **Cache miss**: forwards to the handler, stores the response.
#[derive(Clone)]
pub struct IdempotencyLayer {
    store: Arc<dyn IdempotencyStore>,
    ttl: Duration,
    in_flight_ttl: Duration,
    replay_through_inner: bool,
    fail_closed_on_replay: bool,
    metrics: Option<crate::middleware::MetricsCollector>,
    entropy: Arc<dyn crate::entropy::Entropy>,
}

impl IdempotencyLayer {
    /// Create the layer. The response TTL is the store's
    /// [`IdempotencyStore::default_ttl`]; the in-flight TTL is
    /// [`DEFAULT_IN_FLIGHT_TTL`].
    #[must_use]
    pub fn new(store: Arc<dyn IdempotencyStore>) -> Self {
        let ttl = store.default_ttl();
        Self {
            store,
            ttl,
            in_flight_ttl: DEFAULT_IN_FLIGHT_TTL,
            replay_through_inner: false,
            fail_closed_on_replay: false,
            metrics: None,
            entropy: Arc::new(crate::entropy::OsEntropy),
        }
    }

    /// Inject the entropy source used to mint in-flight lock owner ids.
    ///
    /// Defaults to [`crate::entropy::OsEntropy`]; the framework threads the
    /// app's seeded source here so lock ids replay deterministically under a
    /// fixed simulation seed.
    #[must_use]
    pub fn with_entropy(mut self, entropy: Arc<dyn crate::entropy::Entropy>) -> Self {
        self.entropy = entropy;
        self
    }

    #[must_use]
    pub const fn with_ttl(mut self, ttl: Duration) -> Self {
        self.ttl = ttl;
        self
    }

    #[must_use]
    pub const fn with_in_flight_ttl(mut self, ttl: Duration) -> Self {
        self.in_flight_ttl = ttl;
        self
    }

    #[must_use]
    pub const fn replay_through_inner(mut self) -> Self {
        self.replay_through_inner = true;
        self.fail_closed_on_replay = false;
        self
    }

    #[must_use]
    pub const fn fail_closed_on_replay(mut self) -> Self {
        self.replay_through_inner = false;
        self.fail_closed_on_replay = true;
        self
    }

    #[must_use]
    pub fn with_metrics(mut self, metrics: crate::middleware::MetricsCollector) -> Self {
        self.metrics = Some(metrics);
        self
    }
}

impl<S> Layer<S> for IdempotencyLayer {
    type Service = IdempotencyService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        IdempotencyService {
            inner,
            store: self.store.clone(),
            ttl: self.ttl,
            in_flight_ttl: self.in_flight_ttl,
            replay_through_inner: self.replay_through_inner,
            fail_closed_on_replay: self.fail_closed_on_replay,
            metrics: self.metrics.clone(),
            entropy: self.entropy.clone(),
        }
    }
}

// ── Service ───────────────────────────────────────────────────────────────────

/// Tower [`Service`] produced by [`IdempotencyLayer`].
#[derive(Clone)]
pub struct IdempotencyService<S> {
    inner: S,
    store: Arc<dyn IdempotencyStore>,
    ttl: Duration,
    in_flight_ttl: Duration,
    replay_through_inner: bool,
    fail_closed_on_replay: bool,
    metrics: Option<crate::middleware::MetricsCollector>,
    entropy: Arc<dyn crate::entropy::Entropy>,
}

struct IdempotencyRequestConfig {
    store: Arc<dyn IdempotencyStore>,
    ttl: Duration,
    in_flight_ttl: Duration,
    replay_through_inner: bool,
    fail_closed_on_replay: bool,
    metrics: Option<crate::middleware::MetricsCollector>,
    entropy: Arc<dyn crate::entropy::Entropy>,
}

impl<S> Service<Request<Body>> for IdempotencyService<S>
where
    S: Service<Request<Body>, Response = Response<Body>, Error = std::convert::Infallible>
        + Clone
        + Send
        + 'static,
    S::Future: Send + 'static,
{
    type Response = Response<Body>;
    type Error = std::convert::Infallible;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: Request<Body>) -> Self::Future {
        // Clone-and-swap: the instance that was polled ready becomes `inner`
        // for this call; `self.inner` is replaced with a fresh clone for the
        // next call. This preserves Tower backpressure semantics.
        let clone = self.inner.clone();
        let inner = std::mem::replace(&mut self.inner, clone);
        let config = IdempotencyRequestConfig {
            store: self.store.clone(),
            ttl: self.ttl,
            in_flight_ttl: self.in_flight_ttl,
            replay_through_inner: self.replay_through_inner,
            fail_closed_on_replay: self.fail_closed_on_replay,
            metrics: self.metrics.clone(),
            entropy: self.entropy.clone(),
        };
        Box::pin(handle_idempotent_request(inner, config, req))
    }
}

#[allow(
    clippy::unwrap_used,
    reason = "infallible: response built from static status/body"
)]
fn request_body_too_large_response() -> Response<Body> {
    Response::builder()
        .status(StatusCode::PAYLOAD_TOO_LARGE)
        .body(Body::from(
            "request body too large for idempotency middleware",
        ))
        .unwrap()
}

#[allow(
    clippy::unwrap_used,
    reason = "infallible: response built from static status/body"
)]
fn in_flight_conflict_response() -> Response<Body> {
    Response::builder()
        .status(StatusCode::CONFLICT)
        .header("retry-after", "1")
        .body(Body::from(
            "a request with this idempotency key is already being processed; \
             retry after 1 second",
        ))
        .unwrap()
}

/// `409` for a key in flight, counted as a conflict.
fn in_flight_conflict(metrics: Option<&crate::middleware::MetricsCollector>) -> Response<Body> {
    metrics.inspect(|m| m.record_idempotency_conflict());
    in_flight_conflict_response()
}

#[allow(
    clippy::unwrap_used,
    reason = "infallible: response built from static status/body"
)]
fn replay_requires_inner_stop_response() -> Response<Body> {
    Response::builder()
        .status(StatusCode::CONFLICT)
        .body(Body::from(
            "idempotency replay requires an inner replay stop for this route",
        ))
        .unwrap()
}

#[allow(
    clippy::unwrap_used,
    reason = "infallible: response built from static status/body"
)]
pub(crate) fn persistence_failed_response() -> Response<Body> {
    Response::builder()
        .status(StatusCode::SERVICE_UNAVAILABLE)
        .body(Body::from("idempotency persistence unavailable"))
        .unwrap()
}

struct PreparedIdempotencyRequest {
    idempotency_key: String,
    storage_key: String,
    stale_cookie_storage_key: Option<String>,
    key_context: StorageKeyContext,
    body_hash: Vec<u8>,
    session: Option<crate::session::Session>,
    parts: Parts,
    body_bytes: Bytes,
}

/// The in-flight lock one request holds.
///
/// Dropping it keeps the lock; the store's in-flight TTL then frees the key.
/// This fails closed when the handler future is cancelled or panics: the
/// mutation may have committed, so a retry gets `409` until the TTL expires.
/// Only [`Self::release`] frees the key at once.
struct InFlightLock {
    store: Arc<dyn IdempotencyStore>,
    key: String,
    owner: String,
    /// How long the store keeps the lock. Alias locks use the same time.
    in_flight_ttl: Duration,
}

impl InFlightLock {
    /// Free the key now. Call this only on an outcome observed to completion.
    async fn release(self) {
        if let Err(error) = self.store.unlock(&self.key, &self.owner).await {
            tracing::warn!(
                error = %error,
                "Idempotency unlock failed; the lock expires by its in-flight TTL"
            );
        }
    }

    /// Keep the key locked until the in-flight TTL expires.
    #[allow(
        clippy::unused_self,
        reason = "consumes the lock; the store's TTL frees the key"
    )]
    fn hold_until_ttl(self) {}

    /// Keep the key locked until its record expires, for a record that
    /// committed in the handler's transaction and waits for a session rewrite.
    /// If the hold fails, the lock still expires by its in-flight TTL.
    #[cfg(feature = "db")]
    async fn hold_until_record_expires(&self) {
        let store: &dyn std::any::Any = self.store.as_ref();
        let Some(db) = store.downcast_ref::<DbIdempotencyStore>() else {
            return;
        };
        if let Err(error) = db.hold_until_expiry(&self.key, &self.owner).await {
            tracing::warn!(
                error = %error,
                "Idempotency lock extension failed; the lock expires by its in-flight TTL"
            );
        }
    }

    #[cfg(not(feature = "db"))]
    #[allow(clippy::unused_async, reason = "same signature as the db build")]
    async fn hold_until_record_expires(&self) {}
}

#[derive(Clone)]
pub(crate) struct DeferredIdempotencyCommit {
    inner: Arc<Mutex<Option<DeferredIdempotencyState>>>,
}

struct DeferredIdempotencyState {
    store: Arc<dyn IdempotencyStore>,
    storage_key: String,
    key_context: StorageKeyContext,
    alias_storage_keys: Vec<String>,
    /// Alias keys that `lock.owner` holds. The commit frees them.
    locked_alias_keys: Vec<String>,
    primary_replay_after_guard_denial: bool,
    idempotency_key: String,
    record: IdempotencyRecord,
    body_hash: Vec<u8>,
    ttl: Duration,
    lock: InFlightLock,
}

impl DeferredIdempotencyState {
    /// Return the alias key for a retry. Return `None` if the key is the
    /// primary key or is already registered. This also records a primary
    /// replay after guard denial.
    fn alias_key(
        &mut self,
        session_id: Option<&str>,
        primary_replay_after_guard_denial: bool,
        tenant_override: Option<&str>,
    ) -> Option<String> {
        // Use the override only if this request resolved a tenant. A
        // `public_paths` route resolves no tenant on a retry. A forced tenant
        // would make the alias key differ from the retry's key.
        let tenant_override = tenant_override.filter(|_| self.key_context.tenant.is_some());
        let storage_key = self.key_context.storage_key(session_id, tenant_override);
        if primary_replay_after_guard_denial && storage_key != self.storage_key {
            self.primary_replay_after_guard_denial = true;
        }
        (storage_key != self.storage_key && !self.alias_storage_keys.contains(&storage_key))
            .then_some(storage_key)
    }
}

impl DeferredIdempotencyCommit {
    fn new(state: DeferredIdempotencyState) -> Self {
        Self {
            inner: Arc::new(Mutex::new(Some(state))),
        }
    }

    fn take(&self) -> Option<DeferredIdempotencyState> {
        self.inner
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .take()
    }

    async fn commit_with_final_headers(
        &self,
        headers: &HeaderMap,
    ) -> Result<(), IdempotencyStoreError> {
        let Some(mut state) = self.take() else {
            return Ok(());
        };

        state.record.headers = extract_finalized_session_replay_headers(headers);
        let primary_scope = if state.primary_replay_after_guard_denial {
            FINALIZED_SESSION_OLD_SCOPE
        } else {
            FINALIZED_SESSION_CURRENT_SCOPE
        };
        let primary_record = finalized_session_record(state.record.clone(), primary_scope);
        let alias_record = finalized_session_record(state.record, FINALIZED_SESSION_CURRENT_SCOPE);
        let locked_alias_keys = state.locked_alias_keys;
        let writes = std::iter::once((state.storage_key, primary_record)).chain(
            state
                .alias_storage_keys
                .into_iter()
                .map(|key| (key, alias_record.clone())),
        );
        for (storage_key, record) in writes {
            // Another owner's lock or response fences a key out, and a retry
            // under that key would replay the other response: fail closed as
            // on a store error.
            let written = state
                .store
                .set(
                    &storage_key,
                    &state.lock.owner,
                    record,
                    state.body_hash.clone(),
                    state.ttl,
                )
                .await
                .and_then(stored);
            if let Err(error) = written {
                tracing::error!(
                    idempotency.key = %state.idempotency_key,
                    error = %error,
                    "Deferred idempotency persistence failed after finalized session response; failing closed"
                );
                state.lock.hold_until_ttl();
                return Err(error);
            }
        }
        for key in &locked_alias_keys {
            if let Err(error) = state.store.unlock(key, &state.lock.owner).await {
                tracing::warn!(
                    error = %error,
                    "Idempotency alias unlock failed; the lock expires by its in-flight TTL"
                );
            }
        }
        state.lock.release().await;
        Ok(())
    }

    /// Return the store, key, owner and lock TTL for a new alias. Return
    /// `None` if there is nothing to lock. This function is not async, so no
    /// lock guard stays held across an await.
    fn alias_reservation(
        &self,
        session_id: Option<&str>,
        primary_replay_after_guard_denial: bool,
        tenant_override: Option<&str>,
    ) -> Option<(Arc<dyn IdempotencyStore>, String, String, Duration)> {
        let mut guard = self.inner.lock().unwrap_or_else(PoisonError::into_inner);
        let reservation = guard.as_mut().and_then(|state| {
            let key = state.alias_key(
                session_id,
                primary_replay_after_guard_denial,
                tenant_override,
            )?;
            Some((
                Arc::clone(&state.store),
                key,
                state.lock.owner.clone(),
                state.lock.in_flight_ttl,
            ))
        });
        drop(guard);
        reservation
    }

    /// Lock the alias key that a retry will use, before the session is saved.
    ///
    /// Return an error if another owner holds the key or the store fails.
    /// The caller then rejects the request. If the commit fails, the lock
    /// stays until its TTL ends.
    async fn reserve_session_alias(
        &self,
        session_id: Option<&str>,
        primary_replay_after_guard_denial: bool,
        tenant_override: Option<&str>,
    ) -> Result<(), IdempotencyStoreError> {
        let reservation = self.alias_reservation(
            session_id,
            primary_replay_after_guard_denial,
            tenant_override,
        );
        let Some((store, key, owner, in_flight_ttl)) = reservation else {
            return Ok(());
        };
        if !acquire_lock(store.as_ref(), &key, &owner, in_flight_ttl).await {
            return Err(IdempotencyStoreError::backend(
                "another request holds the session alias idempotency key",
            ));
        }
        let mut guard = self.inner.lock().unwrap_or_else(PoisonError::into_inner);
        if let Some(state) = guard.as_mut() {
            state.locked_alias_keys.push(key.clone());
            state.alias_storage_keys.push(key);
        }
        drop(guard);
        Ok(())
    }

    fn add_session_alias(
        &self,
        session_id: Option<&str>,
        primary_replay_after_guard_denial: bool,
        tenant_override: Option<&str>,
    ) {
        let mut guard = self.inner.lock().unwrap_or_else(PoisonError::into_inner);
        let Some(state) = guard.as_mut() else {
            return;
        };
        if let Some(key) = state.alias_key(
            session_id,
            primary_replay_after_guard_denial,
            tenant_override,
        ) {
            state.alias_storage_keys.push(key);
        }
        drop(guard);
    }

    fn keep_locked_until_ttl(&self) {
        if let Some(state) = self.take() {
            state.lock.hold_until_ttl();
        }
    }
}

/// Take the deferred commit and return the future that stores the record with
/// the response's final headers, or `None` when no commit is pending.
///
/// Synchronous, with a boxed future: `SessionLayer` runs on every request, and
/// an `async fn` awaited there would add its whole state machine to every
/// request's future, not only to the rare one with a pending commit (see
/// `config_alloc_gate`).
pub(crate) fn finalize_deferred_session_commit(
    response: &mut Response<Body>,
) -> Option<IdempotencyFuture<'_, ()>> {
    let commit = response
        .extensions_mut()
        .remove::<DeferredIdempotencyCommit>()?;
    let headers = response.headers();
    Some(Box::pin(async move {
        commit.commit_with_final_headers(headers).await
    }))
}

/// Register the storage key a retry presenting `session_id` will compute as
/// an alias of the primary key, so that retry replays the response instead of
/// re-executing the mutation.
///
/// `tenant_override`, when set, is the tenant `[tenancy] source = "session"`
/// will resolve for that retry — read live from the finalized session data
/// rather than the tenant captured before the handler ran, which a handler
/// that itself changes the tenancy session key (an org switch, a
/// tenant-scoped login) may have just moved. `None` leaves the alias in the
/// same tenant namespace as the primary key, which is correct for every
/// other tenancy source and for a session mutation that doesn't touch the
/// tenancy session key.
pub(crate) fn add_deferred_session_replay_key(
    response: &Response<Body>,
    session_id: Option<&str>,
    primary_replay_after_guard_denial: bool,
    tenant_override: Option<&str>,
) {
    if let Some(commit) = response.extensions().get::<DeferredIdempotencyCommit>() {
        commit.add_session_alias(
            session_id,
            primary_replay_after_guard_denial,
            tenant_override,
        );
    }
}

/// Lock the alias key before the session is saved. A request with the saved
/// session then finds the key in flight. Return `None` if no commit is
/// pending. If the future returns `Err`, reject the request.
///
/// This function copies its inputs. The future must be `Send`, and
/// `Response` is not `Sync`.
pub(crate) fn reserve_deferred_session_replay_key(
    response: &Response<Body>,
    session_id: Option<&str>,
    primary_replay_after_guard_denial: bool,
    tenant_override: Option<&str>,
) -> Option<IdempotencyFuture<'static, ()>> {
    let commit = response
        .extensions()
        .get::<DeferredIdempotencyCommit>()?
        .clone();
    let session_id = session_id.map(str::to_owned);
    let tenant_override = tenant_override.map(str::to_owned);
    Some(Box::pin(async move {
        commit
            .reserve_session_alias(
                session_id.as_deref(),
                primary_replay_after_guard_denial,
                tenant_override.as_deref(),
            )
            .await
    }))
}

pub(crate) fn keep_deferred_session_commit_locked(response: &mut Response<Body>) {
    if let Some(commit) = response
        .extensions_mut()
        .remove::<DeferredIdempotencyCommit>()
    {
        commit.keep_locked_until_ttl();
    }
}

fn request_idempotency_key(req: &Request<Body>) -> Option<String> {
    let key = req
        .headers()
        .get(IDEMPOTENCY_KEY_HEADER)?
        .to_str()
        .unwrap_or("");
    (!key.is_empty()).then(|| key.to_owned())
}

fn in_flight_lock_owner(entropy: &dyn crate::entropy::Entropy) -> String {
    entropy.uuid_v4().to_string()
}

// `clippy::result_large_err` (armed by rustc 1.98) measures the `Err` variant at
// 128 bytes — that is `axum::response::Response`'s own size, not something this
// crate chose. Returning a ready-made rejection response IS the idiom here, and
// boxing it would add an allocation to every rejection path to satisfy a size
// heuristic. Allowed at the site rather than workspace-wide so the lint stays
// armed for error types we do control.
#[allow(clippy::result_large_err)]
async fn prepare_idempotency_request(
    idempotency_key: String,
    req: Request<Body>,
) -> Result<PreparedIdempotencyRequest, Response<Body>> {
    let (mut parts, body) = req.into_parts();
    let key_context = StorageKeyContext::from_parts(idempotency_key.clone(), &parts);
    let session_id = storage_session_id_for_parts(&parts).await;
    let storage_key = key_context.storage_key(session_id.as_deref(), None);
    let stale_cookie_storage_key = stale_cookie_session_id_for_parts(&parts).and_then(|id| {
        let key = key_context.storage_key(Some(&id), None);
        (key != storage_key).then_some(key)
    });
    parts.extensions.insert(IdempotencyContext::new(
        idempotency_key.clone(),
        storage_key.clone(),
    ));
    let session = parts.extensions.get::<crate::session::Session>().cloned();
    let content_type = parts
        .headers
        .get(axum::http::header::CONTENT_TYPE)
        .map(axum::http::HeaderValue::as_bytes);
    let body_limit = parts
        .extensions
        .get::<crate::security::config::UploadConfig>()
        .map_or(DEFAULT_REQUEST_BODY_LIMIT, |c| c.max_request_size_bytes);
    let body_bytes = axum::body::to_bytes(body, body_limit)
        .await
        .map_err(|_| request_body_too_large_response())?;
    let body_hash = compute_body_hash(&body_bytes, content_type);

    Ok(PreparedIdempotencyRequest {
        idempotency_key,
        storage_key,
        stale_cookie_storage_key,
        key_context,
        body_hash,
        session,
        parts,
        body_bytes,
    })
}

async fn lookup_prepared_entry(
    store: &dyn IdempotencyStore,
    prepared: &PreparedIdempotencyRequest,
) -> Result<Option<IdempotencyEntry>, IdempotencyStoreError> {
    if let Some(key) = prepared.stale_cookie_storage_key.as_deref()
        && let Some(entry) = store.get(key).await?
    {
        return Ok(Some(entry));
    }

    store.get(&prepared.storage_key).await
}

/// An upper bound on the deadline of a lock just taken for `in_flight_ttl`,
/// in Unix ms on the app clock: read after the store set the lock, with the
/// same 1 s floor for a zero TTL.
fn lock_deadline_bound_ms(in_flight_ttl: Duration) -> i64 {
    let ttl = if in_flight_ttl.is_zero() {
        Duration::from_secs(1)
    } else {
        in_flight_ttl
    };
    crate::time::ambient_now()
        .timestamp_millis()
        .saturating_add(i64::try_from(ttl.as_millis()).unwrap_or(i64::MAX))
}

/// Acquire the in-flight lock for `key`. A store error counts as "held", so
/// an outage fails closed with `409` and never runs the handler twice.
async fn acquire_lock(
    store: &dyn IdempotencyStore,
    key: &str,
    owner: &str,
    in_flight_ttl: Duration,
) -> bool {
    match store.try_lock(key, owner, in_flight_ttl).await {
        Ok(acquired) => acquired,
        Err(error) => {
            tracing::warn!(
                error = %error,
                "Idempotency lock unavailable; failing closed to prevent duplicate processing"
            );
            false
        }
    }
}

/// A response that another owner's lock or response fenced out was not
/// stored: a retry would replay the other response, or run the mutation
/// again. Callers fail closed on it as on a store error.
pub(crate) fn stored(written: bool) -> Result<(), IdempotencyStoreError> {
    written.then_some(()).ok_or_else(|| {
        IdempotencyStoreError::backend(
            "another request holds the idempotency key or stored its response",
        )
    })
}

/// What became of a lock that the work before the handler may have outlived.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum LockForHandler {
    /// The work was quick: the lock runs as taken.
    Held,
    /// The lock runs a full TTL from now.
    Renewed,
    /// Another owner took the key, or nobody holds it: the handler must not run.
    Lost,
}

/// Renew `owner`'s lock on `key` when the work since `taken_at`, read just
/// before the lock was taken, used more than a tenth of the in-flight TTL.
///
/// A lookup or a copy that slow can outlive the lock, and a retry could then
/// take the key and run alongside this request's handler.
pub(crate) async fn keep_lock_for_handler(
    store: &dyn IdempotencyStore,
    key: &str,
    owner: &str,
    in_flight_ttl: Duration,
    taken_at: Instant,
) -> Result<LockForHandler, IdempotencyStoreError> {
    let ttl = if in_flight_ttl.is_zero() {
        Duration::from_secs(1)
    } else {
        in_flight_ttl
    };
    let elapsed = crate::time::ambient_instant().saturating_duration_since(taken_at);
    if elapsed <= ttl.checked_div(10).unwrap_or_default() {
        return Ok(LockForHandler::Held);
    }
    Ok(if store.renew_lock(key, owner, ttl).await? {
        LockForHandler::Renewed
    } else {
        LockForHandler::Lost
    })
}

/// When the request's lock was taken, for how long, and the bound on its
/// deadline (see [`lock_deadline_bound_ms`]).
struct LockTiming {
    taken_at: Instant,
    in_flight_ttl: Duration,
    deadline_ms: i64,
}

/// Hand the lock back with its deadline bound, renewed when the work before
/// the handler was slow (see [`keep_lock_for_handler`]), or answer `409` for a
/// request whose lock lapsed to another owner. `held` is the outcome of
/// [`carry_stale_key`]; its response passes through.
// `Err` is a ready-made response, as in `prepare_idempotency_request`.
#[allow(clippy::result_large_err)]
async fn renew_slow_lock(
    held: Result<InFlightLock, Response<Body>>,
    timing: LockTiming,
    prepared: &PreparedIdempotencyRequest,
    metrics: Option<&crate::middleware::MetricsCollector>,
) -> Result<(InFlightLock, i64), Response<Body>> {
    let lock = held?;
    let ttl = timing.in_flight_ttl;
    match keep_lock_for_handler(
        lock.store.as_ref(),
        &lock.key,
        &lock.owner,
        ttl,
        timing.taken_at,
    )
    .await
    {
        Ok(LockForHandler::Held) => Ok((lock, timing.deadline_ms)),
        Ok(LockForHandler::Renewed) => Ok((lock, lock_deadline_bound_ms(ttl))),
        Ok(LockForHandler::Lost) => {
            tracing::warn!(
                idempotency.key = %prepared.idempotency_key,
                "Idempotency lock lapsed before the handler ran; returning 409"
            );
            lock.release().await;
            Err(in_flight_conflict(metrics))
        }
        Err(error) => {
            tracing::warn!(
                idempotency.key = %prepared.idempotency_key,
                error = %error,
                "Idempotency lock renewal failed; failing closed"
            );
            lock.hold_until_ttl();
            Err(in_flight_conflict(metrics))
        }
    }
}

/// The key of the stale session cookie is in flight: another attempt holds it.
struct StaleKeyInFlight;

/// Lock the key of the request's stale session cookie, if it has one.
///
/// The lock is held until the request's own key is locked, so a recovery
/// point the first attempt committed under the old key can be carried over
/// (see [`adopt_stale_recovery_point`]).
async fn lock_stale_cookie_key(
    store: &Arc<dyn IdempotencyStore>,
    prepared: &PreparedIdempotencyRequest,
    in_flight_ttl: Duration,
    entropy: &dyn crate::entropy::Entropy,
) -> Result<Option<InFlightLock>, StaleKeyInFlight> {
    let Some(key) = prepared.stale_cookie_storage_key.as_deref() else {
        return Ok(None);
    };

    let owner = in_flight_lock_owner(entropy);
    if acquire_lock(store.as_ref(), key, &owner, in_flight_ttl).await {
        Ok(Some(InFlightLock {
            store: Arc::clone(store),
            key: key.to_owned(),
            owner,
            in_flight_ttl,
        }))
    } else {
        Err(StaleKeyInFlight)
    }
}

/// Lock the stale session cookie's key, if the request has one, then the
/// request's own key. `None` when either is in flight: the caller answers
/// `409`. The second value is the stale key's lock.
async fn lock_request_keys(
    store: &Arc<dyn IdempotencyStore>,
    prepared: &PreparedIdempotencyRequest,
    in_flight_ttl: Duration,
    entropy: &dyn crate::entropy::Entropy,
) -> Option<(InFlightLock, Option<InFlightLock>)> {
    let Ok(stale_lock) = lock_stale_cookie_key(store, prepared, in_flight_ttl, entropy).await
    else {
        tracing::debug!(
            idempotency.key = %prepared.idempotency_key,
            "Stale session cookie idempotency key already in flight — returning 409"
        );
        return None;
    };
    let owner = in_flight_lock_owner(entropy);
    if !acquire_lock(store.as_ref(), &prepared.storage_key, &owner, in_flight_ttl).await {
        release_stale_lock(stale_lock).await;
        tracing::debug!(
            idempotency.key = %prepared.idempotency_key,
            "Idempotency key already in flight — returning 409"
        );
        return None;
    }
    let lock = InFlightLock {
        store: Arc::clone(store),
        key: prepared.storage_key.clone(),
        owner,
        in_flight_ttl,
    };
    Some((lock, stale_lock))
}

async fn release_stale_lock(stale: Option<InFlightLock>) {
    if let Some(stale) = stale {
        stale.release().await;
    }
}

/// Carry the recovery point of the stale cookie's key over to the request's
/// own key, then free the stale key.
///
/// A retry whose session went away runs under a new session's key. Without
/// the copy, its [`IdempotencyTx`] would not see the steps the first attempt
/// committed, and would run them again. A failed copy is a store error: the
/// request fails closed with `500`, and the stale key keeps its point.
///
/// `false` when the stale key's lock expired and another request took the
/// key before the copy: the request must not run, as for a key in flight.
#[cfg(feature = "db")]
async fn adopt_stale_recovery_point(
    stale: Option<InFlightLock>,
    lock: &InFlightLock,
) -> Result<bool, IdempotencyStoreError> {
    let Some(stale) = stale else {
        return Ok(true);
    };
    let store: &dyn std::any::Any = lock.store.as_ref();
    let adopted = match store.downcast_ref::<DbIdempotencyStore>() {
        Some(db) => {
            db.adopt_recovery_point(&stale.key, &stale.owner, &lock.key, &lock.owner)
                .await
        }
        None => Ok(true),
    };
    stale.release().await;
    adopted
}

#[cfg(not(feature = "db"))]
#[allow(clippy::unused_async, reason = "same signature as the db build")]
async fn adopt_stale_recovery_point(
    stale: Option<InFlightLock>,
    _lock: &InFlightLock,
) -> Result<bool, IdempotencyStoreError> {
    release_stale_lock(stale).await;
    Ok(true)
}

/// Tells the middleware whether the handler stored the record in its own
/// transaction, through [`IdempotencyTx::commit`].
#[derive(Default)]
struct TxProbe {
    #[cfg(feature = "db")]
    tx: Option<IdempotencyTx>,
}

impl TxProbe {
    /// Give the handler an [`IdempotencyTx`] when the store is a
    /// [`DbIdempotencyStore`].
    #[cfg(feature = "db")]
    fn attach(
        store: &Arc<dyn IdempotencyStore>,
        prepared: &mut PreparedIdempotencyRequest,
        owner: &str,
        ttl: Duration,
        lock_deadline_ms: i64,
    ) -> Self {
        let store: &dyn std::any::Any = store.as_ref();
        if !store.is::<DbIdempotencyStore>() {
            return Self::default();
        }
        let tx = IdempotencyTx::new(
            prepared.storage_key.clone(),
            owner.to_owned(),
            prepared.body_hash.clone(),
            ttl,
            lock_deadline_ms,
        );
        prepared.parts.extensions.insert(tx.clone());
        Self { tx: Some(tx) }
    }

    #[cfg(not(feature = "db"))]
    fn attach(
        _store: &Arc<dyn IdempotencyStore>,
        _prepared: &mut PreparedIdempotencyRequest,
        _owner: &str,
        _ttl: Duration,
        _lock_deadline_ms: i64,
    ) -> Self {
        Self::default()
    }

    /// `true` when the handler's transaction committed the record.
    #[cfg(feature = "db")]
    async fn committed(&self, store: &Arc<dyn IdempotencyStore>) -> bool {
        let Some(tx) = &self.tx else {
            return false;
        };
        let store: &dyn std::any::Any = store.as_ref();
        match store.downcast_ref::<DbIdempotencyStore>() {
            Some(db) => tx.committed(db).await,
            None => false,
        }
    }

    #[cfg(not(feature = "db"))]
    #[allow(
        clippy::unused_self,
        clippy::unused_async,
        reason = "same signature as the db build"
    )]
    async fn committed(&self, _store: &Arc<dyn IdempotencyStore>) -> bool {
        false
    }
}

fn cacheable_response_record(
    status: u16,
    headers: &HeaderMap,
    body: &Bytes,
    metadata: Vec<(String, Vec<u8>)>,
) -> IdempotencyRecord {
    IdempotencyRecord {
        status,
        headers: extract_replay_headers(headers),
        body: body.to_vec(),
        metadata,
    }
}

async fn handle_idempotent_request<S>(
    mut inner: S,
    config: IdempotencyRequestConfig,
    req: Request<Body>,
) -> Result<Response<Body>, std::convert::Infallible>
where
    S: Service<Request<Body>, Response = Response<Body>, Error = std::convert::Infallible>
        + Send
        + 'static,
    S::Future: Send + 'static,
{
    let IdempotencyRequestConfig {
        store,
        ttl,
        in_flight_ttl,
        replay_through_inner,
        fail_closed_on_replay,
        metrics,
        entropy,
    } = config;

    if !is_mutating_method(req.method()) {
        return inner.call(req).await;
    }

    let Some(idempotency_key) = request_idempotency_key(&req) else {
        return inner.call(req).await;
    };

    let mut prepared = match prepare_idempotency_request(idempotency_key, req).await {
        Ok(prepared) => prepared,
        Err(response) => return Ok(response),
    };

    // ── Cache hit ──────────────────────────────────────────────────────────
    match lookup_prepared_entry(store.as_ref(), &prepared).await {
        Ok(Some(entry)) => {
            return replay_cache_hit(
                &mut inner,
                entry,
                prepared,
                metrics.as_ref(),
                replay_through_inner,
                fail_closed_on_replay,
            )
            .await;
        }
        Ok(None) => {}
        Err(error) => {
            tracing::error!(
                idempotency.key = %prepared.idempotency_key,
                error = %error,
                "Idempotency lookup failed; failing closed"
            );
            return Ok(persistence_failed_response());
        }
    }

    // ── In-flight check (concurrent duplicate) ─────────────────────────────
    let lock_taken_at = crate::time::ambient_instant();
    let Some((lock, stale_lock)) =
        lock_request_keys(&store, &prepared, in_flight_ttl, entropy.as_ref()).await
    else {
        return lock_refused_response(
            &mut inner,
            store.as_ref(),
            prepared,
            metrics.as_ref(),
            replay_through_inner,
            fail_closed_on_replay,
        )
        .await;
    };
    let lock_deadline_ms = lock_deadline_bound_ms(in_flight_ttl);

    // Double-check after acquiring the lock: a concurrent request may have
    // completed between our miss check and lock acquisition.
    match lookup_prepared_entry(store.as_ref(), &prepared).await {
        Ok(Some(entry)) => {
            lock.release().await;
            release_stale_lock(stale_lock).await;
            return replay_cache_hit(
                &mut inner,
                entry,
                prepared,
                metrics.as_ref(),
                replay_through_inner,
                fail_closed_on_replay,
            )
            .await;
        }
        Ok(None) => {}
        Err(error) => {
            lock.hold_until_ttl();
            release_stale_lock(stale_lock).await;
            tracing::error!(
                idempotency.key = %prepared.idempotency_key,
                error = %error,
                "Idempotency lookup failed after lock acquisition; failing closed"
            );
            return Ok(persistence_failed_response());
        }
    }

    let timing = LockTiming {
        taken_at: lock_taken_at,
        in_flight_ttl,
        deadline_ms: lock_deadline_ms,
    };
    let held = carry_stale_key(stale_lock, lock, &prepared, metrics.as_ref()).await;
    let (lock, lock_deadline_ms) =
        match renew_slow_lock(held, timing, &prepared, metrics.as_ref()).await {
            Ok(renewed) => renewed,
            Err(response) => return Ok(response),
        };

    let probe = TxProbe::attach(&store, &mut prepared, &lock.owner, ttl, lock_deadline_ms);
    handle_cache_miss(inner, store, ttl, prepared, metrics.as_ref(), lock, probe).await
}

/// Carry the stale cookie's recovery point over to the request's key (see
/// [`adopt_stale_recovery_point`]) and hand the lock back, or answer for a
/// request that must not run: `409` when another request took the stale
/// key, `500` when the copy failed.
// `Err` is a ready-made response, as in `prepare_idempotency_request`.
#[allow(clippy::result_large_err)]
async fn carry_stale_key(
    stale: Option<InFlightLock>,
    lock: InFlightLock,
    prepared: &PreparedIdempotencyRequest,
    metrics: Option<&crate::middleware::MetricsCollector>,
) -> Result<InFlightLock, Response<Body>> {
    match adopt_stale_recovery_point(stale, &lock).await {
        Ok(true) => Ok(lock),
        Ok(false) => {
            lock.release().await;
            Err(in_flight_conflict(metrics))
        }
        Err(error) => {
            lock.hold_until_ttl();
            tracing::error!(
                idempotency.key = %prepared.idempotency_key,
                error = %error,
                "Idempotency recovery point copy failed; failing closed"
            );
            Err(persistence_failed_response())
        }
    }
}

/// The answer when a key could not be locked.
///
/// A store may refuse the lock because the key holds a stored response:
/// another request stored it and released the key between this request's
/// lookup and its lock. That response is replayed. Otherwise the key is in
/// flight, or the store failed: `409`, and the handler does not run.
async fn lock_refused_response<S>(
    inner: &mut S,
    store: &dyn IdempotencyStore,
    prepared: PreparedIdempotencyRequest,
    metrics: Option<&crate::middleware::MetricsCollector>,
    replay_through_inner: bool,
    fail_closed_on_replay: bool,
) -> Result<Response<Body>, std::convert::Infallible>
where
    S: Service<Request<Body>, Response = Response<Body>, Error = std::convert::Infallible>
        + Send
        + 'static,
    S::Future: Send + 'static,
{
    match lookup_prepared_entry(store, &prepared).await {
        Ok(Some(entry)) => {
            replay_cache_hit(
                inner,
                entry,
                prepared,
                metrics,
                replay_through_inner,
                fail_closed_on_replay,
            )
            .await
        }
        Ok(None) | Err(_) => Ok(in_flight_conflict(metrics)),
    }
}

/// Take what the handler left for the cache in its response extensions: the
/// replay metadata, and whether a committed error response is to be cached.
fn take_replay_markers(extensions: &mut axum::http::Extensions) -> (Vec<(String, Vec<u8>)>, bool) {
    let replay_metadata = extensions
        .remove::<IdempotencyReplayMetadata>()
        .map_or_else(Vec::new, IdempotencyReplayMetadata::into_entries);
    let cache_committed_error = extensions
        .remove::<IdempotencyCacheCommittedErrorResponse>()
        .is_some();
    (replay_metadata, cache_committed_error)
}

async fn handle_cache_miss<S>(
    mut inner: S,
    store: Arc<dyn IdempotencyStore>,
    ttl: Duration,
    prepared: PreparedIdempotencyRequest,
    metrics: Option<&crate::middleware::MetricsCollector>,
    lock: InFlightLock,
    probe: TxProbe,
) -> Result<Response<Body>, std::convert::Infallible>
where
    S: Service<Request<Body>, Response = Response<Body>, Error = std::convert::Infallible>
        + Send
        + 'static,
    S::Future: Send + 'static,
{
    let PreparedIdempotencyRequest {
        idempotency_key,
        storage_key,
        key_context,
        body_hash,
        session,
        parts,
        body_bytes,
        ..
    } = prepared;

    tracing::debug!(
        idempotency.key = %idempotency_key,
        "Idempotency cache miss — forwarding to handler"
    );

    let response = inner
        .call(Request::from_parts(parts, Body::from(body_bytes)))
        .await?;
    let (mut resp_parts, resp_body) = response.into_parts();

    // Collect up to the cache cap; stream oversized bodies through without
    // buffering to avoid materialising large responses in memory.
    let resp_bytes = match collect_response_for_cache(resp_body).await {
        CollectedResponseBody::StreamError(passthrough_body) => {
            lock.release().await;
            tracing::warn!(
                idempotency.key = %idempotency_key,
                "I/O error reading response body; passing the body error through without storing idempotency entry"
            );
            return Ok(Response::from_parts(resp_parts, passthrough_body));
        }
        CollectedResponseBody::TooLarge {
            passthrough_body, ..
        } => {
            // Body exceeded MAX_CACHEABLE_RESPONSE_BODY — stream through.
            lock.release().await;
            tracing::debug!(
                idempotency.key = %idempotency_key,
                limit_bytes = MAX_CACHEABLE_RESPONSE_BODY,
                "Response body exceeded cache limit; streaming through without caching"
            );
            return Ok(Response::from_parts(resp_parts, passthrough_body));
        }
        CollectedResponseBody::Cacheable(bytes) => bytes,
    };

    // The handler stored the record in its own transaction, with its mutation.
    let committed_in_tx = probe.committed(&store).await;
    let (replay_metadata, cache_committed_error) = take_replay_markers(&mut resp_parts.extensions);

    // Cache successful 2xx/3xx responses and explicit "mutation committed"
    // errors; store before unlocking so concurrent duplicates still see a
    // locked key rather than racing to re-execute the handler.
    let status = resp_parts.status.as_u16();
    if (200u32..400).contains(&u32::from(status)) || cache_committed_error || committed_in_tx {
        let session_mutated = if let Some(session) = &session {
            session.has_pending_changes().await
        } else {
            false
        };
        let record =
            cacheable_response_record(status, &resp_parts.headers, &resp_bytes, replay_metadata);
        if session_mutated {
            tracing::debug!(
                idempotency.key = %idempotency_key,
                "Session changed during idempotent request; deferring cache write until SessionLayer finalizes Set-Cookie"
            );
            if committed_in_tx {
                // The committed record has no final Set-Cookie. Keep it hidden
                // until the rewrite ends, even across a crash.
                lock.hold_until_record_expires().await;
            }
            resp_parts.extensions.insert(DeferredIdempotencyCommit::new(
                DeferredIdempotencyState {
                    store,
                    storage_key,
                    key_context,
                    alias_storage_keys: Vec::new(),
                    locked_alias_keys: Vec::new(),
                    primary_replay_after_guard_denial: false,
                    idempotency_key,
                    record,
                    body_hash,
                    ttl,
                    lock,
                },
            ));
            metrics.inspect(|m| m.record_idempotency_miss());
            return Ok(Response::from_parts(resp_parts, Body::from(resp_bytes)));
        }
        if committed_in_tx {
            // Nothing to write: the record committed with the mutation.
        } else if let Err(error) = store
            .set(&storage_key, &lock.owner, record, body_hash, ttl)
            .await
            .and_then(stored)
        {
            tracing::error!(
                idempotency.key = %idempotency_key,
                error = %error,
                "Idempotency persistence failed after handler success; failing closed"
            );
            lock.hold_until_ttl();
            return Ok(persistence_failed_response());
        }
    }
    lock.release().await;

    metrics.inspect(|m| m.record_idempotency_miss());

    // Reconstruct from original parts — preserves set-cookie and extensions.
    Ok(Response::from_parts(resp_parts, Body::from(resp_bytes)))
}

/// Collect response body bytes up to `MAX_CACHEABLE_RESPONSE_BODY`.
///
/// Returns:
/// - `Cacheable(bytes)` — body is within the limit and fully collected
/// - `TooLarge(body)` — body exceeded the limit; the returned `Body` chains the
///   already-read bytes with the remaining stream for pass-through delivery
/// - `StreamError(body)` — reading the body stream failed; the returned `Body`
///   preserves the already-read bytes and then yields the original stream error
enum CollectedResponseBody {
    Cacheable(Bytes),
    TooLarge {
        passthrough_body: Body,
        #[cfg_attr(not(test), allow(dead_code))]
        buffered_len: usize,
    },
    StreamError(Body),
}

async fn collect_response_for_cache(body: Body) -> CollectedResponseBody {
    collect_response_for_cache_with_limit(body, MAX_CACHEABLE_RESPONSE_BODY).await
}

async fn collect_response_for_cache_with_limit(body: Body, limit: usize) -> CollectedResponseBody {
    let mut buf = Vec::<u8>::new();
    let mut data_stream = body.into_data_stream();
    loop {
        match data_stream.next().await {
            None => break,
            Some(Err(err)) => {
                let leading = Bytes::from(buf);
                let passthrough =
                    Body::from_stream(futures::stream::iter(vec![Ok(leading), Err(err)]));
                return CollectedResponseBody::StreamError(passthrough);
            }
            Some(Ok(chunk)) => {
                if chunk.len() > limit.saturating_sub(buf.len()) {
                    let buffered_len = buf.len();
                    let mut leading_chunks = Vec::with_capacity(2);
                    if !buf.is_empty() {
                        leading_chunks.push(Ok::<Bytes, axum::Error>(Bytes::from(buf)));
                    }
                    leading_chunks.push(Ok::<Bytes, axum::Error>(chunk));
                    let passthrough =
                        Body::from_stream(futures::stream::iter(leading_chunks).chain(data_stream));
                    return CollectedResponseBody::TooLarge {
                        passthrough_body: passthrough,
                        buffered_len,
                    };
                }
                buf.extend_from_slice(&chunk);
            }
        }
    }
    CollectedResponseBody::Cacheable(Bytes::from(buf))
}

async fn replay_cache_hit<S>(
    inner: &mut S,
    entry: IdempotencyEntry,
    prepared: PreparedIdempotencyRequest,
    metrics: Option<&crate::middleware::MetricsCollector>,
    replay_through_inner: bool,
    fail_closed_on_replay: bool,
) -> Result<Response<Body>, std::convert::Infallible>
where
    S: Service<Request<Body>, Response = Response<Body>, Error = std::convert::Infallible>
        + Send
        + 'static,
    S::Future: Send + 'static,
{
    let PreparedIdempotencyRequest {
        idempotency_key,
        body_hash,
        mut parts,
        body_bytes,
        ..
    } = prepared;

    if entry.body_hash != body_hash {
        tracing::debug!(
            idempotency.key = %idempotency_key,
            "Idempotency payload mismatch — returning 422"
        );
        #[allow(
            clippy::unwrap_used,
            reason = "infallible: response built from static status/body"
        )]
        let response = Response::builder()
            .status(StatusCode::UNPROCESSABLE_ENTITY)
            .body(Body::from("idempotency key reused with different payload"))
            .unwrap();
        return Ok(response);
    }

    if fail_closed_on_replay {
        tracing::warn!(
            idempotency.key = %idempotency_key,
            "Idempotency cache hit reached a route without an inner replay stop; failing closed"
        );
        return Ok(replay_requires_inner_stop_response());
    }

    tracing::debug!(
        idempotency.key = %idempotency_key,
        idempotency.replayed = true,
        "Idempotency cache hit — replaying stored response"
    );

    if let Some(m) = metrics {
        m.record_idempotency_hit();
    }

    let replay = IdempotencyReplayResponse {
        record: entry.record,
    };
    if replay_through_inner {
        parts.extensions.insert(replay);
        return inner
            .call(Request::from_parts(parts, Body::from(body_bytes)))
            .await;
    }

    Ok(replay.into_response())
}

#[allow(
    clippy::unwrap_used,
    reason = "infallible: response built from static status/body"
)]
fn corrupted_replay_record_response() -> Response<Body> {
    Response::builder()
        .status(StatusCode::INTERNAL_SERVER_ERROR)
        .body(Body::from(
            "stored idempotency replay record is invalid or corrupted",
        ))
        .unwrap()
}

fn response_from_record(record: IdempotencyRecord) -> Response<Body> {
    let mut builder = Response::builder().status(record.status);
    for (name, value) in &record.headers {
        builder = builder.header(name.as_str(), value.as_slice());
    }
    // The status, header names/values, and body all originate from the
    // idempotency store, which may be a custom or corrupted backend. An
    // invalid stored status (e.g. `0` or `> 999`) or invalid header bytes
    // makes the builder stash an error that surfaces here. A corrupted replay
    // record must not crash request handling: fall back to an internal-error
    // response instead of panicking.
    match builder
        .header(X_IDEMPOTENT_REPLAYED, "true")
        .body(Body::from(record.body))
    {
        Ok(response) => response,
        Err(error) => {
            tracing::error!(
                error = %error,
                "Stored idempotency replay record produced an invalid response; \
                 returning 500 instead of replaying"
            );
            corrupted_replay_record_response()
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::http::header::AUTHORIZATION;
    use std::collections::HashMap;
    use std::convert::Infallible;
    use std::sync::Mutex;
    use tower::ServiceExt;

    /// A late `set` from an owner whose lock another owner took writes
    /// nothing; the holder's own `set` is written.
    #[test]
    fn memory_set_skips_while_another_owner_holds_the_key() {
        let store = MemoryIdempotencyStore::new(Duration::from_secs(60));
        let record = |status| IdempotencyRecord {
            status,
            headers: Vec::new(),
            body: Vec::new(),
            metadata: Vec::new(),
        };
        assert!(store.try_lock_now("k", "b", Duration::from_secs(60)));
        store.set_now("k", "a", record(500), Vec::new(), Duration::from_secs(60));
        assert!(store.get_now("k").is_none(), "a does not hold the key");
        store.set_now("k", "b", record(201), Vec::new(), Duration::from_secs(60));
        assert_eq!(store.get_now("k").expect("b's record").record.status, 201);

        // After b releases, a's late write still cannot replace b's entry.
        store.unlock_now("k", "b");
        store.set_now("k", "a", record(500), Vec::new(), Duration::from_secs(60));
        assert_eq!(store.get_now("k").expect("b's record").record.status, 201);
    }

    /// W3 (issue #1797): the in-flight lock owner id is minted from the injected
    /// entropy source, so a fixed seed reproduces the exact lock-owner stream.
    #[test]
    fn in_flight_lock_owner_is_deterministic_under_seeded_entropy() {
        use crate::entropy::SeededEntropy;

        let a = SeededEntropy::new(0x5eed);
        let b = SeededEntropy::new(0x5eed);
        for _ in 0..5 {
            assert_eq!(
                in_flight_lock_owner(&a),
                in_flight_lock_owner(&b),
                "same seed ⇒ identical lock-owner stream"
            );
        }
        // The owner is a well-formed v4 UUID string.
        let owner = in_flight_lock_owner(&SeededEntropy::new(1));
        assert!(uuid::Uuid::parse_str(&owner).is_ok());
        // A different seed diverges.
        assert_ne!(
            in_flight_lock_owner(&SeededEntropy::new(1)),
            in_flight_lock_owner(&SeededEntropy::new(2)),
        );
    }

    #[derive(Clone, Default)]
    struct RecordingStore {
        keys: Arc<Mutex<Vec<String>>>,
    }

    impl RecordingStore {
        fn keys(&self) -> Vec<String> {
            self.keys
                .lock()
                .expect("recording store lock poisoned")
                .clone()
        }

        fn record_key(&self, key: &str) {
            self.keys
                .lock()
                .expect("recording store lock poisoned")
                .push(key.to_owned());
        }
    }

    impl IdempotencyStore for RecordingStore {
        fn get<'a>(&'a self, key: &'a str) -> IdempotencyFuture<'a, Option<IdempotencyEntry>> {
            self.record_key(key);
            ready(None)
        }

        fn set<'a>(
            &'a self,
            key: &'a str,
            _owner: &'a str,
            _record: IdempotencyRecord,
            _body_hash: Vec<u8>,
            _ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            self.record_key(key);
            ready(true)
        }

        fn try_lock<'a>(
            &'a self,
            key: &'a str,
            _owner: &'a str,
            _lock_ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            self.record_key(key);
            ready(true)
        }

        fn unlock<'a>(&'a self, key: &'a str, _owner: &'a str) -> IdempotencyFuture<'a, ()> {
            self.record_key(key);
            ready(())
        }
    }

    #[test]
    fn poisoned_lock_does_not_break_subsequent_requests() {
        use std::panic::{AssertUnwindSafe, catch_unwind};

        let store = MemoryIdempotencyStore::new(Duration::from_secs(60));

        // Simulate a request handler that panics while holding one of the
        // store's internal locks, poisoning it. Without poison recovery this
        // would make every later request that touches `entries` panic too.
        let result = catch_unwind(AssertUnwindSafe(|| {
            let _guard = store
                .entries
                .write()
                .expect("first acquisition is not poisoned");
            panic!("simulated handler panic while holding the idempotency lock");
        }));
        assert!(result.is_err(), "the induced panic should have unwound");
        assert!(
            store.entries.is_poisoned(),
            "holding the write lock across a panic must poison it",
        );

        // A subsequent normal store round-trip must still succeed, proving the
        // `unwrap_or_else(PoisonError::into_inner)` recovery keeps shared state
        // usable for later requests (AC4 lock-poisoning recovery contract).
        let record = IdempotencyRecord {
            status: 200,
            headers: Vec::new(),
            body: b"ok".to_vec(),
            metadata: Vec::new(),
        };
        store.set_now(
            "k",
            "",
            record,
            b"body-hash".to_vec(),
            Duration::from_secs(60),
        );
        let fetched = store.get_now("k");
        assert!(
            fetched.is_some(),
            "store must remain usable after a poisoned lock (into_inner recovery)",
        );
        assert_eq!(
            fetched.expect("entry present after recovery").record.body,
            b"ok",
        );
    }

    fn idempotent_post(path: &str, key: &str, body: &'static str) -> Request<Body> {
        Request::builder()
            .method(Method::POST)
            .uri(path)
            .header(IDEMPOTENCY_KEY_HEADER, key)
            .body(Body::from(body))
            .expect("request builder should be valid")
    }

    fn session_with_user(session_id: &str, user_id: &str) -> crate::session::Session {
        let mut data = HashMap::new();
        data.insert("user_id".to_owned(), user_id.to_owned());
        crate::session::Session::new_for_test(session_id.to_owned(), data)
    }

    fn expected_principal_digest(session_id: Option<&str>) -> String {
        use sha2::Digest as _;
        let mut hasher = sha2::Sha256::new();
        hasher.update(b"authorization:");
        hasher.update(b"\nsession:");
        if let Some(session_id) = session_id {
            hasher.update(session_id.as_bytes());
        }
        hex::encode(hasher.finalize())
    }

    fn expected_storage_key(
        method: &str,
        path: &str,
        session_id: Option<&str>,
        idempotency_key: &str,
    ) -> String {
        use sha2::Digest as _;
        let principal = expected_principal_digest(session_id);
        let mut hasher = sha2::Sha256::new();
        push_storage_key_component(&mut hasher, "method", method.as_bytes());
        push_storage_key_component(&mut hasher, "target", path.as_bytes());
        push_storage_key_component(&mut hasher, "scope-header-count", b"0");
        push_storage_key_component(&mut hasher, "principal", principal.as_bytes());
        push_storage_key_component(&mut hasher, "idempotency-key", idempotency_key.as_bytes());
        format!("v2:{}", hex::encode(hasher.finalize()))
    }

    /// The key a request with a raw `Authorization` header and no session
    /// gets: the session-less key plus the header's digest.
    fn expected_authorization_storage_key(
        method: &str,
        path: &str,
        authorization: &str,
        idempotency_key: &str,
    ) -> String {
        use sha2::Digest as _;
        let principal = expected_principal_digest(None);
        let digest = hex::encode(sha2::Sha256::digest(authorization.as_bytes()));
        let mut hasher = sha2::Sha256::new();
        push_storage_key_component(&mut hasher, "method", method.as_bytes());
        push_storage_key_component(&mut hasher, "target", path.as_bytes());
        push_storage_key_component(&mut hasher, "scope-header-count", b"0");
        push_storage_key_component(&mut hasher, "principal", principal.as_bytes());
        push_storage_key_component(&mut hasher, "authorization", digest.as_bytes());
        push_storage_key_component(&mut hasher, "idempotency-key", idempotency_key.as_bytes());
        format!("v2:{}", hex::encode(hasher.finalize()))
    }

    fn header_scope(authorization: &str) -> BearerScope {
        use sha2::Digest as _;
        BearerScope::Header(hex::encode(sha2::Sha256::digest(authorization.as_bytes())))
    }

    /// Two token clients with no session must not share a cache slot. Before
    /// the bearer component existed both computed the session-less key, so
    /// client B replayed client A's stored response.
    #[test]
    fn storage_key_partitions_by_authorization_header() {
        let a = header_scope("Bearer token-a");
        let b = header_scope("Bearer token-b");
        let key_a = build_storage_key("pay-once", "POST", "/payments", None, None, Some(&a));
        let key_b = build_storage_key("pay-once", "POST", "/payments", None, None, Some(&b));
        let anonymous = build_storage_key("pay-once", "POST", "/payments", None, None, None);

        assert_ne!(
            key_a, key_b,
            "two bearer clients must not share a storage key"
        );
        assert_ne!(
            key_a, anonymous,
            "a bearer client must not share the anonymous slot"
        );
        assert_eq!(
            key_a,
            expected_authorization_storage_key("POST", "/payments", "Bearer token-a", "pay-once")
        );
    }

    /// A cookie-backed session already names the principal, so a raw header
    /// does not split it: session clients keep the keys they had.
    #[test]
    fn storage_key_ignores_raw_authorization_when_a_session_scopes_it() {
        let a = header_scope("Bearer token-a");
        assert_eq!(
            build_storage_key("pay-once", "POST", "/payments", Some("sid"), None, Some(&a)),
            build_storage_key("pay-once", "POST", "/payments", Some("sid"), None, None),
        );
    }

    /// A verified principal scopes the key with or without a session, and two
    /// verified principals never share a slot.
    #[test]
    fn storage_key_partitions_by_verified_principal() {
        let a = BearerScope::Verified("customer-a".to_owned());
        let b = BearerScope::Verified("customer-b".to_owned());
        for session in [None, Some("sid")] {
            let key_a = build_storage_key("pay-once", "POST", "/payments", session, None, Some(&a));
            let key_b = build_storage_key("pay-once", "POST", "/payments", session, None, Some(&b));
            let unscoped = build_storage_key("pay-once", "POST", "/payments", session, None, None);
            assert_ne!(key_a, key_b, "session {session:?}");
            assert_ne!(key_a, unscoped, "session {session:?}");
        }
    }

    /// A verified principal and a header digest that happen to share bytes
    /// land under different labels, so neither can spell the other's slot.
    #[test]
    fn storage_key_bearer_components_are_labelled() {
        let verified = BearerScope::Verified("same".to_owned());
        let header = BearerScope::Header("same".to_owned());
        assert_ne!(
            build_storage_key("k", "POST", "/p", None, None, Some(&verified)),
            build_storage_key("k", "POST", "/p", None, None, Some(&header)),
        );
    }

    /// The verified principal wins over the raw header, so a client that
    /// rotates its token keeps replaying its own response.
    #[test]
    fn bearer_scope_prefers_the_verified_principal() {
        let (mut parts, ()) = Request::builder()
            .method(Method::POST)
            .uri("/payments")
            .header(axum::http::header::AUTHORIZATION, "Bearer rotated-token")
            .body(())
            .expect("request builder should be valid")
            .into_parts();
        assert_eq!(
            bearer_scope_for_parts(&parts),
            Some(header_scope("Bearer rotated-token"))
        );
        parts
            .extensions
            .insert(crate::auth::ApiTokenPrincipal("customer-a".to_owned()));
        assert_eq!(
            bearer_scope_for_parts(&parts),
            Some(BearerScope::Verified("customer-a".to_owned()))
        );
    }

    #[test]
    fn bearer_scope_is_none_without_a_credential() {
        let (parts, ()) = Request::builder()
            .method(Method::POST)
            .uri("/payments")
            .body(())
            .expect("request builder should be valid")
            .into_parts();
        assert_eq!(bearer_scope_for_parts(&parts), None);
    }

    /// An app that does not use tenancy must keep byte-identical storage keys,
    /// so upgrading cannot turn an in-flight client retry into a second
    /// execution of an already-committed mutation.
    #[test]
    fn storage_key_without_a_resolved_tenant_is_unchanged() {
        assert_eq!(
            build_storage_key("pay-once", "POST", "/payments", None, None, None),
            expected_storage_key("POST", "/payments", None, "pay-once")
        );
    }

    /// Two tenants sharing a key, a target and (for a token-authenticated API)
    /// an empty session scope must not share a cache slot.
    #[test]
    fn storage_key_partitions_by_resolved_tenant() {
        let tenant_a = build_storage_key(
            "pay-once",
            "POST",
            "/payments",
            None,
            Some("tenant-a"),
            None,
        );
        let tenant_b = build_storage_key(
            "pay-once",
            "POST",
            "/payments",
            None,
            Some("tenant-b"),
            None,
        );
        let untenanted = build_storage_key("pay-once", "POST", "/payments", None, None, None);

        assert_ne!(tenant_a, tenant_b, "tenants must not share a storage key");
        assert_ne!(tenant_a, untenanted);
        assert_ne!(tenant_b, untenanted);
    }

    /// The tenant component is length-delimited like every other one, so a
    /// tenant id carrying the separator cannot spell another tenant's slot.
    #[test]
    fn storage_key_tenant_component_is_length_delimited() {
        let split = build_storage_key("pay-once", "POST", "/payments", None, Some("a:b"), None);
        let shifted = build_storage_key("pay-once:a", "POST", "/payments", None, Some("b"), None);
        assert_ne!(split, shifted);
    }

    #[test]
    fn idempotency_context_clones_share_mutation_discriminator_sequence() {
        let context = IdempotencyContext::new("client-key".to_owned(), "scoped-key".to_owned());
        let cloned = context.clone();

        assert_eq!(context.next_mutation_discriminator(), "0");
        assert_eq!(cloned.next_mutation_discriminator(), "1");
        assert_eq!(context.next_mutation_discriminator(), "2");
    }

    #[test]
    fn memory_lock_unlock_owned_does_not_release_newer_owner() {
        let store = MemoryIdempotencyStore::new(Duration::from_secs(60));

        assert!(store.try_lock_now("key", "owner-a", Duration::from_millis(5)));
        std::thread::sleep(Duration::from_millis(20));
        assert!(store.try_lock_now("key", "owner-b", Duration::from_secs(60)));

        store.unlock_now("key", "owner-a");
        assert!(
            !store.try_lock_now("key", "owner-c", Duration::from_secs(60)),
            "stale owners must not release a newer in-flight lock"
        );

        store.unlock_now("key", "owner-b");
        assert!(store.try_lock_now("key", "owner-c", Duration::from_secs(60)));
    }

    #[test]
    fn in_flight_lock_held_when_dropped_without_release() {
        // Simulates the inner handler future being cancelled (by the outer
        // request-timeout layer) or unwound by a panic: the lock is dropped
        // without any of the explicit completion paths calling `release`.
        // The lock must stay held so a retry carrying the same Idempotency-Key
        // cannot re-run a mutation whose side effect may already have committed.
        let memory = Arc::new(MemoryIdempotencyStore::new(Duration::from_secs(60)));
        assert!(memory.try_lock_now("key", "owner-a", Duration::from_secs(60)));

        drop(InFlightLock {
            store: memory.clone(),
            key: "key".to_owned(),
            owner: "owner-a".to_owned(),
            in_flight_ttl: Duration::from_secs(60),
        });

        assert!(
            !memory.try_lock_now("key", "owner-b", Duration::from_secs(60)),
            "a cancelled/panicked handler must leave the in-flight lock held until its TTL"
        );
    }

    #[tokio::test]
    async fn in_flight_lock_release_frees_the_key() {
        // The normal completion paths call `release`, which must free the key
        // at once so a subsequent distinct request can proceed.
        let memory = Arc::new(MemoryIdempotencyStore::new(Duration::from_secs(60)));
        assert!(memory.try_lock_now("key", "owner-a", Duration::from_secs(60)));

        InFlightLock {
            store: memory.clone(),
            key: "key".to_owned(),
            owner: "owner-a".to_owned(),
            in_flight_ttl: Duration::from_secs(60),
        }
        .release()
        .await;

        assert!(
            memory.try_lock_now("key", "owner-b", Duration::from_secs(60)),
            "a released lock must free the key"
        );
    }

    #[test]
    fn set_with_extreme_ttl_does_not_panic() {
        // Regression: `Instant::now() + ttl` panics when the sum is not
        // representable. A pathological TTL (configured or attacker-influenced)
        // must clamp rather than crash the process. See saturating_deadline.
        let store = MemoryIdempotencyStore::new(Duration::from_secs(60));
        let record = IdempotencyRecord {
            status: 200,
            headers: Vec::new(),
            body: Vec::new(),
            metadata: Vec::new(),
        };

        // Both of these overflow `Instant + Duration` on the underlying clock.
        for extreme in [Duration::from_secs(u64::MAX), Duration::MAX] {
            store.set_now("extreme-ttl-key", "", record.clone(), Vec::new(), extreme);
            // Entry must be retrievable and (far-future) unexpired.
            assert!(
                store.get_now("extreme-ttl-key").is_some(),
                "entry stored with an extreme TTL should be present and unexpired"
            );
        }

        // The in-flight lock path uses the same arithmetic and must not panic.
        assert!(store.try_lock_now("lock-key", "owner", Duration::MAX));

        // saturating_deadline itself clamps to a representable far future.
        let deadline = saturating_deadline(Duration::MAX);
        assert!(deadline > Instant::now());
    }

    #[tokio::test]
    async fn response_body_stream_errors_are_not_replaced_with_empty_success() {
        let store = Arc::new(MemoryIdempotencyStore::new(Duration::from_secs(60)));
        let service = IdempotencyLayer::new(store).layer(tower::service_fn(
            |_req: Request<Body>| async move {
                let stream = futures::stream::iter(vec![
                    Ok::<Bytes, std::io::Error>(Bytes::from_static(b"partial")),
                    Err(std::io::Error::other("stream failed")),
                ]);
                Ok::<_, Infallible>(
                    Response::builder()
                        .status(StatusCode::OK)
                        .body(Body::from_stream(stream))
                        .expect("response builder should be valid"),
                )
            },
        ));

        let response = service
            .oneshot(idempotent_post("/stream", "stream-key", "same"))
            .await
            .expect("request should complete");

        assert_eq!(response.status(), StatusCode::OK);
        let body = axum::body::to_bytes(response.into_body(), usize::MAX).await;
        assert!(
            body.is_err(),
            "idempotency middleware must preserve response body stream errors"
        );
    }

    #[tokio::test]
    async fn collect_response_checks_chunk_size_before_buffering_past_cap() {
        let chunk = Bytes::from(vec![b'x'; 64]);
        match collect_response_for_cache_with_limit(Body::from(chunk), 10).await {
            CollectedResponseBody::TooLarge {
                passthrough_body,
                buffered_len,
            } => {
                assert_eq!(
                    buffered_len, 0,
                    "single over-cap chunk must not be copied into the cache buffer first"
                );
                let delivered = axum::body::to_bytes(passthrough_body, usize::MAX)
                    .await
                    .expect("passthrough body should collect");
                assert_eq!(delivered.len(), 64);
            }
            CollectedResponseBody::Cacheable(_) => {
                panic!("over-cap chunk must not be considered cacheable")
            }
            CollectedResponseBody::StreamError(_) => panic!("body should not stream-error"),
        }
    }

    #[tokio::test]
    async fn cookie_session_extension_scopes_idempotency_storage_key() {
        let store = Arc::new(MemoryIdempotencyStore::new(Duration::from_secs(60)));
        let mut service = IdempotencyLayer::new(store).layer(tower::service_fn(
            |req: Request<Body>| async move {
                let session = req
                    .extensions()
                    .get::<crate::session::Session>()
                    .cloned()
                    .expect("session extension should be present");
                let user_id = session
                    .get("user_id")
                    .await
                    .expect("session should contain user_id");
                Ok::<_, Infallible>(Response::new(Body::from(user_id)))
            },
        ));

        let mut alice_req = idempotent_post("/orders", "shared-key", "same");
        alice_req
            .extensions_mut()
            .insert(session_with_user("session-alice", "alice"));
        let alice_response = service
            .ready()
            .await
            .expect("service should be ready")
            .call(alice_req)
            .await
            .expect("alice request should complete");
        let alice_body = axum::body::to_bytes(alice_response.into_body(), usize::MAX)
            .await
            .expect("alice body should collect");
        assert_eq!(alice_body, Bytes::from_static(b"alice"));

        let mut bob_req = idempotent_post("/orders", "shared-key", "same");
        bob_req
            .extensions_mut()
            .insert(session_with_user("session-bob", "bob"));
        let bob_response = service
            .ready()
            .await
            .expect("service should be ready")
            .call(bob_req)
            .await
            .expect("bob request should complete");
        assert!(
            bob_response.headers().get(X_IDEMPOTENT_REPLAYED).is_none(),
            "a different cookie-backed session must not replay another user's response"
        );
        let bob_body = axum::body::to_bytes(bob_response.into_body(), usize::MAX)
            .await
            .expect("bob body should collect");
        assert_eq!(bob_body, Bytes::from_static(b"bob"));
    }

    #[tokio::test]
    async fn storage_key_hashes_length_delimited_components() {
        let observed_store = RecordingStore::default();
        let service = IdempotencyLayer::new(Arc::new(observed_store.clone())).layer(
            tower::service_fn(|_req: Request<Body>| async {
                Ok::<_, Infallible>(Response::new(Body::from("ok")))
            }),
        );
        let request = Request::builder()
            .method(Method::POST)
            .uri("/payments")
            .header(IDEMPOTENCY_KEY_HEADER, "pay-once")
            .header(AUTHORIZATION, "Bearer stable-token")
            .body(Body::from("same"))
            .expect("request builder should be valid");

        let response = service
            .oneshot(request)
            .await
            .expect("request should complete");
        let body = axum::body::to_bytes(response.into_body(), usize::MAX)
            .await
            .expect("response body should collect");
        assert_eq!(body, Bytes::from_static(b"ok"));

        let keys = observed_store.keys();
        let storage_key = keys.first().expect("storage key should be recorded");
        assert_eq!(
            storage_key,
            &expected_authorization_storage_key(
                "POST",
                "/payments",
                "Bearer stable-token",
                "pay-once"
            )
        );
        assert!(!storage_key.contains("/payments"));
        assert!(!storage_key.contains("pay-once"));
    }

    #[tokio::test]
    async fn forwarded_request_carries_scoped_idempotency_context() {
        let observed = Arc::new(Mutex::new(None::<(String, String)>));
        let observed_context = observed.clone();
        let service = IdempotencyLayer::new(Arc::new(MemoryIdempotencyStore::new(
            Duration::from_secs(60),
        )))
        .layer(tower::service_fn(move |req: Request<Body>| {
            let observed_context = observed_context.clone();
            async move {
                let context = req
                    .extensions()
                    .get::<IdempotencyContext>()
                    .cloned()
                    .expect("idempotency context should be available to inner handlers");
                *observed_context
                    .lock()
                    .expect("observed context lock poisoned") =
                    Some((context.key().to_owned(), context.scoped_key().to_owned()));
                Ok::<_, Infallible>(Response::new(Body::from("ok")))
            }
        }));
        let request = Request::builder()
            .method(Method::POST)
            .uri("/payments")
            .header(IDEMPOTENCY_KEY_HEADER, "pay-once")
            .header(AUTHORIZATION, "Bearer stable-token")
            .body(Body::from("same"))
            .expect("request builder should be valid");

        let response = service
            .oneshot(request)
            .await
            .expect("request should complete");
        assert_eq!(response.status(), StatusCode::OK);

        let observed = observed
            .lock()
            .expect("observed context lock poisoned")
            .clone()
            .expect("inner handler should record idempotency context");
        assert_eq!(observed.0, "pay-once");
        assert_eq!(
            observed.1,
            expected_authorization_storage_key(
                "POST",
                "/payments",
                "Bearer stable-token",
                "pay-once"
            )
        );
    }

    #[test]
    fn corrupted_stored_record_does_not_panic_on_replay() {
        // A store backend (custom or corrupted) can hand back a record whose
        // stored status is out of the valid HTTP range and whose headers carry
        // invalid bytes. Replaying it must not panic — it must surface an
        // internal-error response instead.
        let corrupted = IdempotencyRecord {
            status: 1000,
            headers: vec![("inv\nalid".to_owned(), vec![0x00, 0x0a])],
            body: b"stored body".to_vec(),
            metadata: Vec::new(),
        };

        let response = IdempotencyReplayResponse { record: corrupted }.into_response();

        assert_eq!(
            response.status(),
            StatusCode::INTERNAL_SERVER_ERROR,
            "a corrupted replay record must fall back to a 500 recovery response"
        );
        assert!(
            response.headers().get(X_IDEMPOTENT_REPLAYED).is_none(),
            "the recovery response must not masquerade as a successful replay"
        );
    }

    #[test]
    fn invalid_stored_status_alone_does_not_panic_on_replay() {
        // Even with otherwise-valid headers, an out-of-range status byte must
        // not crash replay handling.
        let corrupted = IdempotencyRecord {
            status: 0,
            headers: vec![("content-type".to_owned(), b"text/plain".to_vec())],
            body: b"stored body".to_vec(),
            metadata: Vec::new(),
        };

        let response = response_from_record(corrupted);

        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
    }

    /// Issue #3061 AC3: the Redis store runs on a current-thread runtime.
    ///
    /// The store used `block_in_place`, which panics on this runtime flavour.
    /// Redis is not reachable here, so the lookup fails and the middleware
    /// fails closed with `503`. The handler does not run.
    #[cfg(feature = "redis")]
    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn redis_store_on_current_thread_runtime_fails_closed_without_panic() {
        static CALLS: AtomicU64 = AtomicU64::new(0);
        let mut config = crate::config::IdempotencyConfig::default();
        config.redis.url = Some("redis://127.0.0.1:1/".to_owned());
        let store = Arc::new(RedisIdempotencyStore::from_config(&config).expect("lazy client"));
        let app = axum::Router::new()
            .route(
                "/charge",
                axum::routing::post(|| async {
                    CALLS.fetch_add(1, Ordering::SeqCst);
                    "charged"
                }),
            )
            .layer(IdempotencyLayer::new(store));

        let response = app
            .oneshot(idempotent_post("/charge", "current-thread", ""))
            .await
            .expect("infallible");

        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(CALLS.load(Ordering::SeqCst), 0, "the handler must not run");
    }

    /// A store that refuses the lock while a response is stored, like the
    /// database store, and whose first lookup misses because a racing request
    /// stores its response and releases the key just after it.
    struct LateRecordStore {
        inner: MemoryIdempotencyStore,
        raced: std::sync::atomic::AtomicBool,
    }

    impl IdempotencyStore for LateRecordStore {
        fn get<'a>(&'a self, key: &'a str) -> IdempotencyFuture<'a, Option<IdempotencyEntry>> {
            if !self.raced.swap(true, std::sync::atomic::Ordering::SeqCst) {
                let record = IdempotencyRecord {
                    status: 201,
                    headers: Vec::new(),
                    body: b"first".to_vec(),
                    metadata: Vec::new(),
                };
                self.inner.set_now(
                    key,
                    "racer",
                    record,
                    compute_body_hash(b"same", None),
                    Duration::from_secs(60),
                );
                return Box::pin(std::future::ready(Ok(None)));
            }
            self.inner.get(key)
        }

        fn set<'a>(
            &'a self,
            key: &'a str,
            owner: &'a str,
            record: IdempotencyRecord,
            body_hash: Vec<u8>,
            ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            self.inner.set(key, owner, record, body_hash, ttl)
        }

        fn try_lock<'a>(
            &'a self,
            key: &'a str,
            owner: &'a str,
            lock_ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            if self.inner.get_now(key).is_some() {
                return Box::pin(std::future::ready(Ok(false)));
            }
            self.inner.try_lock(key, owner, lock_ttl)
        }

        fn unlock<'a>(&'a self, key: &'a str, owner: &'a str) -> IdempotencyFuture<'a, ()> {
            self.inner.unlock(key, owner)
        }
    }

    /// A lock refused because a response was stored after the lookup replays
    /// that response; the handler does not run and the client gets no `409`.
    #[tokio::test]
    async fn lock_refused_by_a_late_record_replays_it() {
        static CALLS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
        let store = Arc::new(LateRecordStore {
            inner: MemoryIdempotencyStore::new(Duration::from_secs(60)),
            raced: std::sync::atomic::AtomicBool::new(false),
        });
        let service = IdempotencyLayer::new(store).layer(tower::service_fn(
            |_req: Request<Body>| async move {
                CALLS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                Ok::<_, Infallible>(Response::new(Body::from("second")))
            },
        ));

        let response = service
            .oneshot(idempotent_post("/late", "late-key", "same"))
            .await
            .expect("infallible");
        assert_eq!(response.status(), StatusCode::CREATED, "replayed, not 409");
        let body = axum::body::to_bytes(response.into_body(), usize::MAX)
            .await
            .expect("body");
        assert_eq!(&body[..], b"first");
        assert_eq!(
            CALLS.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "the handler did not run"
        );
    }

    /// A store whose lookup after the lock outlives the lock's TTL. With
    /// `steal`, a retry takes the lapsed lock meanwhile.
    struct SlowSecondLookupStore {
        inner: MemoryIdempotencyStore,
        lookups: std::sync::atomic::AtomicUsize,
        steal: bool,
    }

    impl SlowSecondLookupStore {
        fn new(steal: bool) -> Self {
            Self {
                inner: MemoryIdempotencyStore::new(Duration::from_secs(60)),
                lookups: std::sync::atomic::AtomicUsize::new(0),
                steal,
            }
        }
    }

    impl IdempotencyStore for SlowSecondLookupStore {
        fn get<'a>(&'a self, key: &'a str) -> IdempotencyFuture<'a, Option<IdempotencyEntry>> {
            let lookup = self
                .lookups
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Box::pin(async move {
                if lookup == 1 {
                    tokio::time::sleep(Duration::from_millis(150)).await;
                    if self.steal {
                        assert!(
                            self.inner
                                .try_lock_now(key, "retry", Duration::from_secs(60)),
                            "the first lock lapsed"
                        );
                    }
                }
                Ok(self.inner.get_now(key))
            })
        }

        fn set<'a>(
            &'a self,
            key: &'a str,
            owner: &'a str,
            record: IdempotencyRecord,
            body_hash: Vec<u8>,
            ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            self.inner.set(key, owner, record, body_hash, ttl)
        }

        fn try_lock<'a>(
            &'a self,
            key: &'a str,
            owner: &'a str,
            lock_ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            self.inner.try_lock(key, owner, lock_ttl)
        }

        fn unlock<'a>(&'a self, key: &'a str, owner: &'a str) -> IdempotencyFuture<'a, ()> {
            self.inner.unlock(key, owner)
        }

        fn renew_lock<'a>(
            &'a self,
            key: &'a str,
            owner: &'a str,
            lock_ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            self.inner.renew_lock(key, owner, lock_ttl)
        }
    }

    async fn run_after_slow_post_lock_lookup(
        store: Arc<SlowSecondLookupStore>,
        calls: &'static std::sync::atomic::AtomicUsize,
    ) -> StatusCode {
        let service = IdempotencyLayer::new(store)
            .with_in_flight_ttl(Duration::from_millis(50))
            .layer(tower::service_fn(move |_req: Request<Body>| async move {
                calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                Ok::<_, Infallible>(Response::new(Body::from("ran")))
            }));
        service
            .oneshot(idempotent_post("/slow", "slow-key", ""))
            .await
            .expect("infallible")
            .status()
    }

    /// A lock that lapsed to a retry during the lookup after it was taken
    /// does not run the handler: the request gets `409`, and only the retry
    /// that holds the key runs.
    #[tokio::test]
    async fn lock_lapsed_during_the_post_lock_lookup_does_not_run_the_handler() {
        static CALLS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
        let store = Arc::new(SlowSecondLookupStore::new(true));
        let status = run_after_slow_post_lock_lookup(Arc::clone(&store), &CALLS).await;
        assert_eq!(status, StatusCode::CONFLICT);
        assert_eq!(
            CALLS.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "the handler must not run on a lock the retry now holds"
        );
        assert!(
            store
                .inner
                .in_flight
                .read()
                .expect("lock")
                .values()
                .all(|lock| lock.owner == "retry"),
            "the retry keeps its lock"
        );
    }

    /// A slow lookup after the lock, with no retry in between, renews the lock
    /// and runs the handler on a full TTL.
    #[tokio::test]
    async fn slow_post_lock_lookup_renews_the_lock_and_runs_the_handler() {
        static CALLS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
        let store = Arc::new(SlowSecondLookupStore::new(false));
        let status = run_after_slow_post_lock_lookup(store, &CALLS).await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(CALLS.load(std::sync::atomic::Ordering::SeqCst), 1);
    }

    /// The deferred session commit fails closed when another owner's response
    /// fences out a session alias: the response must not go out as stored
    /// while a retry under that alias would replay the other response.
    #[tokio::test]
    async fn deferred_alias_fenced_by_another_owner_fails_closed() {
        let memory = Arc::new(MemoryIdempotencyStore::new(Duration::from_secs(60)));
        let earlier = IdempotencyRecord {
            status: 200,
            headers: Vec::new(),
            body: b"earlier".to_vec(),
            metadata: Vec::new(),
        };
        memory.set_now(
            "alias",
            "earlier",
            earlier,
            Vec::new(),
            Duration::from_secs(60),
        );
        assert!(memory.try_lock_now("primary", "me", Duration::from_secs(60)));
        let store: Arc<dyn IdempotencyStore> = memory.clone();
        let commit = DeferredIdempotencyCommit::new(DeferredIdempotencyState {
            store: Arc::clone(&store),
            storage_key: "primary".to_owned(),
            key_context: StorageKeyContext {
                idempotency_key: "k".to_owned(),
                method: Method::POST,
                target: "/switch".to_owned(),
                tenant: None,
                bearer: None,
            },
            alias_storage_keys: vec!["alias".to_owned()],
            locked_alias_keys: Vec::new(),
            primary_replay_after_guard_denial: false,
            idempotency_key: "k".to_owned(),
            record: IdempotencyRecord {
                status: 200,
                headers: Vec::new(),
                body: b"this one".to_vec(),
                metadata: Vec::new(),
            },
            body_hash: Vec::new(),
            ttl: Duration::from_secs(60),
            lock: InFlightLock {
                store,
                key: "primary".to_owned(),
                owner: "me".to_owned(),
                in_flight_ttl: Duration::from_secs(60),
            },
        });

        let result = commit.commit_with_final_headers(&HeaderMap::new()).await;
        assert!(result.is_err(), "a fenced alias write must fail closed");
        assert_eq!(
            memory.get_now("alias").expect("alias").record.body,
            b"earlier",
            "the other owner's response stays"
        );
        assert!(
            !memory.try_lock_now("primary", "retry", Duration::from_secs(60)),
            "the primary key stays locked until its TTL"
        );
    }

    fn alias_commit(memory: &Arc<MemoryIdempotencyStore>) -> (DeferredIdempotencyCommit, String) {
        assert!(memory.try_lock_now("primary", "me", Duration::from_secs(60)));
        let store: Arc<dyn IdempotencyStore> = memory.clone();
        let key_context = StorageKeyContext {
            idempotency_key: "k".to_owned(),
            method: Method::POST,
            target: "/switch".to_owned(),
            tenant: None,
            bearer: None,
        };
        let alias = key_context.storage_key(Some("sid"), None);
        let commit = DeferredIdempotencyCommit::new(DeferredIdempotencyState {
            store: Arc::clone(&store),
            storage_key: "primary".to_owned(),
            key_context,
            alias_storage_keys: Vec::new(),
            locked_alias_keys: Vec::new(),
            primary_replay_after_guard_denial: false,
            idempotency_key: "k".to_owned(),
            record: IdempotencyRecord {
                status: 200,
                headers: Vec::new(),
                body: b"ok".to_vec(),
                metadata: Vec::new(),
            },
            body_hash: Vec::new(),
            ttl: Duration::from_secs(60),
            lock: InFlightLock {
                store,
                key: "primary".to_owned(),
                owner: "me".to_owned(),
                in_flight_ttl: Duration::from_secs(60),
            },
        });
        (commit, alias)
    }

    /// The alias is locked from reservation, and the commit frees it.
    #[tokio::test]
    async fn reserved_alias_is_locked_until_commit() {
        let memory = Arc::new(MemoryIdempotencyStore::new(Duration::from_secs(60)));
        let (commit, alias) = alias_commit(&memory);

        commit
            .reserve_session_alias(Some("sid"), false, None)
            .await
            .expect("reserve");
        assert!(
            !memory.try_lock_now(&alias, "other", Duration::from_secs(60)),
            "the alias is in flight before the commit"
        );

        commit
            .commit_with_final_headers(&HeaderMap::new())
            .await
            .expect("commit");
        assert!(memory.get_now(&alias).is_some(), "the alias has a response");
        assert!(
            memory.try_lock_now(&alias, "other", Duration::from_secs(60)),
            "the commit frees the alias lock"
        );
        assert!(
            memory.try_lock_now("primary", "other", Duration::from_secs(60)),
            "the commit frees the primary lock"
        );
    }

    /// Another owner's lock on the alias fails the reservation.
    #[tokio::test]
    async fn reserve_fails_when_another_owner_holds_the_alias() {
        let memory = Arc::new(MemoryIdempotencyStore::new(Duration::from_secs(60)));
        let (commit, alias) = alias_commit(&memory);
        assert!(memory.try_lock_now(&alias, "other", Duration::from_secs(60)));

        let result = commit.reserve_session_alias(Some("sid"), false, None).await;

        assert!(result.is_err(), "a held alias must fail closed");
    }

    /// A reservation records a primary replay after guard denial.
    #[tokio::test]
    async fn reserve_records_primary_replay_after_guard_denial() {
        let memory = Arc::new(MemoryIdempotencyStore::new(Duration::from_secs(60)));
        let (commit, _alias) = alias_commit(&memory);

        commit
            .reserve_session_alias(Some("sid"), true, None)
            .await
            .expect("reserve");

        let flag = commit
            .inner
            .lock()
            .expect("lock")
            .as_ref()
            .is_some_and(|state| state.primary_replay_after_guard_denial);
        assert!(flag, "the primary replay flag is set");
    }

    /// A failed commit keeps the alias locked until its TTL.
    #[tokio::test]
    async fn failed_commit_keeps_the_alias_locked() {
        let memory = Arc::new(MemoryIdempotencyStore::new(Duration::from_secs(60)));
        let (commit, alias) = alias_commit(&memory);
        commit
            .reserve_session_alias(Some("sid"), false, None)
            .await
            .expect("reserve");
        // A response from another owner fences the primary write out.
        memory.unlock_now("primary", "me");
        memory.set_now(
            "primary",
            "earlier",
            IdempotencyRecord {
                status: 200,
                headers: Vec::new(),
                body: b"earlier".to_vec(),
                metadata: Vec::new(),
            },
            Vec::new(),
            Duration::from_secs(60),
        );

        let result = commit.commit_with_final_headers(&HeaderMap::new()).await;

        assert!(result.is_err(), "a fenced write must fail closed");
        assert!(
            !memory.try_lock_now(&alias, "other", Duration::from_secs(60)),
            "the alias stays locked until its TTL"
        );
    }

    /// A store whose writes are always fenced out: `set` stores nothing and
    /// returns `false`, as when a newer request took the key.
    struct FencedSetStore {
        inner: MemoryIdempotencyStore,
    }

    impl IdempotencyStore for FencedSetStore {
        fn get<'a>(&'a self, key: &'a str) -> IdempotencyFuture<'a, Option<IdempotencyEntry>> {
            self.inner.get(key)
        }

        fn set<'a>(
            &'a self,
            _key: &'a str,
            _owner: &'a str,
            _record: IdempotencyRecord,
            _body_hash: Vec<u8>,
            _ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            Box::pin(std::future::ready(Ok(false)))
        }

        fn try_lock<'a>(
            &'a self,
            key: &'a str,
            owner: &'a str,
            lock_ttl: Duration,
        ) -> IdempotencyFuture<'a, bool> {
            self.inner.try_lock(key, owner, lock_ttl)
        }

        fn unlock<'a>(&'a self, key: &'a str, owner: &'a str) -> IdempotencyFuture<'a, ()> {
            self.inner.unlock(key, owner)
        }
    }

    /// A fenced-out write after the handler succeeded is not a stored
    /// response: the request fails closed with `503` and keeps its lock, so a
    /// retry gets `409` instead of running the mutation again.
    #[tokio::test]
    async fn fenced_out_write_after_success_fails_closed() {
        static CALLS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
        let store = Arc::new(FencedSetStore {
            inner: MemoryIdempotencyStore::new(Duration::from_secs(60)),
        });
        let service = IdempotencyLayer::new(store).layer(tower::service_fn(
            |_req: Request<Body>| async move {
                CALLS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                Ok::<_, Infallible>(Response::new(Body::from("charged")))
            },
        ));

        let first = service
            .clone()
            .oneshot(idempotent_post("/fenced", "fenced-key", ""))
            .await
            .expect("infallible");
        assert_eq!(first.status(), StatusCode::SERVICE_UNAVAILABLE);
        let retry = service
            .oneshot(idempotent_post("/fenced", "fenced-key", ""))
            .await
            .expect("infallible");
        assert_eq!(retry.status(), StatusCode::CONFLICT);
        assert_eq!(CALLS.load(std::sync::atomic::Ordering::SeqCst), 1);
    }

    /// Characterization for the Echo clone class: `build_storage_key` and
    /// `principal_scope_digest` render their digests through this file's own
    /// `hex_lower`, one of six byte-identical copies across the crate (this
    /// file's test module carries a seventh, test-local copy). Pinned before
    /// every copy is replaced by the `hex` crate (already a dependency,
    /// already used this way in ~20 other places in this crate) at its call
    /// site.
    #[test]
    fn build_storage_key_is_pinned_hex() {
        assert_eq!(
            build_storage_key("pay-once", "POST", "/payments", None, None, None),
            "v2:e776147e12affe592b79542ee766f4171dc7b7a7e50cf234424fc32fa0b3d1ce",
        );
    }
}