refactor(scripts): pass waiting-children as key in addParentJob (#3512)

Author: roggervalf

python/bullmq/flow_producer.py

@@ -91,7 +91,6 @@ async def addNode(self, node: dict, parent: dict, queues_opts: dict, pipe):
 
             await self.scripts.addParentJob(
                 job,
-                wait_children_key,
                 pipe
             )
 

python/bullmq/scripts.py

@@ -34,7 +34,7 @@ def __init__(self, prefix: str, queueName: str, redisConnection: RedisConnection
         self.commands = {
             "addStandardJob": self.redisClient.register_script(self.getScript("addStandardJob-9.lua")),
             "addDelayedJob": self.redisClient.register_script(self.getScript("addDelayedJob-6.lua")),
-            "addParentJob": self.redisClient.register_script(self.getScript("addParentJob-5.lua")),
+            "addParentJob": self.redisClient.register_script(self.getScript("addParentJob-6.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")),
@@ -86,7 +86,7 @@ def mapKey(key):
             return self.keys[key]
         return list(map(mapKey, keys))
 
-    def addJobArgs(self, job: Job, waiting_children_key: str|None):
+    def addJobArgs(self, job: Job):
         #  We are still lacking some arguments here:
         #  ARGV[1] msgpacked arguments array
         #         [9]  repeat job key
@@ -99,7 +99,6 @@ def addJobArgs(self, job: Job, waiting_children_key: str|None):
 
         packedArgs = msgpack.packb(
             [self.keys[""], job.id or "", job.name, job.timestamp, job.parentKey,
-                waiting_children_key,
                 f"{parentKey}:dependencies" if parentKey else None, parent],use_bin_type=True)
 
         return [packedArgs, jsonData, packedOpts]
@@ -131,7 +130,7 @@ def addStandardJob(self, job: Job, timestamp: int, pipe = None):
         """
         keys = self.getKeys(['wait', 'paused', 'meta', 'id', 'completed',
                              'delayed', 'active', 'events', 'marker'])
-        args = self.addJobArgs(job, None)
+        args = self.addJobArgs(job)
         args.append(timestamp)
 
         return self.commands["addStandardJob"](keys=keys, args=args, client=pipe)
@@ -142,7 +141,7 @@ def addDelayedJob(self, job: Job, timestamp: int, pipe = None):
         """
         keys = self.getKeys(['marker', 'meta', 'id',
                             'delayed', 'completed', 'events'])
-        args = self.addJobArgs(job, None)
+        args = self.addJobArgs(job)
         args.append(timestamp)
 
         return self.commands["addDelayedJob"](keys=keys, args=args, client=pipe)
@@ -153,18 +152,18 @@ def addPrioritizedJob(self, job: Job, timestamp: int, pipe = None):
         """
         keys = self.getKeys(['marker', 'meta', 'id', 'prioritized',
                              'delayed', 'completed', 'active', 'events', 'pc'])
-        args = self.addJobArgs(job, None)
+        args = self.addJobArgs(job)
         args.append(timestamp)
 
         return self.commands["addPrioritizedJob"](keys=keys, args=args, client=pipe)
 
-    def addParentJob(self, job: Job, waiting_children_key: str, pipe = None):
+    def addParentJob(self, job: Job, pipe = None):
         """
         Add a job to the queue that is a parent
         """
-        keys = self.getKeys(['meta', 'id', 'delayed', 'completed', 'events'])
-        
-        args = self.addJobArgs(job, waiting_children_key)
+        keys = self.getKeys(['meta', 'id', 'delayed', 'waiting-children', 'completed', 'events'])
+
+        args = self.addJobArgs(job)
 
         return self.commands["addParentJob"](keys=keys, args=args, client=pipe)
 

src/classes/flow-producer.ts

@@ -371,14 +371,10 @@ export class FlowProducer extends EventEmitter {
           const queueKeysParent = new QueueKeys(
             node.prefix || this.opts.prefix,
           );
-          const waitChildrenKey = queueKeysParent.toKey(
-            node.queueName,
-            'waiting-children',
-          );
 
           await job.addJob(<Redis>(multi as unknown), {
             parentDependenciesKey: parent?.parentDependenciesKey,
-            waitChildrenKey,
+            addToWaitingChildren: true,
             parentKey,
           });
 

src/classes/scripts.ts

@@ -169,6 +169,7 @@ export class Scripts {
       queueKeys.meta,
       queueKeys.id,
       queueKeys.delayed,
+      queueKeys['waiting-children'],
       queueKeys.completed,
       queueKeys.events,
     ];
@@ -240,7 +241,6 @@ export class Scripts {
       job.name,
       job.timestamp,
       job.parentKey || null,
-      parentKeyOpts.waitChildrenKey || null,
       parentKeyOpts.parentDependenciesKey || null,
       parent,
       job.repeatJobKey,
@@ -270,7 +270,7 @@ export class Scripts {
 
     let result: string | number;
 
-    if (parentKeyOpts.waitChildrenKey) {
+    if (parentKeyOpts.addToWaitingChildren) {
       result = await this.addParentJob(client, job, encodedOpts, args);
     } else if (typeof opts.delay == 'number' && opts.delay > 0) {
       result = await this.addDelayedJob(client, job, encodedOpts, args);
@@ -746,8 +746,8 @@ export class Scripts {
     return typeof shouldRemove === 'object'
       ? shouldRemove
       : typeof shouldRemove === 'number'
-      ? { count: shouldRemove }
-      : { count: shouldRemove ? 0 : -1 };
+        ? { count: shouldRemove }
+        : { count: shouldRemove ? 0 : -1 };
   }
 
   async moveToFinished(

src/commands/addDelayedJob-6.lua

@@ -21,11 +21,10 @@
             [3]  name
             [4]  timestamp
             [5]  parentKey?
-          x [6]  waitChildrenKey key.
-            [7]  parent dependencies key.
-            [8]  parent? {id, queueKey}
-            [9]  repeat job key
-            [10] deduplication key
+            [6]  parent dependencies key.
+            [7]  parent? {id, queueKey}
+            [8]  repeat job key
+            [9] deduplication key
 
       ARGV[2] Json stringified job data
       ARGV[3] msgpacked options
@@ -50,9 +49,9 @@ local args = cmsgpack.unpack(ARGV[1])
 local data = ARGV[2]
 
 local parentKey = args[5]
-local parent = args[8]
-local repeatJobKey = args[9]
-local deduplicationKey = args[10]
+local parent = args[7]
+local repeatJobKey = args[8]
+local deduplicationKey = args[9]
 local parentData
 
 -- Includes
@@ -73,7 +72,7 @@ local jobCounter = rcall("INCR", idKey)
 local maxEvents = getOrSetMaxEvents(metaKey)
 local opts = cmsgpack.unpack(ARGV[3])
 
-local parentDependenciesKey = args[7]
+local parentDependenciesKey = args[6]
 local timestamp = args[4]
 
 if args[2] == "" then

src/commands/addParentJob-5.lua renamed to src/commands/addParentJob-6.lua

@@ -8,20 +8,20 @@
       KEYS[1] 'meta'
       KEYS[2] 'id'
       KEYS[3] 'delayed'
-      KEYS[4] 'completed'
-      KEYS[5] events stream key
+      KEYS[4] 'waiting-children'
+      KEYS[5] 'completed'
+      KEYS[6] events stream key
 
       ARGV[1] msgpacked arguments array
             [1]  key prefix,
             [2]  custom id (will not generate one automatically)
             [3]  name
             [4]  timestamp
             [5]  parentKey?
-            [6]  waitChildrenKey key.
-            [7]  parent dependencies key.
-            [8]  parent? {id, queueKey}
-            [9]  repeat job key
-            [10] deduplication key
+            [6]  parent dependencies key.
+            [7]  parent? {id, queueKey}
+            [8]  repeat job key
+            [9] deduplication key
 
       ARGV[2] Json stringified job data
       ARGV[3] msgpacked options
@@ -33,8 +33,8 @@
 local metaKey = KEYS[1]
 local idKey = KEYS[2]
 
-local completedKey = KEYS[4]
-local eventsKey = KEYS[5]
+local completedKey = KEYS[5]
+local eventsKey = KEYS[6]
 
 local jobId
 local jobIdKey
@@ -46,9 +46,9 @@ local data = ARGV[2]
 local opts = cmsgpack.unpack(ARGV[3])
 
 local parentKey = args[5]
-local parent = args[8]
-local repeatJobKey = args[9]
-local deduplicationKey = args[10]
+local parent = args[7]
+local repeatJobKey = args[8]
+local deduplicationKey = args[9]
 local parentData
 
 -- Includes
@@ -67,7 +67,7 @@ local jobCounter = rcall("INCR", idKey)
 
 local maxEvents = getOrSetMaxEvents(metaKey)
 
-local parentDependenciesKey = args[7]
+local parentDependenciesKey = args[6]
 local timestamp = args[4]
 if args[2] == "" then
     jobId = jobCounter
@@ -92,7 +92,7 @@ end
 storeJob(eventsKey, jobIdKey, jobId, args[3], ARGV[2], opts, timestamp,
          parentKey, parentData, repeatJobKey)
 
-local waitChildrenKey = args[6]
+local waitChildrenKey = KEYS[4]
 rcall("ZADD", waitChildrenKey, timestamp, jobId)
 rcall("XADD", eventsKey, "MAXLEN", "~", maxEvents, "*", "event",
       "waiting-children", "jobId", jobId)

src/commands/addPrioritizedJob-9.lua

@@ -21,11 +21,10 @@
             [3]  name
             [4]  timestamp
             [5]  parentKey?
-            [6]  waitChildrenKey key.
-            [7]  parent dependencies key.
-            [8]  parent? {id, queueKey}
-            [9]  repeat job key
-            [10] deduplication key
+            [6]  parent dependencies key.
+            [7]  parent? {id, queueKey}
+            [8]  repeat job key
+            [9] deduplication key
 
       ARGV[2] Json stringified job data
       ARGV[3] msgpacked options
@@ -53,9 +52,9 @@ local data = ARGV[2]
 local opts = cmsgpack.unpack(ARGV[3])
 
 local parentKey = args[5]
-local parent = args[8]
-local repeatJobKey = args[9]
-local deduplicationKey = args[10]
+local parent = args[7]
+local repeatJobKey = args[8]
+local deduplicationKey = args[9]
 local parentData
 
 -- Includes
@@ -76,7 +75,7 @@ local jobCounter = rcall("INCR", idKey)
 
 local maxEvents = getOrSetMaxEvents(metaKey)
 
-local parentDependenciesKey = args[7]
+local parentDependenciesKey = args[6]
 local timestamp = args[4]
 if args[2] == "" then
     jobId = jobCounter

src/commands/addStandardJob-9.lua

@@ -31,11 +31,10 @@
             [3]  name
             [4]  timestamp
             [5]  parentKey?
-            [6]  waitChildrenKey key.
-            [7]  parent dependencies key.
-            [8]  parent? {id, queueKey}
-            [9]  repeat job key
-            [10] deduplication key
+            [6]  parent dependencies key.
+            [7]  parent? {id, queueKey}
+            [8]  repeat job key
+            [9] deduplication key
 
       ARGV[2] Json stringified job data
       ARGV[3] msgpacked options
@@ -56,9 +55,9 @@ local data = ARGV[2]
 local opts = cmsgpack.unpack(ARGV[3])
 
 local parentKey = args[5]
-local parent = args[8]
-local repeatJobKey = args[9]
-local deduplicationKey = args[10]
+local parent = args[7]
+local repeatJobKey = args[8]
+local deduplicationKey = args[9]
 local parentData
 
 -- Includes
@@ -80,7 +79,7 @@ local jobCounter = rcall("INCR", KEYS[4])
 local metaKey = KEYS[3]
 local maxEvents = getOrSetMaxEvents(metaKey)
 
-local parentDependenciesKey = args[7]
+local parentDependenciesKey = args[6]
 local timestamp = args[4]
 if args[2] == "" then
     jobId = jobCounter

src/interfaces/parent.ts

@@ -21,7 +21,7 @@ export interface ParentKeys {
 }
 
 export type ParentKeyOpts = {
-  waitChildrenKey?: string;
+  addToWaitingChildren?: boolean;
   parentDependenciesKey?: string;
   parentKey?: string;
 };