Do not retry a write that cannot succeed, on a queue with one writer

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<unknown[]>}
+ */
+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 },
   );
 }

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<unknown>} 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);
+});