oauth-as 0.9.0

// SPDX-License-Identifier: MIT OR Apache-2.0
// Copyright (C) 2026 Matthew Jackson

//! A RUNNABLE conformance harness for the [`Storage`] contract, behind the `test-util` cargo
//! feature (off by default), for a HOST to run from its OWN test suite against its OWN store.
//!
//! Nothing here is re-exported at the crate root, on purpose: `Violation` and `CHECKS` are generic
//! words that only mean something next to the thing they describe, and a host names this surface
//! once, in a test.
//!
//! # Why this exists
//!
//! [`crate::store`] documents the contract in prose, and prose is the version of a contract that
//! nobody has to keep. The load-bearing clause is that `take_device_grant`,
//! `take_refresh_token` and `take_authorization_code` are ATOMIC remove-and-return. A host that
//! implements them as read-then-delete gets three failures that a single-node test suite cannot
//! see and that production will not report:
//!
//! - REFRESH TOKEN DOUBLE SPEND. Two nodes read the same record, both delete it, both rotate. The
//!   attacker and the honest client each end up holding a live chain, and because the honest
//!   client is never locked out, the one observable signal that would have revealed the theft
//!   (RFC 9700 section 4.14.2 reuse detection) never fires.
//! - AUTHORIZATION CODE REPLAY DETECTION SILENTLY DISABLED. Two nodes take the same `Issued`
//!   record, both mint, both write back `Consumed`, last write wins. The server believes the code
//!   was spent once.
//! - DEVICE GRANT DOUBLE ISSUANCE, the same shape at the RFC 8628 redemption.
//!
//! `Storage::claim_replay_id` (compiled in with `client_assertion` or `dpop`) is the same defect
//! shape and is checked the same way: RFC 7523 section 3 and RFC 9449 section 4.3 both make a
//! `jti` single use, and a read-then-write claim tells two concurrent presentations of the SAME
//! assertion that each of them was the first.
//!
//! Nothing inside this crate can detect any of that: the server calls `take_*` and is entitled to
//! believe the answer. So the check has to run where the host's store is, which is what this
//! module is for.
//!
//! # Using it
//!
//! ```toml
//! [dev-dependencies]
//! oauth-as = { version = "*", features = ["test-util"] }
//! ```
//!
//! ```no_run
//! use oauth_as::storage_conformance::StorageConformance;
//!
//! # async fn my_store() -> oauth_as::MemoryStorage { oauth_as::MemoryStorage::new() }
//! # async fn doc() {
//! // The factory MUST return a store that is EMPTY: several checks count records, and a
//! // leftover row from a previous check is indistinguishable from a store that failed to
//! // remove one.
//! let violations = StorageConformance::new(|| async { my_store().await })
//!     // Hand the racers to your own runtime. On a multi-threaded one this is what makes the
//!     // atomicity checks a real race rather than an interleaving.
//!     .with_spawn(|task| {
//!         tokio::spawn(task);
//!     })
//!     .run()
//!     .await;
//! assert!(violations.is_empty(), "{violations:#?}");
//! # }
//! ```
//!
//! It RETURNS the violations rather than panicking, so a host can report them the way it likes:
//! assert on emptiness, print them, feed them to its own reporter, or accept a documented subset.
//! Every violation names a check from [`CHECKS`] plus a human-readable detail.
//!
//! # What the concurrency checks can and cannot prove
//!
//! Read this before quoting a green run at anyone. The honest summary is that this harness proves
//! a store is atomic ACROSS AWAIT POINTS, and cannot prove it is atomic across machines.
//!
//! WHAT IT DOES. Each `take_*` check builds N racing futures. Every racer first parks on a
//! rendezvous gate and does not touch the store until all N have arrived, so the takes are all in
//! flight at once rather than being run one after another. Then:
//!
//! - With [`StorageConformance::with_spawn`], the racers are handed to the HOST'S runtime as
//!   independent tasks. On a multi-threaded runtime that is a genuine data race on real threads,
//!   which is the strongest form of this check and the one worth running in a host's CI.
//! - Without a spawner, the racers are polled concurrently on the caller's own task by a
//!   join combinator in this module. That is interleaving, not parallelism.
//!
//! WHAT THE COOPERATIVE (no spawner) MODE STILL CATCHES, and why it is not a token gesture: a
//! read-then-delete implementation over a network store awaits between the read and the delete,
//! because the read is a round trip. At that await the racer yields, the next racer is polled and
//! performs its own read, and every racer observes the value before any of them removes it. So
//! interleaving is enough to catch the real-world shape of this bug in any store whose operations
//! actually suspend.
//!
//! WHAT NEITHER MODE CAN PROVE:
//!
//! - It cannot prove a store is atomic across PROCESSES or NODES, which is the deployment where
//!   the bug bites. Two racers inside one test process share whatever in-process lock the store
//!   holds; a mutex around a read-then-delete pair will pass this harness and still double-spend
//!   from two nodes. If the store's atomicity comes from a process-local lock rather than from
//!   the DATABASE (`DELETE ... RETURNING`, a conditional update, a compare-and-set), this harness
//!   will not tell you. Run it against the store the way it is deployed, and read the query.
//! - The cooperative mode cannot catch a store that performs read-then-delete with NO suspension
//!   point in between (a purely synchronous `async fn` over an in-process map). Such a store is
//!   in practice atomic anyway, since nothing can interleave with it, but the check passing is
//!   not evidence about the shared-store implementation the host will deploy.
//! - Passing once is not passing always. A race that loses is still a race; N is 8 by default and
//!   raisable with [`StorageConformance::racers`], and a host that cares should run this
//!   repeatedly rather than once.
//! - Nothing here observes the store's isolation level, its retry behaviour, or what it does when
//!   the connection drops mid-operation.
//!
//! If the rendezvous gate cannot be satisfied (a `with_spawn` that runs tasks strictly one after
//! another to completion, so no two racers are ever in flight), the harness reports
//! `harness/race_setup` and the atomicity results in that run mean nothing. That is deliberately a
//! reported violation rather than a silent pass.
//!
//! # Cost when you do not enable it
//!
//! Nothing. `test-util` adds no dependency and no code to a default build; the whole module is
//! behind the feature.

use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use std::time::{Duration, SystemTime};

use crate::authorization::{AuthorizationCodeRecord, AuthorizationCodeState, CodeChallengeMethod};
use crate::client::{Client, ClientAuth, ClientId, DynamicRegistration, SecretHash};
use crate::device::{normalize_user_code, DeviceGrant, DeviceGrantState};
use crate::grant::GrantType;
use crate::scope::ScopeSet;
use crate::store::{Storage, StorageError};
use crate::token::{IssuedToken, RefreshTokenRecord, RefreshTokenState};

/// One way in which a store failed the [`Storage`] contract.
///
/// `check` is one of [`CHECKS`], so a host can group, filter or waive by a stable name; `detail`
/// says what was observed and what was required.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Violation {
    /// The check that failed; always a member of [`CHECKS`].
    pub check: &'static str,
    /// What went wrong, in terms of what was stored and what came back.
    pub detail: String,
}

impl fmt::Display for Violation {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}: {}", self.check, self.detail)
    }
}

/// Every check name [`StorageConformance::run`] can report, so a host can assert that a name it
/// filters on still exists rather than silently waiving a check that was renamed.
///
/// The `claim_replay_id` names are listed unconditionally even though the checks themselves only
/// run when `client_assertion` or `dpop` is compiled in: a host's waiver list should not have to
/// be feature-conditional to be valid.
pub const CHECKS: &[&str] = &[
    HARNESS_RACE_SETUP,
    HARNESS_RACER_PANICKED,
    ROUND_TRIP_CLIENT,
    ROUND_TRIP_DEVICE_GRANT,
    ROUND_TRIP_AUTHORIZATION_CODE,
    ROUND_TRIP_TOKEN,
    ROUND_TRIP_REFRESH_TOKEN,
    ATOMIC_TAKE_DEVICE_GRANT,
    SWAP_APPLIES_ON_MATCH,
    SWAP_HONOURS_EXPECTED,
    SWAP_NEVER_RESURRECTS,
    ATOMIC_TAKE_REFRESH_TOKEN,
    ATOMIC_TAKE_AUTHORIZATION_CODE,
    INDEX_RETIRES_OLD_USER_CODE,
    INDEX_REFUSES_DUPLICATE_USER_CODE,
    INDEX_REFUSAL_WRITES_NOTHING,
    INDEX_CLEARED_BY_TAKE,
    INDEX_NO_NORMALIZATION,
    SWEEP_REMOVES_DEAD,
    SWEEP_KEEPS_LIVE,
    SWEEP_COUNT,
    SWEEP_EMPTY_IS_ZERO,
    REVOKE_FAMILY_REMOVES,
    REVOKE_FAMILY_SPARES_OTHERS,
    REVOKE_FAMILY_COUNT,
    DELETE_CLIENT_CASCADES,
    DELETE_CLIENT_REPORTS,
    DELETE_TOKEN_IDEMPOTENT,
    ATOMIC_CLAIM_REPLAY_ID,
    CLAIM_REPLAY_ID_REFUSES_SECOND,
    SWEEP_RECLAIMS_REPLAY_IDS,
    ATOMIC_TAKE_PUSHED_REQUEST,
    ROUND_TRIP_PUSHED_REQUEST,
    ROUND_TRIP_CONSENT,
    REVOKE_CONSENT_CASCADES,
    REVOKE_CONSENT_SPARES_OTHERS,
    REVOKE_CONSENT_COUNT,
];

const HARNESS_RACE_SETUP: &str = "harness/race_setup";
const HARNESS_RACER_PANICKED: &str = "harness/racer_panicked";
const ROUND_TRIP_CLIENT: &str = "round_trip/client";
const ROUND_TRIP_DEVICE_GRANT: &str = "round_trip/device_grant";
const ROUND_TRIP_AUTHORIZATION_CODE: &str = "round_trip/authorization_code";
const ROUND_TRIP_TOKEN: &str = "round_trip/token";
const ROUND_TRIP_REFRESH_TOKEN: &str = "round_trip/refresh_token";
const ATOMIC_TAKE_DEVICE_GRANT: &str = "atomic_take/take_device_grant";
const SWAP_APPLIES_ON_MATCH: &str = "compare_and_swap_device_grant/applies_when_the_state_matches";
const SWAP_HONOURS_EXPECTED: &str = "compare_and_swap_device_grant/honours_expected";
const SWAP_NEVER_RESURRECTS: &str = "compare_and_swap_device_grant/never_resurrects";
const ATOMIC_TAKE_REFRESH_TOKEN: &str = "atomic_take/take_refresh_token";
const ATOMIC_TAKE_AUTHORIZATION_CODE: &str = "atomic_take/take_authorization_code";
const INDEX_RETIRES_OLD_USER_CODE: &str = "user_code_index/retires_old_entry";
const INDEX_REFUSES_DUPLICATE_USER_CODE: &str = "user_code_index/refuses_duplicate";
const INDEX_REFUSAL_WRITES_NOTHING: &str = "user_code_index/refusal_writes_nothing";
const INDEX_CLEARED_BY_TAKE: &str = "user_code_index/cleared_by_take";
const INDEX_NO_NORMALIZATION: &str = "user_code_index/store_does_not_normalize";
const SWEEP_REMOVES_DEAD: &str = "sweep_expired/removes_dead";
const SWEEP_KEEPS_LIVE: &str = "sweep_expired/keeps_live";
const SWEEP_COUNT: &str = "sweep_expired/count";
const SWEEP_EMPTY_IS_ZERO: &str = "sweep_expired/empty_is_zero";
const REVOKE_FAMILY_REMOVES: &str = "revoke_token_family/removes_the_family";
const REVOKE_FAMILY_SPARES_OTHERS: &str = "revoke_token_family/spares_other_families";
const REVOKE_FAMILY_COUNT: &str = "revoke_token_family/count";
const DELETE_CLIENT_CASCADES: &str = "delete_client/cascades";
const DELETE_CLIENT_REPORTS: &str = "delete_client/reports_whether_it_removed";
const DELETE_TOKEN_IDEMPOTENT: &str = "delete_token/idempotent";
const ATOMIC_CLAIM_REPLAY_ID: &str = "atomic_claim/claim_replay_id";
const CLAIM_REPLAY_ID_REFUSES_SECOND: &str = "claim_replay_id/refuses_a_second_claim";
const SWEEP_RECLAIMS_REPLAY_IDS: &str = "sweep_expired/reclaims_replay_ids";
const ATOMIC_TAKE_PUSHED_REQUEST: &str = "atomic_take/take_pushed_authorization_request";
const ROUND_TRIP_PUSHED_REQUEST: &str = "round_trip/pushed_authorization_request";
const ROUND_TRIP_CONSENT: &str = "round_trip/consent";
const REVOKE_CONSENT_CASCADES: &str = "revoke_consent/cascades";
const REVOKE_CONSENT_SPARES_OTHERS: &str = "revoke_consent/spares_other_subjects";
const REVOKE_CONSENT_COUNT: &str = "revoke_consent/count";

