Skip to content

Commit 6e6881e

Browse files
test(driver-turso): the cross-process autonumber pin starts its writers once and stops waiting on fsync (#23218)
Part of #23182 (row 2) Clause-②: no **Row 2 of #23182 only:** `@objectstack/driver-turso`, `src/turso-remote-autonumber-concurrency.test.ts`, read at 29.3 s in run `38228297391`. Refresh PR #23141 lists it over 10 s in 6 of the 7 hourly full-battery runs that executed it. #23182 remains open for rows 1 and 3 to 6. ## What changed (one file, test-only) The file pins cross-process autonumber atomicity on the remote face: two writer processes and one libSQL database file must give N distinct numbers, monotonic per writer, with evidence that the writers overlapped. Most of its time went to work it does not assert on. Measured at `d4aaeaf0e` on this 4-vCPU container, one round took about 4.4 s: | phase | per round, before | what it was | |:--|--:|:--| | writer start-up | ~2.2 s | `node --import tsx`, then the driver's dependencies (`@objectstack/driver-sql` and what it imports, `@objectstack/spec`, `@libsql/client`) loaded through tsx's hook: ~2 s, against ~1.2 s under plain Node. The package's own source compiled in ~40 ms. | | writes (2 x 200 creates) | ~2.2 s | fsync-bound. A `create` is 3 statements (reserve, insert, read back). A single writer took 5.9 ms per create with `synchronous = FULL` and 1.2 ms with `synchronous = OFF`. | A round whose writers did not overlap also re-spawned both writers. Three changes, none to what is asserted: 1. **Writers start once.** They start once per test and serve every round over stdin/stdout, one JSON line per command and per answer. A retry round costs only its writes. 2. **Dependencies load before tsx.** A writer requires the driver's dependencies with plain Node first, then calls `require('tsx/cjs/api').register()` and requires `src/index.ts`. The source is still compiled by tsx, and the dependencies are the same modules, already in the require cache. `tsx` is already a declared devDependency, and `tsx/cjs/api` is one of its public exports. 3. **No fsync.** Each writer runs `PRAGMA synchronous = OFF` on its own connection. That drops only the disk flush at commit, on a throwaway temp file. The journal mode, the locks and the busy timeout are unchanged, so the cross-process serialisation the pin depends on is unchanged. The round's ready and write times now join the existing measurement line, so a slow CI run can be attributed from its log. Kept as they were: N = 200 per writer, at most 3 rounds, every pin, and the NOT MEASURED skip. N was not reduced: writes are no longer the cost, and a smaller N would only lower how often a broken counter is caught (see the reverse verification below). ## Readings The CI per-file number is vitest's module line (`✓ path (1 test) Nms`). All readings below are from this container under `scripts/pm/os-verify-lock.sh`, on a shared box (other seats' unlocked work was running), with vitest's default reporter (agent-detection variables unset, as in CI). | density | BASE `d4aaeaf0e` | this PR | |:--|--:|--:| | the file alone | 4575, 5256, 4406 ms | 1836, 1840, 1853 ms (`83ece05da`) | | whole `driver-turso` package (88 files) | 5735, 5914 ms | 1912 ms (`83ece05da`) | | package plus `driver-sql`'s suite at once | 10658, 8741, 8973 ms | 4126 ms (`794b74409`, same run as BASE's 8973) | Writer changes per round: 2 to 40 before, 141 to 221 after (unmutated runs). **CI reading for the after-state: NOT MEASURED here.** Job logs and artefacts redirect to an Azure blob host that this container's egress proxy refuses (403). This PR's own Test Core shard carries the file's module line and the measurement line in its log. ## Reverse verification (ablation) **Mutation:** the warm reservation (`UPDATE … SET last_value = last_value + 1 … RETURNING`) split into a `SELECT last_value` and then an `UPDATE … SET last_value = ?` (read-then-write), once with no gap and once with a 1 ms gap. - **Applied** through `scripts/ablation-replace.mjs`: the anchor hit went 1 → 0 and the blob changed. The restore was proven after every leg: blob equal to HEAD and `git diff HEAD` empty. - **Reach:** a tsx child spawned from the package root sees the mutated method, because `nextRemoteSequenceValue.toString()` carries the marker. No `dist` is involved, since both the parent and the writers run `src`. - **Predicted:** red on the "no write failed" pin, with `UNIQUE constraint failed` in a writer's error list, as the file's header recorded. **Observed:** that direction in every red run. | shape | no gap | 1 ms gap | |:--|--:|--:| | this PR (`794b74409` and `83ece05da`) | red 8 of 11 | red 7 of 7 | | this PR's first cut `7e0c6cc50` (before the dependency preload) | red 7 of 10 | red 5 of 5 | | BASE, same box and hour, interleaved with the first cut | red 1 of 5 | not run | A non-atomic counter is caught more often than before, not less: without the fsync the writers interleave far more finely. A zero-gap read-then-write remains a probabilistic catch, as it was at BASE. ## Gates and tests (at `83ece05da` unless named) - **Build:** `pnpm --filter '@objectstack/driver-turso...' build` green, after merging `origin/main`. - **Typecheck and tests:** `pnpm --filter @objectstack/driver-turso typecheck` (`tsc --noEmit`) is green, and its tsconfig includes `src/**/*`, tests included. The package suite ran 88/88 files: 2373 passed, 33 skipped, the same counts as BASE. - **Driver conformance:** `pnpm check:driver-conformance` before the first edit and after the last commit read `50 covered cell(s), 0 in the DEBT ledger, 0 exempt` both times. - **Derived gates:** `node scripts/pm/dispatch-gates.mjs --commands --repo objectstack-ai/objectstack` gave 51 commands, identical to the dispatch-time set, and all ran. The `--ran` verdict reads `51 derived, 51 run, 0 NOT-MEASURED, 0 UNRUN`. Three first answered exit 3 (unmet prerequisites) and were re-run green after their remedy: `check-plugin-teardown-shape --self-test` needed `git fetch --unshallow`, and `check:lean-entry-closure` and `check:dual-build-cjs-loads` needed package `dist`s, restored from the turbo cache. - **Lint, narrowed and proven:** eslint with `--no-inline-config` on the one changed file linted 1 file with 0 errors and 0 warnings. The file is in eslint's population (`--print-config` returns a config for it). The config enables no type-aware linting (the resolved `parserOptions` are `ecmaVersion` and `sourceType`), so this diff cannot move any other file's verdict. - **Bytes:** `check:nul-bytes` is green, and a control-byte self-scan of the file found 0 lines. ## Changeset None. The diff is test-only: the changed file is not in `driver-turso`'s `files[]` (`dist`, `README.md`, `CHANGELOG.md`). Measured on a `dist` built after the merge, four symbols unique to the test (`PRELOADED_DEPENDENCIES`, `WORKER_SOURCE`, `REPLY_DEADLINE_MS`, `turso-autonumber-concurrency-`) have 0 hits, and the positive control `nextRemoteSequenceValue` has 6. Label: `skip-changeset`. ## Acceptance notes - **Import time is shared, not this file's.** The whole vitest process for this one file still takes about 16 to 19 s of wall here, and about 14 to 15 s of that is vitest's import of the driver's source graph. Every file in this package pays that import, and the per-file reading does not count it. - **`@objectstack/core` import cost.** It alone takes about 0.9 s to import under plain Node and dominates a writer's remaining start-up of about 1.2 s. That is an observation about another package's import cost, not a defect; it is not filed. - **The old ablation reading was optimistic.** The file's header recorded the zero-gap ablation as red in 2 of 3 runs at BASE. On this box the same BASE shape was red in 1 of 5. The new shape records its own reading beside the old one. - **The card's pin can only be read after landing:** "the next refresh PR's list no longer names the file". --- _Generated by [Claude Code](https://claude.ai/code/session_01P88ehrZ6hM19hBzdpsQTTo)_ --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent 38a2d3c commit 6e6881e

1 file changed

Lines changed: 198 additions & 84 deletions

File tree

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

Lines changed: 198 additions & 84 deletions
Original file line numberDiff line numberDiff line change
@@ -15,21 +15,52 @@
1515
* read-then-write one. Neither can two `@libsql/client` instances in ONE
1616
* Node process, whose native calls run on the one JS thread.
1717
*
18-
* 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.
18+
* So this file spawns two child processes, each running the package's SOURCE
19+
* compiled by `tsx`, as `packages/spec`'s process-boundary pins do. For each
20+
* round 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 under `node --import tsx`, 200 writes each, the
38+
* default `synchronous = FULL`) took about 4.4 s: about 2.2 s to start each
39+
* writer and about 2.2 s of writes. Of a writer's start-up, the package's own
40+
* source took about 40 ms to compile; the rest was its dependencies' built code
41+
* (`@objectstack/driver-sql` and what it imports, `@objectstack/spec`,
42+
* `@libsql/client`), which loads in about 1.2 s under plain Node and about 2 s
43+
* under tsx's hook. The writes were fsync-bound: one `create` is three
44+
* statements (reserve, insert, read back), and a single writer took 5.9 ms per
45+
* `create` with fsync and 1.2 ms with `synchronous = OFF`. CI's hourly full
46+
* battery measured the file at up to 29.3 s, over 10 s in 6 of the 7 runs that
47+
* executed it, and every retry round paid both start-ups again.
48+
*
49+
* Three changes, none touching what is asserted:
50+
*
51+
* - The writers start ONCE per test and serve every round over stdin/stdout.
52+
* A retry round therefore costs its writes, not two more process start-ups.
53+
* - A writer loads the driver's dependencies with plain Node first, then
54+
* registers tsx and requires the source. The source is still compiled by
55+
* tsx, and the dependencies are the same modules, loaded without the hook.
56+
* - Each writer sets `PRAGMA synchronous = OFF` on its own connection.
57+
* `synchronous` decides only whether SQLite flushes to disk at a commit,
58+
* which matters for a power cut and not for a throwaway temp file. The lock
59+
* protocol is unchanged: the journal mode, the locks each statement takes and
60+
* the busy timeout all stay as they were. What decides atomicity, two
61+
* connections in two processes serialised by the database's write lock, is
62+
* exactly what it was.
63+
*
3364
* # The pins
3465
*
3566
* - N writes from two processes give N distinct numbers — the counter moved
@@ -76,11 +107,19 @@
76107
* what this pin distinguishes is a counter that can hand out one number
77108
* twice, and on a declared-unique column that surfaces as refused writes
78109
* before it surfaces as duplicate rows.
110+
*
111+
* Re-measured on the current shape (writers started once, no fsync) at
112+
* `794b74409`, with the same mutation: red in 6 of 8 runs with no gap and 5 of
113+
* 5 with a 1 ms gap, every time on the same "no write failed" pin. The earlier
114+
* shape, on the same box and day, went red in 1 of 5 runs with no gap. Without
115+
* the fsync the two writers interleave far more finely (about 100 to 200
116+
* writer changes per round, against 2 to 40 before), so a round gives a
117+
* read-then-write counter more chances to hand out one number twice, not fewer.
79118
*/
80119

