feat(job): add exclusive execution option

docs/gitbook/SUMMARY.md

@@ -24,6 +24,7 @@
   * [Stalled Jobs](guide/workers/stalled-jobs.md)
   * [Sandboxed processors](guide/workers/sandboxed-processors.md)
   * [Pausing queues](guide/workers/pausing-queues.md)
+  * [Preserve Order](guide/workers/preserve-order.md)
 * [Jobs](guide/jobs/README.md)
   * [FIFO](guide/jobs/fifo.md)
   * [LIFO](guide/jobs/lifo.md)

docs/gitbook/guide/workers/preserve-order.md

@@ -0,0 +1,15 @@
+# Preserve Order
+
+BullMQ supports preserving execution order in jobs. When _preserveOrder_ option is provided as *true*, jobs will be processed in the same order, indendently of retry strategies. If the current job fails and has a retry strategy, queue will be in rate limit state until the delay is accomplish.
+
+```typescript
+const worker = new Worker('queueName', async (job: Job) => {
+    // do some work
+}, {
+    preserveOrder: true
+    });
+```
+
+{% hint style="warning" %}
+This feature is only allowed when using concurrency 1, any greater value will throw an error:
+{% endhint %}

python/bullmq/job.py

@@ -149,7 +149,7 @@ async def moveToFailed(self, err, token:str, fetchNext:bool = False):
                     )
                 if delay == -1:
                     move_to_failed = True
-                elif delay:
+                elif delay and not self.queue.opts.get("preserveOrder", False):
                     keys, args = self.scripts.moveToDelayedArgs(
                         self.id,
                         round(time.time() * 1000) + delay,
@@ -160,7 +160,10 @@ async def moveToFailed(self, err, token:str, fetchNext:bool = False):
                     await self.scripts.commands["moveToDelayed"](keys=keys, args=args, client=pipe)
                     command = 'delayed'
                 else:
-                    keys, args = self.scripts.retryJobArgs(self.id, self.opts.get("lifo", False), token)
+                    keys, args = self.scripts.retryJobArgs(self.id, self.opts.get("lifo", False), token, {
+                        "preserveOrder": self.queue.opts.get("preserveOrder", False),
+                        "pttl": delay
+                    })
 
                     await self.scripts.commands["retryJob"](keys=keys, args=args, client=pipe)
                     command = 'retryJob'

python/bullmq/scripts.py

@@ -53,7 +53,7 @@ def __init__(self, prefix: str, queueName: str, redisConnection: RedisConnection
             "promote": self.redisClient.register_script(self.getScript("promote-8.lua")),
             "removeJob": self.redisClient.register_script(self.getScript("removeJob-1.lua")),
             "reprocessJob": self.redisClient.register_script(self.getScript("reprocessJob-7.lua")),
-            "retryJob": self.redisClient.register_script(self.getScript("retryJob-10.lua")),
+            "retryJob": self.redisClient.register_script(self.getScript("retryJob-11.lua")),
             "moveJobsToWait": self.redisClient.register_script(self.getScript("moveJobsToWait-7.lua")),
             "saveStacktrace": self.redisClient.register_script(self.getScript("saveStacktrace-1.lua")),
             "updateData": self.redisClient.register_script(self.getScript("updateData-1.lua")),
@@ -250,12 +250,15 @@ def retryJobArgs(self, job_id: str, lifo: bool, token: str, opts: dict = {}):
         keys.append(self.keys['delayed'])
         keys.append(self.keys['prioritized'])
         keys.append(self.keys['pc'])
+        keys.append(self.keys['limiter'])
         keys.append(self.keys['marker'])
 
         push_cmd = "RPUSH" if lifo else "LPUSH"
 
+        pttl = opts.get("pttl", 0)
         args = [self.keys[''], round(time.time() * 1000), push_cmd,
-            job_id, token, "1" if opts.get("skipAttempt") else "0"]
+            job_id, token, "1" if opts.get("preserveOrder") else "0",
+            pttl if pttl > 0 else 0]
 
         return (keys, args)
 

src/classes/job.ts

@@ -683,7 +683,7 @@ export class Job<
 
       if (delay === -1) {
         moveToFailed = true;
-      } else if (delay) {
+      } else if (delay && !opts.preserveOrder) {
         const args = this.scripts.moveToDelayedArgs(
           this.id,
           Date.now() + delay,
@@ -695,7 +695,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, {
+            preserveOrder: opts.preserveOrder,
+            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,17 +920,20 @@ export class Scripts {
       this.queue.keys.delayed,
       this.queue.keys.prioritized,
       this.queue.keys.pc,
+      this.queue.keys.limiter,
       this.queue.keys.marker,
     ];
 
-    const pushCmd = (lifo ? 'R' : 'L') + 'PUSH';
+    const pushCmd = (lifo || opts.preserveOrder ? 'R' : 'L') + 'PUSH';
 
     return keys.concat([
       this.queue.toKey(''),
       Date.now(),
       pushCmd,
       jobId,
       token,
+      opts.preserveOrder ? '1' : '0',
+      opts.pttl && opts.pttl > 0 ? opts.pttl : 0,
     ]);
   }
 

src/classes/worker.ts

@@ -222,6 +222,7 @@ export class Worker<
       stalledInterval: 30000,
       autorun: true,
       runRetryDelay: 15000,
+      preserveOrder: false,
       ...this.opts,
     };
 
@@ -385,6 +386,11 @@ export class Worker<
     ) {
       throw new Error('concurrency must be a finite number greater than 0');
     }
+
+    if (this.opts.preserveOrder && concurrency > 1) {
+      throw new Error('concurrency must be 1 when preserveOrder is enabled');
+    }
+
     this.opts.concurrency = concurrency;
   }
 

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]  preserve order
+      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 {
+  preserveOrder?: boolean;
+  pttl?: number;
+}
+
 export interface MoveToWaitingChildrenOpts {
   child?: {
     id: string;

src/interfaces/worker-options.ts

@@ -58,6 +58,12 @@ export interface WorkerOptions extends QueueBaseOptions {
    */
   maxStalledCount?: number;
 
+  /**
+   * 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.
+   */
+  preserveOrder?: boolean;
+
   /**
    * Number of milliseconds between stallness checks.
    *

tests/test_delay.ts

@@ -301,6 +301,59 @@ describe('Delayed jobs', function () {
     await worker.close();
   });
 
+  describe('when preserveOrder 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,
+        preserveOrder: true,
+        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 },
+        },
+      }));
+      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;

tests/test_worker.ts

@@ -1988,6 +1988,21 @@ describe('workers', function () {
     ).to.throw('drainDelay must be greater than 0');
   });
 
+  describe('when preserveOrder is enabled', () => {
+    it('concurrency cannot be greater than 1', function () {
+      this.timeout(4000);
+      expect(
+        () =>
+          new Worker(queueName, async () => {}, {
+            connection,
+            concurrency: 2,
+            prefix,
+            preserveOrder: true,
+          }),
+      ).to.throw('concurrency must be 1 when preserveOrder is enabled');
+    });
+  });
+
   it('lock extender continues to run until all active jobs are completed when closing a worker', async function () {
     this.timeout(4000);
     let worker;