/// A racer handed to the host's runtime by [`StorageConformance::with_spawn`].
///
/// Boxed because the harness builds N of them and the host's spawner takes one concrete type;
/// `Send` because a multi-threaded runtime may move it between threads, which is exactly the
/// property that makes the spawned mode a real race.
pub type Task = Pin<Box<dyn Future<Output = ()> + Send>>;

type SpawnFn = Arc<dyn Fn(Task) + Send + Sync>;
type BoxTake<T> = Pin<Box<dyn Future<Output = Result<Option<T>, StorageError>> + Send>>;
/// What the racers hand back: one take's answer per racer, in completion order.
type TakeResults<T> = Vec<Result<Option<T>, StorageError>>;

/// The default number of racers per `take_*` check.
const DEFAULT_RACERS: usize = 8;

/// How many times a racer will re-poll while waiting at the rendezvous gate before giving up and
/// declaring the race unsatisfiable. Arrival needs no I/O (the gate is reached before the store is
/// touched), so a spawner that runs its tasks concurrently at all satisfies this in a handful of
/// polls; the budget exists only so a spawner that runs tasks strictly sequentially reports
/// `harness/race_setup` instead of hanging the host's test suite forever.
const GATE_POLL_BUDGET: u32 = 10_000;

/// The [`Storage`] conformance harness. See the module docs, particularly the honest account of
/// what the concurrency checks can and cannot prove.
pub struct StorageConformance<F> {
    new_store: F,
    spawn: Option<SpawnFn>,
    racers: usize,
}

impl<F> StorageConformance<F> {
    /// Build a harness over a factory that returns a FRESH, EMPTY store each time it is called.
    ///
    /// Empty matters: checks count records, and a row left over from an earlier check is
    /// indistinguishable from one the store failed to remove. A factory over a real database
    /// should truncate, or use a fresh schema, rather than reuse.
    pub fn new(new_store: F) -> Self {
        StorageConformance {
            new_store,
            spawn: None,
            racers: DEFAULT_RACERS,
        }
    }

    /// Run the racing takes as independent tasks on the HOST'S runtime, for example
    /// `|task| { tokio::spawn(task); }`.
    ///
    /// This is the mode worth running in CI: on a multi-threaded runtime it makes the `take_*`
    /// checks a genuine parallel race rather than an interleaving. The spawner MUST actually run
    /// the future it is given; one that drops it will hang the checks.
    pub fn with_spawn(mut self, spawn: impl Fn(Task) + Send + Sync + 'static) -> Self {
        self.spawn = Some(Arc::new(spawn));
        self
    }

    /// How many callers race each `take_*`. Eight by default. Values below 2 are raised to 2,
    /// since one racer cannot race.
    pub fn racers(mut self, racers: usize) -> Self {
        self.racers = racers.max(2);
        self
    }
}

