[thinhphucnguyen]

External IdP assertions can be presented to the token endpoint more than
once while they are valid. This adds the cache that remembers the jti of
accepted assertions so the exchange handler can refuse a second use.

What changed:

• replay.rs: the bounded (scope, id) cache from DpopReplayCache,
  generalised so the caller passes the expiry. Same rules as before: bounded
  per scope and in total, and a live entry is never evicted to make room.
• dpop_replay.rs: DpopReplayCache is now a thin wrapper around it. Public
  API and DpopReplayError are unchanged, and I did not touch its tests.
• assertion_replay.rs: AssertionReplayCache, keyed by (domain, issuer,
  jti). An entry lives until exp plus the profile's clock skew, the last
  moment the token can still pass verify_subject_token_jwt. It reads the
  skew from the profile constant, so the two cannot drift apart (the only
  change in verify.rs is pub(crate) on that constant).

Nothing calls the new cache yet. Wiring and config limits come after.

Tested with the new unit tests plus the existing DPoP ones, run one test per
process the way nextest does.

[gtema]

Here we come to the dirty reality. You add a in-process cache which only protects a single node, but the request landing on the other node would work. The WebAuthn has similar issue where a short-living auth state must be shared across all cluster instances. This is done through the Raft storage as the 'ephemeral' value (kept in logs for replay, but not stored in the DB). Here we have a pretty similar usecase.
It doesn't make the PR useless, but first we need to think further on how this should be integrated with each other. One option is to introduce the SingleUseStore trait and have few implementations for it: local and cluster/raft. This PR would then define the trait and implement it for ReplayCache while in the next PR we can add a raft based implementation.
I need to think on this deeper, so for now I would suggest you focus on something else. I will make it a draft for now to signal work in progress

[gtema]

the foundation work is layed in PR 1641. Once it lands you can do following (after rebase):

1. Keep ReplayCache as the sync engine. Its (scope, id, expires, now) shape and error cases already match the contract. Map ReplayError to SingleUseError:

   • Replayed → Replayed
   • CacheFull and ScopeQuotaExceeded → Capacity

   InvalidLifetime is not produced by the engine. The wrapper in step 3 does that.

2. Add the per-domain bound to ReplayCache.

   • Add max_per_domain: Option to the constructor (None for DPoP and TOTP).
   • Track a domain → count map. Decrement it in purge and when an overwrite replaces an expired entry. Increment it only on a fresh insert.
   • Read the domain with single_use::assertion_scope_domain(scope). A scope that doesn't parse, with the bound on, is a caller bug and returns InvalidLifetime.
   • The saved patch local_single_use_store.rs in my scratchpad (/tmp/claude-1001/-home-agoncharov-workspace-github-com-openstack-experimental-keystone-worktrees-replaycache/92371f37-27cd-4162-8a98-9383569b8051/scratchpad/) has the working logic and tests (domain_bound_spans_issuers, malformed scope) to lift.

3. Add LocalSingleUseStore, a thin async wrapper. Hold a ReplayCache and max_lifetime_secs, and impl SingleUseStore:

   • consume calls single_use::check_preconditions(id, expires, now, max_lifetime_secs)?, then the sync engine, then the error mapping.
   • cleanup calls purge_expired and returns Ok.

4. Run the shared suite. conformance::run_suite(|total, per_scope| LocalSingleUseStore::new(...)). The factory must use max_lifetime_secs = conformance::MAX_LIFETIME_SECS (600). Keep the existing replay.rs unit tests as engine tests. Drop any that the suite now duplicates, such as test_entry_dies_exactly_at_expires.

5. AssertionReplayCache:

   • Delete the private scope(). It is already identical to single_use::assertion_scope, so call that instead. Keep test_scope_is_injective, now pointing at the shared helper.
   • Keep the exp + skew + 1 expiry, reading PROFILE_CLOCK_SKEW as now.
   • Constructor takes max_per_issuer, max_per_domain and max_total, passed from assertion_replay_cache_max_entries[_per_domain|_per_issuer].
   • Make it a helper over Arc, so the handler can later swap in the Raft store without a signature change. It should be async, with check_and_insert(domain_id, issuer, jti, exp, now) calling store.consume(...). Map SingleUseError to the section 6 errors when the handler is wired.
   • Reject a jti that is missing, empty or over 256 bytes with invalid_grant before consume. That belongs with the wiring, but note it in the docs now.

6. DpopReplayCache: keep it a sync wrapper over ReplayCache with its public API and DpopReplayError unchanged. It serves resource requests, which stay local by default. For /token, the ADR makes a second use case later. That is not for this PR; only construct it so a second instance is possible.

7. Docs:

   • Module documentation of replay.rs, assertion_replay.rs and dpop_replay.rs says per node only.
   • State that callers record after the mapping match, never before.
   • Reference ADR 0044 sections 1 and 2.
     Both ReplayCache and LocalSingleUseStore should live under crates/core/src/single_use/ since they would be reused by other places and are not oauth2 specific. oauth2_client keeps only thin per-use-case adapters: assertion_replay.rs, dpop_replay.rs