1// Concurrent-writer merge semantics for the facade's merge strategies: lww2// (the default), counter, and g-set columns written by two synced clients.3// Single-session4// behavior is covered by the conformance suite; this file is the multi-client5// half, built on the JazzServer setup from package/access/access_sync_test.ts.6//7// Runs in its own process before the conformance suite (see the8// test:conformance task) because the conformance scalar test deliberately9// wedges the FFI driver last.10import { createDb } from "jazz-tools";11import { deploy, startLocalJazzServer } from "jazz-tools/testing";12import { assert } from "../runtime/test-assert.ts";13import { type ArrayColumn, type IntColumn, s, type StringColumn } from "./mod.ts";1415const counterApp = s.defineApp({16 tallies: s.table({17 name: s.string(),18 total: s.int().default(0).merge("counter") as unknown as IntColumn<false, true>,19 }),20});21const counterPermissions = s.definePermissions(counterApp, ({ policy }) => {22 policy.tallies.allowInsert.always();23 policy.tallies.allowRead.always();24 policy.tallies.allowUpdate.always();25 policy.tallies.allowDelete.always();26});2728// Single-table app: a g-set column destabilizes sibling-table writes in the29// pinned alpha (see the conformance canary), so the g-set coverage stays30// isolated here too.31const gsetApp = s.defineApp({32 tagged: s.table({33 name: s.string(),34 tags: s.array(s.string()).merge("g-set") as unknown as ArrayColumn<"TEXT">,35 }),36});37const gsetPermissions = s.definePermissions(gsetApp, ({ policy }) => {38 policy.tagged.allowInsert.always();39 policy.tagged.allowRead.always();40 policy.tagged.allowUpdate.always();41 policy.tagged.allowDelete.always();42});4344// One column with the implicit default strategy and one with explicit45// .merge("lww"): the same conflict runs through both so the test also pins46// that the default is last-write-wins.47const lwwApp = s.defineApp({48 docs: s.table({49 title: s.string(),50 body: s.string().merge("lww") as unknown as StringColumn<false, false>,51 }),52});53const lwwPermissions = s.definePermissions(lwwApp, ({ policy }) => {54 policy.docs.allowInsert.always();55 policy.docs.allowRead.always();56 policy.docs.allowUpdate.always();57 policy.docs.allowDelete.always();58});5960function secret(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 = 10_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}7980type DeployOptions = Parameters<typeof deploy>[0];81type Harness = {82 server: Awaited<ReturnType<typeof startLocalJazzServer>>;83 client(fill: number): Promise<Awaited<ReturnType<typeof createDb>>>;84 stop(): Promise<void>;85};8687async function bootHarness(88 app: DeployOptions["schema"],89 permissions: DeployOptions["permissions"],90): Promise<Harness> {91 const server = await startLocalJazzServer({ allowLocalFirstAuth: true });92 await deploy({93 appId: server.appId,94 serverUrl: server.url,95 adminSecret: server.adminSecret,96 schema: app,97 permissions,98 });99 const clients: Awaited<ReturnType<typeof createDb>>[] = [];100 return {101 server,102 async client(fill: number) {103 const db = await createDb({104 appId: server.appId,105 serverUrl: server.url,106 secret: secret(fill),107 userBranch: "main",108 driver: { type: "memory" },109 });110 clients.push(db);111 return db;112 },113 async stop() {114 await Promise.allSettled(115 clients.map((db, index) => within(db.logout(), `client ${index} cleanup`, 3_000)),116 );117 await within(server.stop(), "server cleanup", 3_000).catch(() => undefined);118 },119 };120}121122// PINNED (alpha.53): counter columns have no single semantics across replica123// lifetimes. The server merges an update as a replacement when it is causally124// latest and as a sum when updates are concurrent, while live replicas apply125// every update — including the echo of their own reconnected write — as a126// delta. A replica that watched the history read therefore diverges127// permanently from what a fresh boot of the same account reads. Each assert128// below pins one cell of that matrix; when one fails, upstream changed129// counter merge behavior — re-derive the matrix and update the column-palette130// section of docs/data-and-ui.md.131Deno.test("counter merge: concurrent offline updates sum on the server, diverge on a live writer", async () => {132 const harness = await bootHarness(counterApp, counterPermissions);133 try {134 const alice = await harness.client(1);135 const bob = await harness.client(2);136 const row = await within(137 alice.insert(counterApp.tallies, { name: "hits", total: 0 }).wait({ tier: "global" }),138 "seed insert",139 );140 const bobView = await within(141 bob.all(counterApp.tallies.where({ id: row.id }), { tier: "global" }),142 "bob initial read",143 );144 assert(bobView.length === 1 && bobView[0].total === 0, "bob did not sync the seeded row");145146 await alice.disconnect();147 await bob.disconnect();148 const aliceWrite = alice.update(counterApp.tallies, row.id, { total: 5 });149 const bobWrite = bob.update(counterApp.tallies, row.id, { total: 3 });150 await alice.reconnect();151 await bob.reconnect();152 await within(aliceWrite.wait({ tier: "global" }), "alice counter durability");153 await within(bobWrite.wait({ tier: "global" }), "bob counter durability");154155 const read = async (db: typeof alice, label: string) => {156 const rows = await within(157 db.all(counterApp.tallies.where({ id: row.id }), { tier: "global" }),158 `${label} read`,159 );160 return rows[0]?.total;161 };162 const aliceTotal = await read(alice, "alice");163 const bobTotal = await read(bob, "bob");164 const charlie = await harness.client(3);165 const canonical = await read(charlie, "fresh client");166167 assert(168 canonical === 8,169 `counter: concurrent updates no longer sum on the server (fresh client read ${canonical}, ` +170 `not 5 + 3); the merge model changed — remap it`,171 );172 assert(173 bobTotal === 8,174 `counter: second writer read ${bobTotal}, not the canonical sum 8`,175 );176 assert(177 aliceTotal === 13,178 `counter: pinned first-writer divergence changed — read ${aliceTotal}; if 8, upstream ` +179 `stopped double-counting the reconnected writer's own echo: remove the pin`,180 );181 } finally {182 await harness.stop();183 }184});185186Deno.test("counter merge: causally ordered updates replace on the server, accumulate on live replicas", async () => {187 const harness = await bootHarness(counterApp, counterPermissions);188 try {189 const alice = await harness.client(1);190 const bob = await harness.client(2);191 const row = await within(192 alice.insert(counterApp.tallies, { name: "causal", total: 0 }).wait({ tier: "global" }),193 "seed insert",194 );195 await within(196 alice.update(counterApp.tallies, row.id, { total: 2 }).wait({ tier: "global" }),197 "alice update durability",198 );199 const bobBefore = await within(200 bob.all(counterApp.tallies.where({ id: row.id }), { tier: "global" }),201 "bob pre-write read",202 );203 assert(bobBefore[0]?.total === 2, "bob did not sync alice's update before writing");204 await within(205 bob.update(counterApp.tallies, row.id, { total: 3 }).wait({ tier: "global" }),206 "bob update durability",207 );208209 const bobTotal =210 (await within(bob.all(counterApp.tallies.where({ id: row.id })), "bob read"))[0]?.total;211 const aliceTotal = (await within(212 alice.all(counterApp.tallies.where({ id: row.id }), { tier: "global" }),213 "alice read",214 ))[0]?.total;215 const charlie = await harness.client(3);216 const canonical = (await within(217 charlie.all(counterApp.tallies.where({ id: row.id }), { tier: "global" }),218 "fresh client read",219 ))[0]?.total;220221 assert(222 canonical === 3,223 `counter: pinned causal replacement changed — fresh client read ${canonical}; if 5, ` +224 `upstream now sums causally ordered updates like live replicas do: remove the pin`,225 );226 assert(227 aliceTotal === 5 && bobTotal === 5,228 `counter: live replicas no longer accumulate deltas (alice=${aliceTotal}, bob=${bobTotal}, ` +229 `expected both 5 = 2 + 3)`,230 );231 } finally {232 await harness.stop();233 }234});235236Deno.test("g-set columns union concurrent offline updates and converge everywhere", async () => {237 const harness = await bootHarness(gsetApp, gsetPermissions);238 try {239 const alice = await harness.client(1);240 const bob = await harness.client(2);241 const row = await within(242 alice.insert(gsetApp.tagged, { name: "doc", tags: ["a"] }).wait({ tier: "global" }),243 "seed insert",244 );245 const bobView = await within(246 bob.all(gsetApp.tagged.where({ id: row.id }), { tier: "global" }),247 "bob initial read",248 );249 assert(bobView.length === 1, "bob did not sync the seeded row");250251 await alice.disconnect();252 await bob.disconnect();253 const aliceWrite = alice.update(gsetApp.tagged, row.id, { tags: ["a", "b"] });254 const bobWrite = bob.update(gsetApp.tagged, row.id, { tags: ["a", "c"] });255 await alice.reconnect();256 await bob.reconnect();257 await within(aliceWrite.wait({ tier: "global" }), "alice g-set durability");258 await within(bobWrite.wait({ tier: "global" }), "bob g-set durability");259260 const read = async (db: typeof alice, label: string) => {261 const rows = await within(262 db.all(gsetApp.tagged.where({ id: row.id }), { tier: "global" }),263 `${label} read`,264 );265 return [...(rows[0]?.tags ?? [])].sort();266 };267 const aliceTags = await read(alice, "alice");268 const bobTags = await read(bob, "bob");269 const charlie = await harness.client(3);270 const canonical = await read(charlie, "fresh client");271272 const union = JSON.stringify(["a", "b", "c"]);273 for (274 const [label, tags] of [["alice", aliceTags], ["bob", bobTags], ["fresh client", canonical]]275 .map(([l, t]) => [l, JSON.stringify(t)] as const)276 ) {277 assert(278 tags === union,279 `g-set: ${label} read ${tags}, not the union of both writers' elements`,280 );281 }282 } finally {283 await harness.stop();284 }285});286287Deno.test("lww columns resolve concurrent offline updates to one winner everywhere", async () => {288 const harness = await bootHarness(lwwApp, lwwPermissions);289 try {290 const alice = await harness.client(1);291 const bob = await harness.client(2);292 const row = await within(293 alice.insert(lwwApp.docs, { title: "seed-title", body: "seed-body" }).wait({294 tier: "global",295 }),296 "seed insert",297 );298 const bobView = await within(299 bob.all(lwwApp.docs.where({ id: row.id }), { tier: "global" }),300 "bob initial read",301 );302 assert(bobView.length === 1, "bob did not sync the seeded row");303304 await alice.disconnect();305 await bob.disconnect();306 const aliceWrite = alice.update(lwwApp.docs, row.id, {307 title: "alice-title",308 body: "alice-body",309 });310 const bobWrite = bob.update(lwwApp.docs, row.id, { title: "bob-title", body: "bob-body" });311 await alice.reconnect();312 await bob.reconnect();313 await within(aliceWrite.wait({ tier: "global" }), "alice lww durability");314 await within(bobWrite.wait({ tier: "global" }), "bob lww durability");315316 const read = async (db: typeof alice, label: string) => {317 const rows = await within(318 db.all(lwwApp.docs.where({ id: row.id }), { tier: "global" }),319 `${label} read`,320 );321 return { title: rows[0]?.title, body: rows[0]?.body };322 };323 const aliceDoc = await read(alice, "alice");324 const bobDoc = await read(bob, "bob");325 const charlie = await harness.client(3);326 const canonical = await read(charlie, "fresh client");327328 // Bob reconnects after Alice, so his write is the last to arrive and wins329 // on every replica — for the implicit-default column and the explicit330 // "lww" column alike. Unlike counter, live replicas and fresh boots agree.331 for (332 const [label, doc] of [333 ["alice", aliceDoc],334 ["bob", bobDoc],335 ["fresh client", canonical],336 ] as const337 ) {338 assert(339 doc.title === "bob-title" && doc.body === "bob-body",340 `lww: ${label} read ${JSON.stringify(doc)}, not the later-arriving writer's values`,341 );342 }343 } finally {344 await harness.stop();345 }346});