impl<F, Fut, S> StorageConformance<F>
where
    F: Fn() -> Fut,
    Fut: Future<Output = S>,
    S: Storage + 'static,
{
    /// Run every check and return the violations found. An empty vector means the store satisfied
    /// every property this harness can observe, which is NOT the same as "the store is correct":
    /// read the module docs on what the concurrency checks cannot see.
    pub async fn run(&self) -> Vec<Violation> {
        let mut report = Report::default();
        self.round_trip_client(&mut report).await;
        self.round_trip_device_grant(&mut report).await;
        self.round_trip_authorization_code(&mut report).await;
        self.round_trip_token(&mut report).await;
        self.round_trip_refresh_token(&mut report).await;
        self.atomic_take_device_grant(&mut report).await;
        self.compare_and_swap_device_grant(&mut report).await;
        self.atomic_take_refresh_token(&mut report).await;
        self.atomic_take_authorization_code(&mut report).await;
        self.user_code_index(&mut report).await;
        self.sweep(&mut report).await;
        self.revoke_family(&mut report).await;
        self.delete_client(&mut report).await;
        self.delete_token(&mut report).await;
        #[cfg(any(feature = "client_assertion", feature = "dpop"))]
        self.claim_replay_id(&mut report).await;
        #[cfg(feature = "par")]
        self.round_trip_pushed_request(&mut report).await;
        #[cfg(feature = "par")]
        self.atomic_take_pushed_request(&mut report).await;
        #[cfg(feature = "consent")]
        self.consent(&mut report).await;
        report.violations
    }

    /// RFC 7523 section 3 and RFC 9449 section 4.3 both make a `jti` single use, and
    /// `claim_replay_id` is the only thing enforcing it. Same defect shape as the `take_*`
    /// operations, with a worse failure mode: a `take_*` that hands the value out twice at least
    /// produces two token responses somebody might notice, while a claim-if-absent that answers
    /// "you are first" to two callers produces exactly the request the client meant to send,
    /// twice, and nothing anywhere records that it happened.
    #[cfg(any(feature = "client_assertion", feature = "dpop"))]
    async fn claim_replay_id(&self, report: &mut Report) {
        let store = self.store().await;
        let deadline = at(300);

        let results = self
            .race(report, |gate| {
                let store = Arc::clone(&store);
                Box::pin(async move {
                    gate.wait().await;
                    // Mapped to the shape `judge_race` reads: the caller that is told it claimed
                    // the id is the winner, exactly as the caller that receives a taken record is.
                    store
                        .claim_replay_id("jti-race", deadline)
                        .await
                        .map(|claimed| if claimed { Some(()) } else { None })
                })
            })
            .await;
        self.judge_race(
            report,
            ATOMIC_CLAIM_REPLAY_ID,
            "claim on a single-use jti",
            results,
        );

        // Sequential, and it has to hold too: a store can be atomic under a race and still forget
        // what it claimed a moment later.
        let store = self.store().await;
        let first = report.ok(
            CLAIM_REPLAY_ID_REFUSES_SECOND,
            "claim_replay_id",
            store.claim_replay_id("jti-once", deadline).await,
        );
        if first == Some(false) {
            report.fail(
                CLAIM_REPLAY_ID_REFUSES_SECOND,
                "the FIRST claim of an unseen id answered false, so every artifact carrying a jti \
                 is refused as a replay of itself",
            );
        }
        if let Some(second) = report.ok(
            CLAIM_REPLAY_ID_REFUSES_SECOND,
            "claim_replay_id (again)",
            store.claim_replay_id("jti-once", deadline).await,
        ) {
            if second {
                report.fail(
                    CLAIM_REPLAY_ID_REFUSES_SECOND,
                    "the SECOND claim of the same id also answered true: the id is not recorded, \
                     so a client assertion or DPoP proof can be replayed by anyone who observed \
                     one request",
                );
            }
        }
        // A DIFFERENT id must still be claimable: a store that answers false to everything after
        // the first claim would pass the check above and refuse every subsequent request.
        if let Some(other) = report.ok(
            CLAIM_REPLAY_ID_REFUSES_SECOND,
            "claim_replay_id (a different id)",
            store.claim_replay_id("jti-other", deadline).await,
        ) {
            if !other {
                report.fail(
                    CLAIM_REPLAY_ID_REFUSES_SECOND,
                    "a claim of an id that was never claimed answered false",
                );
            }
        }

        // Claims are records too, and the only thing that reclaims them is the host's sweep. A
        // store that never expires them grows once per authenticated request, forever.
        let store = self.store().await;
        let now = at(0);
        if report
            .ok(
                SWEEP_RECLAIMS_REPLAY_IDS,
                "claim_replay_id",
                store.claim_replay_id("jti-sweep", now).await,
            )
            .is_none()
        {
            return;
        }
        if let Some(removed) = report.ok(
            SWEEP_RECLAIMS_REPLAY_IDS,
            "sweep_expired",
            store.sweep_expired(now).await,
        ) {
            if removed != 1 {
                report.fail(
                    SWEEP_RECLAIMS_REPLAY_IDS,
                    format!(
                        "sweep_expired reported {removed} removed with exactly one dead replay id \
                         in the store: claimed ids are records the sweep must reclaim, or the \
                         table grows once per authenticated request forever"
                    ),
                );
            }
        }
    }

    async fn store(&self) -> Arc<S> {
        Arc::new((self.new_store)().await)
    }

    // ------------------------------------------------------------------ round-trip fidelity

    /// A store that silently drops a field passes any test that only checks the key came back.
    /// The fields that matter most here are named in their own checks below: `family_id` is what
    /// RFC 9700 section 4.14.2 reuse revocation walks, and `resource` is the RFC 8707 audience
    /// restriction, so a store that loses either produces tokens that are wider than what was
    /// granted while still looking correct.
    async fn round_trip_client(&self, report: &mut Report) {
        let store = self.store().await;
        let want = sample_client("client-round-trip");
        if report
            .ok(
                ROUND_TRIP_CLIENT,
                "put_client",
                store.put_client(want.clone()).await,
            )
            .is_none()
        {
            return;
        }
        let Some(got) = report.ok(
            ROUND_TRIP_CLIENT,
            "get_client",
            store.get_client(&want.client_id).await,
        ) else {
            return;
        };
        let Some(got) = report.some(ROUND_TRIP_CLIENT, "get_client", got) else {
            return;
        };
        let c = ROUND_TRIP_CLIENT;
        report.same(c, "client_id", &want.client_id, &got.client_id);
        report.same(c, "auth", &want.auth, &got.auth);
        report.same(c, "grant_types", &want.grant_types, &got.grant_types);
        report.same(c, "redirect_uris", &want.redirect_uris, &got.redirect_uris);
        report.same(
            c,
            "allowed_scopes",
            &want.allowed_scopes,
            &got.allowed_scopes,
        );
        report.same(
            c,
            "default_scopes",
            &want.default_scopes,
            &got.default_scopes,
        );
        report.same(c, "name", &want.name, &got.name);
        report.same(c, "registration", &want.registration, &got.registration);
    }

    /// RFC 8628 section 3.3: the user's decision at the verification UI is FIRST-DECISION-WINS, and
    /// `Storage::compare_and_swap_device_grant` is the only thing that makes it so. Three unrelated
    /// actors write one device grant (the polling device, and the user approving or denying), so
    /// without a real compare-and-swap the last writer wins by accident and a DENIAL a human
    /// actually made is silently reverted to `Pending` by a poll that read the grant a moment
    /// earlier.
    ///
    /// A store gets this wrong in two ways, and neither is visible to any other check here, because
    /// both are perfectly ATOMIC. The first is dropping the comparison: `UPDATE ... SET payload =
    /// $1 WHERE device_code = $2`, reporting `rows_affected > 0`, which is one statement, races
    /// nothing, and reinstates exactly the lost update the method exists to prevent.
    ///
    /// The second is the one nobody thinks to test for, and it is worse: a read, a comparison, and
    /// an INSERT-OR-UPDATE. `take_device_grant` is single-use redemption, so a grant redeemed
    /// between that read and that write is gone; an upsert does not fail and does not no-op against
    /// a row that is not there, it puts the grant BACK. An RFC 8628 device code that has already
    /// been exchanged for a token becomes exchangeable a second time. The trait says a swap must
    /// never bring a redeemed grant back; this is the check that holds a store to it.
    async fn compare_and_swap_device_grant(&self, report: &mut Report) {
        let store = self.store().await;
        // Pending, because that is the state both the poll and the verification UI swap AGAINST.
        let pending = DeviceGrant {
            state: DeviceGrantState::Pending,
            ..sample_device_grant("dc-swap", "SWAP-AAAA")
        };
        if report
            .ok(
                SWAP_APPLIES_ON_MATCH,
                "put_device_grant",
                store.put_device_grant(pending.clone()).await,
            )
            .is_none()
        {
            return;
        }

        // 1. The swap a correct store MUST apply: the state is the one the caller read.
        let denied = DeviceGrant {
            state: DeviceGrantState::Denied,
            ..pending.clone()
        };
        let Some(applied) = report.ok(
            SWAP_APPLIES_ON_MATCH,
            "compare_and_swap_device_grant",
            store
                .compare_and_swap_device_grant(&DeviceGrantState::Pending, denied.clone())
                .await,
        ) else {
            return;
        };
        if !applied {
            report.fail(
                SWAP_APPLIES_ON_MATCH,
                "a swap whose expected state matched the stored state reported that it did not \
                 apply; the user's decision at the verification UI would never be recorded",
            );
        }
        match store.get_device_grant(&pending.device_code).await {
            Ok(Some(got)) if got.state == DeviceGrantState::Denied => {}
            Ok(other) => report.fail(
                SWAP_APPLIES_ON_MATCH,
                format!(
                    "a swap that reported success did not change the stored state: read back \
                     {:?}",
                    other.map(|g| g.state)
                ),
            ),
            Err(e) => report.fail(
                SWAP_APPLIES_ON_MATCH,
                format!("get_device_grant failed unexpectedly: {e}"),
            ),
        }

        // 2. The swap a correct store MUST refuse. The stored state has moved on to `Denied`, so a
        // poll still holding `Pending` from its own earlier read must not land. This is the whole
        // of RFC 8628 section 3.3's first-decision-wins property, and a store that ignores
        // `expected` passes every other check in this harness while failing it.
        let repending = DeviceGrant {
            state: DeviceGrantState::Pending,
            ..pending.clone()
        };
        let Some(applied) = report.ok(
            SWAP_HONOURS_EXPECTED,
            "compare_and_swap_device_grant",
            store
                .compare_and_swap_device_grant(&DeviceGrantState::Pending, repending)
                .await,
        ) else {
            return;
        };
        if applied {
            report.fail(
                SWAP_HONOURS_EXPECTED,
                "a swap whose expected state was STALE reported that it applied: the store is not \
                 comparing `expected` against the stored state at all, so the user's decision is \
                 reverted by whichever writer arrives last",
            );
        }
        match store.get_device_grant(&pending.device_code).await {
            Ok(Some(got)) if got.state == DeviceGrantState::Denied => {}
            Ok(other) => report.fail(
                SWAP_HONOURS_EXPECTED,
                format!(
                    "a swap with a stale `expected` overwrote the stored state: the user denied \
                     this grant and it now reads {:?}",
                    other.map(|g| g.state)
                ),
            ),
            Err(e) => report.fail(
                SWAP_HONOURS_EXPECTED,
                format!("get_device_grant failed unexpectedly: {e}"),
            ),
        }

        // 3. Resurrection. The grant is REDEEMED, exactly as `take_device_grant` leaves it after a
        // successful token request, and a swap that was in flight when that happened now lands.
        if report
            .ok(
                SWAP_NEVER_RESURRECTS,
                "take_device_grant",
                store.take_device_grant(&pending.device_code).await,
            )
            .is_none()
        {
            return;
        }
        // Whether the index was ALREADY dirty before the swap ran. A store whose `take` leaves the
        // user-code row behind has a different defect, which `user_code_index/cleared_by_take`
        // owns; without this the swap check would report it a second time under its own name and a
        // host would chase two bugs where there is one.
        let index_already_dirty = matches!(
            store
                .find_device_grant_by_user_code(&normalize_user_code(&pending.user_code))
                .await,
            Ok(Some(_))
        );
        let Some(applied) = report.ok(
            SWAP_NEVER_RESURRECTS,
            "compare_and_swap_device_grant",
            store
                .compare_and_swap_device_grant(&DeviceGrantState::Denied, denied)
                .await,
        ) else {
            return;
        };
        if applied {
            report.fail(
                SWAP_NEVER_RESURRECTS,
                "a swap against a device_code that had already been redeemed reported that it \
                 applied: `Ok(false)` is the only correct answer for a row that is not there",
            );
        }
        match store.get_device_grant(&pending.device_code).await {
            Ok(None) => {}
            Ok(Some(_)) => report.fail(
                SWAP_NEVER_RESURRECTS,
                "a swap brought back a device grant that had been redeemed: the store is writing \
                 through an insert-or-update, so an RFC 8628 single-use device code is now \
                 redeemable a second time",
            ),
            Err(e) => report.fail(
                SWAP_NEVER_RESURRECTS,
                format!("get_device_grant failed unexpectedly: {e}"),
            ),
        }
        // The user-code half of the same resurrection: a store keeping the index as its own row
        // (the ordinary Redis or DynamoDB shape) can put the grant back THERE while the primary
        // lookup stays clean, and the verification UI reads this path.
        match store
            .find_device_grant_by_user_code(&normalize_user_code(&pending.user_code))
            .await
        {
            Ok(None) => {}
            Ok(Some(_)) if index_already_dirty => {}
            Ok(Some(_)) => report.fail(
                SWAP_NEVER_RESURRECTS,
                "a swap put a redeemed grant back into the user-code index: the code a human \
                 typed resolves to a grant that has already been exchanged for a token",
            ),
            Err(e) => report.fail(
                SWAP_NEVER_RESURRECTS,
                format!("find_device_grant_by_user_code failed unexpectedly: {e}"),
            ),
        }
    }

    async fn round_trip_device_grant(&self, report: &mut Report) {
        let store = self.store().await;
        let want = sample_device_grant("dc-round-trip", "RTRT-AAAA");
        if report
            .ok(
                ROUND_TRIP_DEVICE_GRANT,
                "put_device_grant",
                store.put_device_grant(want.clone()).await,
            )
            .is_none()
        {
            return;
        }
        let Some(got) = report.ok(
            ROUND_TRIP_DEVICE_GRANT,
            "get_device_grant",
            store.get_device_grant(&want.device_code).await,
        ) else {
            return;
        };
        let Some(got) = report.some(ROUND_TRIP_DEVICE_GRANT, "get_device_grant", got) else {
            return;
        };
        let c = ROUND_TRIP_DEVICE_GRANT;
        report.same(c, "device_code", &want.device_code, &got.device_code);
        report.same(c, "user_code", &want.user_code, &got.user_code);
        report.same(c, "client_id", &want.client_id, &got.client_id);
        report.same(c, "scope", &want.scope, &got.scope);
        report.same(c, "state", &want.state, &got.state);
        report.same(c, "created_at", &want.created_at, &got.created_at);
        report.same(c, "expires_at", &want.expires_at, &got.expires_at);
        report.same(c, "interval", &want.interval, &got.interval);
        report.same(c, "last_poll_at", &want.last_poll_at, &got.last_poll_at);

        // The same record has to come back through the OTHER read path as well: a store that
        // maintains a second table for the index and populates it from a subset of the columns
        // would pass the primary lookup and hand the verification UI a different grant.
        let Some(found) = report.ok(
            c,
            "find_device_grant_by_user_code",
            store
                .find_device_grant_by_user_code(&normalize_user_code(&want.user_code))
                .await,
        ) else {
            return;
        };
        match found {
            Some(found) => report.same(c, "by-user-code record", &want, &found),
            None => report.fail(
                c,
                "a grant that was just put is not reachable by its normalized user code",
            ),
        }
    }

    async fn round_trip_authorization_code(&self, report: &mut Report) {
        let store = self.store().await;
        let want = sample_authorization_code("code-round-trip");
        if report
            .ok(
                ROUND_TRIP_AUTHORIZATION_CODE,
                "put_authorization_code",
                store.put_authorization_code(want.clone()).await,
            )
            .is_none()
        {
            return;
        }
        // There is no non-destructive read for a code, by design: the server takes it and puts a
        // CONSUMED record back, which is what makes a replay recognisable.
        let Some(got) = report.ok(
            ROUND_TRIP_AUTHORIZATION_CODE,
            "take_authorization_code",
            store.take_authorization_code(&want.code).await,
        ) else {
            return;
        };
        let Some(got) = report.some(
            ROUND_TRIP_AUTHORIZATION_CODE,
            "take_authorization_code",
            got,
        ) else {
            return;
        };
        let c = ROUND_TRIP_AUTHORIZATION_CODE;
        report.same(c, "code", &want.code, &got.code);
        report.same(c, "client_id", &want.client_id, &got.client_id);
        report.same(c, "redirect_uri", &want.redirect_uri, &got.redirect_uri);
        report.same(c, "scope", &want.scope, &got.scope);
        report.same(c, "subject", &want.subject, &got.subject);
        report.same(
            c,
            "code_challenge",
            &want.code_challenge,
            &got.code_challenge,
        );
        report.same(
            c,
            "code_challenge_method",
            &want.code_challenge_method,
            &got.code_challenge_method,
        );
        report.same(c, "resource", &want.resource, &got.resource);
        report.same(c, "expires_at", &want.expires_at, &got.expires_at);
        // `Consumed` carries what the code minted, which is what a replay revokes. A store that
        // flattens the state to a boolean loses the thing the remedy needs.
        report.same(c, "state", &want.state, &got.state);
        // RFC 9396 section 5: the code IS the record of what the resource owner approved, so a
        // store that drops the details here mints a token for a narrower authorization than the
        // user granted, and the client is told nothing about the difference.
        #[cfg(feature = "rar")]
        report.same(
            c,
            "authorization_details",
            &want.authorization_details,
            &got.authorization_details,
        );
        // RFC 9470 section 5: the authentication the host reported at the authorization request,
        // which is what the token minted from this code reports as `auth_time` and `acr`. Dropped,
        // a client that answered an `insufficient_user_authentication` challenge gets a token that
        // claims no step-up happened.
        #[cfg(feature = "consent")]
        report.same(
            c,
            "authentication",
            &want.authentication,
            &got.authentication,
        );
    }

    async fn round_trip_token(&self, report: &mut Report) {
        let store = self.store().await;
        let want = sample_token("at-round-trip", "client-round-trip", Some("fam-round-trip"));
        if report
            .ok(
                ROUND_TRIP_TOKEN,
                "put_token",
                store.put_token(want.clone()).await,
            )
            .is_none()
        {
            return;
        }
        let Some(got) = report.ok(
            ROUND_TRIP_TOKEN,
            "get_token",
            store.get_token(&want.access_token).await,
        ) else {
            return;
        };
        let Some(got) = report.some(ROUND_TRIP_TOKEN, "get_token", got) else {
            return;
        };
        let c = ROUND_TRIP_TOKEN;
        report.same(c, "access_token", &want.access_token, &got.access_token);
        report.same(c, "client_id", &want.client_id, &got.client_id);
        report.same(c, "subject", &want.subject, &got.subject);
        report.same(c, "scope", &want.scope, &got.scope);
        // RFC 8707: the audience the token is restricted to. Dropped here, every token is good at
        // every resource server that trusts this issuer.
        report.same(c, "resource", &want.resource, &got.resource);
        report.same(c, "issued_at", &want.issued_at, &got.issued_at);
        report.same(c, "expires_at", &want.expires_at, &got.expires_at);
        // RFC 9700 section 4.14.2: without this, a detected reuse cannot reach the access tokens
        // the thief already minted.
        report.same(c, "family_id", &want.family_id, &got.family_id);
        // RFC 9449 section 6: the DPoP binding. Dropped, the token is a bearer token again.
        #[cfg(feature = "dpop")]
        report.same(c, "jkt", &want.jkt, &got.jkt);
        // RFC 8705 section 3.1: the mTLS binding, which is the OTHER way this crate sender
        // constrains a token and which is dropped by exactly the same kind of missing column. It
        // went unchecked here for longer than `jkt` did, which is the argument for checking it: a
        // store certified clean by this harness while dropping `x5t_s256` silently unbinds every
        // certificate-bound token it holds, and a resource server that introspects gets a token
        // with no `cnf` at all, which reads as a plain bearer token rather than as an error.
        #[cfg(feature = "mtls")]
        report.same(c, "x5t_s256", &want.x5t_s256, &got.x5t_s256);
        // RFC 9396 section 5: what the resource owner actually approved, beyond the scope string.
        // This crate has twice shipped a path that dropped it, so a store that does the same is
        // precisely the defect this harness exists to catch. A token whose details are gone is a
        // token the resource server can only fall back to `scope` for, which is the coarse
        // permission RAR was adopted to stop relying on.
        #[cfg(feature = "rar")]
        report.same(
            c,
            "authorization_details",
            &want.authorization_details,
            &got.authorization_details,
        );
        // RFC 9470 section 5: the `auth_time` and `acr` an introspecting resource server reads to
        // decide whether the authentication behind this token is strong or fresh enough. Dropped,
        // every token looks like it was minted with no step-up at all.
        #[cfg(feature = "consent")]
        report.same(
            c,
            "authentication",
            &want.authentication,
            &got.authentication,
        );
    }

    async fn round_trip_refresh_token(&self, report: &mut Report) {
        let store = self.store().await;
        let want = sample_refresh("rt-round-trip", "client-round-trip", "fam-round-trip");
        if report
            .ok(
                ROUND_TRIP_REFRESH_TOKEN,
                "put_refresh_token",
                store.put_refresh_token(want.clone()).await,
            )
            .is_none()
        {
            return;
        }
        let Some(got) = report.ok(
            ROUND_TRIP_REFRESH_TOKEN,
            "get_refresh_token",
            store.get_refresh_token(&want.refresh_token).await,
        ) else {
            return;
        };
        let Some(got) = report.some(ROUND_TRIP_REFRESH_TOKEN, "get_refresh_token", got) else {
            return;
        };
        let c = ROUND_TRIP_REFRESH_TOKEN;
        report.same(c, "refresh_token", &want.refresh_token, &got.refresh_token);
        report.same(c, "client_id", &want.client_id, &got.client_id);
        report.same(c, "subject", &want.subject, &got.subject);
        report.same(c, "scope", &want.scope, &got.scope);
        report.same(c, "resource", &want.resource, &got.resource);
        report.same(c, "expires_at", &want.expires_at, &got.expires_at);
        report.same(c, "family_id", &want.family_id, &got.family_id);
        // `Spent` is the whole basis of reuse detection: a store that reads every record back as
        // `Active` turns the RFC 9700 section 4.14.2 remedy off.
        report.same(c, "state", &want.state, &got.state);
        // RFC 9449 section 5: carried across rotation and checked on redemption.
        #[cfg(feature = "dpop")]
        report.same(c, "jkt", &want.jkt, &got.jkt);
        // RFC 8705 section 3.1, and it matters MORE on the refresh record than on the access
        // token: this is the binding the next rotation copies onto the token it mints, so a store
        // that loses it here does not merely unbind one token, it unbinds every token the chain
        // will ever produce.
        #[cfg(feature = "mtls")]
        report.same(c, "x5t_s256", &want.x5t_s256, &got.x5t_s256);
        // RFC 9396 section 6: the refresh record is what the narrowing on the next rotation is
        // measured against. Dropped, the grant carries no details for a rotation to narrow, and
        // the refreshed token silently loses the rich authorization the user approved.
        #[cfg(feature = "rar")]
        report.same(
            c,
            "authorization_details",
            &want.authorization_details,
            &got.authorization_details,
        );
        // RFC 9470: carried across rotation so a client cannot defeat a `max_age` by refreshing.
        #[cfg(feature = "consent")]
        report.same(
            c,
            "authentication",
            &want.authentication,
            &got.authentication,
        );
    }

    // ------------------------------------------------------------------ atomicity

    async fn atomic_take_device_grant(&self, report: &mut Report) {
        let store = self.store().await;
        let grant = sample_device_grant("dc-race", "RACE-AAAA");
        if report
            .ok(
                ATOMIC_TAKE_DEVICE_GRANT,
                "put_device_grant",
                store.put_device_grant(grant).await,
            )
            .is_none()
        {
            return;
        }
        let results = self
            .race(report, |gate| {
                let store = Arc::clone(&store);
                Box::pin(async move {
                    gate.wait().await;
                    store.take_device_grant("dc-race").await
                })
            })
            .await;
        self.judge_race(report, ATOMIC_TAKE_DEVICE_GRANT, "device grant", results);

        // And it is gone by every path afterwards, not merely unavailable to the losers.
        if let Some(again) = report.ok(
            ATOMIC_TAKE_DEVICE_GRANT,
            "get_device_grant after take",
            store.get_device_grant("dc-race").await,
        ) {
            if again.is_some() {
                report.fail(
                    ATOMIC_TAKE_DEVICE_GRANT,
                    "the grant is still readable after take_device_grant returned it",
                );
            }
        }
    }

    async fn atomic_take_refresh_token(&self, report: &mut Report) {
        let store = self.store().await;
        let record = sample_refresh("rt-race", "client-race", "fam-race");
        if report
            .ok(
                ATOMIC_TAKE_REFRESH_TOKEN,
                "put_refresh_token",
                store.put_refresh_token(record).await,
            )
            .is_none()
        {
            return;
        }
        let results = self
            .race(report, |gate| {
                let store = Arc::clone(&store);
                Box::pin(async move {
                    gate.wait().await;
                    store.take_refresh_token("rt-race").await
                })
            })
            .await;
        self.judge_race(report, ATOMIC_TAKE_REFRESH_TOKEN, "refresh record", results);

        if let Some(again) = report.ok(
            ATOMIC_TAKE_REFRESH_TOKEN,
            "get_refresh_token after take",
            store.get_refresh_token("rt-race").await,
        ) {
            if again.is_some() {
                report.fail(
                    ATOMIC_TAKE_REFRESH_TOKEN,
                    "the record is still readable after take_refresh_token returned it",
                );
            }
        }
    }

    async fn atomic_take_authorization_code(&self, report: &mut Report) {
        let store = self.store().await;
        let record = sample_authorization_code("code-race");
        if report
            .ok(
                ATOMIC_TAKE_AUTHORIZATION_CODE,
                "put_authorization_code",
                store.put_authorization_code(record).await,
            )
            .is_none()
        {
            return;
        }
        let results = self
            .race(report, |gate| {
                let store = Arc::clone(&store);
                Box::pin(async move {
                    gate.wait().await;
                    store.take_authorization_code("code-race").await
                })
            })
            .await;
        self.judge_race(
            report,
            ATOMIC_TAKE_AUTHORIZATION_CODE,
            "authorization code record",
            results,
        );

        if let Some(again) = report.ok(
            ATOMIC_TAKE_AUTHORIZATION_CODE,
            "take_authorization_code after take",
            store.take_authorization_code("code-race").await,
        ) {
            if again.is_some() {
                report.fail(
                    ATOMIC_TAKE_AUTHORIZATION_CODE,
                    "a second take_authorization_code returned the record again",
                );
            }
        }
    }

    /// The pushed request has to come back out of the store as it went in, field for field.
    ///
    /// It was the ONE record kind this harness raced and never round-tripped, which made the PAR
    /// path the only one where a store could drop a column and be certified clean. That is the
    /// worst record to have the hole in: RFC 9126 section 2.1 has the AS validate the parameters
    /// at push time, and RFC 9101 section 6.3 has the authorization endpoint use ONLY the pushed
    /// parameters, so a parameter that is validated and then lost is a parameter the client was
    /// told was acceptable and then silently did not get. Losing `code_challenge` in particular is
    /// a silent PKCE downgrade on a request whose whole purpose was to keep it out of the browser.
    ///
    /// There is no non-destructive read for a handle, by design (RFC 9126 section 4 makes it
    /// single use), so the round trip is put-then-take, exactly as the authorization code's is.
    #[cfg(feature = "par")]
    async fn round_trip_pushed_request(&self, report: &mut Report) {
        let store = self.store().await;
        let want = sample_pushed_request("urn:ietf:params:oauth:request_uri:round-trip");
        if report
            .ok(
                ROUND_TRIP_PUSHED_REQUEST,
                "put_pushed_authorization_request",
                store.put_pushed_authorization_request(want.clone()).await,
            )
            .is_none()
        {
            return;
        }
        let Some(got) = report.ok(
            ROUND_TRIP_PUSHED_REQUEST,
            "take_pushed_authorization_request",
            store
                .take_pushed_authorization_request(&want.request_uri)
                .await,
        ) else {
            return;
        };
        let Some(got) = report.some(
            ROUND_TRIP_PUSHED_REQUEST,
            "take_pushed_authorization_request",
            got,
        ) else {
            return;
        };
        let c = ROUND_TRIP_PUSHED_REQUEST;
        report.same(c, "request_uri", &want.request_uri, &got.request_uri);
        // RFC 9126 section 2.2 binds the handle to the client that pushed it, and section 7.5 is
        // the attack that binding prevents; a store that loses it lets a stranger's `/authorize`
        // resolve somebody else's pushed request.
        report.same(c, "client_id", &want.client_id, &got.client_id);
        report.same(c, "response_type", &want.response_type, &got.response_type);
        report.same(c, "redirect_uri", &want.redirect_uri, &got.redirect_uri);
        report.same(c, "scope", &want.scope, &got.scope);
        report.same(c, "state", &want.state, &got.state);
        // RFC 7636 section 4.3: dropped here, the authorization endpoint reads a request with no
        // challenge and the code it mints is redeemable without a verifier.
        report.same(
            c,
            "code_challenge",
            &want.code_challenge,
            &got.code_challenge,
        );
        report.same(
            c,
            "code_challenge_method",
            &want.code_challenge_method,
            &got.code_challenge_method,
        );
        report.same(c, "resource", &want.resource, &got.resource);
        report.same(c, "expires_at", &want.expires_at, &got.expires_at);
        #[cfg(feature = "rar")]
        report.same(
            c,
            "authorization_details",
            &want.authorization_details,
            &got.authorization_details,
        );
        // RFC 9470 section 4. This crate has already shipped a bug where these two were dropped on
        // the PAR path, which disabled step-up for every PAR deployment; a store that drops them
        // reproduces that bug from the other side, and nothing else would report it.
        #[cfg(feature = "consent")]
        report.same(c, "acr_values", &want.acr_values, &got.acr_values);
        #[cfg(feature = "consent")]
        report.same(c, "max_age", &want.max_age, &got.max_age);
    }

    /// RFC 9126 section 4 makes a `request_uri` single use, and
    /// `take_pushed_authorization_request` is the ONLY thing enforcing it. Same defect shape as
    /// the other `take_*` operations: a store that reads then deletes lets two concurrent
    /// `/authorize` hits on one handle both resolve, so one pushed request authorizes twice.
    ///
    /// Worth checking separately rather than assuming a store that got the other three right got
    /// this one right too: this method arrived with the PAR feature, later than the rest, which is
    /// exactly the shape of thing a host adds in a hurry against an existing trait impl.
    #[cfg(feature = "par")]
    async fn atomic_take_pushed_request(&self, report: &mut Report) {
        let store = self.store().await;
        let record = sample_pushed_request("urn:ietf:params:oauth:request_uri:race");
        if report
            .ok(
                ATOMIC_TAKE_PUSHED_REQUEST,
                "put_pushed_authorization_request",
                store.put_pushed_authorization_request(record).await,
            )
            .is_none()
        {
            return;
        }
        let results = self
            .race(report, |gate| {
                let store = Arc::clone(&store);
                Box::pin(async move {
                    gate.wait().await;
                    store
                        .take_pushed_authorization_request("urn:ietf:params:oauth:request_uri:race")
                        .await
                })
            })
            .await;
        self.judge_race(
            report,
            ATOMIC_TAKE_PUSHED_REQUEST,
            "pushed authorization request",
            results,
        );

        if let Some(again) = report.ok(
            ATOMIC_TAKE_PUSHED_REQUEST,
            "take_pushed_authorization_request after take",
            store
                .take_pushed_authorization_request("urn:ietf:params:oauth:request_uri:race")
                .await,
        ) {
            if again.is_some() {
                report.fail(
                    ATOMIC_TAKE_PUSHED_REQUEST,
                    "a second take_pushed_authorization_request returned the handle again",
                );
            }
        }
    }

    /// Consent round trip, and the withdrawal cascade.
    ///
    /// The cascade is the one worth the most care in this whole harness. Withdrawal is what a user
    /// is told stops an application acting for them, so a store that removes the consent row and
    /// leaves the credentials alive has told the user something false, and nothing anywhere
    /// reports it: the endpoint answered 200, the row is gone, and the tokens keep working. That
    /// is strictly worse than a withdrawal that visibly fails.
    ///
    /// So this seeds one of every record kind the contract enumerates for the consent being
    /// withdrawn, AND one of each for a DIFFERENT subject of the same client, then requires the
    /// first set gone, the second set untouched, and the count to match. A store that revokes too
    /// much fails here just as a store that revokes too little does; over-revoking would log a
    /// different user out of an application they never withdrew.
    #[cfg(feature = "consent")]
    async fn consent(&self, report: &mut Report) {
        let store = self.store().await;

        let mine = sample_consent("consent-mine", "subject-conformance");
        let theirs = sample_consent("consent-theirs", "subject-other");
        if report
            .ok(
                ROUND_TRIP_CONSENT,
                "put_consent",
                store.put_consent(mine.clone()).await,
            )
            .is_none()
        {
            return;
        }
        if report
            .ok(
                ROUND_TRIP_CONSENT,
                "put_consent (second subject)",
                store.put_consent(theirs.clone()).await,
            )
            .is_none()
        {
            return;
        }

        // Round trip by id, and by the (client, subject) lookup the remembered-consent path uses.
        // A store whose index disagrees with what it stored answers one and not the other.
        if let Some(Some(back)) = report.ok(
            ROUND_TRIP_CONSENT,
            "get_consent",
            store.get_consent("consent-mine").await,
        ) {
            if *back != mine {
                report.fail(
                    ROUND_TRIP_CONSENT,
                    "get_consent returned a record that differs from the one stored",
                );
            }
        } else {
            report.fail(ROUND_TRIP_CONSENT, "get_consent did not return the record");
        }
        if let Some(found) = report.ok(
            ROUND_TRIP_CONSENT,
            "find_consent",
            store
                .find_consent(&ClientId::new("client-conformance"), "subject-conformance")
                .await,
        ) {
            match found {
                Some(f) if f.consent_id == mine.consent_id => {}
                Some(_) => report.fail(
                    ROUND_TRIP_CONSENT,
                    "find_consent returned a different consent than the one for that subject",
                ),
                None => report.fail(
                    ROUND_TRIP_CONSENT,
                    "find_consent did not find a consent that get_consent can read",
                ),
            }
        }

        // Everything the withdrawal must reach, for the consent being withdrawn and for a
        // bystander subject of the SAME client.
        // The subject has to be overridden on every record, not just the device grant: the shared
        // sample builders all carry one subject, and a "different subject" fixture that is
        // secretly the same subject would make the spares-others check pass for the wrong reason.
        let seed = |subject: &str, tag: &str| {
            let mut token = sample_token(&format!("at-{tag}"), "client-conformance", Some(tag));
            token.subject = Some(subject.to_string());
            let mut refresh = sample_refresh(&format!("rt-{tag}"), "client-conformance", tag);
            refresh.subject = Some(subject.to_string());
            let mut code = sample_authorization_code(&format!("code-{tag}"));
            code.subject = subject.to_string();
            (
                token,
                refresh,
                code,
                sample_approved_device_grant(&format!("dc-{tag}"), &format!("UC{tag}"), subject),
            )
        };
        let (at_mine, rt_mine, code_mine, grant_mine) = seed("subject-conformance", "mine");
        let (at_theirs, rt_theirs, code_theirs, grant_theirs) = seed("subject-other", "theirs");
        for (t, r, c, g) in [
            (&at_mine, &rt_mine, &code_mine, &grant_mine),
            (&at_theirs, &rt_theirs, &code_theirs, &grant_theirs),
        ] {
            // EVERY seed is reported, and each one names the put that did not land. The three
            // below had their `Result` discarded, which made a store that silently failed to
            // persist a fixture PASS this check for the wrong reason: the assertions afterwards
            // are all "the record is gone", and a record that was never written is gone. This is
            // the exported harness, so that false pass would not have misled this repository, it
            // would have certified a stranger's broken store.
            let seeded = report
                .ok(
                    REVOKE_CONSENT_CASCADES,
                    "seeding the records a withdrawal must reach: put_token",
                    store.put_token(t.clone()).await,
                )
                .and(report.ok(
                    REVOKE_CONSENT_CASCADES,
                    "seeding the records a withdrawal must reach: put_refresh_token",
                    store.put_refresh_token(r.clone()).await,
                ))
                .and(report.ok(
                    REVOKE_CONSENT_CASCADES,
                    "seeding the records a withdrawal must reach: put_authorization_code",
                    store.put_authorization_code(c.clone()).await,
                ))
                .and(report.ok(
                    REVOKE_CONSENT_CASCADES,
                    "seeding the records a withdrawal must reach: put_device_grant",
                    store.put_device_grant(g.clone()).await,
                ));
            if seeded.is_none() {
                return;
            }
        }

        let removed = match report.ok(
            REVOKE_CONSENT_CASCADES,
            "revoke_consent",
            store.revoke_consent("consent-mine").await,
        ) {
            Some(n) => n,
            None => return,
        };

        // The four the contract enumerates, plus the consent row itself.
        for (what, gone) in [
            (
                "the access token",
                matches!(store.get_token(&at_mine.access_token).await, Ok(None)),
            ),
            (
                "the refresh record",
                matches!(
                    store.get_refresh_token(&rt_mine.refresh_token).await,
                    Ok(None)
                ),
            ),
            (
                "the unredeemed authorization code",
                matches!(
                    store.take_authorization_code(&code_mine.code).await,
                    Ok(None)
                ),
            ),
            (
                "the approved device grant",
                matches!(
                    store.get_device_grant(&grant_mine.device_code).await,
                    Ok(None)
                ),
            ),
            (
                "the consent record",
                matches!(store.get_consent("consent-mine").await, Ok(None)),
            ),
        ] {
            if !gone {
                report.fail(
                    REVOKE_CONSENT_CASCADES,
                    format!(
                        "revoke_consent left {what} alive, so the user was told this application \
                         was stopped and it was not"
                    ),
                );
            }
        }

        // And nothing belonging to the other subject moved.
        for (what, alive) in [
            (
                "access token",
                matches!(store.get_token(&at_theirs.access_token).await, Ok(Some(_))),
            ),
            (
                "refresh record",
                matches!(
                    store.get_refresh_token(&rt_theirs.refresh_token).await,
                    Ok(Some(_))
                ),
            ),
            (
                "device grant",
                matches!(
                    store.get_device_grant(&grant_theirs.device_code).await,
                    Ok(Some(_))
                ),
            ),
            (
                "consent record",
                matches!(store.get_consent("consent-theirs").await, Ok(Some(_))),
            ),
        ] {
            if !alive {
                report.fail(
                    REVOKE_CONSENT_SPARES_OTHERS,
                    format!(
                        "revoke_consent removed another subject's {what}, logging out a user who \
                         withdrew nothing"
                    ),
                );
            }
        }

        // FOUR: the trait doc is explicit that the consent record itself is not counted, so this
        // is the four credentials. The count is what an operator investigating an incident reads,
        // so a store that reports a number it did not remove is lying to the one person who needs
        // the truth.
        if removed != 4 {
            report.fail(
                REVOKE_CONSENT_COUNT,
                format!(
                    "revoke_consent removed 4 credentials but reported {removed} (the consent \
                     record itself is not counted)"
                ),
            );
        }

        if let Some(second) = report.ok(
            REVOKE_CONSENT_COUNT,
            "revoke_consent (second call)",
            store.revoke_consent("consent-mine").await,
        ) {
            if second != 0 {
                report.fail(
                    REVOKE_CONSENT_COUNT,
                    format!("withdrawing an already-withdrawn consent reported {second}, not 0"),
                );
            }
        }
    }

    /// Exactly one racer may receive the value. More than one IS the double-spend; none means the
    /// value was lost, which is a different bug with the same root (a non-atomic pair of steps).
    fn judge_race<T>(
        &self,
        report: &mut Report,
        check: &'static str,
        what: &str,
        results: TakeResults<T>,
    ) {
        let winners = results.iter().filter(|r| matches!(r, Ok(Some(_)))).count();
        let errors = results.iter().filter(|r| r.is_err()).count();
        if winners > 1 {
            report.fail(
                check,
                format!(
                    "{winners} of {} concurrent takes each received the {what}: the operation is \
                     not an atomic remove-and-return, so this store double-spends single-use \
                     credentials under concurrency",
                    results.len()
                ),
            );
        } else if winners == 0 {
            report.fail(
                check,
                format!(
                    "none of {} concurrent takes received the {what}, though it was stored \
                     beforehand: the value was lost rather than handed to exactly one caller",
                    results.len()
                ),
            );
        }
        if errors > 0 {
            report.fail(
                check,
                format!(
                    "{errors} of {} concurrent takes failed with a StorageError. The server maps \
                     that to server_error, so a legitimate redemption fails under ordinary \
                     contention; a store using optimistic concurrency must retry internally \
                     rather than surface the conflict",
                    results.len()
                ),
            );
        }
    }

    /// Build `racers` futures from `make`, all parked on one rendezvous gate, and run them
    /// concurrently: on the host's runtime when a spawner was installed, otherwise polled together
    /// on this task. See the module docs for what each mode does and does not prove.
    async fn race<T, M>(&self, report: &mut Report, make: M) -> TakeResults<T>
    where
        // Every record this races over is a plain owned value; see `JoinAll` on why `Unpin` costs
        // nothing here.
        T: Send + Unpin + 'static,
        M: Fn(Arc<Gate>) -> BoxTake<T>,
    {
        let n = self.racers;
        let gate = Gate::new(n);
        let futures: Vec<BoxTake<T>> = (0..n).map(|_| make(Arc::clone(&gate))).collect();
        // Racers that never reached their own end. See `RacerGuard`: this is what turns a store
        // that panics under concurrency into a REPORT rather than a hung test run.
        let abandoned = Arc::new(AtomicUsize::new(0));

        let results = match &self.spawn {
            Some(spawn) => {
                let collected: Arc<Mutex<TakeResults<T>>> =
                    Arc::new(Mutex::new(Vec::with_capacity(n)));
                let latch = Latch::new(n);
                for fut in futures {
                    let collected = Arc::clone(&collected);
                    let latch = Arc::clone(&latch);
                    let abandoned = Arc::clone(&abandoned);
                    spawn(Box::pin(async move {
                        let mut guard = RacerGuard {
                            latch,
                            abandoned,
                            finished: false,
                        };
                        let outcome = fut.await;
                        // A poisoned lock means another racer panicked; the recovered guard is
                        // sound here because the vector is only ever pushed to.
                        collected
                            .lock()
                            .unwrap_or_else(|e| e.into_inner())
                            .push(outcome);
                        guard.finished = true;
                    }));
                }
                latch.wait().await;
                let mut guard = collected.lock().unwrap_or_else(|e| e.into_inner());
                std::mem::take(&mut *guard)
            }
            None => JoinAll::new(futures).await,
        };

        let abandoned = abandoned.load(Ordering::SeqCst);
        if abandoned > 0 {
            report.fail(
                HARNESS_RACER_PANICKED,
                format!(
                    "{abandoned} of {n} racers never finished: the store's call panicked, or the \
                     spawner dropped the task before it completed. Whatever the results of this \
                     check say, a store that panics under concurrent access fails the request that \
                     hit it, and on a host that aborts on panic it takes the process with it. The \
                     panic message itself is on the spawner's own reporting path, not here"
                ),
            );
        }

        if gate.unsatisfied() {
            report.fail(
                HARNESS_RACE_SETUP,
                format!(
                    "the {n} racers never overlapped: each gave up waiting for the others, which \
                     means they ran one after another and the atomicity results in this run prove \
                     nothing. A `with_spawn` that runs its task to completion inline does this; \
                     hand the futures to a real runtime instead",
                ),
            );
        }
        results
    }

    // ------------------------------------------------------------------ user-code index

    async fn user_code_index(&self, report: &mut Report) {
        // HALF ONE: a put that CHANGES a grant's user code must retire the old index entry, or the
        // superseded code goes on resolving to the grant.
        let store = self.store().await;
        let ok_first = report
            .ok(
                INDEX_RETIRES_OLD_USER_CODE,
                "put_device_grant",
                store
                    .put_device_grant(sample_device_grant("dc-idx", "AAAA-AAAA"))
                    .await,
            )
            .is_some();
        let ok_second = report
            .ok(
                INDEX_RETIRES_OLD_USER_CODE,
                "put_device_grant (same device_code, new user code)",
                store
                    .put_device_grant(sample_device_grant("dc-idx", "BBBB-BBBB"))
                    .await,
            )
            .is_some();
        if ok_first && ok_second {
            if let Some(found) = report.ok(
                INDEX_RETIRES_OLD_USER_CODE,
                "find_device_grant_by_user_code(new)",
                store.find_device_grant_by_user_code("BBBBBBBB").await,
            ) {
                if found.is_none() {
                    report.fail(
                        INDEX_RETIRES_OLD_USER_CODE,
                        "after a put changed the user code, the NEW code does not resolve",
                    );
                }
            }
            if let Some(found) = report.ok(
                INDEX_RETIRES_OLD_USER_CODE,
                "find_device_grant_by_user_code(old)",
                store.find_device_grant_by_user_code("AAAAAAAA").await,
            ) {
                if found.is_some() {
                    report.fail(
                        INDEX_RETIRES_OLD_USER_CODE,
                        "the OLD user code still resolves after a put changed it: a code the user \
                         was shown and that has been superseded can still be used to approve the \
                         grant",
                    );
                }
            }
        }

        // A take clears the index with the record, in one step. Otherwise a redeemed grant stays
        // reachable by the code a human typed.
        if report
            .ok(
                INDEX_CLEARED_BY_TAKE,
                "take_device_grant",
                store.take_device_grant("dc-idx").await,
            )
            .is_some()
        {
            if let Some(found) = report.ok(
                INDEX_CLEARED_BY_TAKE,
                "find_device_grant_by_user_code after take",
                store.find_device_grant_by_user_code("BBBBBBBB").await,
            ) {
                if found.is_some() {
                    report.fail(
                        INDEX_CLEARED_BY_TAKE,
                        "a taken grant is still reachable by its user code",
                    );
                }
            }
        }

        // HALF TWO: a put whose user code is already indexed for a DIFFERENT device_code must be
        // REFUSED. RFC 8628 section 6.1 makes the user code the credential a human types, so two
        // live grants answering to one code is two devices sharing an identity.
        let store = self.store().await;
        if report
            .ok(
                INDEX_REFUSES_DUPLICATE_USER_CODE,
                "put_device_grant",
                store
                    .put_device_grant(sample_device_grant("dc-first", "CCCC-CCCC"))
                    .await,
            )
            .is_none()
        {
            return;
        }
        let clash = store
            .put_device_grant(sample_device_grant("dc-second", "CCCC-CCCC"))
            .await;
        if clash.is_ok() {
            report.fail(
                INDEX_REFUSES_DUPLICATE_USER_CODE,
                "putting a second grant with a user code already indexed for another device_code \
                 succeeded; it must fail with a StorageError. Repointing the index gives two \
                 devices one identity and orphans the older grant, and it makes the server's \
                 user-code collision retry loop meaningless, since only the store can answer \
                 \"is this code taken\" without a race",
            );
        }

        // The refusal must also be a no-op: a store that writes and then errors leaves the second
        // grant half-present, which is worse than either outcome.
        if let Some(found) = report.ok(
            INDEX_REFUSAL_WRITES_NOTHING,
            "find_device_grant_by_user_code after the refused put",
            store.find_device_grant_by_user_code("CCCCCCCC").await,
        ) {
            match found {
                Some(g) if g.device_code == "dc-first" => {}
                Some(g) => report.fail(
                    INDEX_REFUSAL_WRITES_NOTHING,
                    format!(
                        "the user code now resolves to device_code {:?}, not to the grant that \
                         owned it: the index was repointed by a put that should have written \
                         nothing",
                        g.device_code
                    ),
                ),
                None => report.fail(
                    INDEX_REFUSAL_WRITES_NOTHING,
                    "the user code resolves to nothing after a clashing put: the refused write \
                     removed the index entry belonging to the grant that already owned it",
                ),
            }
        }
        if let Some(found) = report.ok(
            INDEX_REFUSAL_WRITES_NOTHING,
            "get_device_grant(dc-second)",
            store.get_device_grant("dc-second").await,
        ) {
            if found.is_some() {
                report.fail(
                    INDEX_REFUSAL_WRITES_NOTHING,
                    "the clashing grant was persisted even though its user code belonged to \
                     another device_code",
                );
            }
        }

        // Lookups are by NORMALIZED code, and the store does not normalize for the caller. The
        // server normalizes before it ever calls in (RFC 8628 section 6.1); a store that also
        // normalizes would make two different keys collide and would silently accept the display
        // form, which is precisely the input an attacker controls.
        let store = self.store().await;
        if report
            .ok(
                INDEX_NO_NORMALIZATION,
                "put_device_grant",
                store
                    .put_device_grant(sample_device_grant("dc-norm", "WDJB-MJHT"))
                    .await,
            )
            .is_none()
        {
            return;
        }
        if let Some(found) = report.ok(
            INDEX_NO_NORMALIZATION,
            "find_device_grant_by_user_code(normalized)",
            store.find_device_grant_by_user_code("WDJBMJHT").await,
        ) {
            if found.is_none() {
                report.fail(
                    INDEX_NO_NORMALIZATION,
                    "the normalized user code does not resolve, so the store is not indexing what \
                     it was given",
                );
            }
        }
        for probe in ["WDJB-MJHT", "wdjbmjht"] {
            if let Some(found) = report.ok(
                INDEX_NO_NORMALIZATION,
                "find_device_grant_by_user_code(unnormalized)",
                store.find_device_grant_by_user_code(probe).await,
            ) {
                if found.is_some() {
                    report.fail(
                        INDEX_NO_NORMALIZATION,
                        format!(
                            "the store resolved {probe:?}, which is not the normalized key it was \
                             given: it normalizes on the caller's behalf, so two distinct index \
                             keys collide and a lookup the server never intended succeeds"
                        ),
                    );
                }
            }
        }
    }

    // ------------------------------------------------------------------ sweep_expired

    async fn sweep(&self, report: &mut Report) {
        let store = self.store().await;
        let now = at(0);

        // Dead at `now` means `expires_at <= now`, so the boundary record (expires_at == now) is
        // planted deliberately: a store using `<` keeps a record the server treats as expired.
        let mut dead_grant = sample_device_grant("dc-dead", "DEAD-AAAA");
        dead_grant.expires_at = now;
        let mut live_grant = sample_device_grant("dc-live", "LIVE-AAAA");
        live_grant.expires_at = at(600);

        let mut dead_code = sample_authorization_code("code-dead");
        dead_code.expires_at = at_before(1);
        let mut live_code = sample_authorization_code("code-live");
        live_code.expires_at = at(600);

        let mut dead_token = sample_token("at-dead", "client-sweep", Some("fam-sweep"));
        dead_token.expires_at = at_before(1);
        let mut live_token = sample_token("at-live", "client-sweep", Some("fam-sweep"));
        live_token.expires_at = at(600);

        let mut dead_refresh = sample_refresh("rt-dead", "client-sweep", "fam-sweep");
        dead_refresh.expires_at = Some(now);
        let mut live_refresh = sample_refresh("rt-live", "client-sweep", "fam-sweep");
        live_refresh.expires_at = Some(at(600));
        // `None` is a chain with no absolute lifetime and is NOT dead, however old it is.
        let mut endless_refresh = sample_refresh("rt-endless", "client-sweep", "fam-sweep");
        endless_refresh.expires_at = None;

        let c = SWEEP_REMOVES_DEAD;
        let mut planted = true;
        for grant in [dead_grant, live_grant] {
            planted &= report
                .ok(c, "put_device_grant", store.put_device_grant(grant).await)
                .is_some();
        }
        for code in [dead_code, live_code] {
            planted &= report
                .ok(
                    c,
                    "put_authorization_code",
                    store.put_authorization_code(code).await,
                )
                .is_some();
        }
        for token in [dead_token, live_token] {
            planted &= report
                .ok(c, "put_token", store.put_token(token).await)
                .is_some();
        }
        for record in [dead_refresh, live_refresh, endless_refresh] {
            planted &= report
                .ok(
                    c,
                    "put_refresh_token",
                    store.put_refresh_token(record).await,
                )
                .is_some();
        }
        if !planted {
            return;
        }

        let Some(removed) = report.ok(c, "sweep_expired", store.sweep_expired(now).await) else {
            return;
        };

        // Exactly four records were dead: one grant, one code, one access token, one refresh.
        if removed != 4 {
            report.fail(
                SWEEP_COUNT,
                format!(
                    "sweep_expired reported {removed} records removed, but exactly 4 of the 9 \
                     planted records were dead at `now`. The count is what a host schedules its \
                     sweep on, so a wrong one is a store that looks idle while it grows"
                ),
            );
        }

        if let Some(found) = report.ok(
            c,
            "get_device_grant",
            store.get_device_grant("dc-dead").await,
        ) {
            if found.is_some() {
                report.fail(c, "an expired device grant survived the sweep");
            }
        }
        if let Some(found) = report.ok(
            c,
            "find_device_grant_by_user_code",
            store.find_device_grant_by_user_code("DEADAAAA").await,
        ) {
            if found.is_some() {
                report.fail(
                    c,
                    "the user code of a swept grant still resolves: the index outlived the record \
                     it points at",
                );
            }
        }
        if let Some(found) = report.ok(
            c,
            "take_authorization_code",
            store.take_authorization_code("code-dead").await,
        ) {
            if found.is_some() {
                report.fail(c, "an expired authorization code survived the sweep");
            }
        }
        if let Some(found) = report.ok(c, "get_token", store.get_token("at-dead").await) {
            if found.is_some() {
                report.fail(c, "an expired access token survived the sweep");
            }
        }
        if let Some(found) = report.ok(
            c,
            "get_refresh_token",
            store.get_refresh_token("rt-dead").await,
        ) {
            if found.is_some() {
                report.fail(c, "an expired refresh record survived the sweep");
            }
        }

        let k = SWEEP_KEEPS_LIVE;
        if let Some(found) = report.ok(
            k,
            "get_device_grant",
            store.get_device_grant("dc-live").await,
        ) {
            if found.is_none() {
                report.fail(k, "the sweep removed a device grant that had not expired");
            }
        }
        if let Some(found) = report.ok(k, "get_token", store.get_token("at-live").await) {
            if found.is_none() {
                report.fail(k, "the sweep removed an access token that had not expired");
            }
        }
        if let Some(found) = report.ok(
            k,
            "get_refresh_token",
            store.get_refresh_token("rt-live").await,
        ) {
            if found.is_none() {
                report.fail(k, "the sweep removed a refresh record that had not expired");
            }
        }
        if let Some(found) = report.ok(
            k,
            "get_refresh_token(no absolute expiry)",
            store.get_refresh_token("rt-endless").await,
        ) {
            if found.is_none() {
                report.fail(
                    k,
                    "the sweep removed a refresh record whose expires_at is None. A chain with no \
                     absolute lifetime is not dead, and treating None as \"expired at the epoch\" \
                     silently logs every such client out",
                );
            }
        }
        if let Some(found) = report.ok(
            k,
            "take_authorization_code(live)",
            store.take_authorization_code("code-live").await,
        ) {
            if found.is_none() {
                report.fail(
                    k,
                    "the sweep removed an authorization code that had not expired",
                );
            }
        }

        // Safe to call when there is nothing to do. The host runs this on a timer, so an error or
        // a nonzero answer on an idle store is noise a host will learn to ignore.
        let store = self.store().await;
        if let Some(removed) = report.ok(
            SWEEP_EMPTY_IS_ZERO,
            "sweep_expired on an empty store",
            store.sweep_expired(now).await,
        ) {
            if removed != 0 {
                report.fail(
                    SWEEP_EMPTY_IS_ZERO,
                    format!("sweep_expired on an empty store reported {removed} records removed"),
                );
            }
        }
    }

    // ------------------------------------------------------------------ revoke_token_family

    async fn revoke_family(&self, report: &mut Report) {
        let c = REVOKE_FAMILY_REMOVES;
        let store = self.store().await;
        let mut planted = true;
        for (key, family) in [("at-a1", "fam-a"), ("at-a2", "fam-a"), ("at-b", "fam-b")] {
            planted &= report
                .ok(
                    c,
                    "put_token",
                    store
                        .put_token(sample_token(key, "client-fam", Some(family)))
                        .await,
                )
                .is_some();
        }
        // RFC 6749 section 4.4 client credentials produce no refresh chain, so their access tokens
        // carry no family. Planted to prove the revocation does not sweep them up by matching
        // `None` against the family id.
        planted &= report
            .ok(
                c,
                "put_token(no family)",
                store
                    .put_token(sample_token("at-nofam", "client-fam", None))
                    .await,
            )
            .is_some();
        for (key, family) in [("rt-a1", "fam-a"), ("rt-a2", "fam-a"), ("rt-b", "fam-b")] {
            planted &= report
                .ok(
                    c,
                    "put_refresh_token",
                    store
                        .put_refresh_token(sample_refresh(key, "client-fam", family))
                        .await,
                )
                .is_some();
        }
        if !planted {
            return;
        }

        let Some(removed) = report.ok(
            c,
            "revoke_token_family",
            store.revoke_token_family("fam-a").await,
        ) else {
            return;
        };
        if removed != 4 {
            report.fail(
                REVOKE_FAMILY_COUNT,
                format!(
                    "revoke_token_family reported {removed} removed, but the family held 4 \
                     records (2 access tokens and 2 refresh records)"
                ),
            );
        }
        for key in ["at-a1", "at-a2"] {
            if let Some(found) = report.ok(c, "get_token", store.get_token(key).await) {
                if found.is_some() {
                    report.fail(
                        c,
                        format!(
                            "access token {key} carrying the revoked family_id survived. RFC 9700 \
                             section 4.14.2 requires revoking the tokens issued for that \
                             authorization grant, not just the refresh chain, so the thief's \
                             already-minted access tokens stay live"
                        ),
                    );
                }
            }
        }
        for key in ["rt-a1", "rt-a2"] {
            if let Some(found) =
                report.ok(c, "get_refresh_token", store.get_refresh_token(key).await)
            {
                if found.is_some() {
                    report.fail(
                        c,
                        format!("refresh record {key} carrying the revoked family_id survived"),
                    );
                }
            }
        }

        let s = REVOKE_FAMILY_SPARES_OTHERS;
        if let Some(found) = report.ok(s, "get_token", store.get_token("at-b").await) {
            if found.is_none() {
                report.fail(s, "revoking one family removed an access token of another");
            }
        }
        if let Some(found) = report.ok(
            s,
            "get_refresh_token",
            store.get_refresh_token("rt-b").await,
        ) {
            if found.is_none() {
                report.fail(s, "revoking one family removed a refresh record of another");
            }
        }
        if let Some(found) = report.ok(s, "get_token(no family)", store.get_token("at-nofam").await)
        {
            if found.is_none() {
                report.fail(
                    s,
                    "revoking a family removed an access token that carries no family_id at all",
                );
            }
        }

        // It runs on evidence of compromise and must not be turned into an error by a concurrent
        // revocation that got there first.
        match store.revoke_token_family("fam-a").await {
            Ok(0) => {}
            Ok(n) => report.fail(
                REVOKE_FAMILY_COUNT,
                format!("revoking an already-revoked family reported {n} removed, expected 0"),
            ),
            Err(e) => report.fail(
                c,
                format!(
                    "revoking an already-revoked family failed with {e}. Removing records that are \
                     already gone is success: this runs on evidence of compromise"
                ),
            ),
        }
    }

    // ------------------------------------------------------------------ delete_client

    async fn delete_client(&self, report: &mut Report) {
        let c = DELETE_CLIENT_CASCADES;
        let store = self.store().await;
        let doomed = ClientId::new("client-doomed");
        let bystander = ClientId::new("client-bystander");

        let mut planted = true;
        for id in [&doomed, &bystander] {
            planted &= report
                .ok(
                    c,
                    "put_client",
                    store.put_client(sample_client(id.as_str())).await,
                )
                .is_some();
            let mut grant = sample_device_grant(
                &format!("dc-{}", id.as_str()),
                if id == &doomed {
                    "DOOM-AAAA"
                } else {
                    "BYST-AAAA"
                },
            );
            grant.client_id = id.clone();
            planted &= report
                .ok(c, "put_device_grant", store.put_device_grant(grant).await)
                .is_some();
            let mut code = sample_authorization_code(&format!("code-{}", id.as_str()));
            code.client_id = id.clone();
            planted &= report
                .ok(
                    c,
                    "put_authorization_code",
                    store.put_authorization_code(code).await,
                )
                .is_some();
            planted &= report
                .ok(
                    c,
                    "put_token",
                    store
                        .put_token(sample_token(
                            &format!("at-{}", id.as_str()),
                            id.as_str(),
                            Some("fam-cascade"),
                        ))
                        .await,
                )
                .is_some();
            planted &= report
                .ok(
                    c,
                    "put_refresh_token",
                    store
                        .put_refresh_token(sample_refresh(
                            &format!("rt-{}", id.as_str()),
                            id.as_str(),
                            "fam-cascade",
                        ))
                        .await,
                )
                .is_some();
        }
        if !planted {
            return;
        }

        let Some(existed) = report.ok(
            DELETE_CLIENT_REPORTS,
            "delete_client",
            store.delete_client(&doomed).await,
        ) else {
            return;
        };
        if !existed {
            report.fail(
                DELETE_CLIENT_REPORTS,
                "delete_client answered false for a registration that was present",
            );
        }

        if let Some(found) = report.ok(c, "get_client", store.get_client(&doomed).await) {
            if found.is_some() {
                report.fail(c, "the registration survived delete_client");
            }
        }
        // RFC 7592 section 2.3: deleting a registration invalidates what that registration holds.
        // A store that removed only the row leaves a client that no longer exists still calling
        // resource servers until every credential it holds expires on its own.
        if let Some(found) = report.ok(c, "get_token", store.get_token("at-client-doomed").await) {
            if found.is_some() {
                report.fail(
                    c,
                    "an access token issued to the deleted client survived: a client that no \
                     longer exists can still call resource servers",
                );
            }
        }
        if let Some(found) = report.ok(
            c,
            "get_refresh_token",
            store.get_refresh_token("rt-client-doomed").await,
        ) {
            if found.is_some() {
                report.fail(
                    c,
                    "a refresh chain of the deleted client survived, so the deleted client can \
                     mint fresh access tokens indefinitely",
                );
            }
        }
        if let Some(found) = report.ok(
            c,
            "take_authorization_code",
            store.take_authorization_code("code-client-doomed").await,
        ) {
            if found.is_some() {
                report.fail(c, "an authorization code of the deleted client survived");
            }
        }
        if let Some(found) = report.ok(
            c,
            "get_device_grant",
            store.get_device_grant("dc-client-doomed").await,
        ) {
            if found.is_some() {
                report.fail(c, "a device grant of the deleted client survived");
            }
        }
        if let Some(found) = report.ok(
            c,
            "find_device_grant_by_user_code",
            store.find_device_grant_by_user_code("DOOMAAAA").await,
        ) {
            if found.is_some() {
                report.fail(
                    c,
                    "the user-code index entry of the deleted client's device grant survived",
                );
            }
        }

        // The bystander is untouched: a cascade that matches too widely is as wrong as one that
        // matches too narrowly, and far harder to notice.
        if let Some(found) = report.ok(
            c,
            "get_client(bystander)",
            store.get_client(&bystander).await,
        ) {
            if found.is_none() {
                report.fail(c, "delete_client removed a DIFFERENT client's registration");
            }
        }
        if let Some(found) = report.ok(
            c,
            "get_token(bystander)",
            store.get_token("at-client-bystander").await,
        ) {
            if found.is_none() {
                report.fail(c, "delete_client removed another client's access token");
            }
        }
        if let Some(found) = report.ok(
            c,
            "get_refresh_token(bystander)",
            store.get_refresh_token("rt-client-bystander").await,
        ) {
            if found.is_none() {
                report.fail(c, "delete_client removed another client's refresh record");
            }
        }
        if let Some(found) = report.ok(
            c,
            "get_device_grant(bystander)",
            store.get_device_grant("dc-client-bystander").await,
        ) {
            if found.is_none() {
                report.fail(c, "delete_client removed another client's device grant");
            }
        }

        // Removing a client that is already gone is Ok(false), not an error.
        match store.delete_client(&doomed).await {
            Ok(true) => report.fail(
                DELETE_CLIENT_REPORTS,
                "delete_client answered true for a registration that was already gone",
            ),
            Ok(false) => {}
            Err(e) => report.fail(
                DELETE_CLIENT_REPORTS,
                format!("deleting an absent registration failed with {e}, expected Ok(false)"),
            ),
        }
    }

    // ------------------------------------------------------------------ delete_token

    async fn delete_token(&self, report: &mut Report) {
        let c = DELETE_TOKEN_IDEMPOTENT;
        let store = self.store().await;
        if report
            .ok(
                c,
                "put_token",
                store
                    .put_token(sample_token("at-del", "client-del", None))
                    .await,
            )
            .is_none()
        {
            return;
        }
        if report
            .ok(c, "delete_token", store.delete_token("at-del").await)
            .is_none()
        {
            return;
        }
        if let Some(found) = report.ok(c, "get_token", store.get_token("at-del").await) {
            if found.is_some() {
                report.fail(c, "the token is still readable after delete_token");
            }
        }
        // RFC 7009 section 2.2: an invalid token does not cause an error response, so a repeated
        // revocation (which a client is entitled to send) must not fail.
        if let Err(e) = store.delete_token("at-del").await {
            report.fail(
                c,
                format!("deleting an already-deleted token failed with {e}, expected Ok(())"),
            );
        }
        if let Err(e) = store.delete_token("at-never-existed").await {
            report.fail(
                c,
                format!("deleting a token that never existed failed with {e}, expected Ok(())"),
            );
        }
    }
}

