diff --git a/packages/drivers/driver-turso/src/turso-remote-autonumber-concurrency.test.ts b/packages/drivers/driver-turso/src/turso-remote-autonumber-concurrency.test.ts index 953bbb17f39..5ce8fcaa692 100644 --- a/packages/drivers/driver-turso/src/turso-remote-autonumber-concurrency.test.ts +++ b/packages/drivers/driver-turso/src/turso-remote-autonumber-concurrency.test.ts @@ -15,14 +15,15 @@ * read-then-write one. Neither can two `@libsql/client` instances in ONE * Node process, whose native calls run on the one JS thread. * - * So this file spawns two child processes (through `tsx`, over the package's - * SOURCE, the way `packages/spec`'s process-boundary pins do), each holding - * its own `@libsql/client` connection — the native `file:` backend, which is - * libSQL itself, the engine behind a Turso endpoint — to one database file, - * each driving its own `TursoDriver` in REMOTE mode, writing through - * `TursoDriver.create`. A file barrier releases both writers at once, so their - * statements contend on the database's write lock rather than following each - * other through process start-up. + * So this file spawns two child processes, each running the package's SOURCE + * compiled by `tsx`, as `packages/spec`'s process-boundary pins do. For each + * round the parent creates a fresh database, and each writer opens its own + * `@libsql/client` connection to it — the native `file:` backend, which is + * libSQL itself, the engine behind a Turso endpoint — drives its own + * `TursoDriver` in REMOTE mode, and writes through `TursoDriver.create`. The + * parent releases both writers with one `go` written to both pipes in the same + * tick, once both have reported ready, so their statements contend on the + * database's write lock rather than following each other through start-up. * * What is NOT measured here, said plainly: an HTTP `sqld` server. None runs in * this environment, and the transport's HTTP batching and retries are the @@ -30,6 +31,36 @@ * what decides atomicity is the statement, and the statement runs on the same * engine under the same lock here as there. * + * # What the file waits on, and what it no longer waits on + * + * Measured at `d4aaeaf0e` on a 4-vCPU box, one round of the earlier shape + * (writers spawned per round under `node --import tsx`, 200 writes each, the + * default `synchronous = FULL`) took about 4.4 s: about 2.2 s to start each + * writer and about 2.2 s of writes. Of a writer's start-up, the package's own + * source took about 40 ms to compile; the rest was its dependencies' built code + * (`@objectstack/driver-sql` and what it imports, `@objectstack/spec`, + * `@libsql/client`), which loads in about 1.2 s under plain Node and about 2 s + * under tsx's hook. The writes were fsync-bound: one `create` is three + * statements (reserve, insert, read back), and a single writer took 5.9 ms per + * `create` with fsync and 1.2 ms with `synchronous = OFF`. CI's hourly full + * battery measured the file at up to 29.3 s, over 10 s in 6 of the 7 runs that + * executed it, and every retry round paid both start-ups again. + * + * Three changes, none touching what is asserted: + * + * - The writers start ONCE per test and serve every round over stdin/stdout. + * A retry round therefore costs its writes, not two more process start-ups. + * - A writer loads the driver's dependencies with plain Node first, then + * registers tsx and requires the source. The source is still compiled by + * tsx, and the dependencies are the same modules, loaded without the hook. + * - Each writer sets `PRAGMA synchronous = OFF` on its own connection. + * `synchronous` decides only whether SQLite flushes to disk at a commit, + * which matters for a power cut and not for a throwaway temp file. The lock + * protocol is unchanged: the journal mode, the locks each statement takes and + * the busy timeout all stay as they were. What decides atomicity, two + * connections in two processes serialised by the database's write lock, is + * exactly what it was. + * * # The pins * * - N writes from two processes give N distinct numbers — the counter moved @@ -76,11 +107,19 @@ * what this pin distinguishes is a counter that can hand out one number * twice, and on a declared-unique column that surfaces as refused writes * before it surfaces as duplicate rows. + * + * Re-measured on the current shape (writers started once, no fsync) at + * `794b74409`, with the same mutation: red in 6 of 8 runs with no gap and 5 of + * 5 with a 1 ms gap, every time on the same "no write failed" pin. The earlier + * shape, on the same box and day, went red in 1 of 5 runs with no gap. Without + * the fsync the two writers interleave far more finely (about 100 to 200 + * writer changes per round, against 2 to 40 before), so a round gives a + * read-then-write counter more chances to hand out one number twice, not fewer. */ import { describe, it, expect, afterAll } from 'vitest'; import { spawn } from 'node:child_process'; -import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { fileURLToPath } from 'node:url'; @@ -90,12 +129,22 @@ import { TursoDriver } from './index.js'; const HERE = fileURLToPath(new URL('.', import.meta.url)); const PKG_ROOT = join(HERE, '..'); -const DRIVER_ENTRY = new URL('./index.ts', import.meta.url).href; +const DRIVER_ENTRY = fileURLToPath(new URL('./index.ts', import.meta.url)); +/** + * What the driver's source imports at runtime from `node_modules` (its type-only + * imports are erased). The writer loads these BEFORE tsx is registered: they + * are built code, not this package's source, and tsx's require hook roughly + * doubles their load. A dependency missing from this list still loads, under + * tsx, so the list is a cost, never a correctness, question. + */ +const PRELOADED_DEPENDENCIES = ['@libsql/client', '@objectstack/driver-sql', '@objectstack/spec/api', '@objectstack/spec/data']; const WRITES_PER_WORKER = 200; /** Rounds a loaded box gets to let the two writers overlap before the overlap evidence is declared NOT MEASURED. */ const MAX_ROUNDS = 3; const WORKERS = ['A', 'B'] as const; +/** How long a writer may take to answer one command before the round fails with that writer's stderr. */ +const REPLY_DEADLINE_MS = 60_000; const NUMBERED_OBJECT = { name: 'crm_case', @@ -119,75 +168,138 @@ interface WorkerReport { } /** - * The writer. Runs under `tsx` so it imports the driver's SOURCE, like the - * test itself does. It connects, registers the schema (the tables already - * exist — the parent created them — so `initObjects` records metadata and - * runs no DDL), announces readiness, waits for the parent's `go` file, then - * writes as fast as the database lets it and reports on stdout. + * The writer. Loads the driver's dependencies with plain Node, then registers + * tsx through its public CJS API and requires the driver's SOURCE, like the + * test itself does — so the source is compiled by tsx, to CJS (this package + * has no `"type": "module"`), and its requires of those dependencies are cache + * hits. It pays that start-up once, for all the rounds. It reads one JSON + * command per line on stdin and answers one JSON line on stdout: + * + * - `{ open: }` — connect to that round's database, register the + * schema (the tables already exist — the parent created them — so + * `initObjects` records metadata and runs no DDL), answer `{ ready }`. + * - `{ go: }` — write n records as fast as the database lets it, + * disconnect, answer the report. + * + * stdin closing ends the loop, and the process exits. */ const WORKER_SOURCE = ` - import { createClient } from '@libsql/client'; - import { existsSync, writeFileSync } from 'node:fs'; - // tsx compiles this package's .ts to CJS (no "type": "module"), so the - // class may arrive under \`default\` — the spelling packages/spec's - // process-boundary pins use. - const driverModule = await import(${JSON.stringify(DRIVER_ENTRY)}); - const TursoDriver = driverModule.TursoDriver ?? driverModule.default?.TursoDriver; - const { OS_TEST_DB, OS_TEST_DIR, OS_TEST_TAG, OS_TEST_WRITES } = process.env; - const client = createClient({ url: 'file:' + OS_TEST_DB, timeout: 20000 }); - const driver = new TursoDriver({ url: 'libsql://concurrency.probe', client }); - await driver.connect(); - await driver.initObjects([${JSON.stringify(NUMBERED_OBJECT)}]); - writeFileSync(OS_TEST_DIR + '/ready-' + OS_TEST_TAG, String(process.pid)); - const go = OS_TEST_DIR + '/go'; - while (!existsSync(go)) await new Promise((r) => setTimeout(r, 2)); - const numbers = []; - const errors = []; - for (let i = 0; i < Number(OS_TEST_WRITES); i++) { - try { - const row = await driver.create('crm_case', { organization_id: 'orgA', title: OS_TEST_TAG + '-' + i }); - numbers.push(row.case_number); - } catch (e) { - errors.push(String(e && e.message || e)); + import { createRequire } from 'node:module'; + import { createInterface } from 'node:readline'; + const require = createRequire(${JSON.stringify(join(PKG_ROOT, 'package.json'))}); + for (const dependency of ${JSON.stringify(PRELOADED_DEPENDENCIES)}) require(dependency); + require('tsx/cjs/api').register(); + const { TursoDriver } = require(${JSON.stringify(DRIVER_ENTRY)}); + const { createClient } = require('@libsql/client'); + const { OS_TEST_TAG } = process.env; + const reply = (message) => process.stdout.write(JSON.stringify(message) + '\\n'); + let driver = null; + for await (const line of createInterface({ input: process.stdin })) { + const command = JSON.parse(line); + if (command.open) { + const client = createClient({ url: 'file:' + command.open, timeout: 20000 }); + // No disk flush at commit: see "What the file waits on" in the parent. + await client.execute('PRAGMA synchronous = OFF'); + driver = new TursoDriver({ url: 'libsql://concurrency.probe', client }); + await driver.connect(); + await driver.initObjects([${JSON.stringify(NUMBERED_OBJECT)}]); + reply({ ready: OS_TEST_TAG }); + } else if (command.go) { + const numbers = []; + const errors = []; + for (let i = 0; i < command.go; i++) { + try { + const row = await driver.create('crm_case', { organization_id: 'orgA', title: OS_TEST_TAG + '-' + i }); + numbers.push(row.case_number); + } catch (e) { + errors.push(String(e && e.message || e)); + } + } + await driver.disconnect(); + driver = null; + reply({ tag: OS_TEST_TAG, pid: process.pid, numbers, errors }); } } - await driver.disconnect(); - process.stdout.write(JSON.stringify({ tag: OS_TEST_TAG, pid: process.pid, numbers, errors })); `; -function runWorker(env: Record): Promise { - return new Promise((resolve, reject) => { - const child = spawn( - process.execPath, - ['--import', 'tsx', '--input-type=module', '-e', WORKER_SOURCE], - { cwd: PKG_ROOT, env: { ...process.env, ...env }, stdio: ['ignore', 'pipe', 'pipe'] }, - ); - let out = ''; - let err = ''; - child.stdout.on('data', (d) => { out += String(d); }); - child.stderr.on('data', (d) => { err += String(d); }); - child.on('error', reject); - child.on('close', (code) => { - if (code !== 0) { - reject(new Error(`writer ${env.OS_TEST_TAG} exited ${code}\nstderr:\n${err}\nstdout:\n${out}`)); - return; - } +interface Writer { + tag: string; + send(command: Record): void; + /** The writer's next answer; rejects with its stderr if it dies or stays silent past the deadline. */ + next(what: string): Promise; + /** Close stdin and wait for the writer to exit; one still running 10 s later is killed. */ + stop(): Promise; +} + +function startWriter(tag: string): Writer { + const child = spawn( + process.execPath, + ['--input-type=module', '-e', WORKER_SOURCE], + { cwd: PKG_ROOT, env: { ...process.env, OS_TEST_TAG: tag }, stdio: ['pipe', 'pipe', 'pipe'] }, + ); + const inbox: unknown[] = []; + const waiting: Array<{ resolve: (m: unknown) => void; reject: (e: Error) => void }> = []; + let failure: Error | null = null; + let stderr = ''; + let pending = ''; + const fail = (e: Error) => { + failure ??= e; + while (waiting.length > 0) waiting.shift()!.reject(failure); + }; + child.stderr.on('data', (d) => { stderr += String(d); }); + child.stdout.on('data', (d) => { + pending += String(d); + for (let nl = pending.indexOf('\n'); nl >= 0; nl = pending.indexOf('\n')) { + const line = pending.slice(0, nl); + pending = pending.slice(nl + 1); + let message: unknown; try { - resolve(JSON.parse(out) as WorkerReport); + message = JSON.parse(line); } catch (e) { - reject(new Error(`writer ${env.OS_TEST_TAG} wrote no report: ${String(e)}\nstderr:\n${err}\nstdout:\n${out}`)); + fail(new Error(`writer ${tag} wrote a line that is not a report: ${String(e)}\nline: ${line}\nstderr:\n${stderr}`)); + return; } - }); + const waiter = waiting.shift(); + if (waiter) waiter.resolve(message); + else inbox.push(message); + } }); + const exited = new Promise((resolve) => child.on('close', resolve)); + child.on('error', fail); + // A command written to a writer that already died fails here, with its stderr. + child.stdin.on('error', (e) => fail(new Error(`writer ${tag} could not take a command: ${String(e)}\nstderr:\n${stderr}`))); + child.on('close', (code) => fail(new Error(`writer ${tag} exited ${code}\nstderr:\n${stderr}`))); + return { + tag, + send: (command) => { child.stdin.write(JSON.stringify(command) + '\n'); }, + next: (what: string) => { + if (inbox.length > 0) return Promise.resolve(inbox.shift() as T); + if (failure) return Promise.reject(failure); + return new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error(`writer ${tag} sent no ${what} within ${REPLY_DEADLINE_MS / 1000}s\nstderr:\n${stderr}`)), + REPLY_DEADLINE_MS, + ); + waiting.push({ + resolve: (m) => { clearTimeout(timer); resolve(m as T); }, + reject: (e) => { clearTimeout(timer); reject(e); }, + }); + }); + }, + stop: async () => { + child.stdin.end(); + const killer = setTimeout(() => child.kill('SIGKILL'), 10_000); + await exited; + clearTimeout(killer); + }, + }; } -const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); - /** * One round: a fresh database, both writers, the barrier, every hard pin. * Answers how many times the writer changed in the global order. */ -async function round(): Promise { +async function round(writers: Writer[]): Promise { const dir = mkdtempSync(join(tmpdir(), 'turso-autonumber-concurrency-')); dirs.push(dir); const dbPath = join(dir, 'shared.db'); @@ -199,23 +311,15 @@ async function round(): Promise { await parent.connect(); await parent.initObjects([NUMBERED_OBJECT as any]); - // A writer that dies before the barrier surfaces as ITS error, not as a - // readiness timeout. - let died: Error | null = null; - const runs = WORKERS.map((tag) => - runWorker({ OS_TEST_DB: dbPath, OS_TEST_DIR: dir, OS_TEST_TAG: tag, OS_TEST_WRITES: String(WRITES_PER_WORKER) }) - .catch((e: Error) => { died ??= e; throw e; }), - ); // Release both only once both are connected and registered, so the - // writes contend rather than queue behind process start-up. - const deadline = Date.now() + 60_000; - while (!WORKERS.every((tag) => existsSync(join(dir, `ready-${tag}`)))) { - if (died) throw died; - if (Date.now() > deadline) throw new Error('writers did not become ready within 60s'); - await sleep(5); - } - writeFileSync(join(dir, 'go'), '1'); - const reports = await Promise.all(runs); + // writes contend rather than queue behind connection set-up. + const opened = Date.now(); + for (const w of writers) w.send({ open: dbPath }); + await Promise.all(writers.map((w) => w.next('ready'))); + const released = Date.now(); + for (const w of writers) w.send({ go: WRITES_PER_WORKER }); + const reports = await Promise.all(writers.map((w) => w.next('report'))); + const written = Date.now(); // Different processes, and not this one. const pids = reports.map((r) => r.pid); @@ -260,17 +364,27 @@ async function round(): Promise { const inOrder = [...owner.keys()].sort((a, b) => a - b).map((n) => owner.get(n)!); let switches = 0; for (let i = 1; i < inOrder.length; i++) if (inOrder[i] !== inOrder[i - 1]) switches++; - process.stdout.write(`[turso-remote-autonumber-concurrency] pids ${pids.join('/')} · ${total} writes · ${new Set(all).size} distinct · ${switches} writer change(s)\n`); + // The phase times say where a slow run went: "ready" includes both process + // start-ups on the first round only, "writes" is the contention itself. + process.stdout.write( + `[turso-remote-autonumber-concurrency] pids ${pids.join('/')} · ${total} writes · ${new Set(all).size} distinct · ` + + `${switches} writer change(s) · ready ${released - opened}ms · writes ${written - released}ms\n`, + ); return switches; } describe('[#21113] two writers in two processes draw distinct numbers from one database', () => { it(`${WORKERS.length} x ${WRITES_PER_WORKER} creates -> ${WORKERS.length * WRITES_PER_WORKER} distinct, monotonic per writer, and the writers overlapped`, async (ctx) => { + const writers = WORKERS.map(startWriter); const observed: number[] = []; - for (let i = 0; i < MAX_ROUNDS; i++) { - const switches = await round(); - observed.push(switches); - if (switches >= 2) return; + try { + for (let i = 0; i < MAX_ROUNDS; i++) { + const switches = await round(writers); + observed.push(switches); + if (switches >= 2) return; + } + } finally { + await Promise.all(writers.map((w) => w.stop())); } ctx.skip( `NOT MEASURED: in ${MAX_ROUNDS} round(s) the two processes never overlapped on this box ` +