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) <noreply@anthropic.com>

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);
   });
 });
 

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<NextResponse | null> {
+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<CacheLookup> {
   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<boolean> {
+  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<void> {
+  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;