/// Convenience over [`StorageConformance`] for a host with a single-threaded test runtime and no
/// spawner to offer. Read the module docs on what the cooperative mode proves before relying on
/// this one rather than [`StorageConformance::with_spawn`].
pub async fn check_storage<F, Fut, S>(new_store: F) -> Vec<Violation>
where
    F: Fn() -> Fut,
    Fut: Future<Output = S>,
    S: Storage + 'static,
{
    StorageConformance::new(new_store).run().await
}

// ---------------------------------------------------------------------- reporting

#[derive(Default)]
struct Report {
    violations: Vec<Violation>,
}

impl Report {
    fn fail(&mut self, check: &'static str, detail: impl Into<String>) {
        self.violations.push(Violation {
            check,
            detail: detail.into(),
        });
    }

    /// Unwrap a storage result, recording an unexpected failure as a violation rather than
    /// panicking: this harness reports, it does not abort the host's test process.
    fn ok<T>(&mut self, check: &'static str, what: &str, r: Result<T, StorageError>) -> Option<T> {
        match r {
            Ok(v) => Some(v),
            Err(e) => {
                self.fail(check, format!("{what} failed unexpectedly: {e}"));
                None
            }
        }
    }

    fn some<T>(&mut self, check: &'static str, what: &str, v: Option<T>) -> Option<T> {
        if v.is_none() {
            self.fail(
                check,
                format!("{what} returned None for a record that was just stored"),
            );
        }
        v
    }

