feat(flow): support passing a chain option

package.json

@@ -80,7 +80,7 @@
     "@types/msgpack": "^0.0.31",
     "@types/node": "^12.20.25",
     "@types/semver": "^7.3.9",
-    "@types/sinon": "^7.5.2",
+    "@types/sinon": "^10.0.13",
     "@types/uuid": "^3.4.10",
     "@typescript-eslint/eslint-plugin": "^4.32.0",
     "@typescript-eslint/parser": "^5.33.0",
@@ -112,7 +112,7 @@
     "rimraf": "^3.0.2",
     "rrule": "^2.6.9",
     "semantic-release": "^19.0.3",
-    "sinon": "^15.1.0",
+    "sinon": "^18.0.1",
     "test-console": "^2.0.0",
     "ts-mocha": "^10.0.0",
     "ts-node": "^10.7.0",

python/bullmq/scripts.py

@@ -31,10 +31,10 @@ def __init__(self, prefix: str, queueName: str, redisConnection: RedisConnection
         self.redisConnection = redisConnection
         self.redisClient = redisConnection.conn
         self.commands = {
-            "addStandardJob": self.redisClient.register_script(self.getScript("addStandardJob-8.lua")),
-            "addDelayedJob": self.redisClient.register_script(self.getScript("addDelayedJob-6.lua")),
+            "addStandardJob": self.redisClient.register_script(self.getScript("addStandardJob-9.lua")),
+            "addDelayedJob": self.redisClient.register_script(self.getScript("addDelayedJob-7.lua")),
             "addParentJob": self.redisClient.register_script(self.getScript("addParentJob-4.lua")),
-            "addPrioritizedJob": self.redisClient.register_script(self.getScript("addPrioritizedJob-8.lua")),
+            "addPrioritizedJob": self.redisClient.register_script(self.getScript("addPrioritizedJob-9.lua")),
             "changePriority": self.redisClient.register_script(self.getScript("changePriority-7.lua")),
             "cleanJobsInSet": self.redisClient.register_script(self.getScript("cleanJobsInSet-3.lua")),
             "extendLock": self.redisClient.register_script(self.getScript("extendLock-2.lua")),
@@ -119,7 +119,7 @@ def addStandardJob(self, job: Job, timestamp: int, pipe = None):
         Add a standard job to the queue
         """
         keys = self.getKeys(['wait', 'paused', 'meta', 'id',
-                             'completed', 'active', 'events', 'marker'])
+                             'completed', 'active', 'events', 'marker', 'waiting-children'])
         args = self.addJobArgs(job, None)
         args.append(timestamp)
 
@@ -130,7 +130,7 @@ def addDelayedJob(self, job: Job, timestamp: int, pipe = None):
         Add a delayed job to the queue
         """
         keys = self.getKeys(['marker', 'meta', 'id',
-                            'delayed', 'completed', 'events'])
+                            'delayed', 'completed', 'events', 'waiting-children'])
         args = self.addJobArgs(job, None)
         args.append(timestamp)
 
