Do not retry a write that cannot succeed, on a queue with one writer (#145)

Within an hour of the Redis write queue going live it retried two jobs that had
no chance:

  write-failed  UNIQUE constraint failed: authors.slug   attempts=1,2,3
  write-failed  aborted due to timeout  id=30  21:29:27, 21:29:58, 21:30:29

Worker concurrency is 1 and cannot be raised -- SQLite has one writer -- so a
retry is not merely wasted effort. The failing job holds **the cluster's only
writer** for each of its attempts while every other write queues behind it. Job
30 held it for over a minute.

A constraint or syntax error is deterministic: the same statements against the
same data fail identically for ever, so all three attempts were certain to fail
and the second and third bought nothing but blocking. Those now raise BullMQ's
`UnrecoverableError`, which stops the retries -- the caller learns at once and
the writer is released.

Transport failures and timeouts still retry. Those genuinely can go differently
on the next attempt, and retrying them is most of why the queue was wanted.
The classification is `isStatementError`, already used by the folder to decide
when splitting a combined transaction can isolate one bad caller.

The job body moves into an exported `runWriteJob` so the retry decision can be
tested against a fake client, without a broker. The BullMQ plumbing is not the
part with judgement in it.

Not addressed here: three attempts at a thirty-second timeout still costs the
writer ninety seconds. That is a real cost and a defensible policy -- shortening
it trades durability for throughput -- so it wants deciding rather than
smuggling into this change.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: ralyodio

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