    /// Compare one field of a round-tripped record. Field by field rather than whole-record, so
    /// the violation names the field that was dropped instead of printing two records and leaving
    /// the reader to diff them.
    fn same<T: PartialEq + fmt::Debug>(
        &mut self,
        check: &'static str,
        field: &str,
        want: &T,
        got: &T,
    ) {
        if want != got {
            self.fail(
                check,
                format!("field {field} did not survive the round trip: stored {want:?}, read back {got:?}"),
            );
        }
    }
}

// ---------------------------------------------------------------------- concurrency primitives
//
// Hand-written because this crate has no async runtime and no futures library, and gains neither
// for a test-only feature. All three are small enough to read.

/// A rendezvous the racers park on so their `take_*` calls are all in flight at once. Without it,
/// a runtime is free to run each spawned task to completion before starting the next, and a
/// read-then-delete store would pass by never overlapping with itself.
pub(crate) struct Gate {
    target: usize,
    arrived: AtomicUsize,
    open: AtomicBool,
    /// True when a racer gave up waiting: the run's atomicity results prove nothing.
    unsatisfied: AtomicBool,
    waiters: Mutex<Vec<Waker>>,
}

impl Gate {
    pub(crate) fn new(target: usize) -> Arc<Self> {
        Arc::new(Gate {
            target,
            arrived: AtomicUsize::new(0),
            open: AtomicBool::new(false),
            unsatisfied: AtomicBool::new(false),
            waiters: Mutex::new(Vec::new()),
        })
    }

