fix(server): a duplicate of an in-flight txId waits and replays, never runs twice (ADR-0025 amendment)

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
 

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).

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

src/server/mixin.ts

@@ -192,6 +192,9 @@ export function Syncable<Env = unknown, TUser = unknown>() {
       #writesSinceCompaction = 0
       readonly #broadcaster: Broadcaster
       readonly #liveWs = new Set<WebSocket>()
+      /** 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<string, Promise<void>>()
       readonly #api: SyncApi<Env, TUser>
 
       constructor(...args: any[]) {
@@ -705,10 +708,11 @@ export function Syncable<Env = unknown, TUser = unknown>() {
        */
       async #handleMut(ws: WebSocket, f: Extract<ClientFrame, { t: "mut" }>): Promise<void> {
         // 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<ClientFrame, { t: "mut" }>): Promise<void> {
         // 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<Env = unknown, TUser = unknown>() {
 
       /** Run a named command (outside any transaction) and confirm with its result. */
       async #handleCall(ws: WebSocket, f: Extract<ClientFrame, { t: "call" }>): Promise<void> {
-        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<ClientFrame, { t: "call" }>): Promise<void> {
 
         const def = this.#registry.commands.get(f.name)
         if (!def) return this.#rejectTx(ws, f.txId, `unknown command '${f.name}'`, "UNKNOWN_COMMAND")
@@ -842,13 +848,41 @@ export function Syncable<Env = unknown, TUser = unknown>() {
         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<void>): Promise<void> {
+        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<void>((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 } })
       }
 
       #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

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
   }