Repository: taskforcesh/bullmq
Pull request #473 (2021-04-16 01:13:09)

Consider waiting-children state when getState
We should also consider waiting-children when getting an state for a job. Also I was playing with getDependencies and I got an error so I tried to take the opportunity to fix it here.

manast: great thanks!.


Patch:
diff --git a/src/classes/flow.ts b/src/classes/flow.ts
index a03a712db8..ba71e99f04 100644
--- a/src/classes/flow.ts
+++ b/src/classes/flow.ts
@@ -88,7 +88,7 @@ export class Flow extends EventEmitter {
   }
 
   /**
-   * Add a node (job) of a flow to the queue. This method will recursivelly
+   * Add a node (job) of a flow to the queue. This method will recursively
    * add all its children as well. Note that a given job can potentially be
    * a parent and a child job at the same time depending on where it is located
    * in the tree hierarchy.
diff --git a/src/classes/job.ts b/src/classes/job.ts
index a6690af0ce..87458ad4c2 100644
--- a/src/classes/job.ts
+++ b/src/classes/job.ts
@@ -338,6 +338,10 @@ export class Job<T = any, R = any, N extends string = string> {
     return this.isInZSet('delayed');
   }
 
+  isWaitingChildren() {
+    return this.isInZSet('waiting-children');
+  }
+
   isActive() {
     return this.isInList('active');
   }
@@ -379,15 +383,15 @@ export class Job<T = any, R = any, N extends string = string> {
 
     const multi = client.multi();
 
-    await multi.smembers(this.toKey(`${this.id}:`));
-    await multi.hgetall(this.toKey(`${this.id}:`));
+    await multi.hgetall(this.toKey(`${this.id}:processed`));
+    await multi.smembers(this.toKey(`${this.id}:dependencies`));
 
     const [[err1, processed], [err2, unprocessed]] = (await multi.exec()) as [
       [null | Error, string[]],
       [null | Error, object],
     ];
 
-    return { processed, unprocessed: Object(unprocessed).keys() };
+    return { processed, unprocessed };
   }
 
   /**
diff --git a/src/classes/queue.ts b/src/classes/queue.ts
index 47ff032466..414b0589d8 100644
--- a/src/classes/queue.ts
+++ b/src/classes/queue.ts
@@ -221,7 +221,7 @@ export class Queue<
    * and can be slow for very large queues.
    *
    * @param { { force: boolean, count: number }} opts. Use force = true to force obliteration even
-   * with active jobs in the queue. Use count with the maximun number of deleted keys per iteration,
+   * with active jobs in the queue. Use count with the maximum number of deleted keys per iteration,
    * 1000 is the default.
    */
   async obliterate(opts?: { force?: boolean; count?: number }) {
diff --git a/src/classes/scripts.ts b/src/classes/scripts.ts
index 455192c682..34d3dfc5a4 100644
--- a/src/classes/scripts.ts
+++ b/src/classes/scripts.ts
@@ -309,6 +309,7 @@ export class Scripts {
       'active',
       'wait',
       'paused',
+      'waiting-children',
     ].map(function(key: string) {
       return queue.toKey(key);
     });
diff --git a/src/commands/getState-6.lua b/src/commands/getState-7.lua
similarity index 90%
rename from src/commands/getState-6.lua
rename to src/commands/getState-7.lua
index 71ea928d0a..315e2292e8 100644
--- a/src/commands/getState-6.lua
+++ b/src/commands/getState-7.lua
@@ -8,6 +8,7 @@
     KEYS[4] 'active' key
     KEYS[5] 'wait' key
     KEYS[6] 'paused' key
+    KEYS[7] waitChildrenKey key
 
     ARGV[1] job id
   Output:
@@ -54,4 +55,8 @@ if item_in_list(paused_items, ARGV[1]) ~= nil then
   return "waiting"
 end
 
+if redis.call("ZSCORE", KEYS[7], ARGV[1]) ~= false then
+  return "waiting-children"
+end
+
 return "unknown"
diff --git a/src/commands/getStateV2-6.lua b/src/commands/getStateV2-7.lua
similarity index 86%
rename from src/commands/getStateV2-6.lua
rename to src/commands/getStateV2-7.lua
index 8f4b4b2282..1f728b2674 100644
--- a/src/commands/getStateV2-6.lua
+++ b/src/commands/getStateV2-7.lua
@@ -8,6 +8,7 @@
     KEYS[4] 'active' key
     KEYS[5] 'wait' key
     KEYS[6] 'paused' key
+    KEYS[7] waitChildrenKey key
 
     ARGV[1] job id
   Output:
@@ -42,4 +43,8 @@ if redis.call("LPOS", KEYS[6] , ARGV[1]) ~= false then
   return "waiting"
 end
 
+if redis.call("ZSCORE", KEYS[7] , ARGV[1]) ~= false then
+  return "waiting-children"
+end
+
 return "unknown"


Test patch:
diff --git a/src/test/test_flow.ts b/src/test/test_flow.ts
index e4885798ac..8b5266825a 100644
--- a/src/test/test_flow.ts
+++ b/src/test/test_flow.ts
@@ -80,8 +80,10 @@ describe('flows', () => {
     expect(tree).to.have.property('job');
     expect(tree).to.have.property('children');
 
-    const { children } = tree;
+    const { children, job } = tree;
+    const parentState = await job.getState();
 
+    expect(parentState).to.be.eql('waiting-children');
     expect(children).to.have.length(3);
 
     expect(children[0].job.id).to.be.ok;
@@ -195,8 +197,10 @@ describe('flows', () => {
     expect(tree).to.have.property('job');
     expect(tree).to.have.property('children');
 
-    const { children } = tree;
+    const { children, job } = tree;
+    const isWaitingChildren = await job.isWaitingChildren();
 
+    expect(isWaitingChildren).to.be.true;
     expect(children).to.have.length(1);
 
     expect(children[0].job.id).to.be.ok;