Process jobs in parallel with retrying

matt-aitken · matt-aitken · commit b8d73803de79 · 2024-10-04T18:28:39.000-07:00
diff --git a/internal-packages/redis-worker/src/worker.test.ts b/internal-packages/redis-worker/src/worker.test.ts
@@ -7,55 +7,120 @@ import { Logger } from "@trigger.dev/core/logger";
 import { SimpleQueue } from "./queue.js";
 
 describe("Worker", () => {
-  // Tests will be added here
-});
-
-redisTest("concurrency settings", { timeout: 30_000 }, async ({ redisContainer }) => {
-  const processedItems: number[] = [];
-  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: 5000,
-        retry: { maxAttempts: 3 },
+  redisTest("Process items that don't throw", { timeout: 30_000 }, async ({ redisContainer }) => {
+    const processedItems: number[] = [];
+    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: 5000,
+          retry: { maxAttempts: 3 },
+        },
+      },
+      jobs: {
+        testJob: async ({ payload }) => {
+          await new Promise((resolve) => setTimeout(resolve, 30)); // Simulate work
+          processedItems.push(payload.value);
+        },
       },
-    },
-    jobs: {
-      testJob: async ({ payload }) => {
-        await new Promise((resolve) => setTimeout(resolve, 30)); // Simulate work
-        processedItems.push(payload.value);
+      concurrency: {
+        workers: 2,
+        tasksPerWorker: 3,
       },
-    },
-    concurrency: {
-      workers: 2,
-      tasksPerWorker: 3,
-    },
-    logger: new Logger("test", "log"),
+      logger: new Logger("test", "log"),
+    });
+
+    // Enqueue 10 items
+    for (let i = 0; i < 10; i++) {
+      await worker.enqueue({
+        id: `item-${i}`,
+        job: "testJob",
+        payload: { value: i },
+        visibilityTimeoutMs: 5000,
+      });
+    }
+
+    worker.start();
+
+    // Wait for items to be processed
+    await new Promise((resolve) => setTimeout(resolve, 600));
+
+    worker.stop();
+
+    expect(processedItems.length).toBe(10);
+    expect(new Set(processedItems).size).toBe(10); // Ensure all items were processed uniquely
   });
 
-  // Enqueue 10 items
-  for (let i = 0; i < 10; i++) {
-    await worker.enqueue({
-      id: `item-${i}`,
-      job: "testJob",
-      payload: { value: i },
-      visibilityTimeoutMs: 5000,
-    });
-  }
+  redisTest(
+    "Process items that throw an error",
+    { timeout: 30_000 },
+    async ({ redisContainer }) => {
+      const processedItems: number[] = [];
+      const hadAttempt = new Set<string>();
 
-  worker.start();
+      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: 5000,
+            retry: { maxAttempts: 3, minDelayMs: 10 },
+          },
+        },
+        jobs: {
+          testJob: async ({ id, payload }) => {
+            if (!hadAttempt.has(id)) {
+              hadAttempt.add(id);
+              throw new Error("Test error");
+            }
 
-  // Wait for items to be processed
-  await new Promise((resolve) => setTimeout(resolve, 600));
+            await new Promise((resolve) => setTimeout(resolve, 30)); // Simulate work
+            processedItems.push(payload.value);
+          },
+        },
+        concurrency: {
+          workers: 2,
+          tasksPerWorker: 3,
+        },
+        pollIntervalMs: 50,
+        logger: new Logger("test", "error"),
+      });
 
-  worker.stop();
+      // Enqueue 10 items
+      for (let i = 0; i < 10; i++) {
+        await worker.enqueue({
+          id: `item-${i}`,
+          job: "testJob",
+          payload: { value: i },
+          visibilityTimeoutMs: 5000,
+        });
+      }
 
-  expect(processedItems.length).toBe(10);
-  expect(new Set(processedItems).size).toBe(10); // Ensure all items were processed uniquely
+      worker.start();
+
+      // Wait for items to be processed
+      await new Promise((resolve) => setTimeout(resolve, 500));
+
+      worker.stop();
+
+      expect(processedItems.length).toBe(10);
+      expect(new Set(processedItems).size).toBe(10); // Ensure all items were processed uniquely
+    }
+  );
 });
+
+//todo test throwing an error and that retrying works
+//todo test that throwing an error doesn't screw up the other items
+//todo change the processItems to be in parallel using Promise.allResolved
+//process more items when finished
diff --git a/internal-packages/redis-worker/src/worker.ts b/internal-packages/redis-worker/src/worker.ts
@@ -38,6 +38,7 @@ type WorkerOptions<TCatalog extends WorkerCatalog> = {
     workers?: number;
     tasksPerWorker?: number;
   };
