From c57a70ddb95f8ca93d891f483e36610765ecf3db Mon Sep 17 00:00:00 2001 From: Tom McKenzie Date: Wed, 26 Aug 2026 16:18:53 +1000 Subject: [PATCH] =?UTF-8?q?fix(client):=20settle=20in-flight=20mut/call=20?= =?UTF-8?q?across=20an=20unexpected=20close=20=E2=80=94=20hold-and-replay?= =?UTF-8?q?=20+=20typed=20ConnectionLostError?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An in-flight mut/call whose socket dropped unexpectedly was abandoned to its generic 5s confirmation timeout, fully decoupled from the close event. The sharp edge is server-side ordering: recordTx + broadcast run BEFORE the `committed` send, so a drop in that window rolled back a write that had already durably committed (commit -> rollback -> reappear), and the mut promise rejected — with a timeout indistinguishable from a quiet server — for a write that landed. Primary (hold-and-replay): on an unexpected drop with subscriptions active, PARK each in-flight mut/call (swap its generic-timeout timer for a bounded ConnectionLostError timer, retain the encoded frame) and RESEND it on reconnect. The server's dedup table answers the replayed txId with its true recorded outcome — a committed write resolves `committed` (never a timeout rollback) within the dedup window; a rejected one rejects with its real MutationRejectedError; a never-received one executes once. Exactly-once holds either way (ADR-0002 C5). Fallback (typed): where no reconnect can resolve it — a terminal 4xxx close, no active subscriptions, or replay slower than timeoutMs from the drop — the mut/call settles promptly with the new exported ConnectionLostError, instanceof-distinct from MutationRejectedError, TransportClosedError, and the generic timeout. The type is the contract: an app holds its optimistic overlay instead of flashing a rollback. Timeout semantics for a socket that stays open are unchanged. Also (codex adversary review): - unsubscribe() of the last sub while a reconnect is pending cancels the pending timer, clears the reconnecting flag (a handshake already in flight installs without replaying) and settles parked txs typed — a subless reconnect must never replay. - reject a CONCURRENT duplicate in-flight txId loud (MutationRejectedError DUPLICATE_TXID) so a timer's delete-by-key can't evict a different waiter. - export TransportClosedError from the client barrel (declared a public export in ADR-0020 but never wired) so the full error taxonomy is instanceof-usable. New ADR-0021. Fixes #39. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 19 + ...1-in-flight-settlement-unexpected-close.md | 158 +++++++ docs/adr/README.md | 1 + src/client/index.ts | 2 + src/client/transport.ts | 134 +++++- tests/pending-tx-close.test.ts | 398 ++++++++++++++++++ 6 files changed, 709 insertions(+), 3 deletions(-) create mode 100644 docs/adr/0021-in-flight-settlement-unexpected-close.md create mode 100644 tests/pending-tx-close.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 382f16b..b91bce3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,12 @@ While pre-1.0, the public API may change between 0.x releases. Additive and backward-compatible — existing openers compile and run unchanged — but a custom opener should honour it (close its socket, reject) so `close()` during an in-flight handshake frees a still-CONNECTING socket. +- **New export `ConnectionLostError` (ADR-0021, issue #39).** Settles an + in-flight `mut`/`call` whose socket dropped unexpectedly before its receipt + arrived, when no reconnect+replay can resolve the true outcome — catchable and + `instanceof`-distinct from `MutationRejectedError`, `TransportClosedError`, and + the generic confirmation timeout. The type is the contract: an app catches it + to hold its optimistic overlay (the outcome is unknown) rather than roll back. ### Fixed @@ -89,6 +95,19 @@ While pre-1.0, the public API may change between 0.x releases. sub after the socket opens — previously the server persisted a ghost subscription (ADR-0019) with no local consumer until the socket dropped. Pre-existing; surfaced by the SSR lift's adversary review. +- **An in-flight `mut`/`call` is no longer abandoned to its generic timeout when + the socket drops unexpectedly (ADR-0021, issue #39).** Server-side, the durable + commit + broadcast run *before* the `committed` send, so a drop in that window + used to time out and roll back a write that had already succeeded. Now, on an + unexpected drop with subscriptions active, the transport HOLDS each in-flight + `mut`/`call` and REPLAYS it on reconnect; the server's dedup table answers the + replayed `txId` with its true recorded outcome — a committed write resolves + `committed` (never a timeout rollback) within the dedup window, a rejected one + rejects with its real `MutationRejectedError`. Where no reconnect can resolve it + (a terminal 4xxx close, or no active subscriptions), or the replay does not + answer within `timeoutMs` of the drop, the mutation settles promptly with the + new typed `ConnectionLostError` instead of the generic timeout. Timeout + semantics for a socket that stays open are unchanged. ## [0.6.0] — 2026-07-27 diff --git a/docs/adr/0021-in-flight-settlement-unexpected-close.md b/docs/adr/0021-in-flight-settlement-unexpected-close.md new file mode 100644 index 0000000..b0c202a --- /dev/null +++ b/docs/adr/0021-in-flight-settlement-unexpected-close.md @@ -0,0 +1,158 @@ +# 0021 — In-flight mut/call settlement across an unexpected close + +**Status:** Accepted. Fixes issue #39. Builds on ADR-0020 (connect contract), +ADR-0016 (reconnect policy), ADR-0011 (stale-socket receipt semantics), and the +dedup table (ADR-0002 C5, issue #21). Amends none. + +## Context + +`sendMut`/`sendCall` resolve when the server's `committed`/`rejected` for their +`txId` arrives on the same ordered stream (ADR-0002 C1). Each pending waiter is +armed with a single confirmation timeout (`timeoutMs`, default 5 s). Until this +ADR, the UNEXPECTED-close listener never touched `pendingTx`: only the app's own +`close()` cleared it. So a `mut`/`call` in flight when the socket dropped sat, +fully decoupled from the close event, until its own timeout fired. + +The sharp edge is server-side ordering. In `#handleMut`/`#handleCall`, `recordTx` +(the durable dedup entry) and the delta broadcast run **before** the `committed` +send, with no try/catch around the send. A drop in that window means the write +is already durably committed while the client's confirmation times out and +**rolls back a write that succeeded**. The reconnect's resubscribe later +converges the row back, but the app saw commit → rollback → reappear, and the +`mut` promise rejected — with a *generic timeout* indistinguishable from a +server that simply went quiet — for a write that landed. + +Observed 2026-07-29 against 0.6.0 while wiring terminal-close handling in a +shipping product; verified still present on `main` post-#36. ADR-0011's lift +fixed one adjacent edge (a stale socket's *late* receipt now settles its +waiter), but the drop-to-timeout path remained. + +## Decision + +Two complementary mechanisms, primary + fallback. Both hang off the existing +unexpected-close handler and the existing reconnect path — no new timers, no +idle work (ADR-0016 hibernation invariant preserved). + +### 1. Hold-and-replay (primary) — reconcile through the dedup table + +On an unexpected close while subscriptions are active (the same condition under +which ADR-0016 reconnects), the transport does **not** reject the in-flight +`mut`/`call`. It PARKS each one: the generic-timeout timer is swapped for a +bounded `ConnectionLostError` timer, and the encoded frame is retained. When the +reconnect installs a fresh socket, `resendParkedTx()` runs alongside +`resubscribeAll()` and **replays every parked frame**. The server's dedup table +answers each replayed `txId` with its true recorded outcome (`#replayReceipt`): + +- committed before the drop → the client receives `committed` and the mutation + **resolves** — no rollback, the exact case issue #39's second acceptance + bullet demands; +- rejected before the drop → the client receives `rejected` and the mutation + rejects with the true `MutationRejectedError` (and its code, issue #21); +- never received by the server (dropped in transit) → no dedup entry, so the + replay simply EXECUTES now, once. Exactly-once holds either way — this is + precisely what client-generated txIds + dedup were built for (ADR-0002 C5). + +This is "retry-through-dedup": the mutation the app already issued is the retry, +and the server's dedup makes the retry safe. + +### 2. Typed fallback — `ConnectionLostError` when no replay can answer + +A new exported error, `ConnectionLostError` (`instanceof`-distinct from +`MutationRejectedError`, `TransportClosedError`, and the generic timeout +`Error`), settles the waiter when reconnect cannot resolve it: + +- **Terminal close** (ADR-0016 policy returns `null`, e.g. a 4xxx auth + rejection): no reconnect will ever replay, so the parked waiters are failed + immediately after `onClosed`. +- **No active subscriptions**: ADR-0016 gates reconnect on live subscriptions + (to preserve hibernation), so a mutation-only transport that drops can never + replay. Its in-flight `mut`/`call` settle promptly with the typed error — we + do **not** bypass ADR-0016 to reconnect for a replay. This includes the + *transition* case: if the last subscription is `unsubscribe`d while a reconnect + is pending with parked txs, `unsubscribe` cancels a pending reconnect timer, + clears the reconnecting flag (so a handshake already in flight installs without + replaying), and settles the parked txs typed — a reconnect with zero subs must + never replay (codex review). A dial already parked on `open()` is not + force-aborted (that would entangle ADR-0020's epoch/revival machinery); at worst + it installs an idle, demand-less socket the app disposes via `close()`, exactly + as an initial connect can. +- **Replay too slow**: the parked timer bounds the hold to `timeoutMs` from the + drop. If the reconnect+replay hasn't answered by then, the app gets the typed + error rather than an unbounded hang. + +The type is the contract. An app catches `ConnectionLostError` to keep its +optimistic overlay pending (the outcome is genuinely unknown) instead of +flashing a rollback; the resubscribe catch-up then converges the row. + +### Why this shape (a **and** b, not a-only) + +Issue #39 sketched two options: (a) settle immediately with a typed +unknown-outcome error, or (b) reconcile after reconnect via dedup replay. We +chose **(b) as primary with (a) as the bounded fallback** rather than (a) alone: + +- **(a)-only cannot report the true outcome.** It settles every drop as + "unknown", so an app holding its overlay on that signal would be *wrong* for a + write the server actually *rejected* (validation failing exactly as the socket + dropped) — the overlay would linger for a write that never landed, and no + catch-up delta ever arrives to correct it. (b) resolves committed-vs-rejected + from the authoritative record. +- **(b) matches the ethos** (truthfulness, correctness over speed on load-bearing + paths): the app gets ground truth whenever a reconnect is possible, and the + typed error only where truth is genuinely unreachable. +- **(a) is still needed** because (b) is not always possible (terminal close, no + subs) and must be bounded (the parked timer) — and because the *type* is what + lets an app distinguish "unknown" from a real rejection or a quiet server. + +## Consequences & honesty + +- **New export:** `ConnectionLostError` (with optional `txId`). Settles an + in-flight `mut`/`call` only on an *unexpected* drop; never on an app `close()` + (that keeps its existing generic `Error("transport closed")` — the app knows + it closed) and never on a socket that stays open (generic confirmation + timeout, unchanged). +- **Dedup-window honesty — a deployment invariant, not a hope.** A replayed + `txId` only resolves the *true* outcome while its dedup entry survives + (`dedupRetentionMs`, default 1 h); outside that window the server sees it as a + brand-new frame and re-executes it. This is not new to this ADR — it is the + standing dedup contract (ADR-0002 C5): *any* client retry/outbox replay after + retention re-executes, which is exactly why C5 sizes retention to "the maximum + client retry/outbox window". This ADR's automatic replay is bounded to + `timeoutMs` from the drop (the parked timer removes the waiter and stops any + resend after that), so the operative invariant is simply **`timeoutMs` ≤ + `dedupRetentionMs`** — trivially satisfied by the defaults (5 s ≪ 1 h, three + orders of magnitude). A deployment that raises `timeoutMs` above the server's + dedup retention breaks the same at-most-once guarantee any retrying client + already depends on; the client cannot see `dedupRetentionMs` to enforce it, so + it is documented as a co-configuration constraint (codex review). Fully closing + it would need a server frame distinguishing "expired/unknown txId" from + "genuinely new" — a protocol change out of scope here and unnecessary at any + sane configuration. +- **Flapping is bounded.** A waiter already parked keeps the *first* drop's timer + across subsequent drops, so a flapping connection cannot extend the hold past + `timeoutMs` from the first drop. +- **One waiter per txId.** A concurrent reuse of an in-flight `txId` is rejected + loud (`MutationRejectedError` `DUPLICATE_TXID`) rather than allowed to overwrite + the first waiter — otherwise a timer's delete-by-key could evict a *different* + waiter and drop its receipt, and parking would carry the ambiguity across a + reconnect (codex review). A *sequential* retry after settlement is unaffected + (the prior entry is gone) — that is the intended retry-through-dedup path. +- **ADR-0011 not regressed.** A stale socket's late receipt still settles its + waiter (receipt dispatch is `stale`-agnostic; only the cursor is guarded). If a + buffered `committed` arrives before the `close` event, it removes the waiter + from `pendingTx` first, so parking only ever catches genuinely-unsettled + txIds. Double-settle is impossible: the receipt handler and both timers guard + on the map entry, and whichever fires first deletes it. +- **No new timers when idle.** Parking reuses the per-waiter timer slot; replay + rides the existing reconnect. Nothing polls while the DO is quiet. + +## Alternatives considered + +- **(a)-only (reject-with-unknown on every drop).** Simpler, prompt, but discards + the true outcome and shifts a rollback-vs-hold decision onto every app for a + case the dedup table can answer authoritatively. Rejected — see above. +- **Reconnect a subscription-less transport just to replay.** Would bypass + ADR-0016's hibernation gate for a marginal case; the typed fallback covers it + honestly instead. +- **Re-arm a fresh generic timeout on reconnect** (total wait up to 2×timeoutMs). + Rejected: the parked `ConnectionLostError` timer already bounds the hold and + carries the correct, distinguishable meaning ("connection was lost"). diff --git a/docs/adr/README.md b/docs/adr/README.md index c28b1ea..e12e15d 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -28,3 +28,4 @@ explains the displacement. | [0018](./0018-oversize-frames.md) | Oversize frames: client pre-send guard is the rejection surface; outbound is warn-only | Accepted | | [0019](./0019-subscription-persistence-across-hibernation.md) | Subscriptions persist in SQLite and restore on hibernation wake | Accepted | | [0020](./0020-connect-contract-abortable-open.md) | connect() never resolves disconnected; open() is abortable via AbortSignal | Accepted (amends 0016 + 0011 seam) | +| [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) | diff --git a/src/client/index.ts b/src/client/index.ts index ea0af12..507d8b6 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -9,8 +9,10 @@ // server-filtered by a `where` predicate. export { + ConnectionLostError, defaultReconnectDelay, MutationRejectedError, + TransportClosedError, WebSocketTransport, } from "./transport.ts" export type { ReconnectDelayFn, SubHandler, Transport, TransportOptions, WebSocketLike } from "./transport.ts" diff --git a/src/client/transport.ts b/src/client/transport.ts index ae90c6d..b0add28 100644 --- a/src/client/transport.ts +++ b/src/client/transport.ts @@ -107,6 +107,35 @@ export class TransportClosedError extends Error { } } +/** Settles an in-flight `mut`/`call` whose socket dropped UNEXPECTEDLY before the + * server's `committed`/`rejected` receipt arrived — the outcome is genuinely + * unknown to the client (issue #39 / ADR-0021). Distinct from every other + * settlement so an app can hold its optimistic overlay instead of flashing a + * rollback of a write the server may already have committed: + * + * - not a `MutationRejectedError` — the server did NOT reject it; + * - not the generic `Error("confirmation timeout: …")` — the socket died, it + * did not stay open and go quiet; + * - not a `TransportClosedError` — that is a `connect()`-time failure, not a + * settled-in-flight mutation. + * + * It is a FALLBACK: when the transport reconnects (subscriptions active, + * non-terminal close) it first replays each in-flight txId through the server's + * dedup table, so the TRUE outcome (`committed`/`rejected`) wins whenever it + * arrives within the parked window. This error surfaces only when no reconnect + * can resolve it (terminal 4xxx close, no active subscriptions) or the replay + * does not answer before the parked timeout. The type is the contract: an app + * catches it to keep the overlay pending until resubscribe converges the row. */ +export class ConnectionLostError extends Error { + constructor( + message = "connection lost before the server's receipt arrived; outcome unknown", + readonly txId?: string, + ) { + super(message) + this.name = "ConnectionLostError" + } +} + /** Reconnect delay policy (ADR-0016). Called once per reconnect attempt with * the 1-based attempt number (reset to 1 after each successful open) and, * when the drop came from a socket close, that close's code/reason (a failed @@ -168,6 +197,12 @@ interface TxWaiter { resolve: (r: { result?: unknown }) => void reject: (e: Error) => void timer: ReturnType + /** The encoded frame, retained so an unexpected close can replay it through the + * server's dedup table on reconnect (issue #39 / ADR-0021). */ + frame: WireOut + /** True once an unexpected close has swapped this waiter's generic-timeout timer + * for the bounded ConnectionLostError timer and marked it for replay-on-reconnect. */ + parked: boolean } // --- Typed-command projection over the schema Api (`typeof schema`) --------- @@ -401,10 +436,18 @@ export class WebSocketTransport { if (this.ws !== ws) return this.ws = null this.connectPromise = null + if (this.intentionallyClosed) return // close() owns pendingTx teardown // Auto-reconnect on an unexpected drop while subscriptions are active. - if (!this.intentionallyClosed && this.handlers.size > 0) { + if (this.handlers.size > 0) { + // Hold in-flight mut/call for dedup replay on reconnect (issue #39) + // BEFORE scheduling — a terminal policy then fails them typed below. + this.parkPendingForReplay() const { code, reason } = ev as { code?: number; reason?: string } this.scheduleReconnect(code, reason) + } else { + // No subscriptions → ADR-0016 does not reconnect, so no replay can ever + // learn the outcome: settle the in-flight mut/call now, typed (issue #39). + this.failPendingConnectionLost() } }) this.ws = ws @@ -418,6 +461,9 @@ export class WebSocketTransport { if (this.reconnecting) { this.reconnecting = false this.resubscribeAll() + // Replay any in-flight mut/call parked by the drop, so the dedup table + // answers each with its true recorded outcome (issue #39 / ADR-0021). + this.resendParkedTx() } })() this.connectPromise = p @@ -459,6 +505,9 @@ export class WebSocketTransport { // accept-then-close auth rejection): retrying cannot help — surface the // close to the app instead of looping against the DO. this.onClosed?.(closeCode, closeReason) + // No reconnect will ever replay these — settle in-flight mut/call typed now + // rather than let the parked timer wait out timeoutMs (issue #39). + this.failPendingConnectionLost() return } this.reconnectTimer = setTimeout(() => { @@ -492,6 +541,57 @@ export class WebSocketTransport { } } + /** Re-send every parked mut/call frame after a reconnect so the server's dedup + * table answers each replayed txId with its recorded outcome — the + * reconciliation half of issue #39 (ADR-0021). A txId the server never saw + * (dropped before it was received) simply executes now; either way the parked + * waiter settles with the TRUE outcome instead of a spurious rollback. Replay + * stays inside the dedup window: the parked timer bounds the hold to timeoutMs + * (≪ dedupRetentionMs), so a resend only fires while the recorded outcome is + * still present. Runs on the FRESH socket (this.ws already installed). */ + private resendParkedTx(): void { + if (!this.ws) return + for (const [, w] of this.pendingTx) { + if (!w.parked) continue + try { + this.sendRaw(w.frame) + } catch { + /* socket died again mid-resend; the parked timer still bounds the wait */ + } + } + } + + /** An unexpected close with a reconnect pending: HOLD each in-flight mut/call + * for dedup replay rather than let its confirmation timeout fire a spurious + * rollback of a write the server may already have committed (issue #39). Swap + * the generic-timeout timer for a bounded ConnectionLostError timer so the wait + * can't outlast timeoutMs — if the reconnect+replay answers first the true + * outcome wins; otherwise the app gets the typed unknown-outcome error, never a + * plain timeout. Already-parked waiters keep the first drop's bounded timer, so + * a flapping connection can't extend the hold past timeoutMs from the first drop. */ + private parkPendingForReplay(): void { + for (const [txId, w] of this.pendingTx) { + if (w.parked) continue + clearTimeout(w.timer) + w.parked = true + w.timer = setTimeout(() => { + this.pendingTx.delete(txId) + w.reject(new ConnectionLostError(`connection lost before receipt: txId=${txId}`, txId)) + }, this.timeoutMs) + } + } + + /** Settle every in-flight mut/call with the typed unknown-outcome error — used + * when no reconnect will resolve them: a terminal 4xxx close, or an unexpected + * drop with no active subscriptions (ADR-0016 reconnects only for live subs). */ + private failPendingConnectionLost(): void { + for (const [txId, w] of this.pendingTx) { + clearTimeout(w.timer) + w.reject(new ConnectionLostError(`connection lost before receipt: txId=${txId}`, txId)) + } + this.pendingTx.clear() + } + close(): void { this.intentionallyClosed = true this.closeEpoch++ @@ -548,7 +648,25 @@ export class WebSocketTransport { unsubscribe(subId: string): void { this.handlers.delete(subId) - if (this.ws) this.sendFrame({ t: "unsub", subId }) + if (this.ws) { + this.sendFrame({ t: "unsub", subId }) + return + } + // Disconnected/reconnecting and the LAST subscription just went away: ADR-0016 + // does not reconnect for zero subs, so a pending reconnect has nothing to + // resubscribe — and it must never silently reconnect just to REPLAY a parked + // mut/call (that would contradict ADR-0021's no-subs fallback). Cancel a + // still-pending reconnect timer, clear the reconnecting flag so any handshake + // already in flight installs WITHOUT resubscribing or replaying, and settle the + // parked in-flight mut/call typed now (issue #39 / ADR-0021). A dial already + // parked on open() is not force-aborted here — that would entangle ADR-0020's + // epoch/revival machinery; at worst it installs an idle, demand-less socket + // (as an initial connect also can), which the app disposes via close(). + if (this.handlers.size === 0) { + this.clearReconnectTimer() + this.reconnecting = false + this.failPendingConnectionLost() + } } sendMut(frame: Extract): Promise<{ result?: unknown }> { @@ -628,12 +746,22 @@ export class WebSocketTransport { ) } await this.connect() + // A txId identifies exactly ONE in-flight mut/call. A CONCURRENT reuse would + // overwrite the first waiter, then either waiter's timeout timer (deleting by + // key) could evict the other's entry — dropping a receipt and corrupting the + // parked-replay identity across a reconnect (issue #39 / ADR-0021). Reject the + // duplicate loud rather than corrupt state. A SEQUENTIAL retry after the first + // settled is fine: its entry is already gone, so this guard does not fire — + // exactly the retry-through-dedup the server is built to answer. + if (this.pendingTx.has(txId)) { + throw new MutationRejectedError(`txId already in flight: ${txId}`, "DUPLICATE_TXID") + } return new Promise((resolve, reject) => { const timer = setTimeout(() => { this.pendingTx.delete(txId) reject(new Error(`confirmation timeout: txId=${txId}`)) }, this.timeoutMs) - this.pendingTx.set(txId, { resolve, reject, timer }) + this.pendingTx.set(txId, { resolve, reject, timer, frame: encoded, parked: false }) // A socket may refuse a send synchronously (workerd does for frames over // its own cap). Clean up the waiter/timer before rejecting, or the stale // entry lingers with an armed timeout for timeoutMs (codex review). diff --git a/tests/pending-tx-close.test.ts b/tests/pending-tx-close.test.ts new file mode 100644 index 0000000..b71e78d --- /dev/null +++ b/tests/pending-tx-close.test.ts @@ -0,0 +1,398 @@ +import { env, runInDurableObject, SELF } from "cloudflare:test" +import { describe, expect, it } from "vitest" +import { + ConnectionLostError, + MutationRejectedError, + 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" + +// WHY (issue #39 / ADR-0021): an in-flight mut/call whose socket drops +// UNEXPECTEDLY must not be abandoned to its generic 5s confirmation timeout — +// server-side, recordTx + broadcast run BEFORE the `committed` send, so a drop +// in that window means the write is already durably committed while the client's +// optimistic overlay times out and rolls back a write that SUCCEEDED. +// +// The contract these pin: +// - Hold-and-replay (primary): on an unexpected drop with subscriptions active, +// the transport reconnects and REPLAYS each in-flight txId; the server's dedup +// table answers with the TRUE recorded outcome (committed/rejected). A write +// the server committed resolves `committed`, never a timeout rejection. +// - Typed fallback: when no reconnect can resolve it (terminal 4xxx close, or no +// active subscriptions), the in-flight mut/call settles PROMPTLY with a typed, +// distinguishable ConnectionLostError — never the generic timeout. +// - Timeout semantics are UNCHANGED for a socket that stays open. + +const codec = createFrameCodec() + +function noopHandler(): SubHandler { + return { onSnap: () => {}, onSnapEnd: () => {}, onDelta: () => {}, onUptodate: () => {}, onReset: () => {} } +} + +async function waitFor(pred: () => boolean, timeoutMs = 3000): Promise { + const start = Date.now() + while (!pred()) { + if (Date.now() - start > timeoutMs) throw new Error("waitFor timeout") + await new Promise((r) => setTimeout(r, 5)) + } +} + +interface Fake { + ws: WebSocketLike + sent: Array + emit: (type: string, ev: { data?: unknown; code?: number; reason?: string }) => void +} + +function makeFake(): Fake { + const listeners = new Map void>>() + const fake: Fake = { + sent: [], + emit: (type, ev) => { + for (const l of listeners.get(type) ?? []) l(ev) + }, + ws: { + send: (data) => fake.sent.push(codec.decode(data as ArrayBuffer | string) as ClientFrame), + close: () => {}, + addEventListener: (type, l) => { + const arr = listeners.get(type) ?? [] + arr.push(l) + listeners.set(type, arr) + }, + removeEventListener: () => {}, + }, + } + return fake +} + +const mut = (txId: string, key: string, body: string): Extract => ({ + t: "mut", + txId, + collection: "messages", + ops: [{ type: "insert", key, cols: { id: key, body } }], +}) + +// --- Transport-level contract (fake sockets, exact timing control) ---------- + +describe("in-flight mut settlement across an unexpected close (#39) — transport contract", () => { + it("drop-before-server-receives with NO subscriptions settles PROMPTLY with a typed ConnectionLostError", async () => { + const fake = makeFake() + const t = new WebSocketTransport({ url: "wss://x", open: () => fake.ws, timeoutMs: 60_000 }) + + const p = t.sendMut(mut("T1", "a", "hi")) // no subscriptions registered + const outcome = p.then(() => "resolved", (e) => e) + await waitFor(() => fake.sent.some((f) => f.t === "mut")) // the frame left + + fake.emit("close", { code: 1006 }) // unexpected drop, no subs → no reconnect + const r = await outcome + // Settled by the drop, NOT the 60s timeout — and typed, not a generic Error. + expect(r).toBeInstanceOf(ConnectionLostError) + expect(r).not.toBeInstanceOf(MutationRejectedError) + t.close() + }) + + it("holds an in-flight mut across a transient drop and REPLAYS it on reconnect (the resent frame lands on the new socket)", async () => { + const fakes: Array = [] + const t = new WebSocketTransport({ + url: "wss://x", + reconnectDelay: () => 5, + timeoutMs: 60_000, + open: () => { + const f = makeFake() + fakes.push(f) + return f.ws + }, + }) + await t.subscribe("s1", "messages", noopHandler()) + await waitFor(() => fakes.length === 1) + + const p = t.sendMut(mut("T2", "a", "hi")) + let settled = false + const outcome = p.then((v) => ((settled = true), v), (e) => ((settled = true), e)) + await waitFor(() => fakes[0]!.sent.some((f) => f.t === "mut" && (f as { txId?: string }).txId === "T2")) + + // Unexpected drop: the mut is parked, NOT rejected — the app keeps its overlay. + fakes[0]!.emit("close", { code: 1006 }) + // Reconnect opens a fresh socket… + await waitFor(() => fakes.length === 2) + // …which is resubscribed AND replayed the parked mut (retry-through-dedup). + await waitFor(() => fakes[1]!.sent.some((f) => f.t === "mut" && (f as { txId?: string }).txId === "T2")) + expect(fakes[1]!.sent.some((f) => f.t === "sub" && (f as { subId?: string }).subId === "s1")).toBe(true) + expect(settled).toBe(false) // still pending — held for the dedup answer + + // The server's dedup replay answers committed → the promise resolves (no rollback). + fakes[1]!.emit("message", { data: codec.encode({ t: "committed", txId: "T2", seq: "1" } as ServerFrame) }) + await expect(outcome).resolves.toEqual({ result: undefined }) + t.close() + }) + + it("a TERMINAL 4xxx close settles the in-flight mut with ConnectionLostError (and fires onClosed)", async () => { + const fakes: Array = [] + const closed: Array<[number | undefined, string | undefined]> = [] + const t = new WebSocketTransport({ + url: "wss://x", + reconnectDelay: 5, + timeoutMs: 60_000, + onClosed: (code, reason) => closed.push([code, reason]), + open: () => { + const f = makeFake() + fakes.push(f) + return f.ws + }, + }) + await t.subscribe("s1", "messages", noopHandler()) + await waitFor(() => fakes.length === 1) + + const p = t.sendMut(mut("T3", "a", "hi")) + const outcome = p.then(() => "resolved", (e) => e) + await waitFor(() => fakes[0]!.sent.some((f) => f.t === "mut" && (f as { txId?: string }).txId === "T3")) + + fakes[0]!.emit("close", { code: 4403, reason: "removed from this workspace" }) // terminal + const r = await outcome + expect(r).toBeInstanceOf(ConnectionLostError) // no reconnect will ever replay it + await waitFor(() => closed.length === 1) + expect(closed[0]).toEqual([4403, "removed from this workspace"]) + t.close() + }) + + it("unsubscribing the LAST sub while a reconnect is pending settles the parked mut typed and cancels the reconnect", async () => { + const fakes: Array = [] + let opens = 0 + const t = new WebSocketTransport({ + url: "wss://x", + reconnectDelay: () => 30, // long enough to unsubscribe before it fires + timeoutMs: 60_000, + open: () => { + opens++ + const f = makeFake() + fakes.push(f) + return f.ws + }, + }) + await t.subscribe("s1", "messages", noopHandler()) + await waitFor(() => opens === 1) + + const p = t.sendMut(mut("T5", "a", "hi")) + const outcome = p.then(() => "resolved", (e) => e) + await waitFor(() => fakes[0]!.sent.some((f) => f.t === "mut" && (f as { txId?: string }).txId === "T5")) + + fakes[0]!.emit("close", { code: 1006 }) // parked, reconnect scheduled (30ms) + t.unsubscribe("s1") // last sub gone BEFORE the timer fires + + // The parked mut settles typed now — no replay against a subless reconnect. + expect(await outcome).toBeInstanceOf(ConnectionLostError) + // And the reconnect was cancelled: no second socket ever opens. + await new Promise((r) => setTimeout(r, 60)) + expect(opens).toBe(1) + t.close() + }) + + it("unsubscribing the last sub while the reconnect HANDSHAKE is already in flight still settles typed and never replays", async () => { + const fakes: Array = [] + let releaseSecond: (() => void) | null = null + const t = new WebSocketTransport({ + url: "wss://x", + reconnectDelay: () => 5, // let the timer fire, so open() #2 is in flight + timeoutMs: 60_000, + open: () => { + if (fakes.length === 0) { + const f = makeFake() + fakes.push(f) + return f.ws + } + // Second dial (the reconnect): park it on open() so we can unsubscribe + // AFTER the timer fired but BEFORE the socket installs. + return new Promise((res) => { + const f = makeFake() + fakes.push(f) + releaseSecond = () => res(f.ws) + }) + }, + }) + await t.subscribe("s1", "messages", noopHandler()) + await waitFor(() => fakes.length === 1) + + const p = t.sendMut(mut("T6", "a", "hi")) + const outcome = p.then(() => "resolved", (e) => e) + await waitFor(() => fakes[0]!.sent.some((f) => f.t === "mut" && (f as { txId?: string }).txId === "T6")) + + fakes[0]!.emit("close", { code: 1006 }) // park + schedule reconnect + await waitFor(() => releaseSecond !== null) // the reconnect handshake is now in flight + + t.unsubscribe("s1") // last sub gone WHILE open() #2 is parked + // The parked mut settles typed immediately — it does not wait on the handshake. + expect(await outcome).toBeInstanceOf(ConnectionLostError) + + releaseSecond!() // the handshake resolves; the socket may install… + await new Promise((r) => setTimeout(r, 20)) + // …but it MUST NOT resubscribe or replay: no sub frame, no mut frame on it. + expect(fakes[1]!.sent.some((f) => f.t === "mut")).toBe(false) + expect(fakes[1]!.sent.some((f) => f.t === "sub")).toBe(false) + t.close() + }) + + it("rejects a CONCURRENT duplicate in-flight txId loud (never corrupts the first waiter)", async () => { + const fake = makeFake() + const t = new WebSocketTransport({ url: "wss://x", open: () => fake.ws, timeoutMs: 60_000 }) + await t.subscribe("s1", "messages", noopHandler()) + + const first = t.sendMut(mut("DUP", "a", "one")) + await waitFor(() => fake.sent.some((f) => f.t === "mut")) + // Second call reuses the in-flight txId → rejected loud, first left intact. + const second = await t.sendMut(mut("DUP", "b", "two")).then(() => "resolved", (e) => e) + expect(second).toBeInstanceOf(MutationRejectedError) + expect((second as MutationRejectedError).code).toBe("DUPLICATE_TXID") + + // The FIRST waiter still settles from its receipt — not evicted by the second. + fake.emit("message", { data: codec.encode({ t: "committed", txId: "DUP", seq: "1" } as ServerFrame) }) + await expect(first).resolves.toEqual({ result: undefined }) + t.close() + }) + + it("timeout semantics are UNCHANGED for a socket that stays open (generic confirmation timeout, not ConnectionLostError)", async () => { + const fake = makeFake() + const t = new WebSocketTransport({ url: "wss://x", open: () => fake.ws, timeoutMs: 30 }) + await t.subscribe("s1", "messages", noopHandler()) + + const r = await t.sendMut(mut("T4", "a", "hi")).then(() => "resolved", (e) => e) + expect(r).toBeInstanceOf(Error) + expect(r).not.toBeInstanceOf(ConnectionLostError) // the socket never dropped + expect((r as Error).message).toMatch(/confirmation timeout/) + t.close() + }) +}) + +// --- Full-stack reconciliation (real DO, real dedup table) ------------------ +// +// The sharp edge, end-to-end: a wrapper socket SWALLOWS the server's receipt for +// the in-flight txId, then the server-side socket is dropped — reproducing +// "committed server-side, socket died before `committed` arrived". On reconnect +// the transport replays the txId and the REAL dedup table answers with the true +// recorded outcome. + +interface Wrapped extends WebSocketLike { + swallowedReceipt: boolean +} + +/** Wrap a real accepted client socket so that the FIRST `committed`/`rejected` + * for `txId` is dropped before it reaches the transport — the receipt is lost in + * flight while the server has already recorded the outcome. */ +function dropReceiptFor(real: WebSocketLike, txId: string): Wrapped { + const wrapped: Wrapped = { + swallowedReceipt: false, + send: (d) => real.send(d), + close: (code, reason) => real.close(code, reason), + addEventListener: (type, l) => { + if (type !== "message") return real.addEventListener(type, l) + real.addEventListener("message", (ev) => { + if (!wrapped.swallowedReceipt) { + try { + const f = codec.decode(ev.data as ArrayBuffer | string) as ServerFrame + if ((f.t === "committed" || f.t === "rejected") && f.txId === txId) { + wrapped.swallowedReceipt = true + return // swallow: the client never sees this receipt + } + } catch { + /* not a frame we care about; fall through */ + } + } + l(ev) + }) + }, + removeEventListener: (type, l) => real.removeEventListener(type, l), + } + return wrapped +} + +async function openReal(room: string): Promise { + const res = await SELF.fetch(`https://example.com/sync/${room}`, { headers: { Upgrade: "websocket" } }) + const ws = res.webSocket + if (!ws) throw new Error("no webSocket") + ws.accept() + return ws as unknown as WebSocketLike +} + +async function dropServerSockets(room: string): Promise { + await runInDurableObject(env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), (_i, state) => { + for (const sock of state.getWebSockets()) sock.close(1000, "drop") + }) +} + +describe("in-flight mut settlement across an unexpected close (#39) — full-stack via dedup replay", () => { + it("drop-after-commit-before-receipt: the mut RESOLVES committed after reconnect — never a timeout rollback", async () => { + const room = "p39-commit" + const wrappers: Array = [] + const t = new WebSocketTransport({ + url: `https://example.com/sync/${room}`, + reconnectDelay: 20, + timeoutMs: 60_000, // large: prove the settlement is the reconnect+replay, not a timeout + open: async () => { + const real = await openReal(room) + // Only the FIRST socket drops the receipt; the reconnect socket is clean. + if (wrappers.length === 0) { + const w = dropReceiptFor(real, "TC") + wrappers.push(w) + return w + } + return real + }, + }) + await t.connect() + await t.subscribe("s1", "messages", noopHandler()) + + // Fire the mutation. The server commits + records it, broadcasts the delta, + // then sends `committed` — which the wrapper swallows. + const p = t.sendMut(mut("TC", "row", "landed")) + let settled = false + const outcome = p.then((v) => ((settled = true), v), (e) => ((settled = true), e)) + await waitFor(() => wrappers[0]!.swallowedReceipt) // receipt lost in flight + + expect(settled).toBe(false) // not rejected on the drop — held for the true answer + await dropServerSockets(room) // socket dies → parked → reconnect → replay + + // The dedup table answers the replayed txId with the recorded `committed`. + await expect(outcome).resolves.toBeDefined() + // Sanity: the row really is committed server-side. + await runInDurableObject(env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), (_i, s) => { + const rows = Array.from(s.storage.sql.exec("SELECT body FROM messages WHERE id = 'row'")) + expect(rows[0]).toEqual({ body: "landed" }) + }) + t.close() + }) + + it("drop-after-reject-before-receipt: the mut settles with the TRUE MutationRejectedError after reconnect", async () => { + const room = "p39-reject" + const wrappers: Array = [] + const t = new WebSocketTransport({ + url: `https://example.com/sync/${room}`, + reconnectDelay: 20, + timeoutMs: 60_000, + open: async () => { + const real = await openReal(room) + if (wrappers.length === 0) { + const w = dropReceiptFor(real, "TR") + wrappers.push(w) + return w + } + return real + }, + }) + await t.connect() + await t.subscribe("s1", "messages", noopHandler()) + + // FORBIDDEN body → the server records a rejection; the wrapper swallows it. + const p = t.sendMut(mut("TR", "row", "FORBIDDEN")) + const outcome = p.then(() => "resolved", (e) => e) + await waitFor(() => wrappers[0]!.swallowedReceipt) + + await dropServerSockets(room) + + // Replay resolves the "unknown" into the true recorded rejection — not a + // ConnectionLostError, and not a spurious commit. + const r = await outcome + expect(r).toBeInstanceOf(MutationRejectedError) + t.close() + }) +})