Skip to content

Commit 7e0c6cc

Browse files
committed
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 <noreply@anthropic.com>
1 parent d4aaeaf commit 7e0c6cc

1 file changed

Lines changed: 153 additions & 74 deletions

File tree

‎packages/drivers/driver-turso/src/turso-remote-autonumber-concurrency.test.ts‎

Lines changed: 153 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -16,20 +16,45 @@
1616
* Node process, whose native calls run on the one JS thread.
1717
*
1818
* So this file spawns two child processes (through `tsx`, over the package's
19-
* SOURCE, the way `packages/spec`'s process-boundary pins do), each holding
20-
* its own `@libsql/client` connection — the native `file:` backend, which is
21-
* libSQL itself, the engine behind a Turso endpoint — to one database file,
22-
* each driving its own `TursoDriver` in REMOTE mode, writing through
23-
* `TursoDriver.create`. A file barrier releases both writers at once, so their
24-
* statements contend on the database's write lock rather than following each
25-
* other through process start-up.
19+
* SOURCE, the way `packages/spec`'s process-boundary pins do). For each round
20+
* the parent creates a fresh database, and each writer opens its own
21+
* `@libsql/client` connection to it — the native `file:` backend, which is
22+
* libSQL itself, the engine behind a Turso endpoint — drives its own
23+
* `TursoDriver` in REMOTE mode, and writes through `TursoDriver.create`. The
24+
* parent releases both writers with one `go` written to both pipes in the same
25+
* tick, once both have reported ready, so their statements contend on the
26+
* database's write lock rather than following each other through start-up.
2627
*
2728
* What is NOT measured here, said plainly: an HTTP `sqld` server. None runs in
2829
* this environment, and the transport's HTTP batching and retries are the
2930
* concern of the suites that mock `execute`. What a server adds is the network;
3031
* what decides atomicity is the statement, and the statement runs on the same
3132
* engine under the same lock here as there.
3233
*
34+
* # What the file waits on, and what it no longer waits on
35+
*
36+
* Measured at `d4aaeaf0e` on a 4-vCPU box, one round of the earlier shape
37+
* (writers spawned per round, 200 writes each, the default
38+
* `synchronous = FULL`) took about 4.4 s: about 2.2 s to start each writer
39+
* (`tsx` plus the driver's source graph, mostly `@objectstack/driver-sql`),
40+
* and about 2.2 s of writes. The writes were fsync-bound: one `create` is three
41+
* statements (reserve, insert, read back), and a single writer took 5.9 ms per
42+
* `create` with fsync and 1.2 ms with `synchronous = OFF`. CI's hourly full
43+
* battery measured the file at up to 29.3 s, over 10 s in 6 of the 7 runs that
44+
* executed it, and every retry round paid both start-ups again.
45+
*
46+
* Two changes, neither touching what is asserted:
47+
*
48+
* - The writers start ONCE per test and serve every round over stdin/stdout.
49+
* A retry round therefore costs its writes, not two more process start-ups.
50+
* - Each writer sets `PRAGMA synchronous = OFF` on its own connection.
51+
* `synchronous` decides only whether SQLite flushes to disk at a commit,
52+
* which matters for a power cut and not for a throwaway temp file. The lock
53+
* protocol is unchanged: the journal mode, the locks each statement takes and
54+
* the busy timeout all stay as they were. What decides atomicity, two
55+
* connections in two processes serialised by the database's write lock, is
56+
* exactly what it was.
57+
*
3358
* # The pins
3459
*
3560
* - N writes from two processes give N distinct numbers — the counter moved
@@ -80,7 +105,7 @@
80105

81106
import { describe, it, expect, afterAll } from 'vitest';
82107
import { spawn } from 'node:child_process';
83-
import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
108+
import { mkdtempSync, rmSync } from 'node:fs';
84109
import { tmpdir } from 'node:os';
85110
import { join } from 'node:path';
86111
import { fileURLToPath } from 'node:url';
@@ -96,6 +121,8 @@ const WRITES_PER_WORKER = 200;
96121
/** Rounds a loaded box gets to let the two writers overlap before the overlap evidence is declared NOT MEASURED. */
97122
const MAX_ROUNDS = 3;
98123
const WORKERS = ['A', 'B'] as const;
124+
/** How long a writer may take to answer one command before the round fails with that writer's stderr. */
125+
const REPLY_DEADLINE_MS = 60_000;
99126

100127
const NUMBERED_OBJECT = {
101128
name: 'crm_case',
@@ -120,74 +147,132 @@ interface WorkerReport {
120147

121148
/**
122149
* The writer. Runs under `tsx` so it imports the driver's SOURCE, like the
123-
* test itself does. It connects, registers the schema (the tables already
124-
* exist — the parent created them — so `initObjects` records metadata and
125-
* runs no DDL), announces readiness, waits for the parent's `go` file, then
126-
* writes as fast as the database lets it and reports on stdout.
150+
* test itself does, and pays that start-up once for every round. It reads one
151+
* JSON command per line on stdin and answers one JSON line on stdout:
152+
*
153+
* - `{ open: <db path> }` — connect to that round's database, register the
154+
* schema (the tables already exist — the parent created them — so
155+
* `initObjects` records metadata and runs no DDL), answer `{ ready }`.
156+
* - `{ go: <n> }` — write n records as fast as the database lets it,
157+
* disconnect, answer the report.
158+
*
159+
* stdin closing ends the loop, and the process exits.
127160
*/
128161
const WORKER_SOURCE = `
129162
import { createClient } from '@libsql/client';
130-
import { existsSync, writeFileSync } from 'node:fs';
163+
import { createInterface } from 'node:readline';
131164
// tsx compiles this package's .ts to CJS (no "type": "module"), so the
132165
// class may arrive under \`default\` — the spelling packages/spec's
133166
// process-boundary pins use.
134167
const driverModule = await import(${JSON.stringify(DRIVER_ENTRY)});
135168
const TursoDriver = driverModule.TursoDriver ?? driverModule.default?.TursoDriver;
136-
const { OS_TEST_DB, OS_TEST_DIR, OS_TEST_TAG, OS_TEST_WRITES } = process.env;
137-
const client = createClient({ url: 'file:' + OS_TEST_DB, timeout: 20000 });
138-
const driver = new TursoDriver({ url: 'libsql://concurrency.probe', client });
139-
await driver.connect();
140-
await driver.initObjects([${JSON.stringify(NUMBERED_OBJECT)}]);
141-
writeFileSync(OS_TEST_DIR + '/ready-' + OS_TEST_TAG, String(process.pid));
142-
const go = OS_TEST_DIR + '/go';
143-
while (!existsSync(go)) await new Promise((r) => setTimeout(r, 2));
144-
const numbers = [];
145-
const errors = [];
146-
for (let i = 0; i < Number(OS_TEST_WRITES); i++) {
147-
try {
148-
const row = await driver.create('crm_case', { organization_id: 'orgA', title: OS_TEST_TAG + '-' + i });
149-
numbers.push(row.case_number);
150-
} catch (e) {
151-
errors.push(String(e && e.message || e));
169+
const { OS_TEST_TAG } = process.env;
170+
const reply = (message) => process.stdout.write(JSON.stringify(message) + '\\n');
171+
let driver = null;
172+
for await (const line of createInterface({ input: process.stdin })) {
173+
const command = JSON.parse(line);
174+
if (command.open) {
175+
const client = createClient({ url: 'file:' + command.open, timeout: 20000 });
176+
// No disk flush at commit: see "What the file waits on" in the parent.
177+
await client.execute('PRAGMA synchronous = OFF');
178+
driver = new TursoDriver({ url: 'libsql://concurrency.probe', client });
179+
await driver.connect();
180+
await driver.initObjects([${JSON.stringify(NUMBERED_OBJECT)}]);
181+
reply({ ready: OS_TEST_TAG });
182+
} else if (command.go) {
183+
const numbers = [];
184+
const errors = [];
185+
for (let i = 0; i < command.go; i++) {
186+
try {
187+
const row = await driver.create('crm_case', { organization_id: 'orgA', title: OS_TEST_TAG + '-' + i });
188+
numbers.push(row.case_number);
189+
} catch (e) {
190+
errors.push(String(e && e.message || e));
191+
}
192+
}
193+
await driver.disconnect();
194+
driver = null;
195+
reply({ tag: OS_TEST_TAG, pid: process.pid, numbers, errors });
152196
}
153197
}
154-
await driver.disconnect();
155-
process.stdout.write(JSON.stringify({ tag: OS_TEST_TAG, pid: process.pid, numbers, errors }));
156198
`;
157199

158-
function runWorker(env: Record<string, string>): Promise<WorkerReport> {
159-
return new Promise((resolve, reject) => {
160-
const child = spawn(
161-
process.execPath,
162-
['--import', 'tsx', '--input-type=module', '-e', WORKER_SOURCE],
163-
{ cwd: PKG_ROOT, env: { ...process.env, ...env }, stdio: ['ignore', 'pipe', 'pipe'] },
164-
);
165-
let out = '';
166-
let err = '';
167-
child.stdout.on('data', (d) => { out += String(d); });
168-
child.stderr.on('data', (d) => { err += String(d); });
169-
child.on('error', reject);
170-
child.on('close', (code) => {
171-
if (code !== 0) {
172-
reject(new Error(`writer ${env.OS_TEST_TAG} exited ${code}\nstderr:\n${err}\nstdout:\n${out}`));
173-
return;
174-
}
200+
interface Writer {
201+
tag: string;
202+
send(command: Record<string, unknown>): void;
203+
/** The writer's next answer; rejects with its stderr if it dies or stays silent past the deadline. */
204+
next<T>(what: string): Promise<T>;
205+
/** Close stdin and wait for a clean exit; a writer that does not leave is killed. */
206+
stop(): Promise<void>;
207+
}
208+
209+
function startWriter(tag: string): Writer {
210+
const child = spawn(
211+
process.execPath,
212+
['--import', 'tsx', '--input-type=module', '-e', WORKER_SOURCE],
213+
{ cwd: PKG_ROOT, env: { ...process.env, OS_TEST_TAG: tag }, stdio: ['pipe', 'pipe', 'pipe'] },
214+
);
215+
const inbox: unknown[] = [];
216+
const waiting: Array<{ resolve: (m: unknown) => void; reject: (e: Error) => void }> = [];
217+
let failure: Error | null = null;
218+
let stderr = '';
219+
let pending = '';
220+
const fail = (e: Error) => {
221+
failure ??= e;
222+
while (waiting.length > 0) waiting.shift()!.reject(failure);
223+
};
224+
child.stderr.on('data', (d) => { stderr += String(d); });
225+
child.stdout.on('data', (d) => {
226+
pending += String(d);
227+
for (let nl = pending.indexOf('\n'); nl >= 0; nl = pending.indexOf('\n')) {
228+
const line = pending.slice(0, nl);
229+
pending = pending.slice(nl + 1);
230+
let message: unknown;
175231
try {
176-
resolve(JSON.parse(out) as WorkerReport);
232+
message = JSON.parse(line);
177233
} catch (e) {
178-
reject(new Error(`writer ${env.OS_TEST_TAG} wrote no report: ${String(e)}\nstderr:\n${err}\nstdout:\n${out}`));
234+
fail(new Error(`writer ${tag} wrote a line that is not a report: ${String(e)}\nline: ${line}\nstderr:\n${stderr}`));
235+
return;
179236
}
180-
});
237+
const waiter = waiting.shift();
238+
if (waiter) waiter.resolve(message);
239+
else inbox.push(message);
240+
}
181241
});
242+
const exited = new Promise<number | null>((resolve) => child.on('close', resolve));
243+
child.on('error', fail);
244+
child.on('close', (code) => fail(new Error(`writer ${tag} exited ${code}\nstderr:\n${stderr}`)));
245+
return {
246+
tag,
247+
send: (command) => { child.stdin.write(JSON.stringify(command) + '\n'); },
248+
next: <T>(what: string) => {
249+
if (inbox.length > 0) return Promise.resolve(inbox.shift() as T);
250+
if (failure) return Promise.reject(failure);
251+
return new Promise<T>((resolve, reject) => {
252+
const timer = setTimeout(
253+
() => reject(new Error(`writer ${tag} sent no ${what} within ${REPLY_DEADLINE_MS / 1000}s\nstderr:\n${stderr}`)),
254+
REPLY_DEADLINE_MS,
255+
);
256+
waiting.push({
257+
resolve: (m) => { clearTimeout(timer); resolve(m as T); },
258+
reject: (e) => { clearTimeout(timer); reject(e); },
259+
});
260+
});
261+
},
262+
stop: async () => {
263+
child.stdin.end();
264+
const killer = setTimeout(() => child.kill('SIGKILL'), 10_000);
265+
await exited;
266+
clearTimeout(killer);
267+
},
268+
};
182269
}
183270

184-
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
185-
186271
/**
187272
* One round: a fresh database, both writers, the barrier, every hard pin.
188273
* Answers how many times the writer changed in the global order.
189274
*/
190-
async function round(): Promise<number> {
275+
async function round(writers: Writer[]): Promise<number> {
191276
const dir = mkdtempSync(join(tmpdir(), 'turso-autonumber-concurrency-'));
192277
dirs.push(dir);
193278
const dbPath = join(dir, 'shared.db');
@@ -199,23 +284,12 @@ async function round(): Promise<number> {
199284
await parent.connect();
200285
await parent.initObjects([NUMBERED_OBJECT as any]);
201286

202-
// A writer that dies before the barrier surfaces as ITS error, not as a
203-
// readiness timeout.
204-
let died: Error | null = null;
205-
const runs = WORKERS.map((tag) =>
206-
runWorker({ OS_TEST_DB: dbPath, OS_TEST_DIR: dir, OS_TEST_TAG: tag, OS_TEST_WRITES: String(WRITES_PER_WORKER) })
207-
.catch((e: Error) => { died ??= e; throw e; }),
208-
);
209287
// Release both only once both are connected and registered, so the
210-
// writes contend rather than queue behind process start-up.
211-
const deadline = Date.now() + 60_000;
212-
while (!WORKERS.every((tag) => existsSync(join(dir, `ready-${tag}`)))) {
213-
if (died) throw died;
214-
if (Date.now() > deadline) throw new Error('writers did not become ready within 60s');
215-
await sleep(5);
216-
}
217-
writeFileSync(join(dir, 'go'), '1');
218-
const reports = await Promise.all(runs);
288+
// writes contend rather than queue behind connection set-up.
289+
for (const w of writers) w.send({ open: dbPath });
290+
await Promise.all(writers.map((w) => w.next('ready')));
291+
for (const w of writers) w.send({ go: WRITES_PER_WORKER });
292+
const reports = await Promise.all(writers.map((w) => w.next<WorkerReport>('report')));
219293

220294
// Different processes, and not this one.
221295
const pids = reports.map((r) => r.pid);
@@ -266,11 +340,16 @@ async function round(): Promise<number> {
266340

267341
describe('[#21113] two writers in two processes draw distinct numbers from one database', () => {
268342
it(`${WORKERS.length} x ${WRITES_PER_WORKER} creates -> ${WORKERS.length * WRITES_PER_WORKER} distinct, monotonic per writer, and the writers overlapped`, async (ctx) => {
343+
const writers = WORKERS.map(startWriter);
269344
const observed: number[] = [];
270-
for (let i = 0; i < MAX_ROUNDS; i++) {
271-
const switches = await round();
272-
observed.push(switches);
273-
if (switches >= 2) return;
345+
try {
346+
for (let i = 0; i < MAX_ROUNDS; i++) {
347+
const switches = await round(writers);
348+
observed.push(switches);
349+
if (switches >= 2) return;
350+
}
351+
} finally {
352+
await Promise.all(writers.map((w) => w.stop()));
274353
}
275354
ctx.skip(
276355
`NOT MEASURED: in ${MAX_ROUNDS} round(s) the two processes never overlapped on this box ` +

0 commit comments

Comments
 (0)