| author | |
| committer | |
| log | f37f99ca9e04dd0c4d0ee814402ae224175adc8d |
| tree | b308c2b5e9fe6b93fd635cb60ca40ca2645cfaf8 |
| parent | 4befc863f54a3341911c346c8822286ec37e54af |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>2 files changed, 70 insertions(+), 8 deletions(-)
packages/shared/src/nodeSqliteClient.test.ts+47-1| ... | @@ -18,6 +18,8 @@ import * as SqliteClient from "./nodeSqliteClient.ts"; | ... | @@ -18,6 +18,8 @@ import * as SqliteClient from "./nodeSqliteClient.ts"; |
| 18 | const startedWorkers = vi.hoisted((): Array<NodeWorkerThreads.Worker> => []); | 18 | const startedWorkers = vi.hoisted((): Array<NodeWorkerThreads.Worker> => []); |
| 19 | // When on, workers run a script that never announces, like a worker stuck opening the file. | 19 | // When on, workers run a script that never announces, like a worker stuck opening the file. |
| 20 | const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false })); | 20 | const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false })); |
| 21 | // When set, workers throw this at startup, like a checkpoint worker on SQLite before 3.51. | ||
| 22 | const failingWorkers = vi.hoisted(() => ({ message: undefined as string | undefined })); | ||
| 21 | 23 | ||
| 22 | vi.mock("node:worker_threads", async (importOriginal) => { | 24 | vi.mock("node:worker_threads", async (importOriginal) => { |
| 23 | const actual = await importOriginal<typeof NodeWorkerThreads>(); | 25 | const actual = await importOriginal<typeof NodeWorkerThreads>(); |
| ... | @@ -26,7 +28,9 @@ vi.mock("node:worker_threads", async (importOriginal) => { | ... | @@ -26,7 +28,9 @@ vi.mock("node:worker_threads", async (importOriginal) => { |
| 26 | super( | 28 | super( |
| 27 | ...((silentWorkers.on | 29 | ...((silentWorkers.on |
| 28 | ? ["setInterval(() => {}, 1 << 30);", { eval: true }] | 30 | ? ["setInterval(() => {}, 1 << 30);", { eval: true }] |
| 29 | : args) as ConstructorParameters<typeof actual.Worker>), | 31 | : failingWorkers.message !== undefined |
| 32 | ? [`throw new Error(${JSON.stringify(failingWorkers.message)});`, { eval: true }] | ||
| 33 | : args) as ConstructorParameters<typeof actual.Worker>), | ||
| 30 | ); | 34 | ); |
| 31 | startedWorkers.push(this); | 35 | startedWorkers.push(this); |
| 32 | } | 36 | } |
| ... | @@ -762,3 +766,45 @@ it.effect("hands checkpoints back to the writer when the worker stops", () => | ... | @@ -762,3 +766,45 @@ it.effect("hands checkpoints back to the writer when the worker stops", () => |
| 762 | }), | 766 | }), |
| 763 | ), | 767 | ), |
| 764 | ); | 768 | ); |
| 769 | |||
| 770 | it.effect("warns once and hands checkpoints back when the worker fails at startup", () => { | ||
| 771 | const { warnings, layer: logs } = captureWarnings(); | ||
| 772 | const reason = "SQLite 3.50.4 has no NOOP checkpoint"; | ||
| 773 | return Effect.acquireUseRelease( | ||
| 774 | Effect.sync(() => { | ||
| 775 | failingWorkers.message = reason; | ||
| 776 | }), | ||
| 777 | () => | ||
| 778 | withCheckpointClient((sql) => | ||
| 779 | Effect.gen(function* () { | ||
| 780 | const worker = startedWorkers.at(-1)!; | ||
| 781 | // threadId is -1 once the worker has exited and the client has seen "error" and "exit". | ||
| 782 | if (worker.threadId !== -1) { | ||
| 783 | yield* Effect.promise(() => new Promise((resolve) => worker.once("exit", resolve))); | ||
| 784 | } | ||
| 785 | const autocheckpoint = yield* sql<{ | ||
| 786 | readonly wal_autocheckpoint: number; | ||
| 787 | }>`PRAGMA wal_autocheckpoint`; | ||
| 788 | assert.equal(autocheckpoint[0]?.wal_autocheckpoint, 1000); | ||
| 789 | assert.deepEqual(warnings, [ | ||
| 790 | [ | ||
| 791 | "SQLite checkpoint worker unavailable; checkpoints run on the main connection", | ||
| 792 | { reason }, | ||
| 793 | ], | ||
| 794 | ]); | ||
| 795 | }), | ||
| 796 | ), | ||
| 797 | () => | ||
| 798 | Effect.sync(() => { | ||
| 799 | failingWorkers.message = undefined; | ||
| 800 | }), | ||
| 801 | ).pipe(Effect.provide(logs)); | ||
| 802 | }); | ||
| 803 | |||
| 804 | it.effect("does not warn when the client stops its checkpoint worker", () => { | ||
| 805 | const { warnings, layer: logs } = captureWarnings(); | ||
| 806 | return Effect.gen(function* () { | ||
| 807 | yield* withCheckpointClient(() => Effect.void); | ||
| 808 | assert.lengthOf(warnings, 0); | ||
| 809 | }).pipe(Effect.provide(logs)); | ||
| 810 | }); |
packages/shared/src/nodeSqliteClient.ts+23-7| ... | @@ -526,28 +526,36 @@ parentPort.on("message", (request) => { | ... | @@ -526,28 +526,36 @@ parentPort.on("message", (request) => { |
| 526 | /** | 526 | /** |
| 527 | * Moves the writer's WAL checkpoints to a worker. Call the returned function | 527 | * Moves the writer's WAL checkpoints to a worker. Call the returned function |
| 528 | * after each statement on the writer. While the worker runs, the writer only | 528 | * after each statement on the writer. While the worker runs, the writer only |
| 529 | * checkpoints at WRITER_BACKSTOP_FRAMES. If the worker fails or exits, the | 529 | * checkpoints at WRITER_BACKSTOP_FRAMES. If the worker cannot start, fails or |
| 530 | * writer goes back to SQLite's default. | 530 | * exits, the writer goes back to SQLite's default and one warning says why. |
| 531 | */ | 531 | */ |
| 532 | const startCheckpointWorker = Effect.fnUntraced(function* ( | 532 | const startCheckpointWorker = Effect.fnUntraced(function* ( |
| 533 | filename: string, | 533 | filename: string, |
| 534 | writer: NodeSqlite.DatabaseSync, | 534 | writer: NodeSqlite.DatabaseSync, |
| 535 | ) { | 535 | ) { |
| 536 | const warn = (reason: string) => | ||
| 537 | Effect.logWarning( | ||
| 538 | "SQLite checkpoint worker unavailable; checkpoints run on the main connection", | ||
| 539 | { reason }, | ||
| 540 | ); | ||
| 536 | let worker: NodeWorkerThreads.Worker; | 541 | let worker: NodeWorkerThreads.Worker; |
| 537 | try { | 542 | try { |
| 538 | worker = new NodeWorkerThreads.Worker(checkpointWorkerSource, { | 543 | worker = new NodeWorkerThreads.Worker(checkpointWorkerSource, { |
| 539 | eval: true, | 544 | eval: true, |
| 540 | workerData: { filename, frames: CHECKPOINT_FRAMES, idleMs: CHECKPOINT_IDLE_MS }, | 545 | workerData: { filename, frames: CHECKPOINT_FRAMES, idleMs: CHECKPOINT_IDLE_MS }, |
| 541 | }); | 546 | }); |
| 542 | } catch { | 547 | } catch (cause) { |
| 543 | // No worker threads here: the writer keeps checkpointing. | 548 | // No worker threads here: the writer keeps checkpointing. |
| 549 | yield* warn(String(cause)); | ||
| 544 | return undefined; | 550 | return undefined; |
| 545 | } | 551 | } |
| 546 | writer.exec(`PRAGMA wal_autocheckpoint = ${WRITER_BACKSTOP_FRAMES};`); | 552 | writer.exec(`PRAGMA wal_autocheckpoint = ${WRITER_BACKSTOP_FRAMES};`); |
| 553 | // Worker events arrive outside any fiber; this keeps the client's loggers. | ||
| 554 | const runFork = Effect.runForkWith(yield* Effect.context<never>()); | ||
| 547 | let stopped = false; | 555 | let stopped = false; |
| 548 | let checking = false; | 556 | let checking = false; |
| 549 | let checkAgain = false; | 557 | let checkAgain = false; |
| 550 | const handBack = () => { | 558 | const handBack = (reason: string) => { |
| 551 | if (stopped) return; | 559 | if (stopped) return; |
| 552 | stopped = true; | 560 | stopped = true; |
| 553 | try { | 561 | try { |
| ... | @@ -555,10 +563,18 @@ const startCheckpointWorker = Effect.fnUntraced(function* ( | ... | @@ -555,10 +563,18 @@ const startCheckpointWorker = Effect.fnUntraced(function* ( |
| 555 | } catch { | 563 | } catch { |
| 556 | // The writer is already closed. | 564 | // The writer is already closed. |
| 557 | } | 565 | } |
| 566 | runFork(warn(reason)); | ||
| 558 | }; | 567 | }; |
| 559 | worker.on("error", handBack); | 568 | // A worker that throws, such as on SQLite before 3.51, emits "error" before "exit". |
| 560 | worker.on("exit", handBack); | 569 | worker.on("error", (cause) => handBack(cause.message)); |
| 561 | yield* Effect.addFinalizer(() => Effect.promise(() => worker.terminate())); | 570 | worker.on("exit", (code) => handBack(`exited with code ${code}`)); |
| 571 | yield* Effect.addFinalizer(() => | ||
| 572 | Effect.promise(() => { | ||
| 573 | // A normal shutdown: the writer closes next, so there is nothing to hand back or report. | ||
| 574 | stopped = true; | ||
| 575 | return worker.terminate(); | ||
| 576 | }), | ||
| 577 | ); | ||
| 562 | 578 | ||
| 563 | // At most one request is in flight. A statement that runs during it asks for | 579 | // At most one request is in flight. A statement that runs during it asks for |
| 564 | // one more check, so the frames of a last commit before idle still get one. | 580 | // one more check, so the frames of a last commit before idle still get one. |