"""``@idempotent`` — one-line ``Idempotency-Key`` support for a route handler. ## Why this exists ``core/idempotency.py`` gives us a solid primitive, but wiring it into a handler is six lines of boilerplate (declare the ``Header`` param, build the service, ``lookup()``, early-return, ``record()``, ``mark_failed()`` on the error path). That cost meant it got wired into **5 endpoints** — while the mobile write primitive (``useMobileMutation``) mints an ``Idempotency-Key`` for *every* mobile write and attaches it to ~30 endpoints that silently ignore it. Silently is the problem. The mobile hook's contract says "the SAME key is reused across an in-flight retry, so a network blip after the server already processed the write does not double-execute — provided the endpoint honours the key". For most endpoints that proviso was false, so a flaky-network retry could genuinely double-execute a state transition (approve a grant twice → two provider dispatches; advance an offboarding twice; revoke a seat twice). The client-side re-entry guard only covers a double TAP inside one page instance; it cannot cover a retry, a second tab, or a second device. ## Usage ```python @router.post("/grants/{grant_id}/approve") @idempotent("cloud_grant.approve", key_params=("grant_id",)) async def approve_grant( grant_id: uuid.UUID, request: Request, db: AsyncSession = Depends(get_db), ... ): ... return success_response(data=..., request=request) ``` That is the whole change per endpoint. The decorator: * reads ``Idempotency-Key`` off the request (no key → straight through, so behaviour for non-opted-in clients is byte-identical to before); * refuses malformed keys (oversized/non-printable) with a 422 envelope BEFORE any DB work — an oversized key used to land verbatim in a ``String(128)`` column and 500 with a MySQL 1406; * scopes the reservation by ``scope`` plus the values of ``key_params``, so one client key cannot collide across different resources; * binds the key to the request body (SHA-256 of the raw bytes): the same key reused with a DIFFERENT payload answers 422 (``IDEMPOTENCY_PAYLOAD_MISMATCH``) instead of silently replaying the first call's success for a write that never happened. Form-encoded and multipart requests are the one exception — see requirement 4 below; * returns the cached envelope on a successful replay; * raises ``ConflictError`` while a first call is still running (the primitive's semantics — the client retries later with the same key); * marks the reservation ``failed`` when the handler raises, so a legitimate retry is not blocked until TTL. ## Requirements on the decorated handler 1. It must take a ``request: Request`` parameter (all of ours do — it is needed for ``success_response``). The decorator reads the header from it rather than declaring its own ``Header`` param, which keeps the handler's OpenAPI signature untouched. 2. It must return the ``success_response`` dict (or any JSON-able value). ``JSONResponse``/``StreamingResponse`` returns are NOT cacheable — the decorator refuses them loudly at call time rather than caching something that replays wrong. 3. Its caller must commit. The decorator only ``flush()``es (via the service), exactly like the hand-wired call sites: the reservation lands in the same transaction as the work, so a rolled-back handler leaves no completed reservation behind. A handler that commits INTERNALLY (most of ours do) breaks the symmetry of that arrangement on the failure path: the reservation is durable by the time a later line raises, and ``get_db``'s rollback then undoes the ``failed`` flush, leaving the row ``pending`` and every retry answering 409 for the full TTL. The decorator therefore also schedules a deferred release on a fresh session (see ``_deferred_release``). Handlers do not need to know which case they are in. 4. A ``Form``/``UploadFile`` handler gets replay protection but NO payload binding. FastAPI parses those bodies with ``request.form()``, which drains the receive stream without filling Starlette's ``_body`` cache, so the fingerprint read (``request.body()``) would raise ``RuntimeError: Stream consumed`` → 500. The decorator detects the form content types and stores a NULL fingerprint instead, which is the same "not fingerprinted" state every hand-wired call site is in. No wired endpoint takes a form today. 5. A handler with NO body parameter (``cloud_grant.renew`` is the only one) makes the decorator's fingerprint read the FIRST read of the request stream — FastAPI never touched it. A client that hung up would therefore raise ``ClientDisconnect`` before the handler and silently skip work that used to happen, so that read degrades to "no fingerprint" (same NULL state as requirement 4) instead of aborting. """ from __future__ import annotations import asyncio import copy import functools import hashlib import inspect import logging import uuid from collections.abc import Callable, Sequence from typing import Any, TypeVar from fastapi import Request from sqlalchemy.ext.asyncio import AsyncSession from starlette.requests import ClientDisconnect from starlette.responses import Response from app.core.background import spawn from app.core.exceptions import IdempotencyKeyInvalidError from app.core.idempotency import ( IN_PROGRESS_TAKEOVER_MINUTES as _DEFAULT_TAKEOVER, ) from app.core.idempotency import ( IdempotencyService, release_reservation, ) F = TypeVar("F", bound=Callable[..., Any]) _HEADER = "Idempotency-Key" # Must fit ``core_idempotency_records.key`` (String(128)). A longer key # went into the INSERT verbatim and raised MySQL 1406 (strict mode) → an # unhandled DataError 500 on every wired endpoint. _MAX_KEY_LENGTH = 128 # ``core_idempotency_records.scope`` is the OTHER String(128) column, and # it has its own 1406 history (see the model's comment): the composed # ``f"{scope}:{suffix}"`` must fit it too. Longest wired scope today is # 98 chars. _MAX_SCOPE_LENGTH = 128 # ``str(uuid.UUID(...))`` — every wired ``key_param`` resolves to a # resource id or a person id, so ``len(scope) + len(key_params) * (1 + # _UUID_STR_LENGTH)`` is a bound the decorator can check at DECORATION # time (see ``decorator``). Longest wired scope projects to 98. _UUID_STR_LENGTH = 36 # Content types FastAPI parses with ``request.form()`` instead of # ``request.body()``. ``form()`` drains the receive stream WITHOUT # populating Starlette's ``_body`` cache, so a later ``body()`` raises # ``RuntimeError: Stream consumed`` — fingerprinting is skipped for these # rather than 500ing the endpoint (see the wrapper). _FORM_CONTENT_TYPES = frozenset({ "multipart/form-data", "application/x-www-form-urlencoded", }) # Long enough for the request's own transaction to roll back and release the # row lock, short enough that a user retrying by hand is not the one waiting. _RELEASE_DELAY_SECONDS = 1.0 async def _deferred_release( scope: str, key: str, row_id: uuid.UUID | None, ) -> None: """Release THIS request's reservation once the request has unwound.""" await asyncio.sleep(_RELEASE_DELAY_SECONDS) await release_reservation(scope, key, row_id) def _find_kwarg( name: str, kwargs: dict[str, Any], expected_type: type | None = None, ) -> Any: value = kwargs.get(name) if value is not None: return value if expected_type is not None: for candidate in kwargs.values(): if isinstance(candidate, expected_type): return candidate return None def _resolve_path(kwargs: dict[str, Any], path: str) -> Any: """``"current_user.id"`` → ``kwargs["current_user"].id``. Attribute access is what makes an identity-scoped key usable: the raw object's ``str()`` carries a memory address, so using it would produce a unique scope per request and quietly turn the whole guard into a no-op. """ root, _, rest = path.partition(".") value = kwargs.get(root) for attr in filter(None, rest.split(".")): value = getattr(value, attr, None) if value is None: break if value is None: # Loud, not silent: a scope of ``family:None`` would be SHARED by # every caller, so one user's cached response could be replayed to # another. Refusing is the only safe outcome. raise RuntimeError( f"@idempotent: key_param {path!r} resolved to None; the " "reservation scope would be shared across callers" ) return value def _redact(payload: Any, paths: Sequence[str], scope: str = "") -> Any: """Return a copy of ``payload`` with ``paths`` blanked, marking each. Used for responses that carry a **one-time secret**. The first caller still receives the real value; only the copy that gets persisted for replay is stripped. """ if not paths: return payload if not isinstance(payload, dict): # Same reasoning as the path-mismatch branch below: by the time we # get here the handler has run and probably committed, so raising # would turn a succeeded operation into a 500 and strand the # reservation. Cache nothing rather than cache a shape we cannot # prove is secret-free. logging.getLogger(__name__).error( "@idempotent(%s): redact_cached=%s but the handler returned %s; " "caching an empty body instead.", scope, list(paths), type(payload).__name__, ) return {"success": True, "data": None, "redacted_on_replay": True} clone = copy.deepcopy(payload) for path in paths: parts = path.split(".") node: Any = clone for part in parts[:-1]: if not isinstance(node, dict) or part not in node: node = None break node = node[part] leaf = parts[-1] if not isinstance(node, dict) or leaf not in node: # The path stopped matching — a renamed field, a reshaped # envelope. Never fall through to caching the payload: the # whole reason this argument exists is that the response holds # a live credential, and a silent skip would put it back into # `core_idempotency_records` with nothing saying so. # # But do not raise either. By this point the handler has run # and very likely committed, so raising turns a SUCCEEDED # operation into a 500 and (being outside the failure path) # strands the reservation `pending` for the full TTL — the # user pays for a source-code mistake. Instead drop the whole # payload from the cache and shout in the log; the developer # signal already exists in CI (see # `test_the_pat_endpoint_still_declares_the_field_it_redacts`). logging.getLogger(__name__).error( "@idempotent(%s): redact_cached path %r did not match the " "response; caching a stripped body instead of a possible " "secret. Fix the path.", scope, path, ) stripped = {k: v for k, v in clone.items() if k != "data"} stripped["data"] = None stripped["redacted_on_replay"] = True return stripped node[leaf] = None # The client cannot tell "field was null" from "field was # withheld" without this, and the difference decides whether it # tells the user to revoke and start over. node[f"{leaf}_withheld_on_replay"] = True return clone def idempotent( scope: str, *, key_params: Sequence[str] = (), ttl_hours: int | None = None, redact_cached: Sequence[str] = (), takeover_minutes: int | None = _DEFAULT_TAKEOVER, ) -> Callable[[F], F]: """Wrap a route handler with ``Idempotency-Key`` replay protection. ``scope`` is the endpoint family (``"cloud_grant.approve"``). ``key_params`` names handler parameters whose values are appended to the scope — pass the resource id(s) so the same client key used against two different resources reserves two different operations. Dotted paths are supported for attribute access: ``"current_user.id"`` uses the caller's id rather than ``str(AuthUser)``, which would embed a per-process memory address and make every request a fresh scope — silently disabling the dedupe it was added for. Values are stringified; the combined scope must fit ``core_idempotency_records.scope`` (128 chars) — every ``family:uuid`` pair does comfortably. Two guards, neither of which 1406s or silently truncates: the uuid-sized projection is refused at DECORATION time (so a bad wiring fails at import, like the other misuse checks), and a value that turns out longer than a uuid is refused at call time. """ def decorator(func: F) -> F: sig = inspect.signature(func) param_names = list(sig.parameters) for p in key_params: root = p.split(".", 1)[0] if root not in param_names: raise ValueError( f"@idempotent({scope!r}): key_param {p!r} is not a " f"parameter of {func.__qualname__}" ) if "request" not in param_names: raise ValueError( f"@idempotent({scope!r}): {func.__qualname__} must take a " "`request: Request` parameter (the decorator reads the " "Idempotency-Key header from it)" ) # Scope length is statically boundable for the uuid-valued # key_params every wired endpoint uses, so check it HERE, with the # other misuse checks: the next over-long scope then fails at # import (CI, every test run) instead of reaching production and # 500ing on the first request. The request-time check below stays # — it is the only thing that covers a key_param resolving to # something longer than a uuid (a slug, a composed ref), which # nothing here can predict. projected = len(scope) + len(key_params) * (1 + _UUID_STR_LENGTH) if projected > _MAX_SCOPE_LENGTH: raise ValueError( f"@idempotent({scope!r}): scope plus {len(key_params)} " f"uuid key_param(s) composes to {projected} chars, " f"exceeding the {_MAX_SCOPE_LENGTH}-char " "core_idempotency_records.scope column; shorten the scope " "name or use fewer key_params" ) @functools.wraps(func) async def wrapper(*args: Any, **kwargs: Any) -> Any: # Route handlers are always called by FastAPI with keyword # arguments; bind positionals defensively so a direct unit-test # call still works. if args: bound = sig.bind_partial(*args, **kwargs) kwargs = dict(bound.arguments) args = () request: Request | None = _find_kwarg("request", kwargs, Request) db: AsyncSession | None = _find_kwarg("db", kwargs, AsyncSession) # No request to read the header off, or no session to record # against → behave exactly as the undecorated handler. Narrowed # here (not folded into the ``not key`` check) so the rest of # the wrapper can use ``request``/``db`` without an assert. if request is None or db is None: return await func(**kwargs) raw_key = request.headers.get(_HEADER) if not raw_key: return await func(**kwargs) # Before ANY DB work: the key lands verbatim in a String(128) # column, so an oversized/garbage key must be a 422 envelope, # not a MySQL 1406 → 500. Measured on the STRIPPED value — # that is what ``IdempotencyService`` stores, so measuring the # raw header would 422 a key that fits. key = raw_key.strip() if ( not key or len(key) > _MAX_KEY_LENGTH or not key.isprintable() ): raise IdempotencyKeyInvalidError( f"Idempotency-Key must be 1-{_MAX_KEY_LENGTH} printable " "characters" ) # Bind the key to this exact payload. FastAPI parses a JSON / # raw body through ``request.body()``, which Starlette caches on # the Request, so the read below is free and changes nothing # downstream. It is NOT free for a ``Form``/``UploadFile`` # route: FastAPI parses those with ``request.form()``, which # drains the receive stream without populating ``_body``, and # ``body()`` then raises ``RuntimeError: Stream consumed``. Such # a route is left unfingerprinted (NULL, exactly like a # hand-wired call site — the payload-mismatch check is skipped, # everything else still applies) instead of 500ing. media_type = ( request.headers.get("content-type", "") .split(";", 1)[0].strip().lower() ) fingerprint: str | None = None if media_type not in _FORM_CONTENT_TYPES: try: body = await request.body() except (ClientDisconnect, RuntimeError) as exc: # Degrade to "not fingerprinted" (NULL → the payload # comparison is skipped, exactly like a hand-wired # call site) instead of aborting the request. # # This must not abort, because for the one wired # endpoint with NO body parameter # (``cloud_grant.renew``) THIS is the first read of # the stream, so a client that hung up would turn a # renewal that used to happen into nothing at all — # a silent behaviour change caused purely by adding # replay protection. And it cannot mask a disconnect # that SHOULD abort: for every endpoint that takes a # body, FastAPI parses it during dependency solving # BEFORE calling the handler, so a truncated body # has already raised by the time we get here and the # read below only ever hits Starlette's cache. # Server-side, finishing the write is also what the # client wants — it retries with the same key and # gets the cached envelope. logging.getLogger(__name__).warning( "@idempotent(%s): body unreadable (%s: %s); " "recording the reservation without a payload " "fingerprint.", scope, type(exc).__name__, exc, ) body = None if body is not None: fingerprint = hashlib.sha256(body).hexdigest() full_scope = scope if key_params: suffix = ":".join( str(_resolve_path(kwargs, p)) for p in key_params ) full_scope = f"{scope}:{suffix}" if len(full_scope) > _MAX_SCOPE_LENGTH: # Not client input: ``scope`` is authored in source and # the suffix is server-side ids, so this is a # programming error and a 4xx would blame the caller. # Refusing beats the two silent alternatives — a 1406 # 500 on every call, or a truncated scope that merges # two different operations' reservations. raise RuntimeError( f"@idempotent({scope!r}): composed scope is " f"{len(full_scope)} chars, exceeding the " f"{_MAX_SCOPE_LENGTH}-char " "core_idempotency_records.scope column; shorten the " "scope name or use fewer key_params" ) service = IdempotencyService( db, scope=full_scope, key=key, takeover_minutes=takeover_minutes, request_fingerprint=fingerprint, **({"ttl_hours": ttl_hours} if ttl_hours is not None else {}), ) cached = await service.lookup() if cached is not None: return cached # Snapshot the reserved row's id NOW, into a plain local. The # exception path below needs it, and by then the session may be # in pending-rollback (a failed flush inside the handler) or # past a commit — both states make reading it off the ORM # object emit a refresh SELECT that raises, masking the # handler's real exception and skipping the release. Here the # row was just flushed, so the value is present and cheap. reserved = service.reserved_row_id try: result = await func(**kwargs) if isinstance(result, Response): # Inside the try, so the refusal path below also settles # the reservation — the handler already ran (and likely # committed), and a stranded ``pending`` row would block # every retry for the takeover window or full TTL. raise TypeError( f"@idempotent({scope!r}): {func.__qualname__} " f"returned a {type(result).__name__}, which cannot " "be cached for replay. Return the success_response " "dict instead, or drop the decorator." ) except Exception: # Free the reservation so a legitimate retry is not blocked # for the full TTL. TWO paths, because one is not enough: # # * a deferred release on its OWN session — required # because most decorated handlers commit() internally, so # the reservation is already durable when a later line # raises, and `get_db`'s rollback then undoes the flush # below, leaving the row 'pending' and every retry 409; # * in-session flush — right when the handler and the # reservation live or die together. # # The release is deferred rather than awaited inline: the # request's transaction still holds the row lock at this # point, so an immediate UPDATE from a second session would # block on it until this one unwinds. # # Spawn FIRST, off the id snapshotted BEFORE the handler ran: # when the handler failure was DB-shaped (deadlock 1213 / # lock-wait 1205 / IntegrityError) the request session is # poisoned, and anything that touches it here — including # reading ``id`` off the reservation object — raises # PendingRollbackError, which must neither mask the original # exception nor skip the spawn. if reserved is not None: release = _deferred_release(full_scope, key, reserved) try: spawn( release, name=f"idempotency-release:{full_scope}", ) except RuntimeError: # ``create_task`` only raises this with no running # loop, which cannot happen inside an awaited # coroutine — so this branch is unreachable today. # It is kept rather than dropped because without it # ANY future raise out of ``spawn`` (a swapped # implementation, a shutting-down worker) would # propagate from inside this ``except`` block and # REPLACE the handler's real exception — the exact # masking this path exists to prevent. Closing the # coroutine is what makes the branch correct: it was # constructed before ``spawn``, so leaving it # un-awaited would emit a "coroutine was never # awaited" RuntimeWarning on top of the failure. release.close() logging.getLogger(__name__).warning( "@idempotent(%s): could not schedule the " "deferred release.", full_scope, exc_info=True, ) # The ``raise`` below is unconditional, so the caller # never continues on this session — and the session is very # likely ALREADY poisoned (that is the case this branch # exists for). Opening a savepoint on a dead transaction # would raise and mask the handler's real exception, which # is the one thing this path must not do. # savepoint-exempt: handler re-raises unconditionally; session not reused try: await service.mark_failed() except Exception: logging.getLogger(__name__).warning( "@idempotent(%s): in-session mark_failed flush " "failed (session poisoned by the handler error); " "the deferred release covers the cleanup.", full_scope, exc_info=True, ) raise # Persist a redacted copy when the response carries a one-time # secret. THIS caller still gets the real thing; what must not # happen is the secret sitting in `core_idempotency_records` for # the full TTL, readable by anything that can read the database # — which would quietly retract the "the server stores only a # hash" promise the endpoint makes to the user. await service.record(_redact(result, redact_cached, scope)) return result return wrapper # type: ignore[return-value] return decorator