1// Write-verdict harness: the three server-side properties a client-side write2// journal needs to record verdicts without a server ledger.3//4// The server only adjudicates writes made under a STALE policy: the client5// runtime enforces the locally-known policy synchronously at the call site,6// so a server rejection requires the store's permissions to tighten after the7// client last synced them. The fixture models exactly that — permission8// revoked at the store while the writing device is offline.9//10// Properties exercised:11// 1. Correlation — a rejected write surfaces `permission_denied` with the12// originating write's batch id, on both verdict surfaces (`wait()` →13// `PersistedWriteRejectedError`, and the `onMutationError` fallback event14// for writes nobody awaited).15// 2. Re-derivability — a write unacked at disconnect is re-sent on reconnect16// and the rejection fires then; nothing is lost to the disconnect.17// 3. Determinism — the verdict for a batch is stable across further18// reconnects, and an independent runtime issuing the same write against19// the same store state derives the same code. The fallback event itself20// is delivered once per runtime lifetime (not re-fired on later21// reconnects), so a journal must persist the verdict when it arrives;22// re-derivation on demand goes through `wait()`.23//24// Schemas are defined INSIDE the test fn: jazz registers schemas process-25// globally at definition time, and module-level definitions would clobber26// other test files' declared hashes.2728import { assert, assertEquals } from "@std/assert";29import { createDb, schema as s } from "jazz-tools";30import { deploy } from "jazz-tools/testing";31import { definePermissions } from "jazz-tools/permissions";32import { createSyncNode } from "../src/node.ts";33import { classifyMutationError } from "../src/verdict.ts";3435interface MutationErrorEvent {36 code: string;37 reason: string;38 batch: {39 batchId: string;40 mode: string;41 sealed: boolean;42 latestSettlement: { kind: string; batchId: string; code?: string; reason?: string } | null;43 };44}4546interface Rejection {47 outcome: "rejected";48 name?: string;49 batchId?: string;50 code?: string;51 reason?: string;52}5354type Verdict = Rejection | { outcome: "settled" };5556function accountSecret(fill: number): string {57 let binary = "";58 for (const byte of new Uint8Array(32).fill(fill)) binary += String.fromCharCode(byte);59 return btoa(binary).replaceAll("+", "-").replaceAll("/", "_").replace(/=+$/, "");60}6162async function within<T>(operation: Promise<T>, label: string, milliseconds = 20_000): Promise<T> {63 let timer: ReturnType<typeof setTimeout> | undefined;64 try {65 return await Promise.race([66 operation,67 new Promise<never>((_resolve, reject) => {68 timer = setTimeout(() => reject(new Error(`${label} timed out`)), milliseconds);69 }),70 ]);71 } finally {72 if (timer !== undefined) clearTimeout(timer);73 }74}7576/** Settle a write handle into a plain verdict object. */77function settle(handle: { wait(options: { tier: "global" }): Promise<unknown> }): Promise<Verdict> {78 return handle.wait({ tier: "global" }).then(79 (): Verdict => ({ outcome: "settled" }),80 (e: Error & Partial<Rejection>): Verdict => ({81 outcome: "rejected",82 name: e.name,83 batchId: e.batchId,84 code: e.code,85 reason: e.reason,86 }),87 );88}8990async function pollUntil(91 label: string,92 predicate: () => boolean,93 milliseconds = 15_000,94): Promise<void> {95 const deadline = Date.now() + milliseconds;96 while (Date.now() < deadline) {97 if (predicate()) return;98 await new Promise((r) => setTimeout(r, 250));99 }100 throw new Error(`${label} did not become true within ${milliseconds}ms`);101}102103Deno.test({104 name: "write verdict: rejection correlates by batch id, survives reconnect, and is deterministic",105 sanitizeOps: false,106 sanitizeResources: false,107 fn: async () => {108 const tables = { verdict__entries: s.table({ text: s.string() }) };109 const app = s.defineApp(tables);110 const permissive = definePermissions(app, ({ policy, session }) => [111 policy.verdict__entries.allowRead.where({ $createdBy: session.userId }),112 policy.verdict__entries.allowInsert.where({ $createdBy: session.userId }),113 ]);114 const readOnly = definePermissions(app, ({ policy, session }) => [115 policy.verdict__entries.allowRead.where({ $createdBy: session.userId }),116 ]);117118 const node = await createSyncNode({119 appId: crypto.randomUUID(),120 backendSecret: "lofi_backend_verdict",121 adminSecret: "lofi_admin_verdict",122 inMemory: true,123 mesh: "off",124 });125 const target = {126 appId: node.appId,127 serverUrl: node.url.replace(/^ws/, "http"),128 adminSecret: "lofi_admin_verdict",129 };130 try {131 await deploy({ ...target, schema: app, permissions: permissive });132133 // Two devices of one account, both booted while inserts are allowed, so134 // both hold the permissive policy locally.135 const secret = accountSecret(21);136 const connect = () =>137 createDb({138 appId: node.appId,139 serverUrl: target.serverUrl,140 secret,141 userBranch: "main",142 driver: { type: "memory" },143 });144 const deviceA = await connect();145 const deviceB = await connect();146 try {147 await within(148 deviceA.insert(app.verdict__entries, { text: "seed" }).wait({ tier: "global" }),149 "seed write under the permissive policy",150 );151152 // Both devices go offline; the store's policy tightens underneath.153 await deviceA.disconnect();154 await deviceB.disconnect();155 await deploy({ ...target, schema: app, permissions: readOnly });156157 // Offline writes under the stale (permissive) policy: the local158 // runtime accepts them and hands back batch ids.159 const awaited = deviceA.insert(app.verdict__entries, { text: "awaited stale write" });160 const unawaited = deviceA.insert(app.verdict__entries, { text: "un-awaited stale write" });161 const fromB = deviceB.insert(app.verdict__entries, { text: "device B stale write" });162 assert(awaited.batchId !== unawaited.batchId, "distinct writes, distinct batch ids");163164 const events: MutationErrorEvent[] = [];165 deviceA.onMutationError((event) => events.push(event as MutationErrorEvent));166167 // Property 2 — the unacked writes are re-sent on reconnect and the168 // rejection fires then.169 await deviceA.reconnect();170171 // Property 1a — the awaited surface: a typed rejection carrying the172 // originating batch id and a machine-readable code.173 const verdict = await within(settle(awaited), "awaited stale write settles");174 assertEquals(verdict.outcome, "rejected");175 const rejection = verdict as Rejection;176 assertEquals(rejection.name, "PersistedWriteRejectedError");177 assertEquals(rejection.batchId, awaited.batchId, "verdict correlates to the write");178 assertEquals(rejection.code, "permission_denied");179 assertEquals(classifyMutationError(rejection.code!), "permanent");180 assert(rejection.reason && rejection.reason.length > 0, "reason present (diagnostics)");181182 // Property 1b — the fallback event surface for writes nobody awaited:183 // the event carries the batch record, correlating by batch id.184 await pollUntil(185 "onMutationError fires for the un-awaited write",186 () => events.some((e) => e.batch.batchId === unawaited.batchId),187 );188 const event = events.find((e) => e.batch.batchId === unawaited.batchId)!;189 assertEquals(event.code, "permission_denied");190 assertEquals(event.batch.sealed, true);191 assertEquals(event.batch.latestSettlement?.kind, "rejected");192 assertEquals(event.batch.latestSettlement?.code, "permission_denied");193 const eventsAfterFirstCycle = events.length;194195 // Property 3a — the verdict is stable across further reconnects, and196 // re-waiting re-derives it without a new fallback event.197 await deviceA.disconnect();198 await deviceA.reconnect();199 const again = await within(settle(awaited), "re-derived verdict after reconnect");200 assertEquals(again.outcome, "rejected");201 assertEquals((again as Rejection).code, "permission_denied");202 assertEquals((again as Rejection).batchId, awaited.batchId);203 const unawaitedAgain = await within(settle(unawaited), "un-awaited verdict via wait()");204 assertEquals((unawaitedAgain as Rejection).code, "permission_denied");205 // Delivered once per runtime lifetime: the reconnect did not re-fire206 // events for already-settled batches. A journal must persist the207 // verdict when it first arrives.208 await new Promise((r) => setTimeout(r, 1500));209 assertEquals(events.length, eventsAfterFirstCycle, "no event re-fire on reconnect");210211 // Property 3b — an independent runtime, same account, same store212 // state, same write shape: the same code.213 await deviceB.reconnect();214 const verdictB = await within(settle(fromB), "device B stale write settles");215 assertEquals(verdictB.outcome, "rejected");216 assertEquals((verdictB as Rejection).code, "permission_denied");217 assertEquals((verdictB as Rejection).batchId, fromB.batchId);218219 // Once the tightened policy has synced to a device, writes it denies220 // never reach the server again: the local runtime refuses them221 // synchronously at the call site, before any batch exists. These222 // surface as thrown errors to the author, not as write verdicts.223 // (Until that sync lands — fresh boot, first reconnect — the same224 // write is accepted locally and adjudicated by the server, as above.)225 let localRefusal: Error | null = null;226 await pollUntil("local enforcement of the synced read-only policy", () => {227 try {228 deviceA.insert(app.verdict__entries, { text: "refused?" });229 return false;230 } catch (e) {231 localRefusal = e as Error;232 return true;233 }234 });235 assert(236 (localRefusal as unknown as Error).message.includes("policy denied INSERT"),237 `local refusal names the policy: ${(localRefusal as unknown as Error).message}`,238 );239 } finally {240 await Promise.allSettled([241 within(deviceA.logout(), "device A cleanup", 3000),242 within(deviceB.logout(), "device B cleanup", 3000),243 ]);244 }245 } finally {246 await node.stop();247 }248 },249});