@@ -141,7 +141,7 @@ def addPrioritizedJob(self, job: Job, timestamp: int, pipe = None):
         Add a prioritized job to the queue
         """
         keys = self.getKeys(['marker', 'meta', 'id',
-                            'prioritized', 'completed', 'active', 'events', 'pc'])
+                            'prioritized', 'completed', 'active', 'events', 'pc', 'waiting-children'])
         args = self.addJobArgs(job, None)
         args.append(timestamp)
 

src/classes/job.ts

@@ -30,6 +30,7 @@ import {
   removeUndefinedFields,
   optsAsJSON,
   optsFromJSON,
+  getChainKey,
 } from '../utils';
 import { Backoffs } from './backoffs';
 import { Scripts, raw2NextJobData } from './scripts';
@@ -148,6 +149,11 @@ export class Job<
    */
   deduplicationId?: string;
 
+  /**
+   * Chain key.
+   */
+  chainKey?: string;
+
   /**
    * Base repeat job key.
    */
@@ -217,6 +223,10 @@ export class Job<
       ? opts.deduplication.id
       : this.debounceId;
 
+    this.chainKey = opts.chain
+      ? getChainKey(opts.chain, queue.qualifiedName)
+      : undefined;
+
     this.toKey = queue.toKey.bind(queue);
     this.setScripts();
 
@@ -340,6 +350,10 @@ export class Job<
       job.repeatJobKey = json.rjk;
     }
 
+    if (json.chkey) {
+      job.chainKey = json.chkey;
+    }
+
     if (json.deid) {
       job.debounceId = json.deid;
       job.deduplicationId = json.deid;
@@ -447,6 +461,7 @@ export class Job<
       timestamp: this.timestamp,
       failedReason: JSON.stringify(this.failedReason),
       stacktrace: JSON.stringify(this.stacktrace),
+      chainKey: this.chainKey,
       debounceId: this.debounceId,
       deduplicationId: this.deduplicationId,
       repeatJobKey: this.repeatJobKey,
@@ -1062,7 +1077,7 @@ export class Job<
       this.queue.on('closing', onFailed);
 
       const removeListeners = () => {
-        clearInterval(timeout);
+        clearTimeout(timeout);
         queueEvents.removeListener(completedEvent, onCompleted);
         queueEvents.removeListener(failedEvent, onFailed);
         this.queue.removeListener('closing', onFailed);

src/classes/queue-keys.ts

@@ -25,6 +25,7 @@ export class QueueKeys {
       'pc', // priority counter key
       'marker', // marker key
       'de', // deduplication key
+      'ch', // chains key
     ].forEach(key => {
       keys[key] = this.toKey(name, key);
     });

src/classes/queue.ts

@@ -655,6 +655,17 @@ export class Queue<
     return !removed;
   }
 
+  /**
+   * Removes a chain key.
+   *
+   * @param id - identifier
+   */
+  async removeChainKey(id: string): Promise<number> {
+    const client = await this.client;
+
+    return client.del(`${this.keys.ch}:${id}`);
+  }
+
   /**
    * Removes a debounce key.
    * @deprecated use removeDeduplicationKey

src/classes/scripts.ts

@@ -99,6 +99,7 @@ export class Scripts {
       queueKeys.delayed,
       queueKeys.completed,
       queueKeys.events,
+      queueKeys['waiting-children'],
     ];
 
     keys.push(pack(args), job.data, encodedOpts);
@@ -121,6 +122,7 @@ export class Scripts {
       queueKeys.active,
       queueKeys.events,
       queueKeys.pc,
+      queueKeys['waiting-children'],
     ];
 
     keys.push(pack(args), job.data, encodedOpts);
@@ -161,6 +163,7 @@ export class Scripts {
       queueKeys.active,
       queueKeys.events,
       queueKeys.marker,
+      queueKeys['waiting-children'],
     ];
 
     keys.push(pack(args), job.data, encodedOpts);
@@ -176,9 +179,15 @@ export class Scripts {
   ): Promise<string> {
     const queueKeys = this.queue.keys;
 
-    const parent: Record<string, any> = job.parent
-      ? { ...job.parent, fpof: opts.fpof, rdof: opts.rdof, idof: opts.idof }
-      : null;
+    const parent: Record<string, any> =
+      job.parent || job.chainKey
+        ? {
+            ...(job.parent ? job.parent : {}),
+            fpof: opts.fpof,
+            rdof: opts.rdof,
+            idof: opts.idof,
+          }
+        : null;
 
     const args = [
       queueKeys[''],
@@ -191,6 +200,7 @@ export class Scripts {
       parent,
       job.repeatJobKey,
       job.deduplicationId ? `${queueKeys.de}:${job.deduplicationId}` : null,
+      job.chainKey ? job.chainKey : null,
     ];
 
     let encodedOpts;

src/classes/worker.ts

@@ -597,7 +597,7 @@ export class Worker<
         this.blockUntil = await this.waiting;
 
         if (this.blockUntil <= 0 || this.blockUntil - Date.now() < 1) {
-          return this.moveToActive(client, token, this.opts.name);
+          return await this.moveToActive(client, token, this.opts.name);
         }
       } catch (err) {
         // Swallow error if locally paused or closing since we did force a disconnection
@@ -620,7 +620,17 @@ export class Worker<
           this.abortDelayController,
         );
       }
-      return this.moveToActive(client, token, this.opts.name);
+      try {
+        return await this.moveToActive(client, token, this.opts.name);
+      } catch (err) {
+        // Swallow error if locally paused or closing since we did force a disconnection
+        if (
+          !(this.paused || this.closing) &&
+          isNotConnectionError(<Error>err)
+        ) {
+          throw err;
+        }
+      }
     }
   }
 

src/commands/addDelayedJob-6.lua → src/commands/addDelayedJob-7.lua

@@ -14,6 +14,7 @@
       KEYS[4] 'delayed'
       KEYS[5] 'completed'
       KEYS[6] events stream key
+      KEYS[7] 'waiting-children'
 
       ARGV[1] msgpacked arguments array
             [1]  key prefix,
@@ -26,6 +27,7 @@
             [8]  parent? {id, queueKey}
             [9]  repeat job key
             [10] deduplication key
+            [11] chain key
 
       ARGV[2] Json stringified job data
       ARGV[3] msgpacked options
@@ -53,20 +55,26 @@ local parentKey = args[5]
 local parent = args[8]
 local repeatJobKey = args[9]
 local deduplicationKey = args[10]
+local chainKey = args[11]
 local parentData
 
 -- Includes
 --- @include "includes/addDelayMarkerIfNeeded"
+--- @include "includes/addJobInWaitingChildren"
 --- @include "includes/deduplicateJob"
 --- @include "includes/getDelayedScore"
 --- @include "includes/getOrSetMaxEvents"
 --- @include "includes/handleDuplicatedJob"
 --- @include "includes/storeJob"
+--- @include "includes/updateParentInLastJobInChain"
+--- @include "includes/upsertChainKeyIfNeeded"
 
 if parentKey ~= nil then
-    if rcall("EXISTS", parentKey) ~= 1 then return -5 end
+  if rcall("EXISTS", parentKey) ~= 1 then return -5 end
+end
 
-    parentData = cjson.encode(parent)
+if parent ~= nil then
+  parentData = cjson.encode(parent)
 end
 
 local jobCounter = rcall("INCR", idKey)
@@ -99,21 +107,32 @@ end
 -- Store the job.
 local delay, priority = storeJob(eventsKey, jobIdKey, jobId, args[3], ARGV[2],
                                  opts, timestamp, parentKey, parentData,
-                                 repeatJobKey)
+                                 repeatJobKey, chainKey)
+
+local lastJobKeyInChain = upsertChainKeyIfNeeded(chainKey, jobIdKey)
+
+if lastJobKeyInChain then
+  updateParentInLastJobInChain(lastJobKeyInChain, jobIdKey, jobId)
 
-local score, delayedTimestamp = getDelayedScore(delayedKey, timestamp, tonumber(delay))
+  local waitingChildrenKey = KEYS[7]
+
+  addJobInWaitingChildren(waitingChildrenKey, jobIdKey .. ":dependencies", lastJobKeyInChain, jobId,
+    timestamp, eventsKey, maxEvents)
+else
+  local score, delayedTimestamp = getDelayedScore(delayedKey, timestamp, tonumber(delay))
 
-rcall("ZADD", delayedKey, score, jobId)
-rcall("XADD", eventsKey, "MAXLEN", "~", maxEvents, "*", "event", "delayed",
-      "jobId", jobId, "delay", delayedTimestamp)
+  rcall("ZADD", delayedKey, score, jobId)
+  rcall("XADD", eventsKey, "MAXLEN", "~", maxEvents, "*", "event", "delayed",
+        "jobId", jobId, "delay", delayedTimestamp)
 
--- mark that a delayed job is available
-local markerKey = KEYS[1]
-addDelayMarkerIfNeeded(markerKey, delayedKey)
+  -- mark that a delayed job is available
+  local markerKey = KEYS[1]
+  addDelayMarkerIfNeeded(markerKey, delayedKey)
 
--- Check if this job is a child of another job, if so add it to the parents dependencies
-if parentDependenciesKey ~= nil then
+  -- Check if this job is a child of another job, if so add it to the parents dependencies
+  if parentDependenciesKey ~= nil then
     rcall("SADD", parentDependenciesKey, jobIdKey)
+  end
 end
 
 return jobId .. "" -- convert to string

src/commands/addParentJob-4.lua

@@ -51,6 +51,7 @@ local deduplicationKey = args[10]
 local parentData
 
 -- Includes
+--- @include "includes/addJobInWaitingChildren"
 --- @include "includes/deduplicateJob"
 --- @include "includes/getOrSetMaxEvents"
 --- @include "includes/handleDuplicatedJob"
@@ -91,14 +92,9 @@ end
 storeJob(eventsKey, jobIdKey, jobId, args[3], ARGV[2], opts, timestamp,
          parentKey, parentData, repeatJobKey)
 
-local waitChildrenKey = args[6]
-rcall("ZADD", waitChildrenKey, timestamp, jobId)
-rcall("XADD", eventsKey, "MAXLEN", "~", maxEvents, "*", "event",
-      "waiting-children", "jobId", jobId)
+local waitingChildrenKey = args[6]
 
--- Check if this job is a child of another job, if so add it to the parents dependencies
-if parentDependenciesKey ~= nil then
-    rcall("SADD", parentDependenciesKey, jobIdKey)
-end
+addJobInWaitingChildren(waitingChildrenKey, parentDependenciesKey, jobIdKey, jobId,
+  timestamp, eventsKey, maxEvents)
 
 return jobId .. "" -- convert to string