From 558f8c597fe820e196bbbcbb01453086f30134c0 Mon Sep 17 00:00:00 2001 From: Vishal Katyal Date: Mon, 26 Oct 2026 12:52:10 -0400 Subject: [PATCH] fix(rest/nodejs): keep one checkout per Idempotency-Key when requests race createCheckout and updateCheckout read the idempotency record, await the agent profile fetch, then write, with saveCheckout before saveIdempotencyRecord and the record statement being INSERT OR REPLACE. Requests sharing one Idempotency-Key that arrive while that fetch is in flight all find no record, so each persists its own checkout and each answers 201, and a later retry of the key replays whichever wrote last. Claim the key and store the checkout in one transaction, through the runInTransaction helper, with the key statement changed to INSERT ... ON CONFLICT(key) DO NOTHING. The first request to claim the key stores its checkout and answers as before. A request that finds the key already claimed stores nothing and answers with the recorded response when its request hash matches, or 409 when it does not. The transaction keeps a record from being left behind for a checkout that was never stored. completeCheckout and cancelCheckout share the changed statement. Neither has an await between its idempotency read and its record write, so neither reaches the statement with the key already claimed, and their answers do not change. cancelCheckout now stores the canceled checkout and records the key in one transaction as well, as completeCheckout already does, so a storage failure between the two writes cannot leave a canceled checkout whose key answers every retry 409. Tests: test/idempotency_race.test.ts, three cases that hold the mocked profile fetch open until every racer has reached it, each 409 checked for the IDEMPOTENCY_CONFLICT code, and two that plant a storage failure to show a key is never recorded for a checkout that was not stored and a checkout is never canceled without its key. --- rest/nodejs/src/api/checkout.ts | 101 ++++--- rest/nodejs/src/data/transactions.ts | 47 ++- rest/nodejs/test/idempotency_race.test.ts | 342 ++++++++++++++++++++++ 3 files changed, 452 insertions(+), 38 deletions(-) create mode 100644 rest/nodejs/test/idempotency_race.test.ts diff --git a/rest/nodejs/src/api/checkout.ts b/rest/nodejs/src/api/checkout.ts index f8346e4..ce20815 100644 --- a/rest/nodejs/src/api/checkout.ts +++ b/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); }; diff --git a/rest/nodejs/src/data/transactions.ts b/rest/nodejs/src/data/transactions.ts index b805fda..abeba58 100644 --- a/rest/nodejs/src/data/transactions.ts +++ b/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; + }); } /** diff --git a/rest/nodejs/test/idempotency_race.test.ts b/rest/nodejs/test/idempotency_race.test.ts new file mode 100644 index 0000000..9480f16 --- /dev/null +++ b/rest/nodejs/test/idempotency_race.test.ts @@ -0,0 +1,342 @@ +// Copyright 2026 UCP Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Concurrent idempotency: one key raced against itself. +// +// createCheckout and updateCheckout read the record, await the agent profile, +// and then write. Requests sharing one Idempotency-Key that arrive while that +// fetch is in flight all find "no record" before any of them writes. The +// three race tests hold the profile fetch open until every racer has reached +// it, so the interleaving is reached on every run rather than by chance; the +// fourth and fifth plant a storage failure and send no profile. + +import assert from "node:assert/strict"; +import { before, mock, test } from "node:test"; + +import { zValidator } from "@hono/zod-validator"; +import { Hono } from "hono"; + +import { CheckoutService } from "../src/api/checkout"; +import { getProductsDb, getTransactionsDb, initDbs } from "../src/data/db"; +import { + ExtendedCheckoutCreateRequestSchema, + ExtendedCheckoutUpdateRequestSchema, +} from "../src/models"; +import { IdParamSchema, prettyValidation } from "../src/utils/validation"; + +const RACERS = 8; + +const PROFILE = { + ucp: { + version: "2026-04-08", + capabilities: { + "dev.ucp.shopping.order": [ + { + name: "dev.ucp.shopping.order", + config: { webhook_url: "https://agent.example/webhooks" }, + }, + ], + }, + }, +}; + +// The documented form of the header (overview/index.md, "UCP-Agent"): a +// profile URL the business fetches. Only this form awaits a network round +// trip inside the handler. +const AGENT_HEADERS = { + "Content-Type": "application/json", + "UCP-Agent": 'profile="https://agent.example/profiles/shopping-agent.json"', +}; + +function body(quantity: number) { + return { + currency: "USD", + line_items: [{ item: { id: "bouquet_roses" }, quantity }], + payment: {}, + }; +} + +function buildApp() { + const svc = new CheckoutService(); + const app = new Hono<{ Variables: { logger: typeof console } }>(); + app.use(async (c, next) => { + c.set("logger", console); + await next(); + }); + app.post( + "/checkout-sessions", + zValidator("json", ExtendedCheckoutCreateRequestSchema, prettyValidation), + svc.createCheckout + ); + app.put( + "/checkout-sessions/:id", + zValidator("param", IdParamSchema, prettyValidation), + zValidator("json", ExtendedCheckoutUpdateRequestSchema, prettyValidation), + svc.updateCheckout + ); + app.post( + "/checkout-sessions/:id/cancel", + zValidator("param", IdParamSchema, prettyValidation), + svc.cancelCheckout + ); + return app; +} + +before(() => { + initDbs(":memory:", ":memory:"); + getProductsDb() + .prepare( + "INSERT INTO products (id, title, price, image_url) VALUES (?, ?, ?, ?)" + ) + .run("bouquet_roses", "Red Rose", 3500, ""); + getTransactionsDb() + .prepare("INSERT INTO inventory (product_id, quantity) VALUES (?, ?)") + .run("bouquet_roses", 100); +}); + +// Replaces fetch with a profile server that answers only once `expected` +// callers are waiting on it. Every racer is then past its idempotency read +// before any racer can write. The timer is a backstop so that a change which +// stops some racers short of the fetch fails the test instead of hanging it; +// unref is called on it so that it never holds the event loop open on its +// own, and in that case the runner reports the pending test as soon as the +// loop drains. +function gateProfileFetch(expected: number) { + let waiting = 0; + let open!: () => void; + const gate = new Promise((resolve) => { + open = resolve; + }); + const backstop = setTimeout(open, 2000); + backstop.unref(); + mock.method(globalThis, "fetch", async () => { + waiting += 1; + if (waiting === expected) { + clearTimeout(backstop); + open(); + } + await gate; + return new Response(JSON.stringify(PROFILE), { + headers: { "Content-Type": "application/json" }, + }); + }); +} + +function countCheckouts(): number { + return ( + getTransactionsDb() + .prepare("SELECT COUNT(*) AS n FROM checkouts") + .get() as { + n: number; + } + ).n; +} + +// Every 409 must be the idempotency conflict itself: another 409 code (the +// server answers CHECKOUT_NOT_MODIFIABLE and OUT_OF_STOCK with 409 too) must +// not pass as one. +async function assertConflicts(responses: Response[]): Promise { + for (const r of responses) { + if (r.status !== 409) continue; + const body = (await r.clone().json()) as { + messages?: { code?: string }[]; + }; + assert.equal(body.messages?.[0]?.code, "IDEMPOTENCY_CONFLICT"); + } +} + +function summary(responses: Response[]): string { + const counts = new Map(); + for (const r of responses) + counts.set(r.status, (counts.get(r.status) ?? 0) + 1); + return [...counts.entries()] + .sort(([a], [b]) => a - b) + .map(([status, n]) => `${n}x${status}`) + .join(", "); +} + +test("concurrent identical creates under one key persist one checkout", async (t) => { + t.after(() => mock.restoreAll()); + const app = buildApp(); + gateProfileFetch(RACERS); + const persisted = countCheckouts(); + + const responses = await Promise.all( + Array.from({ length: RACERS }, () => + app.request("/checkout-sessions", { + method: "POST", + headers: { ...AGENT_HEADERS, "Idempotency-Key": "race-create-same" }, + body: JSON.stringify(body(1)), + }) + ) + ); + + assert.equal(summary(responses), `${RACERS}x201`); + const ids = new Set(); + for (const r of responses) ids.add(((await r.json()) as { id: string }).id); + assert.equal( + ids.size, + 1, + `identical retries must name ONE checkout; got ${ids.size}` + ); + assert.equal( + countCheckouts() - persisted, + 1, + "only the winning request may persist a checkout" + ); +}); + +test("concurrent differing creates under one key admit exactly one", async (t) => { + t.after(() => mock.restoreAll()); + const app = buildApp(); + gateProfileFetch(RACERS); + const persisted = countCheckouts(); + + const responses = await Promise.all( + Array.from({ length: RACERS }, (_, i) => + app.request("/checkout-sessions", { + method: "POST", + headers: { ...AGENT_HEADERS, "Idempotency-Key": "race-create-differ" }, + body: JSON.stringify(body(i + 1)), + }) + ) + ); + + assert.equal( + summary(responses), + `1x201, ${RACERS - 1}x409`, + "reusing a key with new parameters is a conflict, never a second checkout" + ); + await assertConflicts(responses); + assert.equal(countCheckouts() - persisted, 1); +}); + +test("concurrent differing updates under one key admit exactly one", async (t) => { + t.after(() => mock.restoreAll()); + const app = buildApp(); + const created = await app.request("/checkout-sessions", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body(1)), + }); + assert.equal(created.status, 201); + const { id } = (await created.json()) as { id: string }; + gateProfileFetch(RACERS); + + const responses = await Promise.all( + Array.from({ length: RACERS }, (_, i) => + app.request(`/checkout-sessions/${id}`, { + method: "PUT", + headers: { ...AGENT_HEADERS, "Idempotency-Key": "race-update-differ" }, + body: JSON.stringify(body(i + 1)), + }) + ) + ); + + assert.equal(summary(responses), `1x200, ${RACERS - 1}x409`); + await assertConflicts(responses); + const winner = responses.find((r) => r.status === 200)!; + const winnerQty = ( + (await winner.json()) as { line_items: { quantity: number }[] } + ).line_items[0].quantity; + const stored = JSON.parse( + ( + getTransactionsDb() + .prepare("SELECT data FROM checkouts WHERE id = ?") + .get(id) as { data: string } + ).data + ) as { line_items: { quantity: number }[] }; + assert.equal( + stored.line_items[0].quantity, + winnerQty, + "the stored checkout must be the one the winner was told about" + ); +}); + +// The key is recorded before the checkout is stored, so the two writes must +// be one transaction: a key recorded for a checkout that was never stored +// would replay a 201 naming a checkout that does not exist, on every retry. +test("a key is never recorded for a checkout that was not stored", async (t) => { + const app = buildApp(); + const db = getTransactionsDb(); + // Make the checkout write fail the way a full disk or a locked file would. + db.exec( + "CREATE TRIGGER planted_storage_failure BEFORE INSERT ON checkouts " + + "BEGIN SELECT RAISE(ABORT, 'planted storage failure'); END" + ); + t.after(() => db.exec("DROP TRIGGER IF EXISTS planted_storage_failure")); + const headers = { + "Content-Type": "application/json", + "Idempotency-Key": "create-after-failed-write", + }; + const failed = await app.request("/checkout-sessions", { + method: "POST", + headers, + body: JSON.stringify(body(1)), + }); + assert.equal(failed.status, 400); + + db.exec("DROP TRIGGER planted_storage_failure"); + const retried = await app.request("/checkout-sessions", { + method: "POST", + headers, + body: JSON.stringify(body(1)), + }); + assert.equal(retried.status, 201); + const { id } = (await retried.json()) as { id: string }; + assert.ok( + db.prepare("SELECT id FROM checkouts WHERE id = ?").get(id), + "the retry must name a checkout that exists" + ); +}); + +// cancelCheckout stores the canceled checkout and then records the key, so +// those two writes must be one transaction as well, as they are in +// completeCheckout: a checkout canceled without its key would answer every +// retry of that key 409 CHECKOUT_NOT_MODIFIABLE instead of replaying the +// cancel. +test("a checkout is never canceled without its key being recorded", async (t) => { + const app = buildApp(); + const db = getTransactionsDb(); + const created = await app.request("/checkout-sessions", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body(1)), + }); + assert.equal(created.status, 201); + const { id } = (await created.json()) as { id: string }; + db.exec( + "CREATE TRIGGER planted_storage_failure BEFORE INSERT ON idempotency_keys " + + "BEGIN SELECT RAISE(ABORT, 'planted storage failure'); END" + ); + t.after(() => db.exec("DROP TRIGGER IF EXISTS planted_storage_failure")); + const headers = { "Idempotency-Key": "cancel-after-failed-write" }; + const failed = await app.request(`/checkout-sessions/${id}/cancel`, { + method: "POST", + headers, + }); + assert.equal(failed.status, 500); + + db.exec("DROP TRIGGER planted_storage_failure"); + const retried = await app.request(`/checkout-sessions/${id}/cancel`, { + method: "POST", + headers, + }); + assert.equal(retried.status, 200, await retried.clone().text()); + assert.equal( + ((await retried.json()) as { status: string }).status, + "canceled", + "the retry must cancel the checkout the failed attempt left open" + ); +});