feat: Add an option to not keep failed jobs

hoarder-app · Nov 9, 2024 · 6bfc81e · 6bfc81e
1 parent 7681d18
commit 6bfc81e
Show file tree

Hide file tree

Showing 4 changed files with 107 additions and 27 deletions.
diff --git a/src/options.ts b/src/options.ts
@@ -6,6 +6,7 @@ export interface SqliteQueueOptions {
   defaultJobArgs: {
     numRetries: number;
   };
+  keepFailedJobs: boolean;
 }
 
 export interface EnqueueOptions {

diff --git a/src/queue.test.ts b/src/queue.test.ts
@@ -0,0 +1,98 @@
+/* eslint-disable @typescript-eslint/require-await */
+import { describe, expect, test } from "vitest";
+
+import {
+    buildDBClient,
+    SqliteQueue
+} from "./";
+
+interface Work {
+  increment: number;
+  succeedAfter?: number;
+  blockForSec?: number;
+}
+
+describe("SqliteQueue", () => {
+  test("idempotency keys", async () => {
+    const queue = new SqliteQueue<Work>(
+      "queue1",
+      buildDBClient(":memory:", true),
+      {
+        defaultJobArgs: {
+          numRetries: 0,
+        },
+        keepFailedJobs: false,
+      },
+    );
+
+    await queue.enqueue({ increment: 1 });
+    await queue.enqueue({ increment: 2 }, { idempotencyKey: "2" });
+    await queue.enqueue({ increment: 2 }, { idempotencyKey: "2" });
+    await queue.enqueue({ increment: 2 }, { idempotencyKey: "2" });
+    await queue.enqueue({ increment: 3 }, { idempotencyKey: "3" });
+
+    expect(await queue.stats()).toEqual({
+      pending: 3,
+      running: 0,
+      pending_retry: 0,
+      failed: 0,
+    });
+
+  });
+
+  test("keep failed jobs", async () => {
+    const queueKeep = new SqliteQueue<Work>(
+      "queue1",
+      buildDBClient(":memory:", true),
+      {
+        defaultJobArgs: {
+          numRetries: 0,
+        },
+        keepFailedJobs: true,
+      },
+    );
+
+    const queueDontKeep = new SqliteQueue<Work>(
+      "queue2",
+      buildDBClient(":memory:", true),
+      {
+        defaultJobArgs: {
+          numRetries: 0,
+        },
+        keepFailedJobs: false,
+      },
+    );
+
+    const job1 = await queueKeep.enqueue({ increment: 1 });
+    const job2 = await queueDontKeep.enqueue({ increment: 1 });
+
+    expect(await queueKeep.stats()).toEqual({
+      pending: 1,
+      running: 0,
+      pending_retry: 0,
+      failed: 0,
+    });
+    expect(await queueDontKeep.stats()).toEqual({
+      pending: 1,
+      running: 0,
+      pending_retry: 0,
+      failed: 0,
+    });
+
+    queueKeep.finalize(job1!.id, job1!.allocationId, "failed");
+    queueDontKeep.finalize(job2!.id, job2!.allocationId, "failed");
+
+    expect(await queueKeep.stats()).toEqual({
+      pending: 0,
+      running: 0,
+      pending_retry: 0,
+      failed: 1,
+    });
+    expect(await queueDontKeep.stats()).toEqual({
+      pending: 0,
+      running: 0,
+      pending_retry: 0,
+      failed: 0,
+    });
+  });
+});
diff --git a/src/queue.ts b/src/queue.ts
@@ -136,7 +136,7 @@ export class SqliteQueue<T> {
     alloctionId: string,
     status: "completed" | "pending_retry" | "failed",
   ) {
-    if (status == "completed") {
+    if (status == "completed" || (status == "failed" && !this.options.keepFailedJobs)) {
       await this.db
         .delete(tasksTable)
         .where(

diff --git a/src/runner.test.ts b/src/runner.test.ts
@@ -149,6 +149,7 @@ describe("SqiteQueueRunner", () => {
         defaultJobArgs: {
           numRetries: 0,
         },
+        keepFailedJobs: true,
       },
     );
 
@@ -210,6 +211,7 @@ describe("SqiteQueueRunner", () => {
         defaultJobArgs: {
           numRetries: 2,
         },
+        keepFailedJobs: true,
       },
     );
 
@@ -247,6 +249,7 @@ describe("SqiteQueueRunner", () => {
         defaultJobArgs: {
           numRetries: 1,
         },
+        keepFailedJobs: true,
       },
     );
 
@@ -282,6 +285,7 @@ describe("SqiteQueueRunner", () => {
         defaultJobArgs: {
           numRetries: 1,
         },
+        keepFailedJobs: true,
       },
     );
 
@@ -326,6 +330,7 @@ describe("SqiteQueueRunner", () => {
         defaultJobArgs: {
           numRetries: 0,
         },
+        keepFailedJobs: true,
       },
     );
 
@@ -387,11 +392,13 @@ describe("SqiteQueueRunner", () => {
       defaultJobArgs: {
         numRetries: 0,
       },
+      keepFailedJobs: true,
     });
     const queue2 = new SqliteQueue<Work>("queue2", db, {
       defaultJobArgs: {
         numRetries: 0,
       },
+      keepFailedJobs: true,
     });
 
     const barrier = new Barrier(0);
@@ -439,30 +446,4 @@ describe("SqiteQueueRunner", () => {
     expect(results.numCompleted).toEqual(1000);
     expect(results.numFailed).toEqual(0);
   });
-
-  test("idempotency keys", async () => {
-    const queue = new SqliteQueue<Work>(
-      "queue1",
-      buildDBClient(":memory:", true),
-      {
-        defaultJobArgs: {
-          numRetries: 0,
-        },
-      },
-    );
-
-    await queue.enqueue({ increment: 1 });
-    await queue.enqueue({ increment: 2 }, { idempotencyKey: "2" });
-    await queue.enqueue({ increment: 2 }, { idempotencyKey: "2" });
-    await queue.enqueue({ increment: 2 }, { idempotencyKey: "2" });
-    await queue.enqueue({ increment: 3 }, { idempotencyKey: "3" });
-
-    expect(await queue.stats()).toEqual({
-      pending: 3,
-      running: 0,
-      pending_retry: 0,
-      failed: 0,
-    });
-
-  });
 });