diff --git a/CHANGELOG.md b/CHANGELOG.md index 7df302c..d64d2f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,6 +51,31 @@ While pre-1.0, the public API may change between 0.x releases. ### Fixed +- **Subscriptions survive hibernation wake (ADR-0019; field report against + 0.5.1).** Hibernatable sockets survive a DO eviction by design, but the + subscription registry was instance memory: a wake restored the socket set + and nothing else, so an idle client's live queries went silently dead on a + still-open socket — its own mutations still confirmed while deltas fanned + out to nobody, and the client's only re-subscribe trigger (the close path) + never fired. Subscriptions are now written through to a durable + `_sync_subs` table keyed by a per-socket id tag stamped at accept, and + restored onto surviving sockets during `registerSync` on every wake — + before anything can dispatch or drain. No wire or client change: an + unmodified 0.5.1 client against a fixed server recovers fully. Orphaned + rows (a socket that dies without `webSocketClose`) are swept during the + existing compaction housekeeping; no idle timers (ADR-0006 invariant + intact). Pinned by `tests/hibernation.test.ts` under **real evictions** + (`evictDurableObject`, unlocked by the vitest 4 migration — the + eviction-based wake test issue #29 asked for), in five shapes: eviction + after successful broadcasts, eviction before the first-ever broadcast, + eviction of a cohosted base (tagged-restore branch, host socket untouched), + a restored sub whose collection left the schema (reconciled: `reset` + + row dropped, healthy subs untouched), and a restored predicate that no + longer compiles (socket closed with a non-terminal code so reconnect + re-subscribes — a `reset` would strand the query, adversarial review). + Sockets accepted by a **pre-fix** build that survive an in-place upgrade + carry no id tag and keep the old behavior (dead until reconnect) — a + one-release sharp edge, documented in ADR-0019. - **Rejection `code` now survives tx-dedup replay (#21).** The dedup record persisted only the rejection message, so a client retrying the same `txId` got the reason with no machine-readable `code` — breaking code-based error diff --git a/CLAUDE.md b/CLAUDE.md index d9747e3..7250916 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -83,6 +83,9 @@ Each is load-bearing and guarded by an ADR — read it before you touch the area - **One decision → one ADR.** Conventional commits (`!` = breaking); end with the Co-Authored-By trailer. Commit/push only when asked; `main` is the release branch. -- **Run plans and risky changes past codex (`gpt-5.5`) as an adversary** before - committing — it has caught real bugs here (the cursor-fetch race, the - scroll-back `DuplicateKeySyncError`). +- **Run plans and risky changes past codex (`gpt-5.6-sol`) as an adversary** + before committing — it has caught real bugs here (the cursor-fetch race, the + scroll-back `DuplicateKeySyncError`). Use the strongest available model; a + weaker adversary is a weaker review. Pipe input in (`git diff | codex exec + … -`) or close stdin (`codex exec "…" < /dev/null`) — an open stdin hangs + codex forever. diff --git a/docs/adr/0015-syncable-mixin.md b/docs/adr/0015-syncable-mixin.md index e624e18..5b1a3b3 100644 --- a/docs/adr/0015-syncable-mixin.md +++ b/docs/adr/0015-syncable-mixin.md @@ -223,3 +223,7 @@ tddc design gap; a PR against Actors is possible in principle but out of scope. (e.g. "sync and host sockets coexist"), not by a test that actually tears the instance down and reconstructs it. Upgrading `vitest`/`vitest-pool-workers` to unlock a real wake test is tracked as follow-up work, not bundled here. + *[Superseded: the vitest 4 migration (PR #34) lifted the toolchain blocker, + and ADR-0019's `tests/hibernation.test.ts` now exercises the restore under + real `evictDurableObject` cycles — which promptly exposed that the restore + was incomplete (sockets survived, subscriptions didn't). See ADR-0019.]* diff --git a/docs/adr/0019-subscription-persistence-across-hibernation.md b/docs/adr/0019-subscription-persistence-across-hibernation.md new file mode 100644 index 0000000..108eb7e --- /dev/null +++ b/docs/adr/0019-subscription-persistence-across-hibernation.md @@ -0,0 +1,217 @@ +# ADR-0019: Subscriptions persist in SQLite and restore on hibernation wake + +**Status**: Accepted +**Date**: 2026-07-27 +**Issue**: field report against 0.5.1 (downstream); also resolves [#29](https://github.com/grrowl/tanstack-durable-object-sync/issues/29) (real-eviction wake test) + +## Context + +ADR-0001 D13 chose hibernation-native sockets (`acceptWebSocket`, identity via +`serializeAttachment`); M7 added reconnect catch-up: after a socket **close**, +the client transport re-sends its `sub` frames with `since` cursors +(`resubscribeAll`, `transport.ts`). The unstated assumption: subscription loss +and socket loss coincide. + +Hibernation breaks that assumption by design. An eviction tears down instance +memory — including `SubscriptionRegistry` (`subscriptions.ts`), a +`WeakMap` whose own header said "lost on hibernation and +re-established by the client's `resub` on reconnect" — while hibernatable +sockets **survive**. The client never sees a close, so its only re-subscribe +trigger never fires. After any eviction: + +- the wake restore (`mixin.ts` constructor) re-populates the broadcaster's + socket set from `getWebSockets`, but the subscription registry starts empty; +- mutations still confirm — `committed` flows normally — while + `drainAndBroadcast` fans out to nobody; +- the client's live queries go **silently dead on a still-open socket**, until + an unrelated network drop forces a reconnect. + +Field-reported against 0.5.1. Pinned by `tests/hibernation.test.ts` under a +**real eviction** (`evictDurableObject`, default `webSockets: "hibernate"` — +production semantics), in both field shapes: eviction after successful +broadcasts, and eviction before the first-ever broadcast (which additionally +crosses the never-drained `_sync_meta` state with the restore). This is the +eviction-based wake test issue #29 asked for; the toolchain blocker recorded +in ADR-0015's Consequences ("verified by code-reading only", pinned +`vitest-pool-workers@0.12.21`) was lifted by the vitest 4 migration (PR #34), +and that note is superseded for this path. + +## Decisions + +### D0: Subscriptions are durable per-socket state + +The resub-on-reconnect design stays — it is correct for actual reconnects. +This ADR adds the missing half: on a wake **without** a reconnect, the server +restores what only the server lost. No protocol change, no client change; an +unmodified 0.5.1 client against a fixed server recovers fully. Any design +that needed a client upgrade to stop losing data was rejected on that ground +alone (see rejected alternatives). + +### D1: Storage — `_sync_subs`, written through, codec-encoded predicate + +```sql +CREATE TABLE IF NOT EXISTS _sync_subs ( + socket_id TEXT NOT NULL, + sub_id TEXT NOT NULL, + collection TEXT NOT NULL, + where_ir TEXT, -- tagged-value codec; NULL = unfiltered + PRIMARY KEY (socket_id, sub_id) +) +``` + +Created in `initSchema` (idempotent), so an already-deployed DO picks it up on +its first wake under the new build — the same in-place pattern as +`_sync_seen_tx.error_code` (ADR-0014 era migration). The predicate IR is +stored with the wire **value codec** (`encode`/`decode`, as +`dedup.ts#encodeResult` already does for command results), not `JSON.stringify` +— predicate literals may carry tagged values (dates, bytes) that JSON would +corrupt; the codec round-trips them exactly. + +Write-through happens at the mixin, keyed to the same events that mutate the +in-memory registry: `#handleSub` success upserts (`INSERT OR REPLACE` — a +re-sub on an existing `subId` replaces, matching ADR-0012's cap semantics), +`unsub` deletes the row, `webSocketClose`/`webSocketError` delete the socket's +rows. `SubscriptionRegistry` itself stays a pure in-memory index: its +predicate-floor behavior (ADR-0013) and unit tests are untouched. + +### D2: Socket identity across eviction — a second tag, not the attachment + +Rows must be re-associated with *surviving sockets* on wake. The socket id is +carried as a second tag at accept: `acceptWebSocket(server, [SYNC_TAG, +SOCKET_ID_TAG_PREFIX + id])`. Tags are the only per-socket datum that both +survives hibernation and stays out of the author's attachment; budget is +comfortable (Cloudflare allows 10 tags × 256 bytes; this uses 2). + +**Rejected alternative — the socket attachment** (the pattern a downstream +consumer uses for its own fixed-shape step subscriptions; also Cloudflare's +documented default for per-socket wake state). Steelmanned honestly (this +section was corrected on the simplicity-adversary round): + +- Its genuine advantages, conceded: the attachment dies with the socket — + no id tag, no orphan rows, no sweep; and it is the platform primitive + built for exactly this. +- **Contract stability is the decisive rejection.** The attachment is the + **author's** `TUser` claims (ADR-0001 D13; handlers read it raw via + `deserializeAttachment`, and ADR-0015's second discriminator — the `__pk` + guard — inspects that exact shape). A library envelope changes a documented + contract, needs shape-sniffing to coexist with pre-envelope sockets + surviving an upgrade wake, and every future reader inherits the ambiguity. +- **No honest aggregate bound.** `serializeAttachment` caps at 16,384 bytes + (current Cloudflare docs; an earlier revision of this ADR said 2 KiB — + stale). Most small deployments would fit, but `maxSubsPerSocket` is 256 + (ADR-0012) with unbounded predicate IR, so sufficiently complex apps hit a + cliff where sub #N is rejected *because of its siblings' sizes* — failure + dependent on unrelated state, for bookkeeping reasons. Reject-don't- + degrade cuts both ways. A size-triggered SQL fallback would be two + persistence designs instead of one. Re-serializing the whole set per + `sub`/`unsub` is also O(total state) write amplification. +- Cloudflare's own guidance for state beyond the cap is "use the Storage API + and store the corresponding key as an attachment" — i.e. **this ADR's + architecture is the platform's documented escalation path**; the key rides + a tag instead of the attachment purely for the contract reason above. + +**Rejected alternative — close every sync socket on wake** (the null +hypothesis; ~3 constructor lines would replace this entire ADR, with the +battle-tested reconnect + `resubscribeAll` + `since` path doing all +recovery, and the client staying sole owner of subscription truth): +hibernation follows ~10 seconds of idleness (Cloudflare lifecycle docs), so +this converts *ordinary application quiet* into reconnect waves — every wake +closes every connected client, each reconnect re-upgrades, re-auths, re-subs +and re-snapshots or catches up; a once-a-minute-burst workload would +reconnect each client ~1,440×/day, and the hibernation API's entire purpose +(cheap stable sockets across idle) is nullified by keeping sockets alive +with auto-pong only to slam them shut on wake. Correct but regressive. + +### D3: Restore point — inside `registerSync`, before anything can dispatch + +Restore runs at the end of `#registerSync`, after `initSchema`: for each +restored socket, read its id tag, load its rows, and `#subs.add` each. The +before-anything-dispatches guarantee is **conditional on the author's +documented obligation** (`registerSync` in the constructor's +`blockConcurrencyWhile`); a host that registers late fails loud at +`#registry` on the first frame rather than racing a half-restored registry. + +Two failure paths, corrected on adversarial review (both first-draft versions +sent `reset`, which the client answers by **truncating without +re-subscribing** — a reset here would strand the query dead while its single +cursor kept advancing past unseen changes): + +- **Predicate no longer compiles** (`UnsupportedPredicateError` — a version + change narrowed the evaluator floor): the socket is **closed** with a + non-terminal code (1011) after its durable rows are dropped. The client's + reconnect machinery re-subscribes everything recoverable with `since` + cursors, and the bad predicate flows through `#handleSub`'s pre-existing, + app-visible rejection. One bad predicate costs the socket's other subs a + reconnect round trip — acceptable for a should-never-happen skew path; + fail-loud beats limping. +- **Collection absent from the compiled schema** (an older build persisted + the row, or a mid-life re-registration removed the collection): a + post-restore **reconcile pass** removes the sub (memory + durable) and + sends `reset` — which IS correct here: the collection is gone, truncation + is the terminal state, there is nothing to re-subscribe to. Without this, + the restored sub is a zombie no drain will ever service. The reconcile pass + runs on every `registerSync`, so a mid-life re-registration that drops a + collection sheds its live subs the same way. + +### D4: No new timers; orphan hygiene rides compaction + +A socket that dies without `webSocketClose` ever firing (hard termination, +`webSockets: "close"` eviction of a crashed instance) leaves rows behind. +They are swept in `#maybeCompact`'s existing deferred housekeeping: delete +`_sync_subs` rows whose `socket_id` matches no live socket's id tag. Nothing +polls; the sweep rides work the DO was already awake to do. The no-idle-timer +invariant (ADR-0006) is preserved end to end: persistence writes ride frame +handlers, restore rides construction, hygiene rides compaction. + +**Rejected alternative — wake-time server-prompted resub** (a new server +frame asking surviving sockets to re-send `sub`s): requires a client upgrade +in lockstep (violates D0); turns every wake into per-socket +snapshot/catch-up round trips, which production eviction cadence would +amplify; and grows the wire protocol to route around what is purely +server-side bookkeeping. Recorded for the future, not planned: a +capability-negotiated protocol resub could in principle retire `_sync_subs` +once pre-protocol clients leave the support window (the simplicity round's +strongest long-term candidate) — but even at that end state, per-wake round +trips lose to a free local restore, so it is a machinery-deletion play, not +a runtime win. + +## Consequences + +- Live queries survive eviction on still-open sockets. Both field shapes are + pinned by `tests/hibernation.test.ts` under real evictions — the suite's + first actual eviction-and-reconstruct coverage (closes the gap ADR-0015 + documented; resolves #29). The test premise-guards the harness (asserts the + socket did **not** close on eviction) so a future change in pool-workers + semantics fails loudly instead of greening the pin for the wrong reason. +- Sockets accepted by a **pre-fix** build that survive an in-place upgrade + wake carry no id tag and no rows: they behave exactly as before the fix + (dead until reconnect). Durability begins with sockets accepted by the + fixed build. A one-release sharp edge — documented, not worked around. +- `sub`/`unsub` each cost one extra SQLite write (write-through), bracketed + by the snapshot they already trigger. Negligible. +- `_sync_subs` rows are socket ephemera, not data: `unsub` and close delete + them, re-subs replace in place, and compaction sweeps orphans — so the + table holds exactly the live subscriptions of currently-open sockets, and + is empty when nobody is connected. **No TTL, deliberately**: the sweep is + liveness-keyed, not time-keyed — a TTL shorter than a long-idle-but-alive + socket would delete live subscriptions (re-creating this ADR's bug), and a + longer one is pointless. +- `subscriptions.ts`'s header comment — the stale assumption itself — is + rewritten to point here. +- **Restored subs are grandfathered against a lowered `maxSubsPerSocket`** + (codex review): they passed the cap at accept time; enforcing a newly + lowered cap at restore would kill arbitrary live queries + nondeterministically. New subs enforce the current cap as before. +- **Write-through is not failure-atomic, deliberately** (codex review): a + SQLite failure between the in-memory mutation and the durable write is + handled by the platform's failure model — a failed Durable Object storage + write gates output and resets the instance, so a memory/durable divergence + cannot outlive the instance that created it; the next construction restores + from the durable truth. No rollback code rides the happy path. +- **Residual risk, out of scope**: deltas enqueued in the egress coalescer die + with instance memory *after* the drain cursor has advanced. In production + this window is closed by construction — an armed flush timer prevents + hibernation (broadcast.ts's documented design), so a DO with pending + coalesced deltas cannot be evicted. Only a test-harness forced eviction can + land inside the ~tickMs window; the hibernation tests await delta delivery + before evicting. Flagged rather than engineered around. diff --git a/docs/adr/README.md b/docs/adr/README.md index e9cf56f..112f09d 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -21,7 +21,8 @@ explains the displacement. | [0012](./0012-wire-input-hardening.md) | Wire-input hardening: frame-shape guards, inbound limits, sanitized execute errors | Accepted | | [0013](./0013-predicate-floor-one-evaluator.md) | Filtered-subscription membership: one evaluator is the source of truth; the floor is the verified-agreeing set | Accepted | | [0014](./0014-object-sync-schema.md) | `defineSync`: one schema value, mutations on the collection, commands on the connection | Accepted (supersedes 0001 D11 builder; closes 0010 manifest) | -| [0015](./0015-syncable-mixin.md) | `Syncable` mixin: the sync core as a mixin over any DO base | Accepted (reframes 0001 D13; Actor unsupported) | +| [0015](./0015-syncable-mixin.md) | `Syncable` mixin: the sync core as a mixin over any DO base | Accepted (reframes 0001 D13; Actor unsupported; wake-test gap closed by 0019) | | [0016](./0016-reconnect-policy.md) | Transport reconnect policy: injectable delay function, terminal 4xxx, jittered backoff | Accepted | | [0017](./0017-blob-wire-normalization.md) | BLOB wire normalization: bare ArrayBuffer becomes Uint8Array at emission | Accepted | | [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 | diff --git a/recipes/cohosting.md b/recipes/cohosting.md index 91a7007..d0aba80 100644 --- a/recipes/cohosting.md +++ b/recipes/cohosting.md @@ -76,13 +76,15 @@ Be clear-eyed about what backs the guarantee before you build on it. no socket entry points) were checked by reading partyserver 0.5.8, agents 0.17.3, and think 0.12.1. A newer host version may behave differently; the filtering is an internal behavior of a pre-1.0 package. -- **Not yet exercised end to end.** No test forces a real hibernation eviction, - so the wake-time restore is verified by code reading plus same-instance tag - assertions (the tooling to force an eviction lands in a newer - `vitest-pool-workers` than this repo pins; see ADR-0015). And CI never runs - the real `agents` package, only the fake. If you cohost with a real `Agent` - in production, you are ahead of our own test coverage; we would love to hear - how it goes. +- **Wake restore exercised under real evictions** — `tests/hibernation.test.ts` + forces actual eviction-and-reconstruct cycles (`evictDurableObject`, unlocked + by the vitest 4 migration) and pins that subscriptions survive on surviving + sockets (ADR-0019; the first real eviction promptly exposed that they + hadn't been). The eviction tests run over the bare-DO base; the cohosted + bases are still covered by same-instance tag assertions only. And CI never + runs the real `agents` package, only the fake. If you cohost with a real + `Agent` in production, you are ahead of our own test coverage; we would love + to hear how it goes. ## Using `@cloudflare/actors` diff --git a/src/server/changes.ts b/src/server/changes.ts index 6df9152..74dc8f4 100644 --- a/src/server/changes.ts +++ b/src/server/changes.ts @@ -55,6 +55,17 @@ export function initSchema(sql: SqlStorage): void { if (!seenTxCols.some((c) => c.name === "error_code")) { sql.exec("ALTER TABLE _sync_seen_tx ADD COLUMN error_code TEXT") } + // Durable per-socket subscriptions (ADR-0019): hibernatable sockets survive + // eviction while the in-memory registry does not; these rows let the wake + // restore re-associate subscriptions with surviving sockets. SQL lives in + // subscriptions.ts. + sql.exec(`CREATE TABLE IF NOT EXISTS _sync_subs ( + socket_id TEXT NOT NULL, + sub_id TEXT NOT NULL, + collection TEXT NOT NULL, + where_ir TEXT, + PRIMARY KEY (socket_id, sub_id) + )`) } /** diff --git a/src/server/mixin.ts b/src/server/mixin.ts index f43c7ee..6aaaed7 100644 --- a/src/server/mixin.ts +++ b/src/server/mixin.ts @@ -40,13 +40,28 @@ import { Broadcaster } from "./broadcast.ts" import { decodeResult, encodeResult, lookupTx, recordTx, type SeenTx, sweepDedup } from "./dedup.ts" import { compileSchema, type CompiledSync, type SyncSchema, ValidationError } from "./registry.ts" import { andPredicates, compileSubsetQuery, UnsupportedPredicateError } from "./sql-compiler.ts" -import { SubscriptionRegistry, type Sub } from "./subscriptions.ts" +import { + deletePersistedSocketSubs, + deletePersistedSub, + loadPersistedSubs, + persistSub, + SubscriptionRegistry, + type Sub, + sweepOrphanSubs, +} from "./subscriptions.ts" /** Reserved hibernation tag stamped on every sync socket. The wake-time restore * (`getWebSockets(SYNC_TAG)`) and every handler's socket-ownership check key off * this tag, so tddc never touches a host's sockets and vice versa (ADR-0015). */ export const SYNC_TAG = "_tddc" +/** Second tag stamped at accept: a per-socket id that survives hibernation and + * keys this socket's durable `_sync_subs` rows, so a wake can re-associate + * subscriptions with the surviving socket (ADR-0019). Tags — not the + * attachment, which is contractually the author's claims (ADR-0001 D13, + * ADR-0015) and capped at 16 KiB. Budget: 2 of Cloudflare's 10 tags/socket. */ +const SOCKET_ID_TAG_PREFIX = "_tddc.sid:" + /** Outbound frame-size warning threshold (ADR-0018): observability only, and * deliberately NOT a knob. Inbound is enforced at the edge-cap constant; * outbound is deliberately unenforced — breaking a correct broadcast to save @@ -254,6 +269,68 @@ export function Syncable() { initSchema(this.#sql) ensureTriggers(this.#sql, compiled.collections.values()) this.#compiled = compiled + this.#restoreSubs() + this.#reconcileSubs() + } + + /** Re-associate durable `_sync_subs` rows with the sockets that survived + * a hibernation wake (ADR-0019). Runs inside `registerSync` — the + * author's documented constructor obligation — so the registry is whole + * before any frame dispatch or drain can consult it (the guarantee is + * conditional on that obligation; a frame arriving before registration + * fails loud at `#registry`). Idempotent (a second registerSync re-adds + * the same entries; `add` replaces). + * + * A row whose predicate no longer compiles (a version change narrowed + * the evaluator floor) closes the SOCKET with a non-terminal code + * instead of sending `reset`: the client's `onReset` truncates but does + * not re-subscribe (a reset would strand the query dead while its + * cursor advances — codex review), whereas a close engages the + * reconnect machinery, which re-subscribes everything recoverable and + * routes the bad predicate through `#handleSub`'s pre-existing, + * app-visible rejection. */ + #restoreSubs(): void { + for (const ws of this.#liveWs) { + const sid = this.#socketIdFor(ws) + if (!sid) continue // legacy pre-0.6 socket: nothing was persisted + for (const row of loadPersistedSubs(this.#sql, sid)) { + try { + this.#subs.add(ws, row.subId, row.collection, row.where) + } catch (e) { + if (e instanceof UnsupportedPredicateError) { + console.error( + `restored sub '${row.subId}' on '${row.collection}' no longer compiles (${e.message}); ` + + `closing socket for clean re-subscribe`, + ) + this.#dropSocketSubs(ws) // durable rows too — the retry must re-derive from the client + this.#liveWs.delete(ws) + ws.close(1011, "sync restore failed; resubscribe") + break // remaining rows for this socket are moot + } + throw e + } + } + } + } + + /** Drop subscriptions whose collection is not in the compiled schema — + * in memory AND durably, with a `reset` so the client truncates + * (correct terminal state: the collection is gone, there is nothing to + * re-subscribe to). Covers a wake restoring rows from an older schema + * AND a mid-life re-registration that removed a collection; without + * this, such subs sit in the registry as zombies no drain will ever + * service (codex review). */ + #reconcileSubs(): void { + for (const ws of this.#liveWs) { + const sid = this.#socketIdFor(ws) + for (const sub of this.#subs.forWs(ws)) { + if (this.#registry.collections.has(sub.collection)) continue + console.error(`sub '${sub.subId}' dropped: collection '${sub.collection}' is not registered`) + this.#subs.remove(ws, sub.subId) + if (sid) deletePersistedSub(this.#sql, sid, sub.subId) + this.#send(ws, { t: "reset", sub: sub.subId }) + } + } } // ---- runtime-dispatched overrides -------------------------------------- @@ -307,7 +384,9 @@ export function Syncable() { server.serializeAttachment(attachment) // Tagged accept (SYNC_TAG) + plain attachment (no `__pk`): the two // independent discriminators that keep host and sync sockets apart. - this.ctx.acceptWebSocket(server, [SYNC_TAG]) + // The second tag is this socket's durable id (ADR-0019): tags survive + // hibernation, so a wake can find the socket's `_sync_subs` rows. + this.ctx.acceptWebSocket(server, [SYNC_TAG, SOCKET_ID_TAG_PREFIX + crypto.randomUUID()]) this.#liveWs.add(server) return new Response(null, { status: 101, webSocket: client }) @@ -321,6 +400,18 @@ export function Syncable() { return this.#isBareDO || this.ctx.getTags(ws).includes(SYNC_TAG) } + /** The socket's durable id (ADR-0019), read back from its accept-time + * tag. Undefined for a legacy socket accepted by a pre-0.6 build that + * survived an in-place upgrade wake — such a socket's subscriptions + * were never persisted, so it degrades to pre-fix behavior (dead until + * reconnect) rather than failing. */ + #socketIdFor(ws: WebSocket): string | undefined { + for (const tag of this.ctx.getTags(ws)) { + if (tag.startsWith(SOCKET_ID_TAG_PREFIX)) return tag.slice(SOCKET_ID_TAG_PREFIX.length) + } + return undefined + } + override async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): Promise { if (!this.#isSyncSocket(ws)) { const base = super.webSocketMessage as @@ -382,7 +473,7 @@ export function Syncable() { if (base) base.call(this, ws, code ?? 1000, reason ?? "", wasClean ?? false) return } - this.#subs.removeAll(ws) + this.#dropSocketSubs(ws) this.#liveWs.delete(ws) } @@ -392,10 +483,22 @@ export function Syncable() { if (base) base.call(this, ws, error) return } - this.#subs.removeAll(ws) + this.#dropSocketSubs(ws) this.#liveWs.delete(ws) } + /** Drop a closed/errored socket's subscriptions, in memory AND durably + * (ADR-0019). The `#compiled` guard covers a DO that never called + * `registerSync` (no `initSchema`, no `_sync_subs`): it can have had no + * subscriptions, so there is nothing durable to delete. */ + #dropSocketSubs(ws: WebSocket): void { + this.#subs.removeAll(ws) + if (this.#compiled) { + const sid = this.#socketIdFor(ws) + if (sid) deletePersistedSocketSubs(this.#sql, sid) + } + } + /** Shape-guard: returns true iff `v` is a structurally valid ClientFrame. * (ADR-0012) Runs after decode, before any SQL binding. * @@ -452,9 +555,14 @@ export function Syncable() { switch (frame.t) { case "sub": return this.#handleSub(ws, frame) - case "unsub": + case "unsub": { this.#subs.remove(ws, frame.subId) + if (this.#compiled) { + const sid = this.#socketIdFor(ws) + if (sid) deletePersistedSub(this.#sql, sid, frame.subId) + } return + } case "mut": return this.#handleMut(ws, frame) case "call": @@ -768,6 +876,16 @@ export function Syncable() { compactChanges(this.#sql) pruneChanges(this.#sql, this.changelogRetentionMs, Date.now()) sweepDedup(this.#sql, this.dedupRetentionMs, Date.now()) + // Orphaned subscription rows (socket died without webSocketClose — + // hard termination): hygiene rides compaction, nothing polls + // (ADR-0019 D4). Every id-tagged socket carries SYNC_TAG, so the + // tagged query is the complete live set. + const liveIds = new Set() + for (const ws of this.ctx.getWebSockets(SYNC_TAG)) { + const sid = this.#socketIdFor(ws) + if (sid) liveIds.add(sid) + } + sweepOrphanSubs(this.#sql, liveIds) })(), ) } @@ -835,6 +953,13 @@ export function Syncable() { } throw e } + // Write through (ADR-0019): the accepted sub becomes durable, so a + // hibernation wake can restore it onto the surviving socket. Only + // after `add` succeeds — a rejected predicate must not persist. + { + const sid = this.#socketIdFor(ws) + if (sid) persistSub(this.#sql, sid, frame.subId, frame.collection, frame.where) + } const seq = String(currentSeq(this.#sql)) // Reconnect catch-up: a `since` cursor asks for changes after that point diff --git a/src/server/subscriptions.ts b/src/server/subscriptions.ts index 5eae90c..9bb4192 100644 --- a/src/server/subscriptions.ts +++ b/src/server/subscriptions.ts @@ -1,12 +1,17 @@ -// In-memory subscription registry, keyed by WebSocket. Subscriptions live in -// memory only — they are lost on hibernation and re-established by the client's -// `resub` on reconnect (M7). +// Subscription registry: an in-memory index keyed by WebSocket, backed by +// durable per-socket rows in `_sync_subs` (ADR-0019). Hibernatable sockets +// survive eviction while instance memory does not, so the mixin writes through +// (sub/unsub/close) and restores this index from SQLite on every construction +// — a wake with surviving sockets keeps its live queries. An actual reconnect +// (socket close) still recovers via the client's `sub`+`since` path (M7). // // A subscription carries an optional `where` predicate IR (M5). It is compiled // with @tanstack/db's own evaluator so server-side filtering matches the // client's operator semantics exactly (no second predicate implementation). import { compileSingleRowExpression, toBooleanPredicate } from "@tanstack/db" +import type { SqlStorage } from "@cloudflare/workers-types" +import { decode as decodeValue, encode as encodeValue } from "../wire/codec.ts" import { UnsupportedPredicateError } from "./sql-compiler.ts" export interface Sub { @@ -105,3 +110,63 @@ export class SubscriptionRegistry { return out } } + +// ---- durable persistence for the registry (`_sync_subs`, ADR-0019) --------- +// Table-scoped SQL, dedup.ts-style. The table is created in +// changes.ts#initSchema. `where` IR is stored with the wire value codec (as +// dedup does for command results): predicate literals may carry tagged values +// (dates, bytes) that JSON.stringify would corrupt. + +/** A persisted subscription row, decoded. `where` is the wire-shaped IR. */ +export interface PersistedSub { + subId: string + collection: string + where: unknown +} + +/** Upsert one subscription row. A re-sub on an existing subId replaces (the + * same semantics as SubscriptionRegistry.add / ADR-0012's cap accounting). */ +export function persistSub(sql: SqlStorage, socketId: string, subId: string, collection: string, where: unknown): void { + sql.exec( + "INSERT OR REPLACE INTO _sync_subs(socket_id, sub_id, collection, where_ir) VALUES (?, ?, ?, ?)", + socketId, + subId, + collection, + where === undefined || where === null ? null : encodeValue(where), + ) +} + +export function deletePersistedSub(sql: SqlStorage, socketId: string, subId: string): void { + sql.exec("DELETE FROM _sync_subs WHERE socket_id = ? AND sub_id = ?", socketId, subId) +} + +export function deletePersistedSocketSubs(sql: SqlStorage, socketId: string): void { + sql.exec("DELETE FROM _sync_subs WHERE socket_id = ?", socketId) +} + +export function loadPersistedSubs(sql: SqlStorage, socketId: string): Array { + const rows = Array.from( + sql.exec<{ sub_id: string; collection: string; where_ir: string | null }>( + "SELECT sub_id, collection, where_ir FROM _sync_subs WHERE socket_id = ?", + socketId, + ), + ) + return rows.map((r) => ({ + subId: r.sub_id, + collection: r.collection, + where: r.where_ir === null ? undefined : decodeValue(r.where_ir), + })) +} + +/** Drop rows for sockets that are no longer live — hygiene for sockets that + * died without a webSocketClose (hard termination). Rides compaction + * (ADR-0019 D4); nothing polls. An empty `liveSocketIds` clears the table. */ +export function sweepOrphanSubs(sql: SqlStorage, liveSocketIds: ReadonlySet): void { + if (liveSocketIds.size === 0) { + sql.exec("DELETE FROM _sync_subs") + return + } + const ids = Array.from(liveSocketIds) + const placeholders = ids.map(() => "?").join(",") + sql.exec(`DELETE FROM _sync_subs WHERE socket_id NOT IN (${placeholders})`, ...ids) +} diff --git a/tests/hibernation.test.ts b/tests/hibernation.test.ts new file mode 100644 index 0000000..0f2dbd2 --- /dev/null +++ b/tests/hibernation.test.ts @@ -0,0 +1,298 @@ +import { env, evictDurableObject, runInDurableObject, SELF } from "cloudflare:test" +import { describe, expect, it } from "vitest" +import { createFrameCodec } from "../src/wire/frame-codec.ts" +import { encode as encodeValue } from "../src/wire/codec.ts" +import type { ClientFrame, ServerFrame } from "../src/wire/frames.ts" + +// WHY: hibernatable WebSockets SURVIVE a Durable Object eviction — that is the +// entire point of the hibernation API (ADR-0001 D13). The client therefore +// never sees a `close`, never reconnects, and never re-sends its `sub` frames +// (the transport's only re-subscribe trigger is the close path). If the +// server-side subscription registry lives in instance memory only, a wake +// restores the socket set but not the subscriptions: the client's own +// mutations still confirm (`committed` flows fine) while its live queries go +// silently dead on a still-open socket. Reported against 0.5.1 downstream; +// this pins the contract that a subscription on a surviving socket keeps +// receiving deltas across an eviction-and-wake cycle. +// +// `evictDurableObject` (vitest-pool-workers >= 0.16.20; here since the vitest 4 +// migration, PR #34) defaults to `webSockets: "hibernate"` — production +// semantics: instance memory torn down, durable storage and hibernatable +// sockets preserved. This is the real-eviction test issue #29 asked for. + +const codec = createFrameCodec() + +async function openWs(path: string, headers: Record = {}): Promise { + const res = await SELF.fetch(`https://example.com${path}`, { headers: { Upgrade: "websocket", ...headers } }) + 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)) +} + +function collectUntil(ws: WebSocket, done: (f: ServerFrame) => boolean, timeoutMs = 2000): 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("subscriptions survive hibernation wake", () => { + it("keeps delivering deltas to a pre-eviction subscriber on its surviving socket", async () => { + const room = "hib-sub-survives" + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)) + + // Subscribe on socket A and prove live fan-out works pre-eviction: A's own + // mutation must come back as a delta before its receipt (ADR-0002 C1). + const wsA = await openWs(`/sync/${room}`) + let aClosed = false + wsA.addEventListener("close", () => { + aClosed = true + }) + send(wsA, { t: "sub", subId: "s1", collection: "messages" }) + await collectUntil(wsA, (f) => f.t === "snap-end") + send(wsA, { + t: "mut", + txId: "pre1", + collection: "messages", + ops: [{ type: "insert", key: "pre", cols: { id: "pre", body: "before eviction" } }], + }) + const preFrames = await collectUntil(wsA, (f) => f.t === "committed") + expect(preFrames.some((f) => f.t === "d" && f.key === "pre")).toBe(true) + + // Real eviction: instance memory (incl. any in-memory subscription + // registry) is gone; durable storage and the hibernatable socket survive. + await evictDurableObject(stub) + + // Premise guard: the harness must NOT have closed A. If this fails, the + // eviction semantics changed and the pin below would be red for the wrong + // reason — a closed socket would make the client transport reconnect and + // re-subscribe, masking the registry loss entirely. + expect(aClosed).toBe(false) + + // Wake the DO from a DIFFERENT socket (the woken instance must serve B + // before A has sent anything — A is idle, as in the field report). Start + // collecting on A BEFORE the write so the delta cannot slip past. + const wsB = await openWs(`/sync/${room}`) + const deltaToA = collectUntil(wsA, (f) => f.t === "d" && f.key === "after-wake", 3000) + send(wsB, { + t: "mut", + txId: "post1", + collection: "messages", + ops: [{ type: "insert", key: "after-wake", cols: { id: "after-wake", body: "written after wake" } }], + }) + + // B's path is intact either way — the write commits durably. + await collectUntil(wsB, (f) => f.t === "committed") + + // THE PIN: A's subscription must have survived the wake. Without + // persistence this times out — the woken registry is empty, so the drain + // fans out to nobody while A's socket sits open and silent. + const frames = await deltaToA + const d = frames.find((f) => f.t === "d" && f.key === "after-wake") + expect(d).toBeDefined() + expect(d!.t === "d" && d!.sub).toBe("s1") + expect(aClosed).toBe(false) + + wsA.close(1000, "done") + wsB.close(1000, "done") + }) + + // Downstream field shape (their bump-verification case): a FRESH space that + // hibernates before its first-ever broadcast. Same root cause, but this + // variant also crosses the never-drained state (`_sync_meta` has no drain + // cursor yet) with the wake restore — the general test above cannot catch a + // regression specific to that interaction. + it("survives eviction that happens before the first-ever broadcast", async () => { + const room = "hib-first-broadcast" + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)) + + // Subscribe on a fresh DO: empty snapshot, and deliberately NO mutation — + // nothing has ever been drained or broadcast when the eviction lands. + const wsA = await openWs(`/sync/${room}`) + let aClosed = false + wsA.addEventListener("close", () => { + aClosed = true + }) + send(wsA, { t: "sub", subId: "s1", collection: "messages" }) + await collectUntil(wsA, (f) => f.t === "snap-end") + + await evictDurableObject(stub) + expect(aClosed).toBe(false) // premise guard, as above + + const wsB = await openWs(`/sync/${room}`) + const deltaToA = collectUntil(wsA, (f) => f.t === "d" && f.key === "first", 3000) + send(wsB, { + t: "mut", + txId: "first1", + collection: "messages", + ops: [{ type: "insert", key: "first", cols: { id: "first", body: "first ever write" } }], + }) + await collectUntil(wsB, (f) => f.t === "committed") + + const frames = await deltaToA + expect(frames.some((f) => f.t === "d" && f.key === "first" && f.sub === "s1")).toBe(true) + expect(aClosed).toBe(false) + + wsA.close(1000, "done") + wsB.close(1000, "done") + }) + + // Cohosted base (ADR-0015): the wake restore takes the tagged branch + // (`getWebSockets(SYNC_TAG)`), not the bare-DO catch-all — the branch the + // old host-matrix GAP note could only verify by code-reading. A host socket + // shares the DO across the same eviction; the restore must re-attach the + // sync socket's subscription and must NOT touch the host's socket. + it("restores subscriptions over a cohosted base without touching host sockets", async () => { + const room = "hib-cohosted" + const stub = env.HOST_DO.get(env.HOST_DO.idFromName(room)) + + const hostWs = await openWs(`/host/${room}/_host`) + let hostGot = 0 + hostWs.addEventListener("message", () => { + hostGot += 1 + }) + const wsA = await openWs(`/host/${room}/_sync`, { "x-user": "alice" }) + let aClosed = false + wsA.addEventListener("close", () => { + aClosed = true + }) + send(wsA, { t: "sub", subId: "s1", collection: "messages" }) + await collectUntil(wsA, (f) => f.t === "snap-end") + + await evictDurableObject(stub) + expect(aClosed).toBe(false) // premise guard, as above + + const wsB = await openWs(`/host/${room}/_sync`, { "x-user": "bob" }) + const deltaToA = collectUntil(wsA, (f) => f.t === "d" && f.key === "co", 3000) + send(wsB, { + t: "mut", + txId: "co1", + collection: "messages", + ops: [{ type: "insert", key: "co", cols: { id: "co", body: "cohosted wake" } }], + }) + await collectUntil(wsB, (f) => f.t === "committed") + + const frames = await deltaToA + expect(frames.some((f) => f.t === "d" && f.key === "co" && f.sub === "s1")).toBe(true) + // The sync delta reached A, so the broadcast cycle is complete — and the + // host's socket saw none of it (it survived the same eviction untouched). + // Settle one delivery window first: a frame wrongly fanned out to the host + // could still be in flight when A's delta resolves (codex review). + await new Promise((r) => setTimeout(r, 100)) + expect(hostGot).toBe(0) + + hostWs.close(1000, "done") + wsA.close(1000, "done") + wsB.close(1000, "done") + }) + + // Reconcile-on-restore (ADR-0019 D3, codex review): a durable row whose + // collection is not in the compiled schema (an older build persisted it, + // then the schema dropped the collection) must NOT restore into a zombie + // sub that no drain will ever service. The client gets a `reset` (truncate + // is the correct terminal state — the collection is gone), the row is + // deleted, and healthy subs on the same socket are untouched. + it("resets a restored sub whose collection left the schema; healthy subs unaffected", async () => { + const room = "hib-absent-collection" + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)) + + const wsA = await openWs(`/sync/${room}`) + send(wsA, { t: "sub", subId: "s1", collection: "messages" }) + await collectUntil(wsA, (f) => f.t === "snap-end") + + // Seed a row for THIS socket's durable id, naming a collection the schema + // does not register — as if persisted by an older build whose schema had it. + await runInDurableObject(stub, (_i, s) => { + const tags = s.getWebSockets().flatMap((w) => s.getTags(w)) + const sid = tags.find((t) => t.startsWith("_tddc.sid:"))!.slice("_tddc.sid:".length) + s.storage.sql.exec( + "INSERT INTO _sync_subs(socket_id, sub_id, collection, where_ir) VALUES(?, 's-ghost', 'ghosts', NULL)", + sid, + ) + }) + + await evictDurableObject(stub) + + // Wake the DO; restore + reconcile run in the constructor. Expect the + // ghost sub's reset on the surviving socket… + const resetP = collectUntil(wsA, (f) => f.t === "reset" && f.sub === "s-ghost", 3000) + const wsB = await openWs(`/sync/${room}`) + send(wsB, { + t: "mut", + txId: "g1", + collection: "messages", + ops: [{ type: "insert", key: "g", cols: { id: "g", body: "after ghost" } }], + }) + await resetP + // …while the healthy sub still delivers, and the ghost row is gone. + await collectUntil(wsA, (f) => f.t === "d" && f.key === "g", 3000) + const ghostRows = await runInDurableObject(stub, (_i, s) => + Array.from(s.storage.sql.exec("SELECT sub_id FROM _sync_subs WHERE sub_id = 's-ghost'")), + ) + expect(ghostRows.length).toBe(0) + + wsA.close(1000, "done") + wsB.close(1000, "done") + }) + + // Uncompilable restored predicate (ADR-0019 D3, codex review): `reset` would + // strand the query — the client's onReset truncates but never re-subscribes, + // while its cursor keeps advancing. The server must instead CLOSE the socket + // (non-terminal code): reconnect machinery re-subscribes everything + // recoverable and routes the bad predicate through the normal, app-visible + // sub rejection. Pinned with the ADR-0013 `ne` operator, which the evaluator + // floor rejects. + it("closes the socket when a restored predicate no longer compiles", async () => { + const room = "hib-bad-predicate" + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)) + + const wsA = await openWs(`/sync/${room}`) + const closed = new Promise<{ code: number }>((resolve) => { + wsA.addEventListener("close", (e) => resolve({ code: (e as CloseEvent).code })) + }) + send(wsA, { t: "sub", subId: "s1", collection: "messages" }) + await collectUntil(wsA, (f) => f.t === "snap-end") + + // Corrupt this socket's persisted predicate into IR outside the evaluator + // floor (`ne` — only `not(eq(...))` is supported, ADR-0013). + const badWhere = encodeValue({ + type: "func", + name: "ne", + args: [ + { type: "ref", path: ["body"] }, + { type: "val", value: "x" }, + ], + }) + await runInDurableObject(stub, (_i, s) => { + s.storage.sql.exec("UPDATE _sync_subs SET where_ir = ? WHERE sub_id = 's1'", badWhere) + }) + + await evictDurableObject(stub) + + // Any wake attempts the restore; the corrupt row must close A cleanly. + const wsB = await openWs(`/sync/${room}`) + const { code } = await closed + expect(code).toBe(1011) // non-terminal: a real client reconnects + resubscribes + const rows = await runInDurableObject(stub, (_i, s) => + Array.from(s.storage.sql.exec("SELECT sub_id FROM _sync_subs")), + ) + expect(rows.length).toBe(0) // the retry must re-derive from the client + + wsB.close(1000, "done") + }) +}) diff --git a/tests/host-matrix.test.ts b/tests/host-matrix.test.ts index 7049ff7..f6becd6 100644 --- a/tests/host-matrix.test.ts +++ b/tests/host-matrix.test.ts @@ -15,18 +15,12 @@ import { type FakeHost, testSchema } from "./test-worker.ts" // tables. The fake stands in for the real `agents` package so CI never carries a // ~13 MB pre-1.0 dependency; a real-`agents` smoke gates version bumps. // -// GAP (honest, not faked): none of these tests force a real hibernation -// eviction. They assert the tag partition (`getWebSockets(SYNC_TAG)` vs. -// `getWebSockets(HOST_TAG)`) on the SAME live instance — which pins the -// cohosting contract this file exists for, but not the wake-time restore -// itself (`mixin.ts`'s constructor, ADR-0015 discriminator 1) that reruns -// `getWebSockets(SYNC_TAG)` after a fresh eviction-and-reconstruct cycle. As -// of this repo's pinned `@cloudflare/vitest-pool-workers@0.12.21`, there is -// no in-process way to force that cycle: the `evictDurableObject`/ -// `evictAllDurableObjects` helpers that do it were added in `0.16.20`, which -// requires `vitest@^4` — a major bump out of scope here. The restore is -// verified by code-reading only; see ADR-0015's Consequences for detail and -// the upgrade this is blocked on. +// Scope note: these tests assert the tag partition (`getWebSockets(SYNC_TAG)` +// vs. `getWebSockets(HOST_TAG)`) on the SAME live instance — the cohosting +// contract this file exists for. The wake-time restore across a real +// eviction-and-reconstruct cycle is covered in `tests/hibernation.test.ts` +// (`evictDurableObject`, unlocked by the vitest 4 migration; ADR-0019 — +// whose real evictions exposed and fixed the subscription-registry loss). const codec = createFrameCodec() const HOST_TAG = "__host" diff --git a/tests/maybe-compact.test.ts b/tests/maybe-compact.test.ts index bce1aac..7f35b9f 100644 --- a/tests/maybe-compact.test.ts +++ b/tests/maybe-compact.test.ts @@ -117,4 +117,43 @@ describe("maybeCompact wiring: threshold gate + compaction + retention prune (co }) ws.close() }) + + // ADR-0019 D4: `_sync_subs` rows for a socket that died WITHOUT a + // webSocketClose (hard termination) are orphans — nothing else deletes them. + // Their hygiene rides this same housekeeping path, so it gets the same + // drive-the-real-path treatment: seed an orphan row, cross the threshold, + // poll for the deferred sweep. Precision matters as much as deletion: the + // live socket's own row must survive the same sweep. + it("sweeps orphaned _sync_subs rows while keeping live sockets' rows", async () => { + const room = "mc-orphan-subs" + const ws = await openWs(room) + send(ws, { t: "sub", subId: "s1", collection: "messages" }) + await waitForFrame(ws, (f) => f.t === "snap-end") + + // Seed the orphan: a socket_id no live socket bears. + await runInDurableObject(maint(room), (_i, s) => { + s.storage.sql.exec( + "INSERT INTO _sync_subs(socket_id, sub_id, collection, where_ir) VALUES('dead-socket','s9','messages',NULL)", + ) + }) + + const subRows = (): Promise> => + runInDurableObject(maint(room), (_i, s) => + Array.from(s.storage.sql.exec<{ socket_id: string }>("SELECT socket_id FROM _sync_subs")).map( + (r) => r.socket_id, + ), + ) + expect((await subRows()).includes("dead-socket")).toBe(true) // seeded + expect((await subRows()).length).toBe(2) // orphan + the live sub's write-through row + + // Cross compactionEvery=3 → housekeeping (incl. the sweep) is scheduled. + await insert(ws, "t1", "k1", "1") + await insert(ws, "t2", "k2", "2") + await insert(ws, "t3", "k3", "3") + await waitUntil(async () => { + const ids = await subRows() + return !ids.includes("dead-socket") && ids.length === 1 // orphan gone, live row kept + }) + ws.close() + }) })