fix(rest/nodejs): keep one checkout per Idempotency-Key when requests race

rest/nodejs/src/api/checkout.ts

@@ -19,11 +19,13 @@ import { z } from "zod";
 import { UCP_VERSION } from "../utils/config";
 
 import {
+  claimIdempotencyAndSaveCheckout,
   getCheckoutSession,
   getIdempotencyRecord,
   getInventory,
   getOrder,
   getProduct,
+  type IdempotencyRecord,
   logRequest,
   reserveStock,
   runInTransaction,
@@ -626,6 +628,27 @@ export class CheckoutService {
       : CheckoutResponseStatusSchema.enum.incomplete;
   }
 
+  /**
+   * Answers a request whose Idempotency-Key is already recorded: the recorded
+   * response when the request hash matches, 409 when it does not.
+   */
+  private replayIdempotent(
+    c: Context,
+    record: IdempotencyRecord,
+    requestHash: string,
+    status: 200 | 201
+  ) {
+    if (record.request_hash !== requestHash) {
+      return ucpErrorResponse(
+        c,
+        new IdempotencyConflictError(
+          "Idempotency key reused with different parameters"
+        )
+      );
+    }
+    return c.json(JSON.parse(record.response_body), status);
+  }
+
   createCheckout = async (c: Context) => {
     const idempotencyKey = c.req.header("Idempotency-Key");
     const ucpAgent = c.req.header("UCP-Agent");
@@ -636,15 +659,7 @@ export class CheckoutService {
       requestHash = this.computeHash("create_checkout", request);
       const record = getIdempotencyRecord(idempotencyKey);
       if (record) {
-        if (record.request_hash !== requestHash) {
-          return ucpErrorResponse(
-            c,
-            new IdempotencyConflictError(
-              "Idempotency key reused with different parameters"
-            )
-          );
-        }
-        return c.json(JSON.parse(record.response_body), 201);
+        return this.replayIdempotent(c, record, requestHash, 201);
       }
     }
 
@@ -761,15 +776,25 @@ export class CheckoutService {
       this.validateInventory(checkout);
       checkout.status = this.openStatus(checkout);
 
-      saveCheckout(checkout.id, checkout.status, checkout);
-
       if (idempotencyKey) {
-        saveIdempotencyRecord(
+        // The profile fetch above yielded the event loop, so another request
+        // with this key may have been answered since the read at the top.
+        // Record the key before persisting: if it is already recorded, that
+        // record is the answer and this checkout must not be stored beside it.
+        const winner = claimIdempotencyAndSaveCheckout(
           idempotencyKey,
           requestHash,
           201,
-          JSON.stringify(checkout)
+          JSON.stringify(checkout),
+          checkout.id,
+          checkout.status,
+          checkout
         );
+        if (winner) {
+          return this.replayIdempotent(c, winner, requestHash, 201);
+        }
+      } else {
+        saveCheckout(checkout.id, checkout.status, checkout);
       }
 
       return c.json(checkout, 201);
@@ -810,15 +835,7 @@ export class CheckoutService {
       requestHash = this.computeHash("update_checkout", updateRequest, id);
       const record = getIdempotencyRecord(idempotencyKey);
       if (record) {
-        if (record.request_hash !== requestHash) {
-          return ucpErrorResponse(
-            c,
-            new IdempotencyConflictError(
-              "Idempotency key reused with different parameters"
-            )
-          );
-        }
-        return c.json(JSON.parse(record.response_body), 200);
+        return this.replayIdempotent(c, record, requestHash, 200);
       }
     }
 
@@ -913,15 +930,24 @@ export class CheckoutService {
       this.validateInventory(existing);
       existing.status = this.openStatus(existing);
 
-      saveCheckout(id, existing.status, existing);
-
       if (idempotencyKey) {
-        saveIdempotencyRecord(
+        // Same window as createCheckout: the profile fetch above yielded, so
+        // record the key before persisting and answer with the record of the
+        // winner if there already is one.
+        const winner = claimIdempotencyAndSaveCheckout(
           idempotencyKey,
           requestHash,
           200,
-          JSON.stringify(existing)
+          JSON.stringify(existing),
+          id,
+          existing.status,
+          existing
         );
+        if (winner) {
+          return this.replayIdempotent(c, winner, requestHash, 200);
+        }
+      } else {
+        saveCheckout(id, existing.status, existing);
       }
 
       return c.json(existing, 200);
@@ -1267,16 +1293,21 @@ export class CheckoutService {
     }
 
     checkout.status = CheckoutResponseStatusSchema.enum.canceled;
-    saveCheckout(id, checkout.status, checkout);
 
-    if (idempotencyKey) {
-      saveIdempotencyRecord(
-        idempotencyKey,
-        requestHash,
-        200,
-        JSON.stringify(checkout)
-      );
-    }
+    // Stored and recorded as one transaction, as completeCheckout does: a
+    // checkout canceled without its key would answer every retry of that key
+    // 409 instead of replaying this response.
+    runInTransaction(() => {
+      saveCheckout(id, checkout.status, checkout);
+      if (idempotencyKey) {
+        saveIdempotencyRecord(
+          idempotencyKey,
+          requestHash,
+          200,
+          JSON.stringify(checkout)
+        );
+      }
+    });
 
     return c.json(checkout, 200);
   };

rest/nodejs/src/data/transactions.ts

@@ -151,17 +151,58 @@ export function getIdempotencyRecord(
   return stmt.get(key) as IdempotencyRecord | undefined;
 }
 
+/**
+ * Records the response for an idempotency key, unless one is already
+ * recorded. Returns undefined when this call recorded it. Returns the
+ * existing record when another request has already recorded this key: the
+ * caller lost a concurrent race on the key and must answer with that record,
+ * not with its own result.
+ */
 export function saveIdempotencyRecord(
   key: string,
   requestHash: string,
   status: number,
   responseBody: string
-): void {
+): IdempotencyRecord | undefined {
   const db = getTransactionsDb();
   const stmt = db.prepare(
-    "INSERT OR REPLACE INTO idempotency_keys (key, request_hash, response_status, response_body) VALUES (?, ?, ?, ?)"
+    "INSERT INTO idempotency_keys (key, request_hash, response_status, response_body) VALUES (?, ?, ?, ?) ON CONFLICT(key) DO NOTHING"
   );
-  stmt.run(key, requestHash, status, responseBody);
+  const inserted =
+    stmt.run(key, requestHash, status, responseBody).changes === 1;
+  return inserted ? undefined : getIdempotencyRecord(key);
+}
+
+/**
+ * Claims the idempotency key and persists the checkout in one transaction,
+ * so a record is never left behind for a checkout that was not stored. Both
+ * writes are synchronous, so unlike the read at the top of the handler they
+ * can share a transaction. Returns undefined when this request won and the
+ * checkout is stored. Returns the existing record when another request has
+ * already claimed the key; nothing is written in that case.
+ */
+export function claimIdempotencyAndSaveCheckout(
+  key: string,
+  requestHash: string,
+  status: number,
+  responseBody: string,
+  checkoutId: string,
+  checkoutStatus: string,
+  checkoutObj: ExtendedCheckoutResponse
+): IdempotencyRecord | undefined {
+  return runInTransaction((): IdempotencyRecord | undefined => {
+    const winner = saveIdempotencyRecord(
+      key,
+      requestHash,
+      status,
+      responseBody
+    );
+    if (winner) {
+      return winner;
+    }
+    saveCheckout(checkoutId, checkoutStatus, checkoutObj);
+    return undefined;
+  });
 }
 
 /**