From 7e0c6cc507ca454cf095b008bc39e2783a6b6a2f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 07:56:49 +0000 Subject: [PATCH 1/3] test(driver-turso): start the concurrency writers once and drop the fsync The cross-process autonumber pin spent its wall on two things it does not assert: a tsx start-up per writer per round, and an fsync per commit on a throwaway temp database. The writers now start once and serve every round over stdin/stdout, and each sets synchronous = OFF on its own connection; the lock protocol and every pin are unchanged. Claude-Session: https://claude.ai/code/session_01P88ehrZ6hM19hBzdpsQTTo Co-authored-by: Claude --- ...urso-remote-autonumber-concurrency.test.ts | 227 ++++++++++++------ 1 file changed, 153 insertions(+), 74 deletions(-) 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..c71d71fa93a 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 @@ -16,13 +16,14 @@ * 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. + * SOURCE, the way `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,30 @@ * 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, 200 writes each, the default + * `synchronous = FULL`) took about 4.4 s: about 2.2 s to start each writer + * (`tsx` plus the driver's source graph, mostly `@objectstack/driver-sql`), + * and about 2.2 s of writes. 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. + * + * Two changes, neither 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. + * - 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 @@ -80,7 +105,7 @@ 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'; @@ -96,6 +121,8 @@ 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', @@ -120,74 +147,132 @@ 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. + * test itself does, and pays that start-up once for every round. 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'; + import { createInterface } from 'node:readline'; // 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)); + 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 a clean exit; a writer that does not leave is killed. */ + stop(): Promise; +} + +function startWriter(tag: string): Writer { + const child = spawn( + process.execPath, + ['--import', 'tsx', '--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); + 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 +284,12 @@ 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. + for (const w of writers) w.send({ open: dbPath }); + await Promise.all(writers.map((w) => w.next('ready'))); + for (const w of writers) w.send({ go: WRITES_PER_WORKER }); + const reports = await Promise.all(writers.map((w) => w.next('report'))); // Different processes, and not this one. const pids = reports.map((r) => r.pid); @@ -266,11 +340,16 @@ async function round(): Promise { 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 ` + From 794b7440918a4c0e19e8cd32c1243c668eb0fd55 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 08:28:40 +0000 Subject: [PATCH 2/3] test(driver-turso): load the writers' dependencies before tsx is registered A writer's start-up was about 40 ms of compiling the package's source and about 2 s of loading its dependencies' built code through tsx's require hook, which plain Node loads in about 1.2 s. The writer now requires those dependencies first, then registers tsx through its public CJS API and requires the source, whose requires hit the cache. The phase times join the measurement line. Claude-Session: https://claude.ai/code/session_01P88ehrZ6hM19hBzdpsQTTo Co-authored-by: Claude --- ...urso-remote-autonumber-concurrency.test.ts | 63 +++++++++++++------ 1 file changed, 44 insertions(+), 19 deletions(-) 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 c71d71fa93a..5d6337596ec 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,8 +15,8 @@ * 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). For each round + * 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 @@ -34,19 +34,25 @@ * # 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, 200 writes each, the default - * `synchronous = FULL`) took about 4.4 s: about 2.2 s to start each writer - * (`tsx` plus the driver's source graph, mostly `@objectstack/driver-sql`), - * and about 2.2 s of writes. The writes were fsync-bound: one `create` is three + * (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. * - * Two changes, neither touching what is asserted: + * 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 @@ -115,7 +121,15 @@ 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. */ @@ -146,9 +160,12 @@ interface WorkerReport { } /** - * The writer. Runs under `tsx` so it imports the driver's SOURCE, like the - * test itself does, and pays that start-up once for every round. It reads one - * JSON command per line on stdin and answers one JSON line 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 every round. 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 @@ -159,13 +176,13 @@ interface WorkerReport { * stdin closing ends the loop, and the process exits. */ const WORKER_SOURCE = ` - import { createClient } from '@libsql/client'; + import { createRequire } from 'node:module'; import { createInterface } from 'node:readline'; - // 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 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; @@ -209,7 +226,7 @@ interface Writer { function startWriter(tag: string): Writer { const child = spawn( process.execPath, - ['--import', 'tsx', '--input-type=module', '-e', WORKER_SOURCE], + ['--input-type=module', '-e', WORKER_SOURCE], { cwd: PKG_ROOT, env: { ...process.env, OS_TEST_TAG: tag }, stdio: ['pipe', 'pipe', 'pipe'] }, ); const inbox: unknown[] = []; @@ -286,10 +303,13 @@ async function round(writers: Writer[]): Promise { // Release both only once both are connected and registered, so the // 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); @@ -334,7 +354,12 @@ async function round(writers: Writer[]): 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; } From d87b9f06e4be47beb72545c4824d917486e19565 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 08:39:47 +0000 Subject: [PATCH 3/3] test(driver-turso): record the re-measured reverse verification The read-then-write mutation, re-run on the new shape, goes red in 6 of 8 runs with no gap and 5 of 5 with a 1 ms gap; the earlier shape went red in 1 of 5 with no gap on the same box. A command written to a writer that already died now fails with that writer's stderr. Claude-Session: https://claude.ai/code/session_01P88ehrZ6hM19hBzdpsQTTo Co-authored-by: Claude --- ...urso-remote-autonumber-concurrency.test.ts | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) 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 5d6337596ec..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 @@ -16,8 +16,8 @@ * Node process, whose native calls run on the one JS thread. * * 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 + * 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 @@ -107,6 +107,14 @@ * 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'; @@ -164,8 +172,8 @@ interface WorkerReport { * 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 every round. It reads one JSON command - * per line on stdin and answers one JSON line on stdout: + * 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 @@ -219,7 +227,7 @@ interface Writer { 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 a clean exit; a writer that does not leave is killed. */ + /** Close stdin and wait for the writer to exit; one still running 10 s later is killed. */ stop(): Promise; } @@ -258,6 +266,8 @@ function startWriter(tag: string): Writer { }); 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,