81120
import { describe, it, expect, afterAll } from 'vitest';
82121
import { spawn } from 'node:child_process';
83-
import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
122+
import { mkdtempSync, rmSync } from 'node:fs';
84123
import { tmpdir } from 'node:os';
85124
import { join } from 'node:path';
86125
import { fileURLToPath } from 'node:url';
@@ -90,12 +129,22 @@ import { TursoDriver } from './index.js';
90129

91130
const HERE = fileURLToPath(new URL('.', import.meta.url));
92131
const PKG_ROOT = join(HERE, '..');
93-
const DRIVER_ENTRY = new URL('./index.ts', import.meta.url).href;
132+
const DRIVER_ENTRY = fileURLToPath(new URL('./index.ts', import.meta.url));
133+
/**
134+
* What the driver's source imports at runtime from `node_modules` (its type-only
135+
* imports are erased). The writer loads these BEFORE tsx is registered: they
136+
* are built code, not this package's source, and tsx's require hook roughly
137+
* doubles their load. A dependency missing from this list still loads, under
138+
* tsx, so the list is a cost, never a correctness, question.
139+
*/
140+
const PRELOADED_DEPENDENCIES = ['@libsql/client', '@objectstack/driver-sql', '@objectstack/spec/api', '@objectstack/spec/data'];
94141