    pub(crate) fn wait(self: &Arc<Self>) -> GateWait {
        GateWait {
            gate: Arc::clone(self),
            counted: false,
            budget: GATE_POLL_BUDGET,
        }
    }

    pub(crate) fn unsatisfied(&self) -> bool {
        self.unsatisfied.load(Ordering::SeqCst)
    }

    fn wake_all(&self) {
        let mut waiters = self.waiters.lock().unwrap_or_else(|e| e.into_inner());
        for waker in waiters.drain(..) {
            waker.wake();
        }
    }
}

pub(crate) struct GateWait {
    gate: Arc<Gate>,
    counted: bool,
    budget: u32,
}

impl Future for GateWait {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if !self.counted {
            self.counted = true;
            if self.gate.arrived.fetch_add(1, Ordering::SeqCst) + 1 >= self.gate.target {
                self.gate.open.store(true, Ordering::SeqCst);
                self.gate.wake_all();
                return Poll::Ready(());
            }
        }
        if self.gate.open.load(Ordering::SeqCst) {
            return Poll::Ready(());
        }
        if self.budget == 0 {
            // Nobody else is coming: the racers are being run one at a time. Recorded rather than
            // hung, and reported as `harness/race_setup` so the run is not mistaken for a pass.
            self.gate.unsatisfied.store(true, Ordering::SeqCst);
            return Poll::Ready(());
        }
        self.budget -= 1;
        // Both a registered waker (for a racer parked on another thread) and a self-wake (so a
        // cooperatively polled racer is re-polled and the budget actually counts down).
        self.gate
            .waiters
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(cx.waker().clone());
        if self.gate.open.load(Ordering::SeqCst) {
            return Poll::Ready(());
        }
        cx.waker().wake_by_ref();
        Poll::Pending
    }
}