+  pollIntervalMs?: number;
   logger?: Logger;
 };
 
@@ -48,14 +49,12 @@ class Worker<TCatalog extends WorkerCatalog> {
   private workers: NodeWorker[] = [];
   private isShuttingDown = false;
   private concurrency: Required<NonNullable<WorkerOptions<TCatalog>["concurrency"]>>;
-  private catalog: TCatalog;
 
-  constructor(options: WorkerOptions<TCatalog>) {
-    this.catalog = options.catalog;
+  constructor(private options: WorkerOptions<TCatalog>) {
     this.logger = options.logger ?? new Logger("Worker", "debug");
 
     const schema: QueueCatalogFromWorkerCatalog<TCatalog> = Object.fromEntries(
-      Object.entries(options.catalog).map(([key, value]) => [key, value.schema])
+      Object.entries(this.options.catalog).map(([key, value]) => [key, value.schema])
     ) as QueueCatalogFromWorkerCatalog<TCatalog>;
 
     this.queue = new SimpleQueue({
@@ -89,7 +88,7 @@ class Worker<TCatalog extends WorkerCatalog> {
     payload: z.infer<TCatalog[K]["schema"]>;
     visibilityTimeoutMs?: number;
   }) {
-    const timeout = visibilityTimeoutMs ?? this.catalog[job].visibilityTimeoutMs;
+    const timeout = visibilityTimeoutMs ?? this.options.catalog[job].visibilityTimeoutMs;
     return this.queue.enqueue({
       id,
       job,
@@ -139,34 +138,77 @@ class Worker<TCatalog extends WorkerCatalog> {
   private async processItems(worker: NodeWorker, count: number) {
     if (this.isShuttingDown) return;
 
+    const pollIntervalMs = this.options.pollIntervalMs ?? 1000;
+
     try {
       const items = await this.queue.dequeue(count);
       if (items.length === 0) {
-        setTimeout(() => this.processItems(worker, count), 1000); // Wait before trying again
+        setTimeout(() => this.processItems(worker, count), pollIntervalMs);
         return;
       }
 
-      worker.postMessage({ type: "process", items });
-
-      for (const { id, job, item, visibilityTimeoutMs } of items) {
-        const handler = this.jobs[job as any];
-        if (!handler) {
-          this.logger.error(`No handler found for job type: ${job as string}`);
-          continue;
-        }
-
-        try {
-          await handler({ id, payload: item, visibilityTimeoutMs });
-          await this.queue.ack(id);
-        } catch (error) {
-          this.logger.error(`Error processing item ${id}:`, { error });
-          // Here you might want to implement a retry mechanism or dead-letter queue
-        }
-      }
+      await Promise.all(
+        items.map(async ({ id, job, item, visibilityTimeoutMs }) => {
+          const catalogItem = this.options.catalog[job as any];
+          const handler = this.jobs[job as any];
+          if (!handler) {
+            this.logger.error(`No handler found for job type: ${job as string}`);
+            return;
+          }
+
+          try {
+            await handler({ id, payload: item, visibilityTimeoutMs });
+            await this.queue.ack(id);
+          } catch (error) {
+            this.logger.error(`Error processing item, it threw an error:`, {
+              name: this.options.name,
+              id,
+              job,
+              item,
+              visibilityTimeoutMs,
+              error,
+            });
+            // Requeue the failed item with a delay
+            try {
+              const retryDelay = catalogItem.retry.minDelayMs ?? 1_000;
+              const retryDate = new Date(Date.now() + retryDelay);
+              this.logger.info(`Requeued failed item ${id} with delay`, {
+                name: this.options.name,
+                id,
+                job,
+                item,
+                retryDate,
+                retryDelay,
+                visibilityTimeoutMs,
+              });
+              await this.queue.enqueue({
+                id,
+                job,
+                item,
+                availableAt: retryDate,
+                visibilityTimeoutMs,
+              });
+            } catch (requeueError) {
+              this.logger.error(`Failed to requeue item, threw error:`, {
+                name: this.options.name,
+                id,
+                job,
+                item,
+                visibilityTimeoutMs,
+                error: requeueError,
+              });
+            }
+          }
+        })
+      );
     } catch (error) {
-      this.logger.error("Error dequeuing items:", { error });
-      setTimeout(() => this.processItems(worker, count), 1000); // Wait before trying again
+      this.logger.error("Error dequeuing items:", { name: this.options.name, error });
+      setTimeout(() => this.processItems(worker, count), pollIntervalMs);
+      return;
     }
+
+    // Immediately process next batch because there were items in the queue
+    this.processItems(worker, count);
   }
 
   private setupShutdownHandlers() {
