diff --git a/CHANGELOG.md b/CHANGELOG.md index 6788137..c56d95c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -137,6 +137,15 @@ While pre-1.0, the public API may change between 0.x releases. (ADR-0021) was pending, the replayed frame got `LIMIT_EXCEEDED` and the client rolled back a write that had committed. The check now runs after the lookup, so a resent `mut` gets its stored outcome while that receipt is retained. +- **A `mut`/`call` replayed while the original is still running no longer runs + twice (ADR-0025 amendment).** `authorize`, and a command's `execute`, can + await I/O, and a second frame for the same `txId` could slip in and run its + own attempt. Under ADR-0021's hold-and-replay (the socket drops mid-authorize, + the client replays on reconnect), the replayed mutation failed on its own pk + and the client rolled back a write that had committed. A replayed command ran + its side effect a second time. A duplicate now waits for the running attempt + and replays its outcome. A replayed `committed` also flushes the socket's + pending deltas first, so it can't overtake a buffered delta (ADR-0002 C1). ### Internal diff --git a/docs/adr/0025-empty-op-key-rejection.md b/docs/adr/0025-empty-op-key-rejection.md index 7df3d7d..caef159 100644 --- a/docs/adr/0025-empty-op-key-rejection.md +++ b/docs/adr/0025-empty-op-key-rejection.md @@ -89,3 +89,59 @@ because `authorize` (and a command's `execute`) can await. That needs a per-`txId` reservation, which is a separate decision and is not made here. Tests: `tests/dedup-before-limits.test.ts`. + +## Amendment — 2026-09-25: a duplicate of an in-flight txId waits, then replays + +This closes the gap the previous amendment left open. It is the same rule (a +resent `txId` gets the outcome of the attempt that ran), now also enforced +while that attempt is still running. + +**The gap was real.** `authorize`, and a command's `execute`, can await +non-storage I/O, and while they do the DO's input gate lets other events in. +Reproduced in workerd with an `authorize` parked on a promise that a later +event resolves, and with a plain `setTimeout`. A second frame for the same +`txId` passed the lookup and ran its own attempt. The mutation's duplicate +failed on the pk (`EXECUTE_FAILED`), and a command's side effect ran twice. +ADR-0021's hold-and-replay reaches this: the socket drops mid-`authorize` and +the client replays the `txId` on its new socket. Through the real transport, +the client then rejected, and rolled back, a write that had committed. + +**Decision.** + +1. The DO instance keeps an in-memory map of the `txId`s whose `mut`/`call` + attempt is running. The entry is set before the attempt starts and removed + in a `finally`, so every exit (committed, rejected, thrown) clears it. +2. A frame whose `txId` is recorded replays the receipt, as before. A frame + whose `txId` is in the map waits for that attempt to end, then checks + again. Usually it then finds the receipt and replays it to its own socket. + If the attempt ended without recording an outcome (it threw), the waiter + runs as a fresh attempt. That attempt produced nothing to replay. +3. A replayed `committed` first flushes the socket's pending deltas. It + advances the client's cursor, so it must not overtake a buffered delta + (ADR-0002 C1, as generalised by ADR-0011 C1′). A waiter replays right after + the commit, while that commit's delta is still in the coalescer. A plain + sequential replay can meet the same buffered egress. + +**What this guarantees, and what it does not.** At most one attempt per +`txId` runs at a time on an instance. It does not make an attempt's effects +atomic with its dedup record. A command whose `execute` has external side +effects, or a mutation that fails between commit and `recordTx`, can still +leave effects with no receipt. A later resend then runs again, as it did +before this change. + +**Hibernation and eviction.** The map is not persisted, and it doesn't need +to be. A DO cannot hibernate while a handler is awaiting, because the pending +event keeps it active. If the instance is evicted or reset, the running +attempt dies with the map. Either it recorded an outcome, which a resend +replays from SQLite, or it did not, and the resend runs fresh (see above). + +**Trade-off.** An `authorize` or command that never settles holds its `txId` +forever, and every duplicate of it waits with it. The DO stays active (and +billed) while they wait. The client still settles by its own timeout. The +server does not bound the wait, because a waiter that gave up would have to +answer something before the attempt it waits on has an outcome, and any +answer could contradict that outcome, which is the bug this amendment fixes. +Before this change the same never-settling handler already hung its own +frame and kept the DO active. + +Tests: `tests/inflight-txid.test.ts` (with `GatedTestDO` in the test worker). diff --git a/docs/adr/README.md b/docs/adr/README.md index 592200a..24bfa23 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -31,4 +31,4 @@ explains the displacement. | [0021](./0021-in-flight-settlement-unexpected-close.md) | In-flight mut/call settlement across an unexpected close: hold-and-replay + typed `ConnectionLostError` | Accepted (fixes #39; builds on 0020/0016/0011) | | [0022](./0022-db-version-support-policy.md) | `@tanstack/db` version support: an uncapped floor, tested at both ends | Accepted (supersedes 0011 D5 peer floor in part) | | [0024](./0024-examples-single-tanstack-db.md) | Examples bundle exactly one `@tanstack/db`, and CI proves it | Accepted (examples only) | -| [0025](./0025-empty-op-key-rejection.md) | An empty `mut` op key is rejected with a reply, after the dedup lookup | Accepted (amends 0012 D1 for op keys; makes 0014 D3 true; amended: every per-tx rejection runs after the dedup lookup) | +| [0025](./0025-empty-op-key-rejection.md) | An empty `mut` op key is rejected with a reply, after the dedup lookup | Accepted (amends 0012 D1 for op keys; makes 0014 D3 true; amended: every per-tx rejection runs after the dedup lookup; a duplicate of an in-flight txId waits, then replays) | diff --git a/src/server/mixin.ts b/src/server/mixin.ts index ca04c1d..7e6515e 100644 --- a/src/server/mixin.ts +++ b/src/server/mixin.ts @@ -192,6 +192,9 @@ export function Syncable() { #writesSinceCompaction = 0 readonly #broadcaster: Broadcaster readonly #liveWs = new Set() + /** txIds whose mut/call handler is running now, each mapped to a promise that + * settles when that attempt ends (ADR-0025, in-flight duplicates). */ + readonly #inflight = new Map>() readonly #api: SyncApi constructor(...args: any[]) { @@ -705,10 +708,11 @@ export function Syncable() { */ async #handleMut(ws: WebSocket, f: Extract): Promise { // Dedup first: a resent txId gets its stored outcome, never a fresh - // rejection from a check below (ADR-0025). - const seen = lookupTx(this.#sql, f.txId) - if (seen) return this.#replayReceipt(ws, f.txId, seen) + // rejection from a check in #applyMut (ADR-0025). + return this.#oncePerTx(ws, f.txId, () => this.#applyMut(ws, f)) + } + async #applyMut(ws: WebSocket, f: Extract): Promise { // Inbound limit: reject over-length batches without applying anything // (ADR-0012). Reject-don't-truncate: a partial apply silently drops writes. if (f.ops.length > this.maxOpsPerMutation) { @@ -798,8 +802,10 @@ export function Syncable() { /** Run a named command (outside any transaction) and confirm with its result. */ async #handleCall(ws: WebSocket, f: Extract): Promise { - const seen = lookupTx(this.#sql, f.txId) - if (seen) return this.#replayReceipt(ws, f.txId, seen) + return this.#oncePerTx(ws, f.txId, () => this.#applyCall(ws, f)) + } + + async #applyCall(ws: WebSocket, f: Extract): Promise { const def = this.#registry.commands.get(f.name) if (!def) return this.#rejectTx(ws, f.txId, `unknown command '${f.name}'`, "UNKNOWN_COMMAND") @@ -842,6 +848,29 @@ export function Syncable() { this.#send(ws, { t: "committed", txId: f.txId, seq: commitSeq, result }) } + /** At most one attempt per txId at a time (ADR-0025). A recorded txId replays + * its receipt. A txId whose attempt is still running (its authorize or a + * command's execute awaits, and the input gate lets this frame in) waits + * for that attempt, then checks again, so it replays the outcome rather + * than running a second time. Otherwise reserve the txId and run. */ + async #oncePerTx(ws: WebSocket, txId: string, run: () => Promise): Promise { + for (;;) { + const seen = lookupTx(this.#sql, txId) + if (seen) return this.#replayReceipt(ws, txId, seen) + const running = this.#inflight.get(txId) + if (!running) break + await running + } + let settle!: () => void + this.#inflight.set(txId, new Promise((r) => (settle = r))) + try { + await run() + } finally { + this.#inflight.delete(txId) + settle() + } + } + #rejectTx(ws: WebSocket, txId: string, message: string, code?: string): void { recordTx(this.#sql, txId, false, null, message, code ?? null, null) this.#send(ws, { t: "rejected", txId, error: code ? { code, message } : { message } }) @@ -849,6 +878,11 @@ export function Syncable() { #replayReceipt(ws: WebSocket, txId: string, seen: SeenTx): void { if (seen.ok) { + // `committed` advances the client's cursor: flush this socket's pending + // deltas first (ADR-0002 C1 / 0011 C1′). A duplicate that waited on an + // in-flight attempt replays right after that commit, while the commit's + // delta for this socket is still in the coalescer. + this.#broadcaster.flushOne(ws) this.#send(ws, { t: "committed", txId, seq: seen.cursor ?? "0", result: decodeResult(seen.result) }) } else { // Shape the replay exactly like #rejectTx's original frame — with the diff --git a/tests/env.d.ts b/tests/env.d.ts index 428602f..60451b3 100644 --- a/tests/env.d.ts +++ b/tests/env.d.ts @@ -8,6 +8,7 @@ declare namespace Cloudflare { MAINT_DO: DurableObjectNamespace SLOW_DO: DurableObjectNamespace LIMITS_DO: DurableObjectNamespace + GATED_DO: DurableObjectNamespace HOST_DO: DurableObjectNamespace HOST_OPTIN_DO: DurableObjectNamespace } diff --git a/tests/inflight-txid.test.ts b/tests/inflight-txid.test.ts new file mode 100644 index 0000000..1bd3555 --- /dev/null +++ b/tests/inflight-txid.test.ts @@ -0,0 +1,329 @@ +import { env, runInDurableObject, SELF } from "cloudflare:test" +import { describe, expect, it } from "vitest" +import { type SubHandler, WebSocketTransport, type WebSocketLike } from "../src/client/transport.ts" +import { createFrameCodec } from "../src/wire/frame-codec.ts" +import type { ClientFrame, ServerFrame } from "../src/wire/frames.ts" +import type { GatedApi, GatedTestDO } from "./test-worker.ts" + +// WHY: a txId runs its handler at most once, even when a second frame for it +// arrives while the first is still in flight (ADR-0025 amendment). `authorize` +// (and a command's `execute`) can await I/O, and while it does the DO's input +// gate is open, so a duplicate frame is dispatched and passes the dedup lookup, +// because nothing is recorded yet. ADR-0021's hold-and-replay produces exactly +// that duplicate: the socket drops mid-authorize and the client replays the txId +// on its new socket. Without the in-flight guard the replay runs the handler a +// second time: a mutation's replay fails (pk conflict → EXECUTE_FAILED) and the +// client rolls back a write that committed; a command's side effect runs twice. +// +// GatedTestDO parks `authorize` on a gate the test releases from a separate +// event, so "the first frame is still in flight" is a fact, not a timing guess. + +const codec = createFrameCodec() + +function stub(room: string): DurableObjectStub { + return env.GATED_DO.get(env.GATED_DO.idFromName(room)) as DurableObjectStub +} + +async function openWs(room: string): Promise { + const res = await SELF.fetch(`https://example.com/gated/${room}`, { headers: { Upgrade: "websocket" } }) + expect(res.status).toBe(101) + const ws = res.webSocket + if (!ws) throw new Error("no webSocket on 101 response") + ws.accept() + return ws +} + +function send(ws: WebSocket, frame: ClientFrame): void { + ws.send(codec.encode(frame)) +} + +/** Resolve with the receipt for `txId` on `ws`. Listens from the call, so call + * it BEFORE the frame that triggers the receipt can be answered. */ +function receiptFor(ws: WebSocket, txId: string, timeoutMs = 3000): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(`no receipt for ${txId}`)), timeoutMs) + const onMsg = (e: MessageEvent): void => { + const f = codec.decode(e.data as ArrayBuffer) as ServerFrame + if ((f.t === "committed" || f.t === "rejected") && f.txId === txId) { + clearTimeout(timer) + ws.removeEventListener("message", onMsg) + resolve(f) + } + } + ws.addEventListener("message", onMsg) + }) +} + +async function waitFor(pred: () => Promise | boolean, timeoutMs = 3000): Promise { + const start = Date.now() + while (!(await pred())) { + if (Date.now() - start > timeoutMs) throw new Error("waitFor timeout") + await new Promise((r) => setTimeout(r, 5)) + } +} + +const parked = (room: string, tag: string): Promise => + runInDurableObject(stub(room), (i: GatedTestDO) => i.gateParked(tag)) +const release = (room: string, tag: string): Promise => + runInDurableObject(stub(room), (i: GatedTestDO) => i.gateRelease(tag)) + +/** Frames on one socket dispatch in order and each handler runs until its first + * await, so once an `echo` sent AFTER `frame` is answered, `frame` has been + * dispatched and is parked wherever it awaits. */ +async function sendAndDispatch(ws: WebSocket, frame: ClientFrame, barrierId: string): Promise { + const barrier = receiptFor(ws, barrierId) + send(ws, frame) + send(ws, { t: "call", txId: barrierId, name: "echo", args: null }) + expect((await barrier).t).toBe("committed") +} + +const gatedInsert = (txId: string, id: string, tag: string): ClientFrame => ({ + t: "mut", + txId, + collection: "gated", + ops: [{ type: "insert", key: id, cols: { id, body: tag } }], +}) + +describe("an in-flight txId runs once; a duplicate replays its outcome (ADR-0025 amendment)", () => { + it("mut: a duplicate on another socket while T's authorize is parked gets T's committed; T runs once", async () => { + const room = "ift-mut" + const ws1 = await openWs(room) + const ws2 = await openWs(room) + const r1 = receiptFor(ws1, "T") + const r2 = receiptFor(ws2, "T") + + send(ws1, gatedInsert("T", "row", "g-mut")) + await waitFor(async () => (await parked(room, "g-mut")) === 1) + await sendAndDispatch(ws2, gatedInsert("T", "row", "g-mut"), "B1") + + // The duplicate is waiting on T, not running its own authorize. + expect(await parked(room, "g-mut")).toBe(1) + await release(room, "g-mut") + + const [a, b] = await Promise.all([r1, r2]) + expect(a).toMatchObject({ t: "committed", txId: "T" }) + expect(b).toMatchObject({ t: "committed", txId: "T", seq: (a as { seq: string }).seq }) + const rows = await runInDurableObject(stub(room), (_i, s) => Array.from(s.storage.sql.exec("SELECT id FROM gated"))) + expect(rows).toEqual([{ id: "row" }]) + ws1.close() + ws2.close() + }) + + it("call: a duplicate while T is in flight gets T's result; the side effect runs once", async () => { + const room = "ift-call" + const ws1 = await openWs(room) + const ws2 = await openWs(room) + const r1 = receiptFor(ws1, "C") + const r2 = receiptFor(ws2, "C") + + send(ws1, { t: "call", txId: "C", name: "bump", args: { tag: "g-call" } }) + await waitFor(async () => (await parked(room, "g-call")) === 1) + await sendAndDispatch(ws2, { t: "call", txId: "C", name: "bump", args: { tag: "g-call" } }, "B2") + + expect(await parked(room, "g-call")).toBe(1) + await release(room, "g-call") + + const [a, b] = await Promise.all([r1, r2]) + expect(a).toMatchObject({ t: "committed", txId: "C", result: { n: 1 } }) + expect(b).toMatchObject({ t: "committed", txId: "C", result: { n: 1 } }) + const n = await runInDurableObject(stub(room), (_i, s) => Array.from(s.storage.sql.exec("SELECT n FROM bumps"))[0]) + expect(n).toEqual({ n: 1 }) + ws1.close() + ws2.close() + }) + + it("a rejected T: the duplicate gets the same rejection, and a later resend is not wedged", async () => { + const room = "ift-reject" + const ws1 = await openWs(room) + const ws2 = await openWs(room) + const r1 = receiptFor(ws1, "R") + const r2 = receiptFor(ws2, "R") + + // T is rejected after its authorize parks: its execute hits an existing row. + await runInDurableObject(stub(room), (_i, s) => { + s.storage.sql.exec("INSERT INTO gated(id, body) VALUES ('dup', 'seed')") + }) + send(ws1, gatedInsert("R", "dup", "g-rej")) + await waitFor(async () => (await parked(room, "g-rej")) === 1) + await sendAndDispatch(ws2, gatedInsert("R", "dup", "g-rej"), "B3") + await release(room, "g-rej") + + const [a, b] = await Promise.all([r1, r2]) + expect(a).toMatchObject({ t: "rejected", txId: "R", error: { code: "EXECUTE_FAILED" } }) + expect(b).toEqual(a) + + // T has settled: a resend is answered from the dedup table, promptly. + const r3 = receiptFor(ws1, "R", 1000) + send(ws1, gatedInsert("R", "dup", "g-rej")) + expect(await r3).toEqual(a) + expect(await parked(room, "g-rej")).toBe(1) + ws1.close() + ws2.close() + }) + + it("the duplicate's socket closes before T settles: T still commits and the DO keeps serving", async () => { + const room = "ift-closed" + const ws1 = await openWs(room) + const ws2 = await openWs(room) + // ws2 is subscribed, so T's commit buffers a delta for it: the waiter's + // replay then flushes (and sends) to a socket the server has seen close. + const subReady = receiptOrFrame(ws2, (f) => f.t === "snap-end") + send(ws2, { t: "sub", subId: "s2", collection: "gated" }) + await subReady + const r1 = receiptFor(ws1, "K") + + send(ws1, gatedInsert("K", "k", "g-closed")) + await waitFor(async () => (await parked(room, "g-closed")) === 1) + await sendAndDispatch(ws2, gatedInsert("K", "k", "g-closed"), "B4") + ws2.close() + // Server-side barrier: only ws1 is still OPEN on the DO. + await waitFor(() => + runInDurableObject(stub(room), (_i, state) => state.getWebSockets().filter((w) => w.readyState === WebSocket.OPEN).length === 1), + ) + await release(room, "g-closed") + + expect(await r1).toMatchObject({ t: "committed", txId: "K" }) + const r2 = receiptFor(ws1, "E") + send(ws1, { t: "call", txId: "E", name: "echo", args: 1 }) + expect(await r2).toMatchObject({ t: "committed", result: { echoed: 1 } }) + ws1.close() + }) + + it("C1: on a subscribed duplicate socket, T's delta arrives before T's replayed committed", async () => { + const room = "ift-c1" + const ws1 = await openWs(room) + const ws2 = await openWs(room) + const subReady = receiptOrFrame(ws2, (f) => f.t === "snap-end") + send(ws2, { t: "sub", subId: "s2", collection: "gated" }) + await subReady + + // Record ws2's frames from here; stop at T's receipt. + const frames = framesUntil(ws2, (f) => f.t === "committed" && f.txId === "T1") + send(ws1, gatedInsert("T1", "c1", "g-c1")) + await waitFor(async () => (await parked(room, "g-c1")) === 1) + await sendAndDispatch(ws2, gatedInsert("T1", "c1", "g-c1"), "B5") + await release(room, "g-c1") + + const seen = await frames + const delta = seen.findIndex((f) => f.t === "d" && f.key === "c1") + expect(delta).toBeGreaterThanOrEqual(0) + expect(delta).toBeLessThan(seen.length - 1) // the delta precedes the committed + ws1.close() + ws2.close() + }) +}) + +/** Resolve once a frame matching `done` arrives on `ws`. */ +function receiptOrFrame(ws: WebSocket, done: (f: ServerFrame) => boolean): Promise { + return framesUntil(ws, done).then(() => undefined) +} + +/** Collect every frame on `ws` up to and including the first matching `done`. */ +function framesUntil(ws: WebSocket, done: (f: ServerFrame) => boolean, timeoutMs = 3000): Promise> { + return new Promise((resolve, reject) => { + const out: Array = [] + const timer = setTimeout(() => reject(new Error(`timeout; got [${out.map((f) => f.t).join(",")}]`)), timeoutMs) + const onMsg = (e: MessageEvent): void => { + out.push(codec.decode(e.data as ArrayBuffer) as ServerFrame) + if (done(out[out.length - 1]!)) { + clearTimeout(timer) + ws.removeEventListener("message", onMsg) + resolve(out) + } + } + ws.addEventListener("message", onMsg) + }) +} + +describe("a replayed committed flushes the socket's pending deltas first (ADR-0002 C1)", () => { + // SlowTickDO: the coalescer tick is 30 s, so a delta for a socket that is not + // the writer stays buffered unless something flushes it. A plain sequential + // resend on that socket must not let `committed` (cursor-advancing) overtake it. + it("a sequential replay on a subscribed socket delivers the buffered delta before committed", async () => { + const room = "ift-slow" + const open = async (): Promise => { + const res = await SELF.fetch(`https://example.com/slow/${room}`, { headers: { Upgrade: "websocket" } }) + const ws = res.webSocket! + ws.accept() + return ws + } + const ws1 = await open() + const ws2 = await open() + const subReady = receiptOrFrame(ws2, (f) => f.t === "snap-end") + send(ws2, { t: "sub", subId: "s2", collection: "messages" }) + await subReady + + const frames = framesUntil(ws2, (f) => f.t === "committed" && f.txId === "S1") + const r1 = receiptFor(ws1, "S1") + const m: ClientFrame = { t: "mut", txId: "S1", collection: "messages", ops: [{ type: "insert", key: "s", cols: { id: "s", body: "v" } }] } + send(ws1, m) + expect((await r1).t).toBe("committed") + send(ws2, m) // resend on the subscribed socket: a dedup replay + + const seen = await frames + const delta = seen.findIndex((f) => f.t === "d" && f.key === "s") + expect(delta).toBeGreaterThanOrEqual(0) + expect(delta).toBeLessThan(seen.length - 1) + ws1.close() + ws2.close() + }) +}) + +// --- Full stack: the real transport's hold-and-replay (ADR-0021) ----------- + +function noopHandler(): SubHandler { + return { onSnap: () => {}, onSnapEnd: () => {}, onDelta: () => {}, onUptodate: () => {}, onReset: () => {} } +} + +describe("hold-and-replay across a drop mid-authorize (ADR-0021 × ADR-0025 amendment)", () => { + it("the replayed mut resolves committed, never a rollback of the write that committed", async () => { + const room = "ift-client" + const sockets: Array<{ sent: Array }> = [] + const t = new WebSocketTransport({ + url: `https://example.com/gated/${room}`, + reconnectDelay: 20, + timeoutMs: 60_000, + open: async () => { + const res = await SELF.fetch(`https://example.com/gated/${room}`, { headers: { Upgrade: "websocket" } }) + const real = res.webSocket + if (!real) throw new Error("no webSocket") + real.accept() + const log = { sent: [] as Array } + sockets.push(log) + return { + send: (d) => { + log.sent.push(codec.decode(d as ArrayBuffer) as ClientFrame) + real.send(d) + }, + close: (code, reason) => real.close(code, reason), + addEventListener: (type, l) => real.addEventListener(type, l as never), + removeEventListener: (type, l) => real.removeEventListener(type, l as never), + } satisfies WebSocketLike + }, + }) + await t.connect() + await t.subscribe("s1", "gated", noopHandler()) // hold-and-replay needs a live sub + + const outcome = t.sendMut(gatedInsert("TX", "row", "g-client") as Extract).then( + (v) => ({ ok: true as const, v }), + (e: unknown) => ({ ok: false as const, e }), + ) + await waitFor(async () => (await parked(room, "g-client")) === 1) + + // The socket drops while TX's authorize is parked. The transport holds TX, + // reconnects, and replays it on the new socket. + await runInDurableObject(stub(room), (_i, state) => { + for (const s of state.getWebSockets()) s.close(1000, "drop") + }) + await waitFor(() => sockets.length === 2 && sockets[1]!.sent.some((f) => f.t === "mut")) + // Barrier: an echo sent after the replay is answered, so the replay is dispatched. + await t.call.echo(null) + + await release(room, "g-client") + const r = await outcome + expect(r.ok ? "committed" : String(r.e)).toBe("committed") + const rows = await runInDurableObject(stub(room), (_i, s) => Array.from(s.storage.sql.exec("SELECT id FROM gated"))) + expect(rows).toEqual([{ id: "row" }]) + t.close() + }) +}) diff --git a/tests/test-worker.ts b/tests/test-worker.ts index 07a7da5..e1d68ca 100644 --- a/tests/test-worker.ts +++ b/tests/test-worker.ts @@ -251,6 +251,86 @@ export class LimitsTestDO extends SyncTestDO { protected override readonly maxSubsPerSocket = 2 } +// ---- In-flight duplicate fixtures (inflight-txid.test.ts) ----------------- +// +// An `authorize` that awaits a real non-storage promise, released by a LATER +// event (the test's runInDurableObject). While it is parked the input gate is +// open, so a second frame for the same txId can be dispatched: the window an +// I/O-bound authorize opens in production (ADR-0025 amendment, in-flight). + +/** Gates keyed by a tag the test chooses; module-scoped because `authorize` + * has no instance. `parked` counts authorize calls that reached the gate. */ +const gates = new Map void> }>() + +function gateFor(tag: string) { + let g = gates.get(tag) + if (!g) gates.set(tag, (g = { parked: 0, open: false, waiters: [] })) + return g +} + +async function parkAt(tag: string): Promise { + const g = gateFor(tag) + g.parked++ + if (!g.open) await new Promise((r) => g.waiters.push(r)) +} + +export const gatedSchema = sync.schema({ + collections: { + gated: sync.collection({ + pk: "id", + mutations: { + insert: { + authorize: ({ op }) => parkAt(op.cols.body), + execute: ({ op, sql }) => { + sql.exec("INSERT INTO gated(id, body) VALUES (?, ?)", op.cols.id, op.cols.body) + }, + }, + }, + }), + }, + commands: { + // A side effect that must run once per txId: bumps a counter, returns it. + bump: sync.command<{ tag: string }>()({ + authorize: ({ args }) => parkAt(args.tag), + execute: ({ sql }) => { + sql.exec("UPDATE bumps SET n = n + 1") + return { n: Array.from(sql.exec("SELECT n FROM bumps"))[0]!.n as number } + }, + }), + echo: sync.command()(({ args }) => ({ echoed: args })), + }, +}) + +export type GatedApi = typeof gatedSchema + +export class GatedTestDO extends SyncDurableObject { + constructor(ctx: DurableObjectState, env: unknown) { + super(ctx, env) + ctx.blockConcurrencyWhile(async () => { + this.sql.exec(`CREATE TABLE IF NOT EXISTS gated (id TEXT PRIMARY KEY, body TEXT)`) + this.sql.exec(`CREATE TABLE IF NOT EXISTS bumps (n INTEGER NOT NULL)`) + if (Array.from(this.sql.exec("SELECT n FROM bumps")).length === 0) this.sql.exec("INSERT INTO bumps(n) VALUES (0)") + this.registerSync(gatedSchema) + }) + } + + protected override parseAttachment(): Claims { + return { userId: "anon" } + } + + /** How many authorize calls have reached the gate `tag`. */ + gateParked(tag: string): number { + return gateFor(tag).parked + } + + /** Open the gate `tag`: release every parked authorize, and let later ones through. */ + gateRelease(tag: string): void { + const g = gateFor(tag) + g.open = true + for (const w of g.waiters.splice(0)) w() + } +} + // ---- Host-matrix fixtures (host-matrix.test.ts) --------------------------- // // A fake partyserver-like host base and the Syncable mixin applied over it. The @@ -365,6 +445,7 @@ interface Env { MAINT_DO: DurableObjectNamespace SLOW_DO: DurableObjectNamespace LIMITS_DO: DurableObjectNamespace + GATED_DO: DurableObjectNamespace HOST_DO: DurableObjectNamespace HOST_OPTIN_DO: DurableObjectNamespace } @@ -388,6 +469,10 @@ export default { const name = url.pathname.slice("/limits/".length) || "default" return env.LIMITS_DO.get(env.LIMITS_DO.idFromName(name)).fetch(req) } + if (url.pathname.startsWith("/gated/")) { + const name = url.pathname.slice("/gated/".length) || "default" + return env.GATED_DO.get(env.GATED_DO.idFromName(name)).fetch(req) + } // Host-matrix routes: /host//... and /host-optin//... forward the // WHOLE request so the DO sees the trailing /_sync or /_host discriminator. if (url.pathname.startsWith("/host/")) { diff --git a/vitest.config.ts b/vitest.config.ts index 2c63231..dac75c2 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -24,6 +24,7 @@ export default defineConfig({ MAINT_DO: { className: "MaintTestDO", useSQLite: true }, SLOW_DO: { className: "SlowTickDO", useSQLite: true }, LIMITS_DO: { className: "LimitsTestDO", useSQLite: true }, + GATED_DO: { className: "GatedTestDO", useSQLite: true }, HOST_DO: { className: "SyncOverHostDO", useSQLite: true }, HOST_OPTIN_DO: { className: "SyncOverHostOptInDO", useSQLite: true }, },