/// Counts spawned racers to completion. Waker-based rather than spinning: past the gate a racer is
/// doing the store's real work, which may be a network round trip.
pub(crate) struct Latch {
    remaining: AtomicUsize,
    waker: Mutex<Option<Waker>>,
}

impl Latch {
    pub(crate) fn new(target: usize) -> Arc<Self> {
        Arc::new(Latch {
            remaining: AtomicUsize::new(target),
            waker: Mutex::new(None),
        })
    }

    pub(crate) fn done(&self) {
        if self.remaining.fetch_sub(1, Ordering::SeqCst) == 1 {
            if let Some(waker) = self.waker.lock().unwrap_or_else(|e| e.into_inner()).take() {
                waker.wake();
            }
        }
    }

    pub(crate) fn wait(self: &Arc<Self>) -> LatchWait {
        LatchWait {
            latch: Arc::clone(self),
        }
    }
}

/// Releases one count of the [`Latch`] when a spawned racer's task ends, HOWEVER it ends.
///
/// The reason it is a `Drop` guard rather than a call at the bottom of the task: a racer whose
/// store call panics never reaches the bottom of the task, so a plain `latch.done()` there leaves
/// the latch one short and [`StorageConformance::run`] parked forever. A host whose store panics
/// under concurrency would then get a hung test run, which is the worst diagnostic available: it
/// names nothing, it points at nothing, and it looks like the harness is broken rather than the
/// store. `Drop` runs during the unwind, so the latch is released and the harness reports.
///
/// `finished` distinguishes the two ways a task can end. It is set as the LAST statement of the
/// task, so an unwind (or a spawner that dropped the future before it completed) leaves it false
/// and the racer is counted as abandoned.
struct RacerGuard {
    latch: Arc<Latch>,
    abandoned: Arc<AtomicUsize>,
    finished: bool,
}

impl Drop for RacerGuard {
    fn drop(&mut self) {
        if !self.finished {
            self.abandoned.fetch_add(1, Ordering::SeqCst);
        }
        self.latch.done();
    }
}

pub(crate) struct LatchWait {
    latch: Arc<Latch>,
}

impl Future for LatchWait {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if self.latch.remaining.load(Ordering::SeqCst) == 0 {
            return Poll::Ready(());
        }
        *self.latch.waker.lock().unwrap_or_else(|e| e.into_inner()) = Some(cx.waker().clone());
        // Re-check after registering, or a racer that finished in between would leave this parked
        // with nobody left to wake it.
        if self.latch.remaining.load(Ordering::SeqCst) == 0 {
            return Poll::Ready(());
        }
        Poll::Pending
    }
}

/// Polls every racer on ONE task, in order, on every wake. This is the cooperative mode: it
/// interleaves rather than parallelizes, which is enough to catch a read-then-delete store that
/// suspends between the read and the delete (any store that talks to a database does).
pub(crate) struct JoinAll<T> {
    futures: Vec<Option<Pin<Box<dyn Future<Output = T> + Send>>>>,
    done: Vec<Option<T>>,
}

