JobLock option reEnqueue to skip trying job again

README.md

@@ -50,7 +50,7 @@ async function boot() {
     add: {
       plugins: ["JobLock"],
       pluginOptions: {
-        JobLock: {},
+        JobLock: { reEnqueue: true },
       },
       perform: async (a, b) => {
         await new Promise((resolve) => {

__tests__/plugins/jobLock.ts

@@ -18,6 +18,13 @@ const jobs = {
       return answer;
     },
   },
+  withoutReEnqueue: {
+    plugins: ["JobLock"],
+    pluginOptions: { JobLock: { reEnqueue: false } },
+    perform: async () => {
+      return "hi";
+    },
+  },
 };
 
 describe("plugins", () => {
@@ -204,6 +211,57 @@ describe("plugins", () => {
       worker2.start();
     });
 
+    test("can be configured not to re-enqueue a duplicate task", async (done) => {
+      let count = 0;
+      worker1 = new Worker(
+        {
+          connection: specHelper.cleanConnectionDetails(),
+          timeout: specHelper.timeout,
+          queues: [specHelper.queue],
+        },
+        jobs
+      );
+      worker2 = new Worker(
+        {
+          connection: specHelper.cleanConnectionDetails(),
+          timeout: specHelper.timeout,
+          queues: [specHelper.queue],
+        },
+        jobs
+      );
+
+      worker1.on("error", (error) => {
+        throw error;
+      });
+      worker2.on("error", (error) => {
+        throw error;
+      });
+
+      await worker1.connect();
+      await worker2.connect();
+
+      const onComplete = async () => {
+        count++;
+        expect(count).toBe(1);
+        await worker1.end();
+        await worker2.end();
+
+        const timestamps = await queue.timestamps();
+        expect(timestamps).toEqual([]);
+
+        done();
+      };
+
+      worker1.on("success", onComplete);
+      worker2.on("success", onComplete);
+
+      await queue.enqueue(specHelper.queue, "withoutReEnqueue");
+      await queue.enqueue(specHelper.queue, "withoutReEnqueue");
+
+      worker1.start();
+      worker2.start();
+    });
+
     test("will run 2 jobs with the different args at the same time", async (done) => {
       worker1 = new Worker(
         {

__tests__/utils/specHelper.ts

@@ -39,9 +39,7 @@ const SpecHelper = {
 
   cleanup: async function () {
     const keys = await this.redis.keys(this.namespace + "*");
-    if (keys.length > 0) {
-      await this.redis.del(keys);
-    }
+    if (keys.length > 0) await this.redis.del(keys);
   },
 
   disconnect: async function () {

src/plugins/JobLock.ts

@@ -25,7 +25,15 @@ export class JobLock extends Plugin {
     if (lockedByMe && lockedByMe.toString().toLowerCase() === "ok") {
       return true;
     } else {
-      await this.reEnqueue();
+      const options = this.job.pluginOptions;
+      const toReEnqueue = options.JobLock
+        ? options.JobLock.reEnqueue !== null &&
+          options.JobLock.reEnqueue !== undefined
+          ? options.JobLock.reEnqueue
+          : true
+        : true;
+
+      if (toReEnqueue) await this.reEnqueue();
       return false;
     }
   }