From 77bb2e654935cebbc0b7c0547a2e3422807bfbf7 Mon Sep 17 00:00:00 2001 From: Tom McKenzie Date: Wed, 26 Aug 2026 15:44:07 +1000 Subject: [PATCH] =?UTF-8?q?test:=20promote=20scratch=20pins=20=E2=80=94=20?= =?UTF-8?q?host-owned=20alarm,=20runSyncedWrite=20capture/broadcast=20seam?= =?UTF-8?q?s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Promote two scratch verification files into the permanent suite by topic (the scratch files themselves are not added). Zero product code changes. host-matrix.test.ts (ADR-0015, host-owned alarm slot): - a full maybeCompact housekeeping cycle arms no alarm - a host-owned alarm survives a maybeCompact housekeeping cycle Both drive a real drain -> #maybeCompact -> waitUntil cycle (MaintTestDO, compactionEvery=3) and poll for the collapse before asserting the alarm slot. server-write.test.ts (ADR-0006, runSyncedWrite capture-vs-broadcast): - a Drizzle-style direct ctx.storage.sql.exec inside runSyncedWrite broadcasts (folds the retired handle-identity pin: the callback arg IS ctx.storage.sql) - a write OUTSIDE runSyncedWrite is trigger-captured but NOT broadcast until the next drain (asserts the dark half the suite only stated) - a multi-statement runSyncedWrite reaches the client as one wire batch: all deltas, then a single uptodate at one seq (trimmed) Retired without promotion: the callback-handle pin (folded above) and the 30-rapid-updates coalescing pin (already covered by coalesce.test.ts, whose final-value assertion needs no strengthening). Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/host-matrix.test.ts | 82 ++++++++++++++++++++++ tests/server-write.test.ts | 139 +++++++++++++++++++++++++++++++++++++ 2 files changed, 221 insertions(+) diff --git a/tests/host-matrix.test.ts b/tests/host-matrix.test.ts index f6becd6..a03530d 100644 --- a/tests/host-matrix.test.ts +++ b/tests/host-matrix.test.ts @@ -1,3 +1,4 @@ +import type { SqlStorage } from "@cloudflare/workers-types" import { env, runInDurableObject, SELF } from "cloudflare:test" import { describe, expect, it } from "vitest" import { createFrameCodec } from "../src/wire/frame-codec.ts" @@ -297,3 +298,84 @@ describe("Syncable over a partyserver-like host (ADR-0015)", () => { onWs.close() }) }) + +// The final cohosting-proof leg (ADR-0015 §"The cohosting proof"): "tddc defines +// no `alarm()`. Compaction rides `ctx.waitUntil` (`mixin.ts`, `#maybeCompact`), so +// the DO's single alarm slot stays wholly owned by the host." A partyserver host, +// an Agent's scheduler, or a cron `alarm` can therefore claim that one slot without +// contending with sync housekeeping. +// +// What these pin is the OBSERVABLE contract, not the internal mechanism: +// 1. A housekeeping cycle leaves NO alarm armed — so it can never wake an idle DO +// (the exact hazard ADR-0006/0015 cite for choosing `waitUntil` over an alarm). +// 2. A host's pre-set alarm SURVIVES a housekeeping cycle unchanged. This is the +// stronger guard: the DO exposes a SINGLE alarm slot, so any use of it by tddc +// would overwrite the host's value — its exact survival proves tddc never +// touched the slot, even transiently. +// That the housekeeping actually RAN (the deferred `waitUntil` → compaction path, vs. +// a no-op) is pinned in depth by `tests/maybe-compact.test.ts`; here we only re-drive +// it far enough to poll its collapse, then assert the alarm invariant. +// +// Vehicle: MaintTestDO (compactionEvery=3). The mixin's compaction threshold is a +// protected knob a non-DO host cannot override through the mixin's public type +// (ADR-0015), and configure() carries no such option — so a low-threshold bare-DO +// subclass is the only in-suite way to fire `#maybeCompact` after a few writes. The +// property is base-independent: `#maybeCompact` is one shared mixin method, never an +// `alarm()`. + +type MaintApi = { runSyncedWrite: (fn: (sql: SqlStorage) => T) => T } +const maintApi = (i: unknown): MaintApi => i as MaintApi +const maint = (room: string) => env.MAINT_DO.get(env.MAINT_DO.idFromName(room)) + +const getAlarm = (room: string): Promise => + runInDurableObject(maint(room), (_i, s) => s.storage.getAlarm()) + +/** Total rows in the change log — collapses to a stable value once the deferred + * compaction runs, so it is the effect we poll for (ADR-0009 latest-op-per-key). */ +const changeRowCount = (room: string): Promise => + runInDurableObject(maint(room), (_i, s) => + Array.from(s.storage.sql.exec<{ n: number }>("SELECT count(*) AS n FROM _sync_changes"))[0]!.n, + ) + +async function waitForCount(pred: () => Promise, timeoutMs = 2000): Promise { + const start = Date.now() + while (!(await pred())) { + if (Date.now() - start > timeoutMs) throw new Error("waitForCount timeout") + await new Promise((r) => setTimeout(r, 10)) + } +} + +/** Three server writes to one key cross compactionEvery=3: the third schedules the + * `waitUntil` housekeeping, which collapses the key's three change rows to one. + * Polling for that collapse proves the real deferred path ran (not just that a + * function exists). registerSync already ran in the DO constructor (ADR-0007). */ +async function driveHousekeepingCycle(room: string): Promise { + await runInDurableObject(maint(room), (instance) => { + maintApi(instance).runSyncedWrite((sql) => sql.exec("INSERT INTO messages(id,body) VALUES('hk','1')")) + maintApi(instance).runSyncedWrite((sql) => sql.exec("UPDATE messages SET body='2' WHERE id='hk'")) + maintApi(instance).runSyncedWrite((sql) => sql.exec("UPDATE messages SET body='3' WHERE id='hk'")) + }) + await waitForCount(async () => (await changeRowCount(room)) === 1) +} + +describe("host-owned alarm slot (ADR-0015)", () => { + it("a full maybeCompact housekeeping cycle arms no alarm", async () => { + const room = "alarm-none" + expect(await getAlarm(room)).toBeNull() // nothing scheduled at construction + await driveHousekeepingCycle(room) // real drain → #maybeCompact → waitUntil + // The housekeeping path never touched the alarm slot: it stays the host's. + expect(await getAlarm(room)).toBeNull() + }) + + it("a host-owned alarm survives a maybeCompact housekeeping cycle", async () => { + const room = "alarm-host" + // The host framework claims the DO's single alarm slot for its own schedule. + const when = Date.now() + 60 * 60 * 1000 + await runInDurableObject(maint(room), (_i, s) => s.storage.setAlarm(when)) + expect(await getAlarm(room)).toBe(when) + + await driveHousekeepingCycle(room) + // tddc's housekeeping neither cleared nor rescheduled the host's alarm. + expect(await getAlarm(room)).toBe(when) + }) +}) diff --git a/tests/server-write.test.ts b/tests/server-write.test.ts index aa87e87..bac167e 100644 --- a/tests/server-write.test.ts +++ b/tests/server-write.test.ts @@ -3,6 +3,8 @@ import type { SqlStorage } from "@cloudflare/workers-types" import { env, runInDurableObject, SELF } from "cloudflare:test" import { describe, expect, it } from "vitest" import { doCollectionOptions, type WebSocketLike, WebSocketTransport } from "../src/client/index.ts" +import { createFrameCodec } from "../src/wire/frame-codec.ts" +import type { ClientFrame, ServerFrame } from "../src/wire/frames.ts" import type { TestApi } from "./test-worker.ts" // WHY: server-originated writes (an agent inserting a row, a webhook, a cron @@ -14,11 +16,16 @@ import type { TestApi } from "./test-worker.ts" // runSyncedWrite is protected (subclass-facing); reach it in the test via the // in-DO instance. registerSync already ran in the DO constructor (ADR-0007). +// `drainAndBroadcast` is the documented manual-drain trigger (ADR-0006), re-aliased +// protected on SyncDurableObject — reached the same way to prove the dark half. type ServerApi = { runSyncedWrite: (fn: (sql: SqlStorage) => T) => T + drainAndBroadcast: () => void } const api = (i: unknown): ServerApi => i as unknown as ServerApi +const codec = createFrameCodec() + function realTransport(room: string): WebSocketTransport { return new WebSocketTransport({ url: `https://example.com/sync/${room}`, @@ -40,6 +47,42 @@ async function waitFor(pred: () => boolean, timeoutMs = 3000): Promise { } } +// A raw codec-level socket, for the wire-framing assertion (SW-D) that the +// collection abstraction hides: the batch's delta/`uptodate` boundary and seq. +async function openRawWs(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 +} + +/** Record every frame arriving on `ws` from now on. */ +function recordFrames(ws: WebSocket): Array { + const out: Array = [] + ws.addEventListener("message", (e: MessageEvent) => out.push(codec.decode(e.data as ArrayBuffer) as ServerFrame)) + return out +} + +/** Subscribe and await `snap-end`, so a later `recordFrames` sees only the + * frames the write under test produces (the empty snapshot is drained first). + * The listener is attached BEFORE the `sub` is sent, so a fast `snap-end` can't + * slip through in the gap. */ +async function subscribeRaw(ws: WebSocket, subId: string): Promise { + await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("no snap-end")), 2000) + const onMsg = (e: MessageEvent): void => { + if ((codec.decode(e.data as ArrayBuffer) as ServerFrame).t === "snap-end") { + clearTimeout(timer) + ws.removeEventListener("message", onMsg) + resolve() + } + } + ws.addEventListener("message", onMsg) + ws.send(codec.encode({ t: "sub", subId, collection: "messages" } satisfies ClientFrame)) + }) +} + describe("runSyncedWrite (ADR-0006) — server-originated writes", () => { it("broadcasts a server-originated insert to a connected client", async () => { const room = "rsw-live" @@ -60,6 +103,102 @@ describe("runSyncedWrite (ADR-0006) — server-originated writes", () => { t.close() }) + it("a Drizzle-style direct ctx.storage.sql.exec inside runSyncedWrite still broadcasts (and the handle IS ctx.storage.sql)", async () => { + // A Drizzle durable-object driver writes through ctx.storage.sql directly and + // ignores the handle runSyncedWrite passes. The write must still fire the CDC + // triggers and broadcast, because runSyncedWrite's handle IS ctx.storage.sql — + // the same connection, no wrapper (the fold of the retired handle-identity pin). + const room = "rsw-direct" + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)) + const t = realTransport(room) + await t.connect() + const messages = createCollection(doCollectionOptions({ transport: t, table: "messages", getKey: (m) => m.id })) + await messages.preload() + + await runInDurableObject(stub, (instance, state) => { + api(instance).runSyncedWrite((sql) => { + expect(sql).toBe(state.storage.sql) // folded: same handle, so a direct exec is captured too + state.storage.sql.exec("INSERT INTO messages(id,body) VALUES(?,?)", "drz1", "via-ctx-storage") + }) + }) + + await waitFor(() => messages.get("drz1") !== undefined) + expect(messages.get("drz1")).toMatchObject({ id: "drz1", body: "via-ctx-storage" }) + t.close() + }) + + it("a write OUTSIDE runSyncedWrite is trigger-captured but NOT broadcast until the next drain", async () => { + // ADR-0006's dark half: a raw sql.exec fires the CDC triggers (the change + // lands in _sync_changes) but is invisible to clients until some later drain + // flushes the backlog. The suite states this invariant; this asserts it — + // capture happens, broadcast does not, until drainAndBroadcast is called. + const room = "rsw-outside" + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)) + const t = realTransport(room) + await t.connect() + const messages = createCollection(doCollectionOptions({ transport: t, table: "messages", getKey: (m) => m.id })) + await messages.preload() + + const captured = await runInDurableObject(stub, (_instance, state) => { + state.storage.sql.exec("INSERT INTO messages(id,body) VALUES(?,?)", "raw1", "unsynced") + return Array.from(state.storage.sql.exec("SELECT key, op FROM _sync_changes WHERE key='raw1'")) as Array<{ + key: string + op: string + }> + }) + expect(captured).toEqual([{ key: "raw1", op: "insert" }]) // trigger fired: captured… + + // …but nothing broadcasts it: the client stays empty across several coalescer + // ticks (tickMs=50) — no drain path runs, so it is never enqueued. + await new Promise((r) => setTimeout(r, 300)) + expect(messages.get("raw1")).toBeUndefined() + + // A manual drainAndBroadcast (as any later mutation would) delivers the backlog. + await runInDurableObject(stub, (instance) => api(instance).drainAndBroadcast()) + await waitFor(() => messages.get("raw1") !== undefined) + expect(messages.get("raw1")).toMatchObject({ id: "raw1", body: "unsynced" }) + t.close() + }) + + it("a multi-statement runSyncedWrite reaches the client as one batch: all deltas, then a single uptodate at one seq", async () => { + // The whole write commits and broadcasts atomically: every delta precedes a + // single `uptodate` boundary, and all share one seq (one stream position). A + // key touched twice in the batch (m2: insert then update) coalesces to one + // delta carrying its latest value. Asserted at the wire, which the collection hides. + const room = "rsw-batch" + const ws = await openRawWs(room) + await subscribeRaw(ws, "s1") + const frames = recordFrames(ws) + + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)) + await runInDurableObject(stub, (instance) => { + api(instance).runSyncedWrite((sql) => { + sql.exec("INSERT INTO messages(id,body) VALUES(?,?)", "m1", "one") + sql.exec("INSERT INTO messages(id,body) VALUES(?,?)", "m2", "two") + sql.exec("UPDATE messages SET body=? WHERE id=?", "two!", "m2") + sql.exec("INSERT INTO messages(id,body) VALUES(?,?)", "m3", "three") + }) + }) + + await waitFor(() => frames.some((f) => f.t === "uptodate")) + await new Promise((r) => setTimeout(r, 100)) // settle: catch any (wrong) trailing frame + const kinds = frames.map((f) => f.t) + // Exactly one uptodate, and it is the LAST frame — every delta precedes it. + expect(kinds.filter((k) => k === "uptodate")).toHaveLength(1) + expect(kinds[kinds.length - 1]).toBe("uptodate") + + const deltas = frames.filter((f) => f.t === "d") as Array> + // Exactly one delta per key — m2's insert+update collapsed to one (not two). + expect(deltas).toHaveLength(3) + expect(new Set(deltas.map((d) => d.key))).toEqual(new Set(["m1", "m2", "m3"])) + const m2 = deltas.find((d) => d.key === "m2")! + expect((m2.cols as { body: string }).body).toBe("two!") + // All deltas carry the single batch seq == the uptodate seq (one position). + const up = frames.find((f) => f.t === "uptodate") as Extract + for (const d of deltas) expect(d.seq).toBe(up.seq) + ws.close() + }) + it("a write to an idle DO (no subscribers) reaches a later client via snapshot", async () => { const room = "rsw-idle" const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room))