impl<T> JoinAll<T> {
    pub(crate) fn new(futures: Vec<Pin<Box<dyn Future<Output = T> + Send>>>) -> Self {
        let mut done = Vec::with_capacity(futures.len());
        done.resize_with(futures.len(), || None);
        JoinAll {
            futures: futures.into_iter().map(Some).collect(),
            done,
        }
    }
}

// `T: Unpin` is not a restriction in practice: T is the take's `Result`, and the futures
// themselves are boxed (and so `Unpin`) precisely so this combinator can be written without
// unsafe code.
impl<T: Unpin> Future for JoinAll<T> {
    type Output = Vec<T>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Vec<T>> {
        let JoinAll { futures, done } = self.get_mut();
        let mut pending = false;
        for (slot, out) in futures.iter_mut().zip(done.iter_mut()) {
            if let Some(fut) = slot {
                match fut.as_mut().poll(cx) {
                    Poll::Ready(v) => {
                        *out = Some(v);
                        *slot = None;
                    }
                    Poll::Pending => pending = true,
                }
            }
        }
        if pending {
            return Poll::Pending;
        }
        Poll::Ready(done.iter_mut().filter_map(Option::take).collect())
    }
}

// ---------------------------------------------------------------------- fixtures
//
// Every field carries a DISTINCTIVE value, because the failure this harness exists to catch is a
// store that silently drops one. Timestamps are whole seconds from a fixed base rather than
// `SystemTime::now()`: sweeps are then deterministic, and a store whose column has one-second
// resolution is not failed for something that is not a contract violation.

const BASE_SECS: u64 = 1_800_000_000;

fn at(offset_secs: u64) -> SystemTime {
    SystemTime::UNIX_EPOCH + Duration::from_secs(BASE_SECS + offset_secs)
}

fn at_before(offset_secs: u64) -> SystemTime {
    SystemTime::UNIX_EPOCH + Duration::from_secs(BASE_SECS - offset_secs)
}

fn scopes(s: &str) -> ScopeSet {
    // The literals below are this module's own and are all valid RFC 6749 section 3.3 tokens.
    ScopeSet::parse(s).unwrap_or_else(|_| ScopeSet::empty())
}

/// The RFC 9396 section 2 `authorization_details` every record fixture carries, as the raw text a
/// client would push.
///
/// Non-empty on purpose, and that is the whole point of it existing. An empty
/// `AuthorizationDetails` is the DEFAULT, so a fixture carrying one cannot tell a store that
/// preserves the field from a store that drops it: both read back empty. This crate has already
/// been bitten twice by RAR details being dropped on a feature-gated path, which is exactly the
/// defect a host's store can have and exactly what this harness exists to make visible.
///
/// One element with several section 2.2 common fields, so a store that truncates the JSON, keeps
/// only the `type`, or round-trips it through a lossy column is caught as well as one that drops
/// the column outright.
#[cfg(feature = "rar")]
const AUTHORIZATION_DETAILS_JSON: &str = r#"[{"type":"conformance-fixture","locations":["https://rs-one.example/"],"actions":["read","write"],"identifier":"account-4711"}]"#;

/// The parsed form of [`AUTHORIZATION_DETAILS_JSON`]. Parsed rather than constructed because
/// `AuthorizationDetail`'s members are the RFC's, not this module's, and the parser is the only
/// thing that has to agree with them.
///
/// A parse failure would leave the fixture EMPTY and the round-trip check unable to see a dropped
/// field, which is silent, so `the_fixtures_carry_the_fields_the_round_trip_checks_exist_for` in
/// `src/tests/storage_conformance.rs` pins it as non-empty rather than trusting the literal.
#[cfg(feature = "rar")]
fn sample_authorization_details() -> crate::rar::AuthorizationDetails {
    crate::rar::AuthorizationDetails::parse(AUTHORIZATION_DETAILS_JSON)
        .unwrap_or_else(|_| crate::rar::AuthorizationDetails::none())
}

/// What the host reported about how it authenticated the user, on every record that carries it.
///
/// `Some`, not `None`, for the reason [`sample_authorization_details`] is non-empty: `None` is the
/// default, so a `None` fixture certifies a store that drops the field. RFC 9470 section 5 is
/// answered from this, so a store that loses it has disabled step-up authentication for the whole
/// deployment while every request continues to succeed.
///
/// `acr` is set as well as `auth_time`: a store that persists the timestamp and drops the class
/// (two columns, one migration) satisfies `max_age` and silently fails every `acr_values` request.
#[cfg(feature = "consent")]
fn sample_authentication() -> Option<Box<crate::consent::Authentication>> {
    Some(Box::new(crate::consent::Authentication {
        auth_time: at_before(120),
        acr: Some("urn:conformance:acr:multi-factor".into()),
    }))
}

fn sample_client(client_id: &str) -> Client {
    Client {
        client_id: ClientId::new(client_id),
        auth: ClientAuth::ConfidentialSecretHash {
            hash: SecretHash::sha256("conformance-secret"),
        },
        grant_types: vec![
            GrantType::AuthorizationCode,
            GrantType::RefreshToken,
            GrantType::DeviceCode,
        ],
        redirect_uris: vec![
            "https://app.example/cb".to_string(),
            "https://app.example/cb2".to_string(),
        ],
        allowed_scopes: scopes("read write admin"),
        default_scopes: scopes("read"),
        name: Some("conformance client".to_string()),
        registration: Some(Box::new(DynamicRegistration {
            registration_access_token_hash: SecretHash::sha256("conformance-rat"),
            client_id_issued_at: Some(BASE_SECS),
            client_secret_expires_at: Some(0),
            token_endpoint_auth_method: "client_secret_basic".to_string(),
        })),
    }
}

fn sample_device_grant(device_code: &str, user_code: &str) -> DeviceGrant {
    DeviceGrant {
        device_code: device_code.to_string(),
        user_code: user_code.to_string(),
        client_id: ClientId::new("client-conformance"),
        scope: scopes("read write"),
        state: DeviceGrantState::Approved {
            subject: "subject-conformance".to_string(),
        },
        created_at: at_before(30),
        expires_at: at(600),
        interval: Duration::from_secs(7),
        last_poll_at: Some(at_before(5)),
    }
}

/// An APPROVED device grant for `subject`. Approved rather than pending on purpose: the consent
/// cascade must reach a grant the user already approved but whose device has not polled yet,
/// which is precisely the window where a withdrawal that misses it hands out a token AFTER the
/// user withdrew. A pending grant is not part of any consent and must survive.
#[cfg(feature = "consent")]
fn sample_approved_device_grant(device_code: &str, user_code: &str, subject: &str) -> DeviceGrant {
    DeviceGrant {
        state: DeviceGrantState::Approved {
            subject: subject.to_string(),
        },
        ..sample_device_grant(device_code, user_code)
    }
}

/// A consent for `subject`, with a scope and a resource so `covers` has something to answer about
/// and a round trip has something to lose.
#[cfg(feature = "consent")]
fn sample_consent(consent_id: &str, subject: &str) -> crate::consent::ConsentRecord {
    crate::consent::ConsentRecord {
        consent_id: consent_id.into(),
        client_id: ClientId::new("client-conformance"),
        subject: subject.into(),
        scope: scopes("read write"),
        resource: vec!["https://rs-one.example/".to_string()],
        granted_at: at_before(60),
        // See `sample_authentication`. `None` here would have made `round_trip/consent` unable to
        // see a store that drops the RFC 9470 step-up state the consent was granted under.
        authentication: sample_authentication(),
    }
}

/// A pushed authorization request, complete enough that a store dropping a field on the way
/// through is visible rather than plausible.
#[cfg(feature = "par")]
fn sample_pushed_request(request_uri: &str) -> crate::par::PushedAuthorizationRequest {
    crate::par::PushedAuthorizationRequest {
        request_uri: request_uri.to_string(),
        client_id: ClientId::new("client-conformance"),
        response_type: Some("code".to_string()),
        redirect_uri: Some("https://app.example/cb".to_string()),
        scope: Some("read write".to_string()),
        state: Some("state-conformance".to_string()),
        code_challenge: Some("E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM".to_string()),
        code_challenge_method: Some("S256".to_string()),
        resource: vec!["https://rs-one.example/".to_string()],
        // Populated for the same reason `acr_values` and `max_age` below are: RFC 9101 section 6.3
        // has the authorization endpoint use ONLY the pushed parameters, so a detail lost between
        // the push and the read is a detail the client was told was acceptable and then did not
        // get. `None` is the default and could not have shown that.
        #[cfg(feature = "rar")]
        authorization_details: Some(AUTHORIZATION_DETAILS_JSON.to_string()),
        // RFC 9470 s4. Populated rather than `None` for this fixture's stated reason: a store that
        // drops one of them on the way through has disabled step-up for every PAR request, and the
        // point of this record is that such a drop is visible.
        #[cfg(feature = "consent")]
        acr_values: Some("urn:acr:phr".to_string()),
        #[cfg(feature = "consent")]
        max_age: Some("300".to_string()),
        expires_at: at(60),
    }
}

fn sample_authorization_code(code: &str) -> AuthorizationCodeRecord {
    AuthorizationCodeRecord {
        code: code.to_string(),
        client_id: ClientId::new("client-conformance"),
        redirect_uri: "https://app.example/cb".to_string(),
        scope: scopes("read write"),
        subject: "subject-conformance".to_string(),
        code_challenge: "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM".to_string(),
        code_challenge_method: CodeChallengeMethod::S256,
        resource: vec![
            "https://rs-one.example/".to_string(),
            "https://rs-two.example/".to_string(),
        ],
        #[cfg(feature = "rar")]
        authorization_details: sample_authorization_details(),
        expires_at: at(60),
        state: AuthorizationCodeState::Consumed {
            access_token: Some("at-minted-by-this-code".to_string()),
            refresh_token: Some("rt-minted-by-this-code".to_string()),
        },
        #[cfg(feature = "consent")]
        authentication: sample_authentication(),
    }
}

fn sample_token(access_token: &str, client_id: &str, family_id: Option<&str>) -> IssuedToken {
    IssuedToken {
        access_token: access_token.to_string(),
        client_id: ClientId::new(client_id),
        subject: Some("subject-conformance".to_string()),
        scope: scopes("read write"),
        resource: vec![
            "https://rs-one.example/".to_string(),
            "https://rs-two.example/".to_string(),
        ],
        #[cfg(feature = "rar")]
        authorization_details: sample_authorization_details(),
        issued_at: at_before(10),
        expires_at: at(3600),
        family_id: family_id.map(str::to_string),
        // RFC 9449 s6: the key this token is bound to. A store that drops it turns a
        // sender-constrained token back into a bearer token, and nothing on the token plane
        // notices, because a token that verifies with no binding is exactly what a bearer token
        // is.
        #[cfg(feature = "dpop")]
        jkt: Some("0ZcOCORZNYy-DWpqq30jZyJGHTN0d2HglBV3uiguA4I".into()),
        // RFC 8705 s3.1: the client certificate this token is bound to. Dropping it has the same
        // consequence as dropping `jkt` above, by the other mechanism: a certificate-bound token
        // silently becomes a bearer token, and nothing on the token plane can tell.
        #[cfg(feature = "mtls")]
        x5t_s256: Some(Box::new(crate::mtls::CertificateThumbprint::from_der(
            b"conformance-fixture-certificate",
        ))),
        #[cfg(feature = "consent")]
        authentication: sample_authentication(),
    }
}

fn sample_refresh(refresh_token: &str, client_id: &str, family_id: &str) -> RefreshTokenRecord {
    RefreshTokenRecord {
        refresh_token: refresh_token.to_string(),
        client_id: ClientId::new(client_id),
        subject: Some("subject-conformance".to_string()),
        scope: scopes("read write"),
        resource: vec![
            "https://rs-one.example/".to_string(),
            "https://rs-two.example/".to_string(),
        ],
        #[cfg(feature = "rar")]
        authorization_details: sample_authorization_details(),
        expires_at: Some(at(86_400)),
        family_id: family_id.to_string(),
        state: RefreshTokenState::Spent,
        // RFC 9449 s5. Dropped here, a stolen refresh token can be re-bound to the thief's key on
        // the next rotation, which leaves the attacker holding a provable token and the victim
        // holding the key that gets refused.
        #[cfg(feature = "dpop")]
        jkt: Some("0ZcOCORZNYy-DWpqq30jZyJGHTN0d2HglBV3uiguA4I".into()),
        // RFC 8705 s3.1: the client certificate this token is bound to. Dropping it has the same
        // consequence as dropping `jkt` above, by the other mechanism: a certificate-bound token
        // silently becomes a bearer token, and nothing on the token plane can tell.
        #[cfg(feature = "mtls")]
        x5t_s256: Some(Box::new(crate::mtls::CertificateThumbprint::from_der(
            b"conformance-fixture-certificate",
        ))),
        #[cfg(feature = "consent")]
        authentication: sample_authentication(),
    }
}

#[cfg(test)]
#[path = "tests/storage_conformance.rs"]
mod tests;