1// Behavioral coverage for the two batch-fate codes beyond permission_denied2// that the pinned alpha's server can emit (see src/verdict.ts for the full3// code-space audit):4//5// - `transaction_conflict`: two devices stage a transactional update to the6// same row from the same visible frontier; the authority accepts the first7// seal and rejects the second ("row visible parent changed since transaction8// write was staged"). The fate is final — the engine does not rebase or9// retry the losing batch — and deterministic across reconnects.10// - `permissions_head_missing`: a store whose schema is deployed without a11// permissions head fails closed and rejects every write, on both verdict12// surfaces, with the batch correlated. Publishing permissions afterwards13// does not revisit the persisted fate; only new batches see the new head.14//15// `invalid_batch_submission` has no test here because it is unreachable from16// the public API: every trigger is a sealed-envelope invariant the client17// runtime itself computes, and the one author-reachable shape — committing an18// empty transaction — throws synchronously before any batch exists (pinned19// below).20//21// Scope note: the conflict is staged ONLINE. A transactional batch committed22// while offline does not re-submit its seal on reconnect at the pinned alpha,23// so the authority never adjudicates it — the write stays unapplied and24// `wait()` fails with a runtime error rather than a verdict. That degradation25// is an upstream client defect at the pin, not a settlement contract, so it26// is documented here rather than pinned as a test.27//28// Schemas are defined INSIDE the test fns: jazz registers schemas process-29// globally at definition time, and module-level definitions would clobber30// other test files' declared hashes.3132import { assert, assertEquals, assertThrows } from "@std/assert";33import { createDb, schema as s } from "jazz-tools";34import { deploy } from "jazz-tools/testing";35import { definePermissions } from "jazz-tools/permissions";36import { createSyncNode } from "../src/node.ts";37import { classifyMutationError } from "../src/verdict.ts";3839interface MutationErrorEvent {40 code: string;41 reason: string;42 batch: {43 batchId: string;44 mode: string;45 sealed: boolean;46 latestSettlement: { kind: string; batchId: string; code?: string; reason?: string } | null;47 };48}4950interface Rejection {51 outcome: "rejected";52 name?: string;53 batchId?: string;54 code?: string;55 reason?: string;56}5758type Verdict = Rejection | { outcome: "settled" };5960function accountSecret(fill: number): string {61 let binary = "";62 for (const byte of new Uint8Array(32).fill(fill)) binary += String.fromCharCode(byte);63 return btoa(binary).replaceAll("+", "-").replaceAll("/", "_").replace(/=+$/, "");64}6566async function within<T>(operation: Promise<T>, label: string, milliseconds = 20_000): Promise<T> {67 let timer: ReturnType<typeof setTimeout> | undefined;68 try {69 return await Promise.race([70 operation,71 new Promise<never>((_resolve, reject) => {72 timer = setTimeout(() => reject(new Error(`${label} timed out`)), milliseconds);73 }),74 ]);75 } finally {76 if (timer !== undefined) clearTimeout(timer);77 }78}7980/** Settle a write handle into a plain verdict object. */81function settle(handle: { wait(options: { tier: "global" }): Promise<unknown> }): Promise<Verdict> {82 return handle.wait({ tier: "global" }).then(83 (): Verdict => ({ outcome: "settled" }),84 (e: Error & Partial<Rejection>): Verdict => ({85 outcome: "rejected",86 name: e.name,87 batchId: e.batchId,88 code: e.code,89 reason: e.reason,90 }),91 );92}9394async function pollUntil(95 label: string,96 predicate: () => Promise<boolean>,97 milliseconds = 20_000,98): Promise<void> {99 const deadline = Date.now() + milliseconds;100 while (Date.now() < deadline) {101 if (await predicate()) return;102 await new Promise((r) => setTimeout(r, 300));103 }104 throw new Error(`${label} did not become true within ${milliseconds}ms`);105}106107Deno.test({108 name:109 "write verdict: a losing concurrent transaction settles transaction_conflict, final and deterministic",110 sanitizeOps: false,111 sanitizeResources: false,112 fn: async () => {113 const tables = { conflict__rows: s.table({ text: s.string() }) };114 const app = s.defineApp(tables);115 const permissions = definePermissions(app, ({ policy, session }) => [116 policy.conflict__rows.allowRead.where({ $createdBy: session.userId }),117 policy.conflict__rows.allowInsert.where({ $createdBy: session.userId }),118 policy.conflict__rows.allowUpdate119 .whereOld({ $createdBy: session.userId })120 .whereNew({ $createdBy: session.userId }),121 ]);122123 const node = await createSyncNode({124 appId: crypto.randomUUID(),125 backendSecret: "lofi_backend_conflict",126 adminSecret: "lofi_admin_conflict",127 inMemory: true,128 mesh: "off",129 });130 const target = {131 appId: node.appId,132 serverUrl: node.url.replace(/^ws/, "http"),133 adminSecret: "lofi_admin_conflict",134 };135 try {136 await deploy({ ...target, schema: app, permissions });137 const secret = accountSecret(34);138 const connect = () =>139 createDb({140 appId: node.appId,141 serverUrl: target.serverUrl,142 secret,143 userBranch: "main",144 driver: { type: "memory" },145 });146 const deviceA = await connect();147 const deviceB = await connect();148 try {149 const seed = await within(150 deviceA.insert(app.conflict__rows, { text: "seed" }).wait({ tier: "global" }),151 "seed write settles",152 );153 await pollUntil(154 "device B observes the seed",155 async () =>156 (await deviceB.all(app.conflict__rows.where({ id: seed.id }), { tier: "global" }))157 .length === 1,158 );159160 // Both devices stage a transactional update to the same row from the161 // same visible frontier, while connected.162 const txA = deviceA.beginTransaction();163 txA.update(app.conflict__rows, seed.id, { text: "A wins" });164 const txB = deviceB.beginTransaction();165 txB.update(app.conflict__rows, seed.id, { text: "B loses" });166167 // A's seal reaches the authority first and advances the frontier.168 const handleA = txA.commit();169 assertEquals((await within(settle(handleA), "A's transaction settles")).outcome, "settled");170171 // B's seal validates against a frontier that has moved.172 const handleB = txB.commit();173 assert(handleA.batchId !== handleB.batchId, "distinct batches");174 const verdict = await within(settle(handleB), "B's transaction settles");175 assertEquals(verdict.outcome, "rejected");176 const rejection = verdict as Rejection;177 assertEquals(rejection.name, "PersistedWriteRejectedError");178 assertEquals(rejection.code, "transaction_conflict");179 assertEquals(rejection.batchId, handleB.batchId, "verdict correlates to the losing batch");180 assert(rejection.reason && rejection.reason.length > 0, "reason present (diagnostics)");181 assertEquals(classifyMutationError(rejection.code!), "permanent");182183 // The fate is final and deterministic: re-waiting across a reconnect184 // re-derives the same code, and the engine never rebases or retries185 // the losing write — the row converges to the winner everywhere.186 await deviceB.disconnect();187 await deviceB.reconnect();188 const again = await within(settle(handleB), "re-derived verdict after reconnect");189 assertEquals((again as Rejection).code, "transaction_conflict");190 assertEquals((again as Rejection).batchId, handleB.batchId);191 await pollUntil("device B converges on the winning write", async () => {192 const rows = await deviceB.all(app.conflict__rows.where({ id: seed.id }), {193 tier: "global",194 });195 return rows.length === 1 && rows[0].text === "A wins";196 });197 const onA = await deviceA.all(app.conflict__rows.where({ id: seed.id }), {198 tier: "global",199 });200 assertEquals(onA[0].text, "A wins");201202 // The one author-reachable malformed-submission shape never leaves203 // the client: committing an empty transaction throws synchronously,204 // so `invalid_batch_submission` cannot be provoked from this surface.205 assertThrows(() => deviceA.beginTransaction().commit());206 } finally {207 await Promise.allSettled([208 within(deviceA.logout(), "device A cleanup", 3000),209 within(deviceB.logout(), "device B cleanup", 3000),210 ]);211 }212 } finally {213 await node.stop();214 }215 },216});217218Deno.test({219 name:220 "write verdict: a store without a permissions head rejects writes with permissions_head_missing",221 sanitizeOps: false,222 sanitizeResources: false,223 fn: async () => {224 const tables = { headless__rows: s.table({ text: s.string() }) };225 const app = s.defineApp(tables);226 const permissions = definePermissions(app, ({ policy, session }) => [227 policy.headless__rows.allowRead.where({ $createdBy: session.userId }),228 policy.headless__rows.allowInsert.where({ $createdBy: session.userId }),229 ]);230231 const node = await createSyncNode({232 appId: crypto.randomUUID(),233 backendSecret: "lofi_backend_headless",234 adminSecret: "lofi_admin_headless",235 inMemory: true,236 mesh: "off",237 });238 const target = {239 appId: node.appId,240 serverUrl: node.url.replace(/^ws/, "http"),241 adminSecret: "lofi_admin_headless",242 };243 try {244 // Schema only — no permissions head. The store fails closed.245 await deploy({ ...target, schema: app });246 const device = await createDb({247 appId: node.appId,248 serverUrl: target.serverUrl,249 secret: accountSecret(35),250 userBranch: "main",251 driver: { type: "memory" },252 });253 try {254 const events: MutationErrorEvent[] = [];255 device.onMutationError((event) => events.push(event as MutationErrorEvent));256257 const awaited = device.insert(app.headless__rows, { text: "awaited" });258 const unawaited = device.insert(app.headless__rows, { text: "un-awaited" });259260 // Awaited surface: typed rejection, batch-correlated.261 const verdict = await within(settle(awaited), "awaited write settles");262 assertEquals(verdict.outcome, "rejected");263 const rejection = verdict as Rejection;264 assertEquals(rejection.name, "PersistedWriteRejectedError");265 assertEquals(rejection.code, "permissions_head_missing");266 assertEquals(rejection.batchId, awaited.batchId);267 assertEquals(classifyMutationError(rejection.code!), "permanent");268269 // Fallback surface for the un-awaited write: same code, full batch270 // record with the persisted settlement.271 await pollUntil(272 "onMutationError fires for the un-awaited write",273 () => Promise.resolve(events.some((e) => e.batch.batchId === unawaited.batchId)),274 );275 const event = events.find((e) => e.batch.batchId === unawaited.batchId)!;276 assertEquals(event.code, "permissions_head_missing");277 assertEquals(event.batch.latestSettlement?.kind, "rejected");278 assertEquals(event.batch.latestSettlement?.code, "permissions_head_missing");279280 // Publishing permissions afterwards does not revisit the persisted281 // fate; a new batch under the new head settles.282 await deploy({ ...target, schema: app, permissions });283 const after = await within(settle(awaited), "re-derived verdict after the head appears");284 assertEquals((after as Rejection).code, "permissions_head_missing");285 await pollUntil("a fresh write settles under the new head", async () => {286 const verdict = await within(287 settle(device.insert(app.headless__rows, { text: "fresh" })),288 "fresh write settles",289 );290 return verdict.outcome === "settled";291 });292 } finally {293 await within(device.logout(), "device cleanup", 3000).catch(() => {});294 }295 } finally {296 await node.stop();297 }298 },299});