feat(job): add exclusive execution option

src/classes/job.ts

@@ -38,6 +38,7 @@ import type { QueueEvents } from './queue-events';
 const logger = debuglog('bull');
 
 const optsDecodeMap = {
+  ee: 'exclusiveExecution',
   fpof: 'failParentOnFailure',
   idof: 'ignoreDependencyOnFailure',
   kl: 'keepLogs',
@@ -679,7 +680,7 @@ export class Job<
 
       if (delay === -1) {
         moveToFailed = true;
-      } else if (delay) {
+      } else if (delay && !this.opts.exclusiveExecution) {
         const args = this.scripts.moveToDelayedArgs(
           this.id,
           Date.now() + delay,
@@ -691,7 +692,10 @@ export class Job<
       } else {
         // Retry immediately
         (<any>multi).retryJob(
-          this.scripts.retryJobArgs(this.id, this.opts.lifo, token),
+          this.scripts.retryJobArgs(this.id, this.opts.lifo, token, {
+            exclusiveExecution: this.opts.exclusiveExecution,
+            pttl: delay,
+          }),
         );
         command = 'retryJob';
       }

src/classes/scripts.ts

@@ -23,6 +23,7 @@ import {
   WorkerOptions,
   KeepJobs,
   MoveToDelayedOpts,
+  RetryOpts,
 } from '../interfaces';
 import {
   JobState,
@@ -907,6 +908,7 @@ export class Scripts {
     jobId: string,
     lifo: boolean,
     token: string,
+    opts: RetryOpts,
   ): (string | number)[] {
     const keys: (string | number)[] = [
       this.queue.keys.active,
@@ -918,6 +920,7 @@ export class Scripts {
       this.queue.keys.delayed,
       this.queue.keys.prioritized,
       this.queue.keys.pc,
+      this.queue.keys.limiter,
       this.queue.keys.marker,
     ];
 
@@ -929,6 +932,8 @@ export class Scripts {
       pushCmd,
       jobId,
       token,
+      opts.exclusiveExecution ? '1' : '0',
+      opts.pttl && opts.pttl > 0 ? opts.pttl : 0,
     ]);
   }
 

src/commands/moveToDelayed-7.lua

@@ -56,7 +56,7 @@ if rcall("EXISTS", jobKey) == 1 then
     if ARGV[7] == "0" then
         rcall("HINCRBY", jobKey, "atm", 1)
     end
-    
+
     rcall("HSET", jobKey, "delay", ARGV[6])
 
     local maxEvents = getOrSetMaxEvents(metaKey)

src/commands/retryJob-10.lua → src/commands/retryJob-11.lua

@@ -11,13 +11,16 @@
       KEYS[7]  delayed key
       KEYS[8]  prioritized key
       KEYS[9]  'pc' priority counter
-      KEYS[10] 'marker'
+      KEYS[10] limiter key
+      KEYS[11] 'marker'
 
       ARGV[1]  key prefix
       ARGV[2]  timestamp
-      ARGV[3]  pushCmd
+      ARGV[3]  pushCmd - lifo -> RPUSH - fifo -> LPUSH
       ARGV[4]  jobId
       ARGV[5]  token
+      ARGV[6]  exclusive execution
+      ARGV[7]  rate limit pttl
 
     Events:
       'waiting'
@@ -36,7 +39,7 @@ local rcall = redis.call
 --- @include "includes/promoteDelayedJobs"
 
 local target, paused = getTargetQueueList(KEYS[5], KEYS[2], KEYS[3])
-local markerKey = KEYS[10]
+local markerKey = KEYS[11]
 
 -- Check if there are delayed jobs that we can move to wait.
 -- test example: when there are delayed jobs between retries
@@ -67,6 +70,13 @@ if rcall("EXISTS", KEYS[4]) == 1 then
 
   rcall("HINCRBY", KEYS[4], "atm", 1)
 
+  if ARGV[6] == "1" then
+    local pttl = tonumber(ARGV[7])
+    if pttl > 0 then
+      rcall("SET", KEYS[10], 999999, "PX", pttl)
+    end
+  end
+
   local maxEvents = getOrSetMaxEvents(KEYS[5])
 
   -- Emit waiting event

src/interfaces/minimal-job.ts

@@ -8,6 +8,11 @@ export interface MoveToDelayedOpts {
   skipAttempt?: boolean;
 }
 
+export interface RetryOpts {
+  exclusiveExecution?: boolean;
+  pttl?: number;
+}
+
 export interface MoveToWaitingChildrenOpts {
   child?: {
     id: string;

src/types/job-options.ts

@@ -1,6 +1,12 @@
 import { BaseJobOptions } from '../interfaces';
 
 export type JobsOptions = BaseJobOptions & {
+  /**
+   * If true, it will rate limit the queue when moving this job into delayed.
+   * Will stop rate limiting the queue until this job is moved to completed or failed.
+   */
+  exclusiveExecution?: boolean;
+
   /**
    * If true, moves parent to failed.
    */
@@ -21,6 +27,12 @@ export type JobsOptions = BaseJobOptions & {
  * These fields are the ones stored in Redis with smaller keys for compactness.
  */
 export type RedisJobOptions = BaseJobOptions & {
+  /**
+   * If true, it will rate limit the queue when moving this job into delayed.
+   * Will stop rate limiting the queue until this job is moved to completed or failed.
+   */
+  ee?: boolean;
+
   /**
    * If true, moves parent to failed.
    */

tests/test_delay.ts

@@ -222,6 +222,57 @@ describe('Delayed jobs', function () {
     await worker.close();
   });
 
+  describe('when exclusive execution is provided', function () {
+    it('should process delayed jobs waiting to be finished in correct order ', async function () {
+      this.timeout(4000);
+      const numJobs = 12;
+
+      const worker = new Worker(queueName, async (job: Job) => {}, {
+        autorun: false,
+        connection,
+        prefix,
+      });
+
+      worker.on('failed', function (job, err) {});
+
+      const orderList: number[] = [];
+      let count = 0;
+      const completed = new Promise<void>((resolve, reject) => {
+        worker.on('completed', async function (job) {
+          try {
+            count++;
+            orderList.push(job.data.order as number);
+            if (count == numJobs) {resolve();}
+          } catch (err) {
+            reject(err);
+          }
+        });
+      });
+
+      const jobs = Array.from(Array(numJobs).keys()).map(index => ({
+        name: 'test',
+        data: { order: numJobs - index },
+        opts: {
+          delay: (numJobs - index) * 150,
+          attempts: 1,
+          backoff: { type: 'fixed', delay: 200 },
+          exclusiveExecution: true,
+        },
+      }));
+      const expectedOrder = Array.from(Array(numJobs).keys()).map(
+        index => index + 1,
+      );
+
+      await queue.addBulk(jobs);
+      worker.run();
+      await completed;
+
+      expect(orderList).to.eql(expectedOrder);
+
+      await worker.close();
+    });
+  });
+
   it('should process delayed jobs with several workers respecting delay', async function () {
     this.timeout(30000);
     let count = 0;