From ae468edca29f7b583f36b9f9211908e7a59528d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 5 Jun 2026 13:44:22 +0200 Subject: [PATCH] fix(idempotency): claim the key before running the side-effect findCached returned null and the response was only persisted after the handler completed, so two concurrent requests carrying the same Idempotency-Key both passed the lookup and both executed the side-effect (duplicate writes for the very clients idempotency protects). Claim the key by inserting a pending sentinel row under the existing (userId, key, method, path) unique constraint before the handler runs. A racing request either finds the pending row or loses the insert race and is refused with 409. The claim is promoted to the completed response on a cachable result, released on a non-cachable result or a secret-shaped body, and released on a handler throw. The pending row carries a short claim TTL so a crashed handler self-heals and the existing expiry-sweep cleanup still applies. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/lib/__tests__/idempotency.test.ts | 124 +++++++++++++++++- src/lib/idempotency.ts | 176 +++++++++++++++++++++++--- 2 files changed, 276 insertions(+), 24 deletions(-) diff --git a/src/lib/__tests__/idempotency.test.ts b/src/lib/__tests__/idempotency.test.ts index 1ab4b1fb8..6883dc2c8 100644 --- a/src/lib/__tests__/idempotency.test.ts +++ b/src/lib/__tests__/idempotency.test.ts @@ -6,7 +6,10 @@ vi.mock("@/lib/db", () => ({ idempotencyKey: { findUnique: vi.fn(), create: vi.fn(), + update: vi.fn(), + updateMany: vi.fn(), delete: vi.fn(), + deleteMany: vi.fn(), }, apiToken: { findUnique: vi.fn(), @@ -54,6 +57,12 @@ beforeEach(() => { vi.resetAllMocks(); vi.mocked(prisma.idempotencyKey.findUnique).mockResolvedValue(null); vi.mocked(prisma.idempotencyKey.create).mockResolvedValue({} as never); + vi.mocked(prisma.idempotencyKey.updateMany).mockResolvedValue({ + count: 1, + } as never); + vi.mocked(prisma.idempotencyKey.deleteMany).mockResolvedValue({ + count: 1, + } as never); }); describe("withIdempotency", () => { @@ -80,18 +89,29 @@ describe("withIdempotency", () => { }); const wrapped = withIdempotency<[NextRequest]>(handler, async () => "u-1"); - // First call: nothing cached → handler runs → response stored. + // First call: nothing cached → key claimed (create) → handler runs → + // claim promoted to the completed response (updateMany). const req1 = makeRequest("POST", { "idempotency-key": "abc-12345678" }); const res1 = await wrapped(req1); const body1 = await res1.json(); expect(res1.status).toBe(201); expect(body1).toEqual({ data: { result: 1 }, error: null }); expect(handler).toHaveBeenCalledTimes(1); + // Claim inserted before the handler ran (pending sentinel). expect(prisma.idempotencyKey.create).toHaveBeenCalledTimes(1); + const claim = ( + vi.mocked(prisma.idempotencyKey.create).mock.calls[0][0] as { + data: { responseStatus: number; responseBody: string }; + } + ).data; + expect(claim.responseStatus).toBe(0); + expect(claim.responseBody).toBe(""); + // Completed response promoted via updateMany. + expect(prisma.idempotencyKey.updateMany).toHaveBeenCalledTimes(1); // Capture the persisted body for replay. const persistedBody = ( - vi.mocked(prisma.idempotencyKey.create).mock.calls[0][0] as { + vi.mocked(prisma.idempotencyKey.updateMany).mock.calls[0][0] as { data: { responseBody: string; responseStatus: number }; } ).data; @@ -199,6 +219,89 @@ describe("withIdempotency", () => { }); }); +// A3 — concurrent same-key requests must not both run the side-effect. +// The key is claimed (pending row) before the handler runs; a racing +// request either sees the pending row or loses the insert race, and +// must be refused with 409 instead of executing a second time. +describe("withIdempotency concurrency claim (A3)", () => { + beforeEach(() => { + vi.resetAllMocks(); + vi.mocked(prisma.idempotencyKey.findUnique).mockResolvedValue(null); + vi.mocked(prisma.idempotencyKey.create).mockResolvedValue({} as never); + vi.mocked(prisma.idempotencyKey.updateMany).mockResolvedValue({ + count: 1, + } as never); + vi.mocked(prisma.idempotencyKey.deleteMany).mockResolvedValue({ + count: 1, + } as never); + }); + + it("returns 409 without running the handler when a pending claim exists", async () => { + // A concurrent request already claimed the key — findUnique returns + // the pending sentinel row (responseStatus 0, not yet expired). + vi.mocked(prisma.idempotencyKey.findUnique).mockResolvedValueOnce({ + id: "idem-pending", + userId: "u-1", + key: "abc-12345678", + method: "POST", + path: "/api/example", + responseStatus: 0, + responseBody: "", + expiresAt: new Date(Date.now() + 60_000), + createdAt: new Date(), + } as never); + + const handler = vi.fn(async () => + NextResponse.json({ data: "ok", error: null }, { status: 201 }), + ); + const wrapped = withIdempotency<[NextRequest]>(handler, async () => "u-1"); + const res = await wrapped( + makeRequest("POST", { "idempotency-key": "abc-12345678" }), + ); + + expect(res.status).toBe(409); + expect(handler).not.toHaveBeenCalled(); + expect(prisma.idempotencyKey.create).not.toHaveBeenCalled(); + }); + + it("returns 409 when a racing request wins the claim insert (P2002)", async () => { + // No row at lookup time, but the claim insert collides with a + // concurrent insert under the unique constraint. + vi.mocked(prisma.idempotencyKey.create).mockRejectedValueOnce( + Object.assign(new Error("unique"), { code: "P2002" }), + ); + + const handler = vi.fn(async () => + NextResponse.json({ data: "ok", error: null }, { status: 201 }), + ); + const wrapped = withIdempotency<[NextRequest]>(handler, async () => "u-1"); + const res = await wrapped( + makeRequest("POST", { "idempotency-key": "abc-12345678" }), + ); + + expect(res.status).toBe(409); + expect(handler).not.toHaveBeenCalled(); + expect(prisma.idempotencyKey.updateMany).not.toHaveBeenCalled(); + }); + + it("releases the claim and re-throws when the handler throws", async () => { + const boom = new Error("handler exploded"); + const handler = vi.fn(async () => { + throw boom; + }); + const wrapped = withIdempotency<[NextRequest]>(handler, async () => "u-1"); + + await expect( + wrapped(makeRequest("POST", { "idempotency-key": "abc-12345678" })), + ).rejects.toBe(boom); + + expect(prisma.idempotencyKey.create).toHaveBeenCalledTimes(1); + // Claim released so a retry isn't locked out for the pending window. + expect(prisma.idempotencyKey.deleteMany).toHaveBeenCalledTimes(1); + expect(prisma.idempotencyKey.updateMany).not.toHaveBeenCalled(); + }); +}); + // Audit C-4 / phase P2: defaultUserIdResolver must support both cookie // sessions AND Bearer tokens. Without the Bearer fallback, idempotency // silently turned off for the iOS / external-ingest paths it was built for. @@ -282,6 +385,12 @@ describe("withIdempotency body-content exclusion (P12)", () => { vi.resetAllMocks(); vi.mocked(prisma.idempotencyKey.findUnique).mockResolvedValue(null); vi.mocked(prisma.idempotencyKey.create).mockResolvedValue({} as never); + vi.mocked(prisma.idempotencyKey.updateMany).mockResolvedValue({ + count: 1, + } as never); + vi.mocked(prisma.idempotencyKey.deleteMany).mockResolvedValue({ + count: 1, + } as never); }); it.each([ @@ -293,12 +402,15 @@ describe("withIdempotency body-content exclusion (P12)", () => { "sk-ant- Anthropic key", '{"data":{"echoed":"sk-ant-api03-xyz"},"error":null}', ], - ])("does NOT cache responses containing %s", async (_label, body) => { + ])("does NOT persist responses containing %s", async (_label, body) => { const handler = vi.fn(async () => new NextResponse(body, { status: 201 })); const wrapped = withIdempotency<[NextRequest]>(handler, async () => "u-1"); await wrapped(makeRequest("POST", { "idempotency-key": "abc-12345678" })); expect(handler).toHaveBeenCalledTimes(1); - expect(prisma.idempotencyKey.create).not.toHaveBeenCalled(); + // The claim row is still inserted, but the secret-shaped body must + // never be promoted into it — the claim is released instead. + expect(prisma.idempotencyKey.updateMany).not.toHaveBeenCalled(); + expect(prisma.idempotencyKey.deleteMany).toHaveBeenCalledTimes(1); }); it.each([ @@ -316,7 +428,7 @@ describe("withIdempotency body-content exclusion (P12)", () => { async () => "u-1", ); await wrapped(makeRequest("POST", { "idempotency-key": "key-12345678" })); - expect(prisma.idempotencyKey.create).toHaveBeenCalledTimes(1); + expect(prisma.idempotencyKey.updateMany).toHaveBeenCalledTimes(1); }, ); @@ -327,7 +439,7 @@ describe("withIdempotency body-content exclusion (P12)", () => { ); const wrapped = withIdempotency<[NextRequest]>(handler, async () => "u-1"); await wrapped(makeRequest("POST", { "idempotency-key": "abc-12345678" })); - expect(prisma.idempotencyKey.create).toHaveBeenCalledTimes(1); + expect(prisma.idempotencyKey.updateMany).toHaveBeenCalledTimes(1); }); }); diff --git a/src/lib/idempotency.ts b/src/lib/idempotency.ts index 6f4db9f56..27af41c1e 100644 --- a/src/lib/idempotency.ts +++ b/src/lib/idempotency.ts @@ -14,10 +14,26 @@ import { prisma } from "@/lib/db"; import { getSession } from "@/lib/auth/session"; import { hashToken } from "@/lib/auth/hmac"; import { annotate } from "@/lib/logging/context"; +import { isP2002 } from "@/lib/prisma-errors"; const TTL_MS = 24 * 60 * 60 * 1000; +/** + * Claim TTL — how long an in-flight "pending" row is honoured before a + * retry is allowed to re-run the handler. Bounds the blast radius of a + * crashed handler that never wrote its result: the key self-heals after + * this window instead of being locked for the full 24h response TTL. + * Sized above the longest realistic write-handler latency. + */ +const PENDING_TTL_MS = 2 * 60 * 1000; const SUPPORTED_METHODS = new Set(["POST", "PUT", "PATCH", "DELETE"]); +/** + * Sentinel `responseStatus` for a claimed-but-not-yet-completed row. A + * real HTTP status is never 0, so `findCached` can distinguish an + * in-flight claim from a cached response without a schema change. + */ +const PENDING_STATUS = 0; + const KEY_REGEX = /^[A-Za-z0-9_\-:.]{8,128}$/; export interface IdempotencyContext { @@ -36,12 +52,21 @@ function getIdempotencyKey(request: Request | NextRequest): string | null { } /** - * Look up a cached response for a (userId, key, method, path) tuple. - * Returns the cached NextResponse or null. + * Outcome of a key lookup: + * - `{ kind: "replay" }` — a completed response is cached; replay it. + * - `{ kind: "pending" }` — another request holds an in-flight claim + * for this key; the caller must NOT run the + * handler again. + * - `null` — no live row; the caller may claim + run. */ -async function findCached( - ctx: IdempotencyContext, -): Promise { +type CacheLookup = { kind: "replay"; response: NextResponse } | { kind: "pending" } | null; + +/** + * Look up the state of a (userId, key, method, path) tuple. Distinguishes + * a completed cached response (replay) from an in-flight claim (pending) + * via the `PENDING_STATUS` sentinel. + */ +async function findCached(ctx: IdempotencyContext): Promise { const row = await prisma.idempotencyKey.findUnique({ where: { userId_key_method_path: { @@ -55,13 +80,20 @@ async function findCached( if (!row) return null; if (row.expiresAt <= new Date()) { - // Stale — purge and fall through. + // Stale (completed row past its TTL, or a crashed claim past the + // pending window) — purge and fall through so the retry re-runs. await prisma.idempotencyKey .delete({ where: { id: row.id } }) .catch(() => {}); return null; } + if (row.responseStatus === PENDING_STATUS) { + // A concurrent request claimed this key and is still running its + // side-effect. Signal the caller to refuse rather than double-execute. + return { kind: "pending" }; + } + let parsed: unknown = null; try { parsed = JSON.parse(row.responseBody); @@ -75,12 +107,63 @@ async function findCached( }); // Replay original status to keep client behaviour byte-identical. - return NextResponse.json(parsed, { - status: row.responseStatus, - headers: { - "X-Idempotent-Replay": "true", - }, - }); + return { + kind: "replay", + response: NextResponse.json(parsed, { + status: row.responseStatus, + headers: { + "X-Idempotent-Replay": "true", + }, + }), + }; +} + +/** + * Atomically claim a key by inserting a `PENDING_STATUS` row under the + * `(userId, key, method, path)` unique constraint BEFORE the handler + * runs. Returns `true` when this caller won the claim, `false` when a + * concurrent request already holds it (unique-constraint collision) so + * the wrapper can refuse with 409 instead of double-executing the + * side-effect. The claim carries a short `PENDING_TTL_MS` so a crashed + * handler self-heals. + */ +async function claimKey(ctx: IdempotencyContext): Promise { + try { + await prisma.idempotencyKey.create({ + data: { + userId: ctx.userId, + key: ctx.key, + method: ctx.method, + path: ctx.path, + responseStatus: PENDING_STATUS, + responseBody: "", + expiresAt: new Date(Date.now() + PENDING_TTL_MS), + }, + }); + return true; + } catch (err) { + // Another request inserted the claim first — refuse this one. + if (isP2002(err)) return false; + throw err; + } +} + +/** + * Release a claim that produced a non-cachable result (or threw) so a + * later retry gets a fresh attempt rather than a stuck pending row. + */ +async function releaseClaim(ctx: IdempotencyContext): Promise { + await prisma.idempotencyKey + .deleteMany({ + where: { + userId: ctx.userId, + key: ctx.key, + method: ctx.method, + path: ctx.path, + responseStatus: PENDING_STATUS, + }, + }) + .catch(() => {}); } async function persistCached( @@ -95,21 +178,27 @@ async function persistCached( .text() .catch(() => "")); + // Promote the claimed pending row to the completed response, extending + // the TTL from the short claim window to the full 24h replay window. + // The claim is held under the unique constraint, so this caller owns + // the row — a plain update, not an upsert. await prisma.idempotencyKey - .create({ - data: { + .updateMany({ + where: { userId: ctx.userId, key: ctx.key, method: ctx.method, path: ctx.path, + }, + data: { responseStatus: response.status, responseBody: body, expiresAt: new Date(Date.now() + TTL_MS), }, }) .catch(() => { - // Concurrent insert with the same key — the other writer wins. - // Ignore because a future replay will hit their cached value. + // The claim row vanished (e.g. purged as stale) — a future replay + // will simply miss and re-run; never throw out of the cache path. }); } @@ -213,9 +302,52 @@ export function withIdempotency< }; const cached = await findCached(ctx); - if (cached) return cached; + if (cached?.kind === "replay") return cached.response; + if (cached?.kind === "pending") { + // A concurrent request is mid-flight on this exact key. Refuse + // rather than run the side-effect a second time. The client should + // retry after the in-flight request lands, at which point the + // completed row replays. + annotate({ + action: { name: "idempotency.inflight_conflict" }, + meta: { method: ctx.method, path: ctx.path }, + }); + return NextResponse.json( + { + data: null, + error: { message: "A request with this Idempotency-Key is already in progress" }, + }, + { status: 409, headers: { "X-Idempotent-Replay": "false" } }, + ); + } - const response = await handler(...args); + // Claim the key before running the handler. If a racing request beats + // us to the insert, treat it as the in-flight conflict above — only + // one caller may execute the side-effect for a given key. + const won = await claimKey(ctx); + if (!won) { + annotate({ + action: { name: "idempotency.inflight_conflict" }, + meta: { method: ctx.method, path: ctx.path }, + }); + return NextResponse.json( + { + data: null, + error: { message: "A request with this Idempotency-Key is already in progress" }, + }, + { status: 409, headers: { "X-Idempotent-Replay": "false" } }, + ); + } + + let response: Response; + try { + response = await handler(...args); + } catch (err) { + // Handler threw — release the claim so a retry isn't locked out for + // the full pending window, then re-throw to the error envelope. + await releaseClaim(ctx); + throw err; + } if (isCachableStatus(response.status)) { // Defence-in-depth: never persist a body that carries a freshly-issued @@ -235,7 +367,15 @@ export function withIdempotency< const text = await cloned.text(); if (!SECRET_PATTERN.test(text)) { await persistCached(ctx, response, text); + } else { + // Secret-shaped body — drop the claim so the key isn't left + // pending (it was never going to cache). + await releaseClaim(ctx); } + } else { + // Non-cachable status (401/403/408/429/5xx) — release the claim so a + // retry gets a fresh attempt rather than a stuck pending row. + await releaseClaim(ctx); } return response;