95142
const WRITES_PER_WORKER = 200;
96143
/** Rounds a loaded box gets to let the two writers overlap before the overlap evidence is declared NOT MEASURED. */
97144
const MAX_ROUNDS = 3;
98145
const WORKERS = ['A', 'B'] as const;
146+
/** How long a writer may take to answer one command before the round fails with that writer's stderr. */
147+
const REPLY_DEADLINE_MS = 60_000;
99148

100149
const NUMBERED_OBJECT = {
101150
name: 'crm_case',
@@ -119,75 +168,138 @@ interface WorkerReport {
119168
}
120169

121170
/**
122-
* 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.
171+
* The writer. Loads the driver's dependencies with plain Node, then registers
172+
* tsx through its public CJS API and requires the driver's SOURCE, like the
173+
* test itself does — so the source is compiled by tsx, to CJS (this package
174+
* has no `"type": "module"`), and its requires of those dependencies are cache
175+
* hits. It pays that start-up once, for all the rounds. It reads one JSON
176+
* command per line on stdin and answers one JSON line on stdout:
177+
*
178+
* - `{ open: <db path> }` — connect to that round's database, register the
179+
* schema (the tables already exist — the parent created them — so
180+
* `initObjects` records metadata and runs no DDL), answer `{ ready }`.
181+
* - `{ go: <n> }` — write n records as fast as the database lets it,
182+
* disconnect, answer the report.
183+
*
184+
* stdin closing ends the loop, and the process exits.
127185
*/
128186
const WORKER_SOURCE = `
129-
import { createClient } from '@libsql/client';
130-
import { existsSync, writeFileSync } from 'node:fs';
131-
// tsx compiles this package's .ts to CJS (no "type": "module"), so the
132-
// class may arrive under \`default\` — the spelling packages/spec's
133-
// process-boundary pins use.
134-
const driverModule = await import(${JSON.stringify(DRIVER_ENTRY)});
135-
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));
187+
import { createRequire } from 'node:module';
188+
import { createInterface } from 'node:readline';
189+
const require = createRequire(${JSON.stringify(join(PKG_ROOT, 'package.json'))});
190+
for (const dependency of ${JSON.stringify(PRELOADED_DEPENDENCIES)}) require(dependency);
191+
require('tsx/cjs/api').register();
192+
const { TursoDriver } = require(${JSON.stringify(DRIVER_ENTRY)});
193+
const { createClient } = require('@libsql/client');
194+
const { OS_TEST_TAG } = process.env;
195+
const reply = (message) => process.stdout.write(JSON.stringify(message) + '\\n');
196+
let driver = null;
197+
for await (const line of createInterface({ input: process.stdin })) {
198+
const command = JSON.parse(line);
199+
if (command.open) {
200+
const client = createClient({ url: 'file:' + command.open, timeout: 20000 });
201+
// No disk flush at commit: see "What the file waits on" in the parent.
202+
await client.execute('PRAGMA synchronous = OFF');
203+
driver = new TursoDriver({ url: 'libsql://concurrency.probe', client });
204+
await driver.connect();
205+
await driver.initObjects([${JSON.stringify(NUMBERED_OBJECT)}]);
206+
reply({ ready: OS_TEST_TAG });
207+
} else if (command.go) {
208+
const numbers = [];
209+
const errors = [];
210+
for (let i = 0; i < command.go; i++) {
211+
try {
212+
const row = await driver.create('crm_case', { organization_id: 'orgA', title: OS_TEST_TAG + '-' + i });
213+
numbers.push(row.case_number);
214+
} catch (e) {
215+
errors.push(String(e && e.message || e));
216+
}
217+
}
218+
await driver.disconnect();
219+
driver = null;
220+
reply({ tag: OS_TEST_TAG, pid: process.pid, numbers, errors });
152221
}
153222
}
154-
await driver.disconnect();
155-
process.stdout.write(JSON.stringify({ tag: OS_TEST_TAG, pid: process.pid, numbers, errors }));
156223
`;
157224

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-
}
225+
interface Writer {
226+
tag: string;
227+
send(command: Record<string, unknown>): void;
228+
/** The writer's next answer; rejects with its stderr if it dies or stays silent past the deadline. */
229+
next<T>(what: string): Promise<T>;
230+
/** Close stdin and wait for the writer to exit; one still running 10 s later is killed. */
231+
stop(): Promise<void>;
232+
}
233+
234+
function startWriter(tag: string): Writer {
235+
const child = spawn(
236+
process.execPath,
237+
['--input-type=module', '-e', WORKER_SOURCE],
238+
{ cwd: PKG_ROOT, env: { ...process.env, OS_TEST_TAG: tag }, stdio: ['pipe', 'pipe', 'pipe'] },
239+
);
240+
const inbox: unknown[] = [];
241+
const waiting: Array<{ resolve: (m: unknown) => void; reject: (e: Error) => void }> = [];
242+
let failure: Error | null = null;
243+
let stderr = '';
244+
let pending = '';
245+
const fail = (e: Error) => {
246+
failure ??= e;
247+
while (waiting.length > 0) waiting.shift()!.reject(failure);
248+
};
249+
child.stderr.on('data', (d) => { stderr += String(d); });
250+
child.stdout.on('data', (d) => {
251+
pending += String(d);
252+
for (let nl = pending.indexOf('\n'); nl >= 0; nl = pending.indexOf('\n')) {
253+
const line = pending.slice(0, nl);
254+
pending = pending.slice(nl + 1);
255+
let message: unknown;
175256
try {
176-
resolve(JSON.parse(out) as WorkerReport);
257+
message = JSON.parse(line);
177258
} catch (e) {
178-
reject(new Error(`writer ${env.OS_TEST_TAG} wrote no report: ${String(e)}\nstderr:\n${err}\nstdout:\n${out}`));
259+
fail(new Error(`writer ${tag} wrote a line that is not a report: ${String(e)}\nline: ${line}\nstderr:\n${stderr}`));
260+
return;
179261
}
180-
});
262+
const waiter = waiting.shift();
263+
if (waiter) waiter.resolve(message);
264+
else inbox.push(message);
265+
}
181266
});
267+
const exited = new Promise<number | null>((resolve) => child.on('close', resolve));
268+
child.on('error', fail);
269+
// A command written to a writer that already died fails here, with its stderr.
270+
child.stdin.on('error', (e) => fail(new Error(`writer ${tag} could not take a command: ${String(e)}\nstderr:\n${stderr}`)));
271+
child.on('close', (code) => fail(new Error(`writer ${tag} exited ${code}\nstderr:\n${stderr}`)));
272+
return {
273+
tag,
274+
send: (command) => { child.stdin.write(JSON.stringify(command) + '\n'); },
275+
next: <T>(what: string) => {
276+
if (inbox.length > 0) return Promise.resolve(inbox.shift() as T);
277+
if (failure) return Promise.reject(failure);
278+
return new Promise<T>((resolve, reject) => {
279+
const timer = setTimeout(
280+
() => reject(new Error(`writer ${tag} sent no ${what} within ${REPLY_DEADLINE_MS / 1000}s\nstderr:\n${stderr}`)),
281+
REPLY_DEADLINE_MS,
282+
);
283+
waiting.push({
284+
resolve: (m) => { clearTimeout(timer); resolve(m as T); },
285+
reject: (e) => { clearTimeout(timer); reject(e); },
286+
});
287+
});
288+
},
289+
stop: async () => {
290+
child.stdin.end();
291+
const killer = setTimeout(() => child.kill('SIGKILL'), 10_000);
292+
await exited;
293+
clearTimeout(killer);
294+
},
295+
};
182296
}
183297

