fix(redis-worker): stop losing dead-lettered messages when the DLQ write fails

The message was removed from in-flight before the dead-letter entry was written,
and the write went through a pipeline whose per-command errors were never
inspected. A failed write therefore lost the message silently: gone from
in-flight, absent from the dead-letter queue, invisible to the reclaim loop.

Write the entry first and only complete the message once it lands. On failure
the message stays in-flight, so the reclaim loop picks it up and it is retried
rather than dropped. A persistently failing write now loops visibly instead of
discarding work.

Author: matt-aitken

packages/redis-worker/src/fair-queue/index.ts

@@ -1498,14 +1498,28 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
       originalTimestamp: storedMessage.timestamp,
     };
 
-    // Complete in visibility manager
-    await this.visibilityManager.complete(storedMessage.id, storedMessage.queueId);
-
     // Add to DLQ
     const pipeline = this.redis.pipeline();
     pipeline.zadd(dlqKey, dlqMessage.deadLetteredAt, storedMessage.id);
     pipeline.hset(dlqDataKey, storedMessage.id, JSON.stringify(dlqMessage));
-    await pipeline.exec();
+    const dlqResults = await pipeline.exec();
+
+    const dlqErrors = (dlqResults ?? [])
+      .map(([error]) => error)
+      .filter((error): error is Error => Boolean(error));
+
+    if (dlqErrors.length > 0) {
+      this.logger.error("Failed to write message to DLQ, leaving it in-flight to be reclaimed", {
+        messageId: storedMessage.id,
+        queueId: storedMessage.queueId,
+        tenantId: storedMessage.tenantId,
+        errors: dlqErrors.map((error) => error.message),
+      });
+      return;
+    }
+
+    // Complete in visibility manager
+    await this.visibilityManager.complete(storedMessage.id, storedMessage.queueId);
 
     this.telemetry.recordDLQ();
 

packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts

@@ -1517,6 +1517,81 @@ describe("FairQueue", () => {
       }
     );
 
+    redisTest(
+      "should keep a message in-flight when the DLQ write fails",
+      { timeout: 15000 },
+      async ({ redisOptions }) => {
+        const started: string[] = [];
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const scheduler = new DRRScheduler({
+          redis: redisOptions,
+          keys,
+          quantum: 10,
+          maxDeficit: 100,
+        });
+
+        const queue = new TestFairQueueHelper(redisOptions, keys, {
+          scheduler,
+          payloadSchema: TestPayloadSchema,
+          shardCount: 1,
+          consumerCount: 1,
+          consumerIntervalMs: 20,
+          visibilityTimeoutMs: 60000,
+          reclaimIntervalMs: 60000,
+          concurrencyGroups: [
+            {
+              name: "tenant",
+              extractGroupId: (q) => q.tenantId,
+              getLimit: async () => 5,
+              defaultLimit: 5,
+            },
+          ],
+          startConsumers: false,
+        });
+
+        const redis = createRedisClient(redisOptions);
+        const dlqKey = keys.deadLetterQueueKey("t1");
+
+        try {
+          await redis.set(dlqKey, "not-a-zset");
+
+          queue.onMessage(async (ctx) => {
+            started.push(ctx.message.payload.value);
+            await ctx.fail(new Error("boom"));
+          });
+
+          await queue.enqueue({
+            queueId: "tenant:t1:queue:q1",
+            tenantId: "t1",
+            payload: { value: "msg-0" },
+          });
+
+          queue.start();
+
+          await vi.waitFor(
+            () => {
+              expect(started).toHaveLength(1);
+            },
+            { timeout: 10000 }
+          );
+
+          await vi.waitFor(
+            async () => {
+              expect(await redis.scard(keys.concurrencyKey("tenant", "t1"))).toBe(0);
+            },
+            { timeout: 5000 }
+          );
+
+          expect(await redis.zcard(keys.inflightKey(0))).toBe(1);
+        } finally {
+          await redis.del(dlqKey);
+          await redis.quit();
+          await queue.close();
+        }
+      }
+    );
+
     redisTest(
       "should release the concurrency slot even when the reclaim requeue fails",
       { timeout: 15000 },