diff --git a/packages/db/src/writeQueue.js b/packages/db/src/writeQueue.js index c709176..0392045 100644 --- a/packages/db/src/writeQueue.js +++ b/packages/db/src/writeQueue.js @@ -1,6 +1,6 @@ -import { Queue, QueueEvents, Worker } from 'bullmq'; +import { Queue, QueueEvents, UnrecoverableError, Worker } from 'bullmq'; -import { createWriteFolder } from './writeFolder.js'; +import { createWriteFolder, isStatementError } from './writeFolder.js'; /** * The write path, moved out of the process and into Redis. @@ -346,19 +346,55 @@ async function releaseQueue(url, prefix) { * @param {{ url: string, prefix?: string, onEvent?: ((event: object) => void)|null }} opts * @returns {import('bullmq').Worker} */ +/** + * Run one write job, and decide whether failing it is worth another attempt. + * + * Separated from the `Worker` so it can be tested without a broker: the retry + * decision is the part with the judgement in it, and the BullMQ plumbing is not. + * + * **A retry is only worth a slot if the next attempt could go differently.** At + * concurrency 1 this is not merely wasted effort, it is head-of-line blocking: + * a failing job holds the *cluster's only writer* for each of its attempts, and + * every other write waits behind it. + * + * Both halves of that were seen in production within an hour of this queue being + * switched on. A `UNIQUE constraint failed: authors.slug` job retried three + * times and could never have succeeded -- the same statements against the same + * data fail identically for ever. Separately, a timing-out job ran from + * 21:29:27 to 21:30:29, three attempts at thirty seconds, with the writer held + * throughout. + * + * So a constraint or syntax error is answered with `UnrecoverableError`, which + * tells BullMQ not to retry: the caller learns at once and the writer is + * released. Transport failures and timeouts still retry, because those can + * genuinely go differently on the next attempt -- that is the whole reason the + * queue was wanted. + * + * @param {import('@libsql/client').Client} client + * @param {Array<{ sql: string, args?: unknown[] }>} statements + * @returns {Promise} + */ +export async function runWriteJob(client, statements) { + if (!statements || statements.length === 0) return []; + + try { + const results = await client.batch(statements, 'write'); + return Array.from(results ?? []).map(encodeResult); + } catch (err) { + if (isStatementError(err)) { + throw new UnrecoverableError(err instanceof Error ? err.message : String(err)); + } + throw err; + } +} + export function createWriteWorker(client, opts) { const connection = connectionFor(opts.url); const prefix = opts.prefix ?? '{rssamplifier}'; return new Worker( WRITE_QUEUE, - async (job) => { - const statements = (job.data.statements ?? []).map(decodeStatement); - if (statements.length === 0) return []; - - const results = await client.batch(statements, 'write'); - return Array.from(results ?? []).map(encodeResult); - }, + (job) => runWriteJob(client, (job.data.statements ?? []).map(decodeStatement)), { connection, prefix, concurrency: 1 }, ); } diff --git a/packages/db/test/write-job-retries.test.js b/packages/db/test/write-job-retries.test.js new file mode 100644 index 0000000..ec20bd7 --- /dev/null +++ b/packages/db/test/write-job-retries.test.js @@ -0,0 +1,97 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; + +import { UnrecoverableError } from 'bullmq'; + +import { runWriteJob } from '../src/writeQueue.js'; + +/** + * Which write failures are worth another attempt. + * + * At worker concurrency 1 — which is not tunable, because SQLite has one writer + * — a retry does not merely waste effort. The failing job holds the cluster's + * only writer for every attempt while every other write queues behind it. + * + * Both failure modes below were seen in production within an hour of the queue + * being switched on: a constraint violation retried three times to no possible + * effect, and a timing-out job that held the writer for over a minute. + */ + +/** @param {(statements: unknown[]) => Promise} batch */ +const clientWith = (batch) => ({ batch: (statements) => batch(statements) }); + +test('a constraint violation is not retried', async () => { + // Deterministic: the same statements against the same data fail the same way + // for ever. This is the exact error that retried three times in production. + const client = clientWith(async () => { + throw new Error('SQLITE_CONSTRAINT: SQLite error: UNIQUE constraint failed: authors.slug'); + }); + + await assert.rejects( + () => runWriteJob(client, [{ sql: 'insert into authors values (1)' }]), + (err) => { + assert.ok(err instanceof UnrecoverableError, 'must tell BullMQ to stop retrying'); + assert.match(err.message, /authors\.slug/, 'the reason must survive'); + return true; + }, + ); +}); + +test('a syntax error is not retried either', async () => { + const client = clientWith(async () => { + throw new Error('SQLITE_ERROR: no such column: nope'); + }); + + await assert.rejects( + () => runWriteJob(client, [{ sql: 'select nope' }]), + (err) => err instanceof UnrecoverableError, + ); +}); + +test('a timeout IS retried, because the next attempt can differ', async () => { + // The distinction that matters. Refusing to retry these would throw away the + // main thing the queue was wanted for. + const client = clientWith(async () => { + throw new Error('The operation was aborted due to timeout'); + }); + + await assert.rejects( + () => runWriteJob(client, [{ sql: 'update feeds set x = 1' }]), + (err) => { + assert.ok(!(err instanceof UnrecoverableError), 'a timeout must stay retryable'); + assert.match(err.message, /timeout/); + return true; + }, + ); +}); + +test('a transport failure is retried', async () => { + const client = clientWith(async () => { + throw new Error('fetch failed'); + }); + + await assert.rejects( + () => runWriteJob(client, [{ sql: 'update feeds set x = 1' }]), + (err) => !(err instanceof UnrecoverableError), + ); +}); + +test('an empty job does no work and touches no client', async () => { + let called = false; + const client = clientWith(async () => { + called = true; + return []; + }); + + assert.deepEqual(await runWriteJob(client, []), []); + assert.equal(called, false, 'an empty job must not take the writer at all'); +}); + +test('a successful job returns one encoded result per statement', async () => { + const client = clientWith(async (statements) => + statements.map(() => ({ rows: [], rowsAffected: 1, columns: [], columnTypes: [] })), + ); + + const results = await runWriteJob(client, [{ sql: 'a' }, { sql: 'b' }]); + assert.equal(results.length, 2); +});