184-
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
185-
186298
/**
187299
* One round: a fresh database, both writers, the barrier, every hard pin.
188300
* Answers how many times the writer changed in the global order.
189301
*/
190-
async function round(): Promise<number> {
302+
async function round(writers: Writer[]): Promise<number> {
191303
const dir = mkdtempSync(join(tmpdir(), 'turso-autonumber-concurrency-'));
192304
dirs.push(dir);
193305
const dbPath = join(dir, 'shared.db');
@@ -199,23 +311,15 @@ async function round(): Promise<number> {
199311
await parent.connect();
200312
await parent.initObjects([NUMBERED_OBJECT as any]);
201313

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-
);
209314
// 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);
315+
// writes contend rather than queue behind connection set-up.
316+
const opened = Date.now();
317+
for (const w of writers) w.send({ open: dbPath });
318+
await Promise.all(writers.map((w) => w.next('ready')));
319+
const released = Date.now();
320+
for (const w of writers) w.send({ go: WRITES_PER_WORKER });
321+
const reports = await Promise.all(writers.map((w) => w.next<WorkerReport>('report')));
322+
const written = Date.now();
219323

220324
// Different processes, and not this one.
221325
const pids = reports.map((r) => r.pid);
@@ -260,17 +364,27 @@ async function round(): Promise<number> {
260364
const inOrder = [...owner.keys()].sort((a, b) => a - b).map((n) => owner.get(n)!);
261365
let switches = 0;
262366
for (let i = 1; i < inOrder.length; i++) if (inOrder[i] !== inOrder[i - 1]) switches++;
263-
process.stdout.write(`[turso-remote-autonumber-concurrency] pids ${pids.join('/')} · ${total} writes · ${new Set(all).size} distinct · ${switches} writer change(s)\n`);
367+
// The phase times say where a slow run went: "ready" includes both process
368+
// start-ups on the first round only, "writes" is the contention itself.
369+
process.stdout.write(
370+
`[turso-remote-autonumber-concurrency] pids ${pids.join('/')} · ${total} writes · ${new Set(all).size} distinct · ` +
371+
`${switches} writer change(s) · ready ${released - opened}ms · writes ${written - released}ms\n`,
372+
);
264373
return switches;
265374
}
266375

267376
describe('[#21113] two writers in two processes draw distinct numbers from one database', () => {
268377
it(`${WORKERS.length} x ${WRITES_PER_WORKER} creates -> ${WORKERS.length * WRITES_PER_WORKER} distinct, monotonic per writer, and the writers overlapped`, async (ctx) => {
378+
const writers = WORKERS.map(startWriter);
269379
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;
380+
try {
381+
for (let i = 0; i < MAX_ROUNDS; i++) {
382+
const switches = await round(writers);
383+
observed.push(switches);
384+
if (switches >= 2) return;
385+
}
386+
} finally {
387+
await Promise.all(writers.map((w) => w.stop()));
274388
}
275389
ctx.skip(
276390
`NOT MEASURED: in ${MAX_ROUNDS} round(s) the two processes never overlapped on this box ` +

0 commit comments

Comments
 (0)