Redis pubsub to redrive from the worker

Author: matt-aitken

internal-packages/redis-worker/src/worker.test.ts

@@ -4,6 +4,7 @@ import { describe } from "node:test";
 import { expect } from "vitest";
 import { z } from "zod";
 import { Worker } from "./worker.js";
+import Redis from "ioredis";
 
 describe("Worker", () => {
   redisTest("Process items that don't throw", { timeout: 30_000 }, async ({ redisContainer }) => {
@@ -191,6 +192,89 @@ describe("Worker", () => {
     }
   );
 
+  redisTest(
+    "Redrive an item from DLQ and process it successfully",
+    { timeout: 30_000 },
+    async ({ redisContainer }) => {
+      const processedItems: number[] = [];
+      const failedItemId = "fail-then-redrive-item";
+      let attemptCount = 0;
+
+      const worker = new Worker({
+        name: "test-worker",
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        catalog: {
+          testJob: {
+            schema: z.object({ value: z.number() }),
+            visibilityTimeoutMs: 1000,
+            retry: { maxAttempts: 3, minTimeoutInMs: 10, maxTimeoutInMs: 50 },
+          },
+        },
+        jobs: {
+          testJob: async ({ id, payload }) => {
+            if (id === failedItemId && attemptCount < 3) {
+              attemptCount++;
+              throw new Error("Temporary failure");
+            }
+            processedItems.push(payload.value);
+          },
+        },
+        concurrency: {
+          workers: 1,
+          tasksPerWorker: 1,
+        },
+        pollIntervalMs: 50,
+        logger: new Logger("test", "error"),
+      });
+
+      try {
+        // Enqueue the item that will fail 3 times
+        await worker.enqueue({
+          id: failedItemId,
+          job: "testJob",
+          payload: { value: 999 },
+        });
+
+        worker.start();
+
+        // Wait for the item to be processed and moved to DLQ
+        await new Promise((resolve) => setTimeout(resolve, 1000));
+
+        // Check that the item is in the DLQ
+        let dlqSize = await worker.queue.sizeOfDeadLetterQueue();
+        expect(dlqSize).toBe(1);
+
+        // Create a Redis client to publish the redrive message
+        const redisClient = new Redis({
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        });
+
+        // Publish redrive message
+        await redisClient.publish("test-worker:redrive", JSON.stringify({ id: failedItemId }));
+
+        // Wait for the item to be redrived and processed
+        await new Promise((resolve) => setTimeout(resolve, 1000));
+
+        // Check that the item was processed successfully
+        expect(processedItems).toEqual([999]);
+
+        // Check that the DLQ is now empty
+        dlqSize = await worker.queue.sizeOfDeadLetterQueue();
+        expect(dlqSize).toBe(0);
+
+        await redisClient.quit();
+      } finally {
+        worker.stop();
+      }
+    }
+  );
+
   //todo test that throwing an error doesn't screw up the other items
   //todo process more items when finished
 

internal-packages/redis-worker/src/worker.ts

@@ -7,6 +7,8 @@ import { Worker as NodeWorker } from "worker_threads";
 import { z } from "zod";
 import { SimpleQueue } from "./queue.js";
 
+import Redis from "ioredis";
+
 type WorkerCatalog = {
   [key: string]: {
     schema: z.ZodFirstPartySchemaTypes | z.ZodDiscriminatedUnion<any, any>;
@@ -42,6 +44,8 @@ type WorkerOptions<TCatalog extends WorkerCatalog> = {
 };
 
 class Worker<TCatalog extends WorkerCatalog> {
+  private subscriber: Redis;
+
   queue: SimpleQueue<QueueCatalogFromWorkerCatalog<TCatalog>>;
   private jobs: WorkerOptions<TCatalog>["jobs"];
   private logger: Logger;
@@ -55,7 +59,7 @@ class Worker<TCatalog extends WorkerCatalog> {
     const schema: QueueCatalogFromWorkerCatalog<TCatalog> = Object.fromEntries(
       Object.entries(this.options.catalog).map(([key, value]) => [key, value.schema])
     ) as QueueCatalogFromWorkerCatalog<TCatalog>;
-
+    //
     this.queue = new SimpleQueue({
       name: options.name,
       redisOptions: options.redisOptions,
@@ -74,6 +78,9 @@ class Worker<TCatalog extends WorkerCatalog> {
     }
 
     this.setupShutdownHandlers();
+
+    this.subscriber = new Redis(options.redisOptions);
+    this.setupSubscriber();
   }
 
   enqueue<K extends keyof TCatalog>({
@@ -240,6 +247,32 @@ class Worker<TCatalog extends WorkerCatalog> {
     this.processItems(worker, count);
   }
 
+  private setupSubscriber() {
+    const channel = `${this.options.name}:redrive`;
+    this.subscriber.subscribe(channel, (err) => {
+      if (err) {
+        this.logger.error(`Failed to subscribe to ${channel}`, { error: err });
+      } else {
+        this.logger.log(`Subscribed to ${channel}`);
+      }
+    });
+
+    this.subscriber.on("message", this.handleRedriveMessage.bind(this));
+  }
+
+  private async handleRedriveMessage(channel: string, message: string) {
+    try {
+      const { id } = JSON.parse(message);
+      if (typeof id !== "string") {
+        throw new Error("Invalid message format: id must be a string");
+      }
+      await this.queue.redriveFromDeadLetterQueue(id);
+      this.logger.log(`Redrived item ${id} from Dead Letter Queue`);
+    } catch (error) {
+      this.logger.error("Error processing redrive message", { error, message });
+    }
+  }
+
   private setupShutdownHandlers() {
     process.on("SIGTERM", this.shutdown.bind(this));
     process.on("SIGINT", this.shutdown.bind(this));
@@ -254,8 +287,10 @@ class Worker<TCatalog extends WorkerCatalog> {
       worker.terminate();
     }
 
+    await this.subscriber.unsubscribe();
+    await this.subscriber.quit();
     await this.queue.close();
-    this.logger.log("All workers shut down.");
+    this.logger.log("All workers and subscribers shut down.");
   }
 
   public start() {