Ensure cron jobs get rescheduled even if the handler throws an error

Author: ericallam

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

@@ -52,7 +52,7 @@ describe("Worker with cron", () => {
 
       await setTimeout(6_000);
 
-      expect(processedItems.length).toBe(2);
+      expect(processedItems.length).toBeGreaterThanOrEqual(2);
 
       const secondItem = processedItems[1];
       expect(secondItem?.timestamp).toBeGreaterThan(firstItem!.timestamp);
@@ -62,4 +62,69 @@ describe("Worker with cron", () => {
       await worker.stop();
     }
   );
+
+  redisTest(
+    "continues processing cron items even when job handler throws errors",
+    { timeout: 180_000 },
+    async ({ redisContainer }) => {
+      const processedItems: CronSchema[] = [];
+      let executionCount = 0;
+
+      const worker = new Worker({
+        name: "test-worker-error",
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        catalog: {
+          cronJob: {
+            cron: "*/3 * * * * *", // Every 3 seconds
+            schema: CronSchema,
+            visibilityTimeoutMs: 5000,
+            retry: { maxAttempts: 1 }, // Only try once to fail faster
+            jitter: 100,
+          },
+        },
+        jobs: {
+          cronJob: async ({ payload }) => {
+            executionCount++;
+            await setTimeout(30); // Simulate work
+
+            // Throw error on first and third execution
+            if (executionCount === 1 || executionCount === 3) {
+              throw new Error(`Simulated error on execution ${executionCount}`);
+            }
+
+            processedItems.push(payload);
+          },
+        },
+        concurrency: {
+          workers: 2,
+          tasksPerWorker: 3,
+        },
+        logger: new Logger("test", "debug"),
+      }).start();
+
+      // Wait long enough for 4 executions (12 seconds + buffer)
+      await setTimeout(14_000);
+
+      // Should have at least 4 executions total
+      expect(executionCount).toBeGreaterThanOrEqual(4);
+
+      // Should have 2 successful items (executions 2 and 4)
+      expect(processedItems.length).toBeGreaterThanOrEqual(2);
+
+      // Verify that some executions failed (execution count > successful count)
+      // This proves that errors occurred but cron scheduling continued
+      expect(executionCount).toBeGreaterThan(processedItems.length);
+
+      // Verify that successful executions still have correct structure
+      const firstSuccessful = processedItems[0];
+      expect(firstSuccessful?.timestamp).toBeGreaterThan(0);
+      expect(firstSuccessful?.cron).toBe("*/3 * * * * *");
+
+      await worker.stop();
+    }
+  );
 });

packages/redis-worker/src/worker.ts

@@ -590,7 +590,13 @@ class Worker<TCatalog extends WorkerCatalog> {
             attempt: newAttempt,
             errorMessage,
           });
+
           await this.queue.moveToDeadLetterQueue(id, errorMessage);
+
+          if (catalogItem.cron) {
+            await this.rescheduleCronJob(job, catalogItem, item);
+          }
+
           return;
         }