Repository: taskforcesh/bullmq
Pull request #445 (2021-03-31 02:14:17)

Refactor getState method
This is an enhancement for getState method, trying to not call redis per state check, we can just do it in one script #143 

ccollie: > This is an enhancement for getState method, trying to not call redis per state check, we can just do it in one script #143

Here's my take:

https://github.com/ccollie/toro-api/blob/master/src/server/commands/getJobState-6.lua

manast: Even faster but only works on redis 6.0.6+ https://redis.io/commands/lpos
roggervalf: Thank you guys, I updated this PR, now supporting lpos, I think it is easy to follow. What do you think?
ccollie: > Thank you guys, I updated this PR, now supporting lpos, I think it is easy to follow. What do you think?

I would suggest you do a version check using `INFO` and act accordingly. Looking at [GCP](https://cloud.google.com/memorystore/docs/redis/supported-versions#current_versions), they're currently at `5.x`

ccollie: > > Thank you guys, I updated this PR, now supporting lpos, I think it is easy to follow. What do you think?
> 
> I would suggest you do a version check using `INFO` and act accordingly. Looking at [GCP](https://cloud.google.com/memorystore/docs/redis/supported-versions#current_versions), they're currently at `5.x`

Nevermind. I see you did that in Typescript

roggervalf: @manast when you get a chance, please some :eyes: 

Patch:
diff --git a/README.md b/README.md
index a8069c8698..308a519f46 100644
--- a/README.md
+++ b/README.md
@@ -25,7 +25,7 @@
     </a>
   </p>
   <p>
-    <em>Follow <a href="https://twitter.com/manast">@manast</a> for Bull news and updates!</em>
+    <em>Follow <a href="https://twitter.com/manast">@manast</a> for *important* Bull/BullMQ news and updates!</em>
   </p>
 </div>
 
@@ -37,7 +37,7 @@ You can find tutorials and news in this blog: https://blog.taskforce.sh/
 
 [<img src="http://taskforce.sh/assets/logo_square.png" width="100" alt="Taskforce.sh, Inc" style="padding: 100px"/>](https://taskforce.sh)
 
-Super charge your queues with a profesional front end and optional Redis hosting:
+Supercharge your queues with a profesional front end and optional Redis hosting:
 - Get a complete overview of all your queues.
 - Inspect jobs, search, retry, or promote delayed jobs.
 - Metrics and statistics.
diff --git a/docs/gitbook/CHANGELOG.md b/docs/gitbook/CHANGELOG.md
deleted file mode 100644
index 32a0fd2361..0000000000
--- a/docs/gitbook/CHANGELOG.md
+++ /dev/null
@@ -1,69 +0,0 @@
-## [1.15.1](https://github.com/taskforcesh/bullmq/compare/v1.15.0...v1.15.1) (2021-03-19)
-
-
-### Bug Fixes
-
-* **obliterate:** safer implementation ([82f571f](https://github.com/taskforcesh/bullmq/commit/82f571f2548c61c776b897fd1c5050bb09c8afca))
-
-# [1.15.0](https://github.com/taskforcesh/bullmq/compare/v1.14.8...v1.15.0) (2021-03-18)
-
-
-### Features
-
-* add method to "obliterate" a queue, fixes [#430](https://github.com/taskforcesh/bullmq/issues/430) ([624be0e](https://github.com/taskforcesh/bullmq/commit/624be0ed48159c2aa405025938925a723330e0c2))
-
-## [1.14.8](https://github.com/taskforcesh/bullmq/compare/v1.14.7...v1.14.8) (2021-03-06)
-
-
-### Bug Fixes
-
-* specify promise type to make TS 4.1 and 4.2 happy. ([#418](https://github.com/taskforcesh/bullmq/issues/418)) ([702f609](https://github.com/taskforcesh/bullmq/commit/702f609b410d8b0652c2d0504a8a67526966fdc3))
-
-## [1.14.7](https://github.com/taskforcesh/bullmq/compare/v1.14.6...v1.14.7) (2021-02-16)
-
-
-### Bug Fixes
-
-* remove "client" property of QueueBaseOptions ([#324](https://github.com/taskforcesh/bullmq/issues/324)) ([e0b9e71](https://github.com/taskforcesh/bullmq/commit/e0b9e71c4da4a93af54c4386af461c61ab5f146c))
-
-## [1.14.6](https://github.com/taskforcesh/bullmq/compare/v1.14.5...v1.14.6) (2021-02-16)
-
-
-### Bug Fixes
-
-* remove next job in removeRepeatableByKey fixes [#165](https://github.com/taskforcesh/bullmq/issues/165) ([fb3a7c2](https://github.com/taskforcesh/bullmq/commit/fb3a7c2f429d535dd9f038687d7230d61201defc))
-
-## [1.14.5](https://github.com/taskforcesh/bullmq/compare/v1.14.4...v1.14.5) (2021-02-16)
-
-
-### Bug Fixes
-
-* add jobId support to repeatable jobs fixes [#396](https://github.com/taskforcesh/bullmq/issues/396) ([c2dc669](https://github.com/taskforcesh/bullmq/commit/c2dc6693a4546e547245bc7ec1e71b4841829619))
-
-## [1.14.4](https://github.com/taskforcesh/bullmq/compare/v1.14.3...v1.14.4) (2021-02-08)
-
-
-### Bug Fixes
-
-* reconnect at start fixes [#337](https://github.com/taskforcesh/bullmq/issues/337) ([fb33772](https://github.com/taskforcesh/bullmq/commit/fb3377280b3bda04a15a62d2901bdd78b869e08c))
-
-## [1.14.3](https://github.com/taskforcesh/bullmq/compare/v1.14.2...v1.14.3) (2021-02-07)
-
-
-### Bug Fixes
-
-* **worker:** avoid possible infinite loop fixes [#389](https://github.com/taskforcesh/bullmq/issues/389) ([d05566e](https://github.com/taskforcesh/bullmq/commit/d05566ec0153f31a1257f7338399fdb55c959487))
-
-## [1.14.2](https://github.com/taskforcesh/bullmq/compare/v1.14.1...v1.14.2) (2021-02-02)
-
-
-### Bug Fixes
-
-* improve job timeout notification by giving the job name and id in the error message ([#387](https://github.com/taskforcesh/bullmq/issues/387)) ([ca886b1](https://github.com/taskforcesh/bullmq/commit/ca886b1f854051aed0888f5b872a64b052b2383e))
-
-## [1.14.1](https://github.com/taskforcesh/bullmq/compare/v1.14.0...v1.14.1) (2021-02-01)
-
-
-### Bug Fixes
-
-* job finish queue events race condition ([355bca5](https://github.com/taskforcesh/bullmq/commit/355bca5ee128bf4ff37608746f9c6f7cca580eb0))
diff --git a/docs/gitbook/SUMMARY.md b/docs/gitbook/SUMMARY.md
index 4d785062b4..8f9bf74f32 100644
--- a/docs/gitbook/SUMMARY.md
+++ b/docs/gitbook/SUMMARY.md
@@ -22,6 +22,8 @@
   * [Delayed](guide/jobs/delayed.md)
   * [Repeatable](guide/jobs/repeatable.md)
   * [Prioritized](guide/jobs/proritized.md)
+  * [Adding bulks](guide/jobs/adding-bulks.md)
+  * [Parent - Child Jobs](guide/jobs/parent-child-jobs.md)
   * [Stalled](guide/jobs/stalled.md)
   * [Getters](guide/jobs/getters.md)
 * [Rate limiting](guide/rate-limiting.md)
diff --git a/docs/gitbook/guide/jobs/adding-bulks.md b/docs/gitbook/guide/jobs/adding-bulks.md
new file mode 100644
index 0000000000..1f4aefbc4d
--- /dev/null
+++ b/docs/gitbook/guide/jobs/adding-bulks.md
@@ -0,0 +1,22 @@
+# Adding bulks
+
+Sometimes it is necessary to add a complete bulk of jobs atomically. For example there could be a requirement that all the jobs must be placed in the queue or none of them. Also, adding a bulk of jobs can be faster since it reduces the roundtrips to Redis:
+
+```typescript
+import { Queue } from 'bullmq'
+
+const queue = new Queue('paint')
+
+const jobs = await queue.addBulk(
+  [
+    { name, data: { paint: 'car' } },
+    { name, data: { paint: 'house' } },
+    { name, data: { paint: 'boat' } },
+  ],
+);
+```
+
+This call can only succeed or fail, and all the or none of the jobs will be added.
+
+
+
diff --git a/docs/gitbook/guide/jobs/parent-child-jobs.md b/docs/gitbook/guide/jobs/parent-child-jobs.md
new file mode 100644
index 0000000000..9119afc86f
--- /dev/null
+++ b/docs/gitbook/guide/jobs/parent-child-jobs.md
@@ -0,0 +1,73 @@
+# Parent - Child Jobs
+
+BullMQ supports Parent - Child relationships between jobs. The basic idea is that a parent job will only start to be processed when all its child jobs have been processed successfully. A part from that, a parent or a child job are no different from regular jobs.
+
+{% hint style="warning" %}
+Currently, parent jobs can only be created using the "addBulk" method. But new methods for managing parents will be added soon.
+{% endhint %}
+
+A parent job can be created implicitly using the "addBulk" method. This method will atomically create all the children and the parent jobs. After this call, as soon as all the children are completed, the parent job will be moved to the wait status and processed as a regular job:
+
+```typescript
+import { Queue } from 'bullmq'
+
+const queue = new Queue('steps')
+
+const jobs = await queue.addBulk(
+  [
+    { name, data: { paint: 'ceiling' } },
+    { name, data: { paint: 'walls' } },
+    { name, data: { fix: 'floor' } },
+  ],
+  {
+    parent: {
+      name: 'renovate-interior',
+      queue: 'renovate-queue',
+      data: { billingInfo: 'foobar' },
+    },
+  },
+);
+```
+
+The above call will return instances for all the jobs added to the queue. Note that the parent queue does not need to be the same queue as the one used for the children.
+
+When the parent job is processed it is possible to access the results generated by its child jobs. For example lets assume the following worker for the child jobs:
+
+```typescript
+import { Worker } from "bullmq"
+
+const stepsQueue = new Worker('steps', async (job) => {
+  await performStep(job.data);
+  
+  if(job.data.paint) {
+    return 2500;
+  } else if( job.data.fix ) {
+    return 1750;
+  }
+});
+```
+
+we can implement a parent worker that sums the costs of the children's jobs using the "getChildrenValues" method:
+
+```typescript
+import { Worker } from "bullmq"
+
+const stepsQueue = new Worker('renovate-interior', async job => {
+  const childrenValues = job.getChildrenValues();
+
+  // childrenValues is an object mapping job keys with their values
+
+  const totalCosts = Object(childrenValues)
+    .values()
+    .reduce((prev, cur) => prev + cur, 0);
+
+  await sendInvoice(totalCosts);
+});
+```
+
+{% hint style="info" %}
+Parent-child jobs is a pretty new feature so expect more functionality landing in BullMQ on the following months.
+{% endhint %}
+
+
+
diff --git a/src/classes/flow.ts b/src/classes/flow.ts
new file mode 100644
index 0000000000..204ee02383
--- /dev/null
+++ b/src/classes/flow.ts
@@ -0,0 +1,210 @@
+import uuid = require('uuid');
+
+import { Redis, Pipeline } from 'ioredis';
+
+import { EventEmitter } from 'events';
+import { QueueBaseOptions } from '../interfaces';
+import { RedisConnection } from './redis-connection';
+import { KeysMap, QueueKeys } from './queue-keys';
+import { FlowJob } from '@src/interfaces/flow-job';
+import { Job } from './job';
+
+interface JobNode {
+  job: Job;
+  children?: JobNode[];
+}
+
+/**
+ * This class allows to add jobs into one or several queues
+ * with dependencies between them in such a way that it is possible
+ * to build complex flows.
+ */
+export class Flow extends EventEmitter {
+  toKey: (name: string, type: string) => string;
+  keys: KeysMap;
+  closing: Promise<void>;
+  queueKeys: QueueKeys;
+
+  protected connection: RedisConnection;
+
+  constructor(public opts: QueueBaseOptions = {}) {
+    super();
+
+    this.opts = {
+      prefix: 'bull',
+      ...opts,
+    };
+
+    this.connection = new RedisConnection(opts.connection);
+    this.connection.on('error', this.emit.bind(this, 'error'));
+
+    this.queueKeys = new QueueKeys(opts.prefix);
+  }
+
+  /**
+   * Adds a flow.
+   *
+   * A flow is a tree-like structure of jobs that depend on each other.
+   * Whenever the children of a given parent are completed, the parent
+   * will be processed, being able to access the children's result data.
+   *
+   * All Jobs can be in different queues, either children or parents,
+   * however this call would be atomic, either it fails and no jobs will
+   * be added to the queues, or it succeeds and all jobs will be added.
+   *
+   * @param flow An object with a tree-like structure where children jobs
+   * will be processed before their parents.
+   */
+  async add(flow: FlowJob) {
+    if (this.closing) {
+      return;
+    }
+    const client = await this.connection.client;
+    const multi = client.multi();
+
+    const jobsTree = this.addNode(multi, flow);
+
+    const result = await multi.exec();
+
+    const updateJobIds = (
+      jobsTree: JobNode,
+      result: [Error, string][],
+      index: number,
+    ) => {
+      // TODO: Can we safely ignore result errors? how could they happen in the
+      // first place?
+      jobsTree.job.id = result[index][1];
+      const children = jobsTree.children;
+      if (children) {
+        for (let i = 0; i < children.length; i++) {
+          updateJobIds(children[i], result, index + i + 1);
+        }
+      }
+    };
+
+    updateJobIds(jobsTree, result, 0);
+
+    return jobsTree;
+  }
+
+  /**
+   * Add a node (job) of a flow to the queue. This method will recursivelly
+   * 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.
+   *
+   * @param multi ioredis pipeline
+   * @param node the node representing a job to be added to some queue
+   * @param parent Parent data sent to children to create the "links" to their parent
+   * @returns
+   */
+  private addNode(
+    multi: Pipeline,
+    node: FlowJob,
+    parent?: {
+      parentOpts: {
+        id: string;
+        queue: string;
+      };
+      parentDependenciesKey: string;
+    },
+  ): JobNode {
+    const queue = this.queueFromNode(node, new QueueKeys(node.prefix));
+    const parentId = node.children && uuid.v4();
+
+    const job = new Job(
+      queue,
+      node.name,
+      node.data,
+      {
+        ...node.opts,
+        parent: parent?.parentOpts,
+      },
+      node.opts?.jobId || parentId,
+    );
+
+    if (node.children) {
+      // Create parent job, will be a job in status "waiting-children".
+      const queueKeysParent = new QueueKeys(node.prefix);
+      const waitChildrenKey = queueKeysParent.toKey(
+        node.queueName,
+        'waiting-children',
+      );
+
+      job.addJob(<Redis>(multi as unknown), {
+        parentDependenciesKey: parent?.parentDependenciesKey,
+        waitChildrenKey,
+      });
+
+      const parentDependenciesKey = `${queueKeysParent.toKey(
+        node.queueName,
+        parentId,
+      )}:dependencies`;
+
+      const children = this.addChildren(multi, node.children, {
+        parentOpts: {
+          id: parentId,
+          queue: queueKeysParent.getPrefixedQueueName(node.queueName),
+        },
+        parentDependenciesKey,
+      });
+
+      return { job, children };
+    } else {
+      job.addJob(<Redis>(multi as unknown), {
+        parentDependenciesKey: parent?.parentDependenciesKey,
+      });
+
+      return { job };
+    }
+  }
+
+  private addChildren(
+    multi: Pipeline,
+    nodes: FlowJob[],
+    parent: {
+      parentOpts: {
+        id: string;
+        queue: string;
+      };
+      parentDependenciesKey: string;
+    },
+  ) {
+    return nodes.map(node => this.addNode(multi, node, parent));
+  }
+
+  /**
+   * Helper factory method that creates a queue-like object
+   * required to create jobs in any queue.
+   *
+   * @param node
+   * @param queueKeys
+   * @returns
+   */
+  private queueFromNode(node: FlowJob, queueKeys: QueueKeys) {
+    return {
+      client: this.connection.client,
+      name: node.queueName,
+      keys: queueKeys.getKeys(node.queueName),
+      toKey: (type: string) => queueKeys.toKey(node.queueName, type),
+      opts: {},
+      closing: this.closing,
+      waitUntilReady: async () => this.connection.client,
+      removeListener: this.removeListener.bind(this) as any,
+      emit: this.emit.bind(this) as any,
+      on: this.on.bind(this) as any,
+      redisVersion: this.connection.redisVersion,
+    };
+  }
+
+  close() {
+    if (!this.closing) {
+      this.closing = this.connection.close();
+    }
+    return this.closing;
+  }
+
+  disconnect() {
+    return this.connection.disconnect();
+  }
+}
diff --git a/src/classes/index.ts b/src/classes/index.ts
index b7b8c99ced..1f77b3bbc1 100644
--- a/src/classes/index.ts
+++ b/src/classes/index.ts
@@ -11,3 +11,4 @@ export * from './scripts';
 export * from './worker';
 export * from './child-pool';
 export * from './sandbox';
+export * from './flow';
diff --git a/src/classes/job.ts b/src/classes/job.ts
index e2263f8bc8..be2c609550 100644
--- a/src/classes/job.ts
+++ b/src/classes/job.ts
@@ -3,8 +3,9 @@ import { debuglog } from 'util';
 import { RetryErrors } from '../enums';
 import { BackoffOptions, JobsOptions, WorkerOptions } from '../interfaces';
 import { errorObject, isEmpty, tryCatch } from '../utils';
-import { Backoffs, QueueBase, QueueEvents } from './';
-import { Scripts } from './scripts';
+import { Backoffs, QueueEvents } from './';
+import { MinimalQueue, ParentOpts, Scripts } from './scripts';
+import { fromPairs } from 'lodash';
 
 const logger = debuglog('bull');
 
@@ -39,7 +40,7 @@ export class Job<T = any, R = any, N extends string = string> {
   private discarded: boolean;
 
   constructor(
-    private queue: QueueBase,
+    private queue: MinimalQueue,
     public name: N,
     public data: T,
     public opts: JobsOptions = {},
@@ -61,7 +62,7 @@ export class Job<T = any, R = any, N extends string = string> {
   }
 
   static async create<T = any, R = any, N extends string = string>(
-    queue: QueueBase,
+    queue: MinimalQueue,
     name: N,
     data: T,
     opts?: JobsOptions,
@@ -76,7 +77,7 @@ export class Job<T = any, R = any, N extends string = string> {
   }
 
   static async createBulk<T = any, R = any, N extends string = string>(
-    queue: QueueBase,
+    queue: MinimalQueue,
     jobs: {
       name: N;
       data: T;
@@ -86,7 +87,8 @@ export class Job<T = any, R = any, N extends string = string> {
     const client = await queue.client;
 
     const jobInstances = jobs.map(
-      job => new Job<T, R, N>(queue, job.name, job.data, job.opts),
+      job =>
+        new Job<T, R, N>(queue, job.name, job.data, job.opts, job.opts?.jobId),
     );
 
     const multi = client.multi();
@@ -104,7 +106,7 @@ export class Job<T = any, R = any, N extends string = string> {
     return jobInstances;
   }
 
-  static fromJSON(queue: QueueBase, json: any, jobId?: string) {
+  static fromJSON(queue: MinimalQueue, json: any, jobId?: string) {
     const data = JSON.parse(json.data || '{}');
     const opts = JSON.parse(json.opts || '{}');
 
@@ -136,7 +138,7 @@ export class Job<T = any, R = any, N extends string = string> {
   }
 
   static async fromId(
-    queue: QueueBase,
+    queue: MinimalQueue,
     jobId: string,
   ): Promise<Job | undefined> {
     // jobId can be undefined if moveJob returns undefined
@@ -345,12 +347,47 @@ export class Job<T = any, R = any, N extends string = string> {
   }
 
   async getState() {
-    if (await this.isCompleted()) return 'completed';
-    if (await this.isFailed()) return 'failed';
-    if (await this.isDelayed()) return 'delayed';
-    if (await this.isActive()) return 'active';
-    if (await this.isWaiting()) return 'waiting';
-    return 'unknown';
+    return Scripts.getState(this.queue, this.id);
+  }
+
+  /**
+   * Get this jobs children result values if any.
+   *
+   * @returns Object mapping children job keys with their values.
+   */
+  async getChildrenValues() {
+    const client = await this.queue.client;
+
+    const result = (await client.hgetall(
+      this.toKey(`${this.id}:processed`),
+    )) as Object;
+
+    if (result) {
+      return fromPairs(
+        Object.entries(result).map(([k, v]) => [k, JSON.parse(v)]),
+      );
+    }
+  }
+
+  /**
+   * Get children job keys if this job is a parent and has children.
+   *
+   * @returns dependencies separated by processed and unprocessed.
+   */
+  async getDependencies() {
+    const client = await this.queue.client;
+
+    const multi = client.multi();
+
+    await multi.smembers(this.toKey(`${this.id}:`));
+    await multi.hgetall(this.toKey(`${this.id}:`));
+
+    const [[err1, processed], [err2, unprocessed]] = (await multi.exec()) as [
+      [null | Error, string[]],
+      [null | Error, object],
+    ];
+
+    return { processed, unprocessed: Object(unprocessed).keys() };
   }
 
   /**
@@ -471,19 +508,22 @@ export class Job<T = any, R = any, N extends string = string> {
   }
 
   private async isInList(list: string) {
-    return Scripts.isJobInList(
-      await this.queue.client,
-      this.queue.toKey(list),
-      this.id,
-    );
+    return Scripts.isJobInList(this.queue, this.queue.toKey(list), this.id);
   }
 
-  private addJob(client: Redis): string {
+  addJob(client: Redis, parentOpts?: ParentOpts): string {
     const queue = this.queue;
 
     const jobData = this.asJSON();
 
-    return Scripts.addJob(client, queue, jobData, this.opts, this.id);
+    return Scripts.addJob(
+      client,
+      queue,
+      jobData,
+      this.opts,
+      this.id,
+      parentOpts,
+    );
   }
 
   private saveAttempt(multi: Pipeline, err: Error) {
diff --git a/src/classes/queue-base.ts b/src/classes/queue-base.ts
index 4b102aea6b..55c5e20a06 100644
--- a/src/classes/queue-base.ts
+++ b/src/classes/queue-base.ts
@@ -2,9 +2,11 @@ import { EventEmitter } from 'events';
 import { QueueBaseOptions } from '../interfaces';
 import { RedisConnection } from './redis-connection';
 import * as IORedis from 'ioredis';
+import { KeysMap, QueueKeys } from './queue-keys';
 
 export class QueueBase extends EventEmitter {
-  keys: { [index: string]: string };
+  toKey: (type: string) => string;
+  keys: KeysMap;
   closing: Promise<void>;
 
   protected connection: RedisConnection;
@@ -23,42 +25,19 @@ export class QueueBase extends EventEmitter {
     this.connection = new RedisConnection(opts.connection);
     this.connection.on('error', this.emit.bind(this, 'error'));
 
-    const keys: { [index: string]: string } = {};
-    [
-      '',
-      'active',
-      'wait',
-      'waiting',
-      'paused',
-      'resumed',
-      'id',
-      'delayed',
-      'priority',
-      'stalled-check',
-      'completed',
-      'failed',
-      'stalled',
-      'repeat',
-      'limiter',
-      'drained',
-      'progress',
-      'meta',
-      'events',
-      'delay',
-    ].forEach(key => {
-      keys[key] = this.toKey(key);
-    });
-    this.keys = keys;
-  }
-
-  toKey(type: string) {
-    return `${this.opts.prefix}:${this.name}:${type}`;
+    const queueKeys = new QueueKeys(opts.prefix);
+    this.keys = queueKeys.getKeys(name);
+    this.toKey = (type: string) => queueKeys.toKey(name, type);
   }
 
   get client(): Promise<IORedis.Redis> {
     return this.connection.client;
   }
 
+  get redisVersion(): string {
+    return this.connection.redisVersion;
+  }
+
   async waitUntilReady() {
     return this.client;
   }
diff --git a/src/classes/queue-getters.ts b/src/classes/queue-getters.ts
index b6c3833d9c..00346099a7 100644
--- a/src/classes/queue-getters.ts
+++ b/src/classes/queue-getters.ts
@@ -25,6 +25,7 @@ export class QueueGetters extends QueueBase {
         case 'failed':
         case 'delayed':
         case 'repeat':
+        case 'waiting-children':
           return callback(key, count ? 'zcard' : 'zrange');
         case 'active':
         case 'wait':
@@ -91,10 +92,18 @@ export class QueueGetters extends QueueBase {
     return this.getJobCountByTypes('waiting', 'paused');
   }
 
+  getWaitingChildrenCount() {
+    return this.getJobCountByTypes('waiting-children', 'paused');
+  }
+
   getWaiting(start = 0, end = 1) {
     return this.getJobs(['waiting'], start, end, true);
   }
 
+  getWaitingChildren(start = 0, end = 1) {
+    return this.getJobs(['waiting-children'], start, end, true);
+  }
+
   getActive(start = 0, end = 1) {
     return this.getJobs(['active'], start, end, true);
   }
diff --git a/src/classes/queue-keys.ts b/src/classes/queue-keys.ts
new file mode 100644
index 0000000000..06fae0ddfd
--- /dev/null
+++ b/src/classes/queue-keys.ts
@@ -0,0 +1,43 @@
+export type KeysMap = { [index in string]: string };
+
+export class QueueKeys {
+  constructor(public readonly prefix: string = 'bull') {}
+
+  getKeys(name: string): KeysMap {
+    const keys: { [index: string]: string } = {};
+    [
+      '',
+      'active',
+      'wait',
+      'waiting',
+      'paused',
+      'resumed',
+      'id',
+      'delayed',
+      'priority',
+      'stalled-check',
+      'completed',
+      'failed',
+      'stalled',
+      'repeat',
+      'limiter',
+      'drained',
+      'progress',
+      'meta',
+      'events',
+      'delay',
+    ].forEach(key => {
+      keys[key] = this.toKey(name, key);
+    });
+
+    return keys;
+  }
+
+  toKey(name: string, type: string) {
+    return `${this.getPrefixedQueueName(name)}:${type}`;
+  }
+
+  getPrefixedQueueName(name: string) {
+    return `${this.prefix}:${name}`;
+  }
+}
diff --git a/src/classes/queue.ts b/src/classes/queue.ts
index 36e3d62880..47ff032466 100644
--- a/src/classes/queue.ts
+++ b/src/classes/queue.ts
@@ -1,3 +1,4 @@
+import { Parent } from '../interfaces/parent';
 import { get } from 'lodash';
 import { v4 } from 'uuid';
 import { JobsOptions, QueueOptions, RepeatOptions } from '../interfaces';
@@ -85,7 +86,9 @@ export class Queue<
    * @param jobs: [] The array of jobs to add to the queue. Each job is defined by 3
    * properties, 'name', 'data' and 'opts'. They follow the same signature as 'Queue.add'.
    */
-  async addBulk(jobs: { name: N; data: T; opts?: JobsOptions }[]) {
+  async addBulk<ParentDataType>(
+    jobs: { name: N; data: T; opts?: JobsOptions }[],
+  ) {
     return Job.createBulk(
       this,
       jobs.map(job => ({
diff --git a/src/classes/redis-connection.ts b/src/classes/redis-connection.ts
index b7aa915209..562b470007 100644
--- a/src/classes/redis-connection.ts
+++ b/src/classes/redis-connection.ts
@@ -10,6 +10,7 @@ export class RedisConnection extends EventEmitter {
   private _client: IORedis.Redis;
   private initializing: Promise<IORedis.Redis>;
   private closing: boolean;
+  private version: string;
 
   constructor(private opts?: ConnectionOptions) {
     super();
@@ -75,10 +76,10 @@ export class RedisConnection extends EventEmitter {
     await load(this._client);
 
     if (opts && opts.skipVersionCheck !== true && !this.closing) {
-      const version = await this.getRedisVersion();
-      if (semver.lt(version, RedisConnection.minimumVersion)) {
+      this.version = await this.getRedisVersion();
+      if (semver.lt(this.version, RedisConnection.minimumVersion)) {
         throw new Error(
-          `Redis version needs to be greater than ${RedisConnection.minimumVersion} Current: ${version}`,
+          `Redis version needs to be greater than ${RedisConnection.minimumVersion} Current: ${this.version}`,
         );
       }
     }
@@ -133,4 +134,8 @@ export class RedisConnection extends EventEmitter {
       }
     }
   }
+
+  get redisVersion(): string {
+    return this.version;
+  }
 }
diff --git a/src/classes/scripts.ts b/src/classes/scripts.ts
index 1ecc5cb294..455192c682 100644
--- a/src/classes/scripts.ts
+++ b/src/classes/scripts.ts
@@ -6,27 +6,61 @@
 'use strict';
 
 import { Redis } from 'ioredis';
+import * as semver from 'semver';
 import {
   JobsOptions,
   QueueSchedulerOptions,
   WorkerOptions,
 } from '../interfaces';
 import { array2obj } from '../utils';
-import { Queue, QueueBase, QueueScheduler, Worker } from './';
+import { QueueBase, QueueScheduler, Worker } from './';
 import { Job, JobJson } from './job';
 
+export type MinimalQueue = Pick<
+  QueueBase,
+  | 'client'
+  | 'toKey'
+  | 'keys'
+  | 'opts'
+  | 'closing'
+  | 'waitUntilReady'
+  | 'removeListener'
+  | 'emit'
+  | 'on'
+  | 'redisVersion'
+>;
+
+export type ParentOpts = {
+  waitChildrenKey?: string;
+  parentDependenciesKey?: string;
+};
+
 export class Scripts {
-  static async isJobInList(client: Redis, listKey: string, jobId: string) {
-    const result = await (<any>client).isJobInList([listKey, jobId]);
-    return result === 1;
+  static async isJobInList(
+    queue: MinimalQueue,
+    listKey: string,
+    jobId: string,
+  ) {
+    const client = await queue.client;
+    let result;
+    if (semver.lt(queue.redisVersion, '6.0.6')) {
+      result = await (<any>client).isJobInList([listKey, jobId]);
+    } else {
+      result = await (<any>client).lpos(listKey, jobId);
+    }
+    return Number.isInteger(result);
   }
 
   static addJob(
     client: Redis,
-    queue: QueueBase,
+    queue: MinimalQueue,
     job: JobJson,
     opts: JobsOptions,
     jobId: string,
+    parentOpts: ParentOpts = {
+      waitChildrenKey: null,
+      parentDependenciesKey: null,
+    },
   ) {
     const queueKeys = queue.keys;
     let keys = [
@@ -38,6 +72,8 @@ export class Scripts {
       queueKeys.priority,
       queueKeys.events,
       queueKeys.delay,
+      parentOpts.waitChildrenKey,
+      parentOpts.parentDependenciesKey,
     ];
 
     const args = [
@@ -54,10 +90,11 @@ export class Scripts {
     ];
 
     keys = keys.concat(<string[]>args);
+
     return (<any>client).addJob(keys);
   }
 
-  static async pause(queue: Queue, pause: boolean) {
+  static async pause(queue: MinimalQueue, pause: boolean) {
     const client = await queue.client;
 
     var src = 'wait',
@@ -74,7 +111,7 @@ export class Scripts {
     return (<any>client).pause(keys.concat([pause ? 'paused' : 'resumed']));
   }
 
-  static async remove(queue: QueueBase, jobId: string) {
+  static async remove(queue: MinimalQueue, jobId: string) {
     const client = await queue.client;
 
     const keys = [
@@ -92,7 +129,7 @@ export class Scripts {
   }
 
   static async extendLock(
-    queue: QueueBase,
+    queue: MinimalQueue,
     jobId: string,
     token: string,
     duration: number,
@@ -109,7 +146,7 @@ export class Scripts {
   }
 
   static async updateProgress(
-    queue: QueueBase,
+    queue: MinimalQueue,
     job: Job,
     progress: number | object,
   ) {
@@ -123,7 +160,7 @@ export class Scripts {
   }
 
   static moveToFinishedArgs(
-    queue: QueueBase,
+    queue: MinimalQueue,
     job: Job,
     val: any,
     propVal: string,
@@ -164,13 +201,15 @@ export class Scripts {
       queueKeys[''],
       token,
       opts.lockDuration,
+      job.opts?.parent?.id,
+      job.opts?.parent?.queue,
     ];
 
     return keys.concat(args);
   }
 
   static async moveToFinished(
-    queue: QueueBase,
+    queue: MinimalQueue,
     job: Job,
     val: any,
     propVal: string,
@@ -202,14 +241,16 @@ export class Scripts {
   static finishedErrors(code: number, jobId: string, command: string) {
     switch (code) {
       case -1:
-        return new Error('Missing key for job ' + jobId + ' ' + command);
+        return new Error(`Missing key for job ${jobId} ${command}`);
       case -2:
-        return new Error('Missing lock for job ' + jobId + ' ' + command);
+        return new Error(`Missing lock for job ${jobId} ${command}`);
+      case -3:
+        return new Error(`Job is not in the active list ${jobId} ${command}`);
     }
   }
 
   static moveToCompleted(
-    queue: QueueBase,
+    queue: MinimalQueue,
     job: Job,
     returnvalue: any,
     removeOnComplete: boolean | number,
@@ -229,7 +270,7 @@ export class Scripts {
   }
 
   static moveToFailedArgs(
-    queue: QueueBase,
+    queue: MinimalQueue,
     job: Job,
     failedReason: string,
     removeOnFailed: boolean | number,
@@ -248,7 +289,7 @@ export class Scripts {
     );
   }
 
-  static async isFinished(queue: QueueBase, jobId: string) {
+  static async isFinished(queue: MinimalQueue, jobId: string) {
     const client = await queue.client;
 
     const keys = ['completed', 'failed'].map(function(key: string) {
@@ -258,8 +299,32 @@ export class Scripts {
     return (<any>client).isFinished(keys.concat([jobId]));
   }
 
+  static async getState(queue: MinimalQueue, jobId: string) {
+    const client = await queue.client;
+
+    const keys = [
+      'completed',
+      'failed',
+      'delayed',
+      'active',
+      'wait',
+      'paused',
+    ].map(function(key: string) {
+      return queue.toKey(key);
+    });
+
+    if (semver.lt(queue.redisVersion, '6.0.6')) {
+      return (<any>client).getState(keys.concat([jobId]));
+    }
+    return (<any>client).getStateV2(keys.concat([jobId]));
+  }
+
   // Note: We have an issue here with jobs using custom job ids
-  static moveToDelayedArgs(queue: QueueBase, jobId: string, timestamp: number) {
+  static moveToDelayedArgs(
+    queue: MinimalQueue,
+    jobId: string,
+    timestamp: number,
+  ) {
     //
     // Bake in the job id first 12 bits into the timestamp
     // to guarantee correct execution order of delayed jobs
@@ -284,7 +349,7 @@ export class Scripts {
   }
 
   static async moveToDelayed(
-    queue: QueueBase,
+    queue: MinimalQueue,
     jobId: string,
     timestamp: number,
   ) {
@@ -303,7 +368,7 @@ export class Scripts {
   }
 
   static async cleanJobsInSet(
-    queue: QueueBase,
+    queue: MinimalQueue,
     set: string,
     timestamp: number,
     limit = 0,
@@ -319,7 +384,7 @@ export class Scripts {
     ]);
   }
 
-  static retryJobArgs(queue: QueueBase, job: Job) {
+  static retryJobArgs(queue: MinimalQueue, job: Job) {
     const jobId = job.id;
 
     const keys = ['active', 'wait', jobId].map(function(name) {
@@ -348,7 +413,7 @@ export class Scripts {
    * -2 means the job was not found in the expected set
    */
   static async reprocessJob(
-    queue: QueueBase,
+    queue: MinimalQueue,
     job: Job,
     state: 'failed' | 'completed',
   ) {
@@ -406,7 +471,7 @@ export class Scripts {
   //  It checks if the job in the top of the delay set should be moved back to the
   //  top of the  wait queue (so that it will be processed as soon as possible)
   //
-  static async updateDelaySet(queue: QueueBase, delayedTimestamp: number) {
+  static async updateDelaySet(queue: MinimalQueue, delayedTimestamp: number) {
     const client = await queue.client;
 
     const keys: (string | number)[] = [
@@ -424,7 +489,7 @@ export class Scripts {
     return (<any>client).updateDelaySet(keys.concat(args));
   }
 
-  static async promote(queue: QueueBase, jobId: string) {
+  static async promote(queue: MinimalQueue, jobId: string) {
     const client = await queue.client;
 
     const keys = [
@@ -473,7 +538,7 @@ export class Scripts {
   }
 
   static async obliterate(
-    queue: Queue,
+    queue: MinimalQueue,
     opts: { force: boolean; count: number },
   ) {
     const client = await queue.client;
diff --git a/src/commands/addJob-8.lua b/src/commands/addJob-10.lua
similarity index 83%
rename from src/commands/addJob-8.lua
rename to src/commands/addJob-10.lua
index 819c0fa91d..8c9ca40abe 100644
--- a/src/commands/addJob-8.lua
+++ b/src/commands/addJob-10.lua
@@ -23,6 +23,8 @@
       KEYS[6] 'priority'
       KEYS[7] events stream key
       KEYS[8] delay stream key
+      KEYS[9] waitChildrenKey key.
+      KEYS[10] parent dependencies key.
 
       ARGV[1]  key prefix,
       ARGV[2]  custom id (will not generate one automatically)
@@ -58,7 +60,12 @@ rcall("HMSET", jobIdKey, "name", ARGV[3], "data", ARGV[4], "opts", ARGV[5],
 
 -- Check if job is delayed
 local delayedTimestamp = tonumber(ARGV[8])
-if (delayedTimestamp ~= 0) then
+
+-- Check if job is a parent, if so add to the parents set
+local waitChildrenKey = KEYS[9]
+if waitChildrenKey ~= "" then
+    rcall("ZADD", waitChildrenKey, ARGV[6], jobId)
+elseif (delayedTimestamp ~= 0) then
     local timestamp = delayedTimestamp * 0x1000 + bit.band(jobCounter, 0xfff)
     rcall("ZADD", KEYS[5], timestamp, jobId)
     rcall("XADD", KEYS[7], "*", "event", "delayed", "jobId", jobId, "delay",
@@ -100,6 +107,14 @@ else
     rcall("XADD", KEYS[7], "*", "event", "waiting", "jobId", jobId)
 end
 
+-- Check if this job is a child of another job, if so add it to the parents dependencies
+-- TODO: Should not be possible to add a child job to a parent that is not in the "waiting-children" status.
+-- fail in this case.
+local parentDependenciesKey = KEYS[10]
+if parentDependenciesKey ~= "" then
+    rcall("SADD", KEYS[10], jobIdKey)
+end
+
 local maxEvents = rcall("HGET", KEYS[3], "opts.maxLenEvents")
 if (maxEvents) then rcall("XTRIM", KEYS[7], "MAXLEN", "~", maxEvents) end
 
diff --git a/src/commands/getState-6.lua b/src/commands/getState-6.lua
new file mode 100644
index 0000000000..71ea928d0a
--- /dev/null
+++ b/src/commands/getState-6.lua
@@ -0,0 +1,57 @@
+--[[
+  Get a job state
+
+  Input: 
+    KEYS[1] 'completed' key,
+    KEYS[2] 'failed' key
+    KEYS[3] 'delayed' key
+    KEYS[4] 'active' key
+    KEYS[5] 'wait' key
+    KEYS[6] 'paused' key
+
+    ARGV[1] job id
+  Output:
+    'completed'
+    'failed'
+    'delayed'
+    'active'
+    'waiting'
+    'unknown'
+]]
+if redis.call("ZSCORE", KEYS[1], ARGV[1]) ~= false then
+  return "completed"
+end
+
+if redis.call("ZSCORE", KEYS[2], ARGV[1]) ~= false then
+  return "failed"
+end
+
+if redis.call("ZSCORE", KEYS[3], ARGV[1]) ~= false then
+  return "delayed"
+end
+
+local function item_in_list (list, item)
+  for _, v in pairs(list) do
+    if v == item then
+      return 1
+    end
+  end
+  return nil
+end
+
+local active_items = redis.call("LRANGE", KEYS[4] , 0, -1)
+if item_in_list(active_items, ARGV[1]) ~= nil then
+  return "active"
+end
+
+local wait_items = redis.call("LRANGE", KEYS[5] , 0, -1)
+if item_in_list(wait_items, ARGV[1]) ~= nil then
+  return "waiting"
+end
+
+local paused_items = redis.call("LRANGE", KEYS[6] , 0, -1)
+if item_in_list(paused_items, ARGV[1]) ~= nil then
+  return "waiting"
+end
+
+return "unknown"
diff --git a/src/commands/getStateV2-6.lua b/src/commands/getStateV2-6.lua
new file mode 100644
index 0000000000..8f4b4b2282
--- /dev/null
+++ b/src/commands/getStateV2-6.lua
@@ -0,0 +1,45 @@
+--[[
+  Get a job state
+
+  Input: 
+    KEYS[1] 'completed' key,
+    KEYS[2] 'failed' key
+    KEYS[3] 'delayed' key
+    KEYS[4] 'active' key
+    KEYS[5] 'wait' key
+    KEYS[6] 'paused' key
+
+    ARGV[1] job id
+  Output:
+    'completed'
+    'failed'
+    'delayed'
+    'active'
+    'waiting'
+    'unknown'
+]]
+if redis.call("ZSCORE", KEYS[1], ARGV[1]) ~= false then
+  return "completed"
+end
+
+if redis.call("ZSCORE", KEYS[2], ARGV[1]) ~= false then
+  return "failed"
+end
+
+if redis.call("ZSCORE", KEYS[3], ARGV[1]) ~= false then
+  return "delayed"
+end
+
+if redis.call("LPOS", KEYS[4] , ARGV[1]) ~= false then
+  return "active"
+end
+
+if redis.call("LPOS", KEYS[5] , ARGV[1]) ~= false then
+  return "waiting"
+end
+
+if redis.call("LPOS", KEYS[6] , ARGV[1]) ~= false then
+  return "waiting"
+end
+
+return "unknown"
diff --git a/src/commands/moveToFinished-7.lua b/src/commands/moveToFinished-7.lua
index 14357899d0..7460ee61ff 100644
--- a/src/commands/moveToFinished-7.lua
+++ b/src/commands/moveToFinished-7.lua
@@ -25,20 +25,24 @@
       ARGV[9]  keys prefix
       ARGV[10] lock token
       ARGV[11] lock duration in milliseconds
+      ARGV[12] parentId
+      ARGV[13] parentQueue
 
      Output:
       0 OK
       -1 Missing key.
       -2 Missing lock.
+      -3 Job not in active set
 
      Events:
       'completed/failed'
 ]] local rcall = redis.call
 
-if rcall("EXISTS", KEYS[3]) == 1 then -- // Make sure job exists
+local jobIdKey = KEYS[3]
+if rcall("EXISTS",jobIdKey) == 1 then -- // Make sure job exists
 
     if ARGV[10] ~= "0" then
-      local lockKey = KEYS[3] .. ':lock'
+      local lockKey = jobIdKey .. ':lock'
       if rcall("GET", lockKey) == ARGV[10] then
           rcall("DEL", lockKey)
       else
@@ -46,20 +50,54 @@ if rcall("EXISTS", KEYS[3]) == 1 then -- // Make sure job exists
       end
     end
 
+    local jobId = ARGV[1]
+
     -- Remove from active list (if not active we shall return error)
-    local numRemovedElements = rcall("LREM", KEYS[1], -1, ARGV[1])
+    local numRemovedElements = rcall("LREM", KEYS[1], -1, jobId)
 
-    -- What if we just ignore this? I think it is good to know regardless.
-    --[[   if(numRemovedElements < 1) then
-    return -2
-  end
- ]]
+    --[[ 
+    -- Currently it breaks some tests but this should be a sensible error.
+    if(numRemovedElements < 1) then
+        return -3
+    end
+    ]]
+    
+    -- If job has a parent we need to 
+    -- 1) remove this job id from parents dependencies
+    -- 2) move the job Id to parent "processed" set
+    -- 3) push the results into parent "results" list
+    -- 4) if parent's dependencies is empty, then move parent to "wait/paused". Note it may be a different queue!.
+    -- NOTE: Priorities not supported yet for parent jobs.
+    local parentId = ARGV[12]
+    if parentId ~= "" and ARGV[5] == "completed" then 
+        local parentQueue = ARGV[13]
+        local parentKey =  parentQueue .. ":" .. parentId
+        local dependenciesSet = parentKey .. ":dependencies"
+        local result = rcall("SREM", dependenciesSet, jobIdKey)
+        if result == 1 then 
+            local processedSet = parentKey .. ":processed"
+            rcall("HSET", processedSet, jobIdKey, ARGV[4])
+            if rcall("SCARD", dependenciesSet) == 0 then 
+                rcall("ZREM", parentQueue .. ":waiting-children", parentId)
+
+                if rcall("HEXISTS", parentQueue .. ":meta", "paused") ~= 1 then
+                    rcall("RPUSH", parentQueue .. ":wait", parentId)
+                else
+                    rcall("RPUSH", parentQueue .. ":paused", parentId)
+                end
+
+                local parentEventStream = parentKey .. ":events"
+                rcall("XADD", parentEventStream, "*", "event", "active", "jobId", parentId, "prev", "waiting-children")
+            end
+        end
+    end
+    
     -- Remove job?
     local removeJobs = tonumber(ARGV[6])
     if removeJobs ~= 1 then
         -- Add to complete/failed set
-        rcall("ZADD", KEYS[2], ARGV[2], ARGV[1])
-        rcall("HMSET", KEYS[3], ARGV[3], ARGV[4], "finishedOn", ARGV[2]) -- "returnvalue" / "failedReason" and "finishedOn"
+        rcall("ZADD", KEYS[2], ARGV[2], jobId)
+        rcall("HMSET", jobIdKey, ARGV[3], ARGV[4], "finishedOn", ARGV[2]) -- "returnvalue" / "failedReason" and "finishedOn"
 
         -- Remove old jobs?
         if removeJobs and removeJobs > 1 then
@@ -73,11 +111,11 @@ if rcall("EXISTS", KEYS[3]) == 1 then -- // Make sure job exists
             rcall("ZREMRANGEBYRANK", KEYS[2], 0, -removeJobs)
         end
     else
-        local jobLogKey = KEYS[3] .. ':logs'
-        rcall("DEL", KEYS[3], jobLogKey)
+        local jobLogKey = jobIdKey .. ':logs'
+        rcall("DEL", jobIdKey, jobLogKey)
     end
 
-    rcall("XADD", KEYS[6], "*", "event", ARGV[5], "jobId", ARGV[1], ARGV[3],
+    rcall("XADD", KEYS[6], "*", "event", ARGV[5], "jobId", jobId, ARGV[3],
           ARGV[4])
 
     -- Try to get next job to avoid an extra roundtrip if the queue is not closing,
diff --git a/src/interfaces/flow-job.ts b/src/interfaces/flow-job.ts
new file mode 100644
index 0000000000..2a1b4f0a04
--- /dev/null
+++ b/src/interfaces/flow-job.ts
@@ -0,0 +1,10 @@
+import { JobsOptions } from './jobs-options';
+
+export interface FlowJob {
+  name: string;
+  queueName: string;
+  data?: any;
+  prefix?: string;
+  opts?: Omit<JobsOptions, 'parent'>;
+  children?: FlowJob[];
+}
diff --git a/src/interfaces/jobs-options.ts b/src/interfaces/jobs-options.ts
index be2669e2a9..715f61a935 100644
--- a/src/interfaces/jobs-options.ts
+++ b/src/interfaces/jobs-options.ts
@@ -85,6 +85,14 @@ export interface JobsOptions {
    */
   stackTraceLimit?: number;
 
+  /**
+   *
+   */
+  parent?: {
+    id: string;
+    queue: string; // Queue name including prefix
+  };
+
   /**
    * Internal property used by repeatable jobs.
    */
diff --git a/src/interfaces/parent.ts b/src/interfaces/parent.ts
new file mode 100644
index 0000000000..576f2c0bdc
--- /dev/null
+++ b/src/interfaces/parent.ts
@@ -0,0 +1,12 @@
+import { JobsOptions } from './jobs-options';
+
+/**
+ * Describes the parent for a Job.
+ */
+export interface Parent<T> {
+  name: string;
+  prefix?: string;
+  queue?: string;
+  data?: T;
+  opts?: JobsOptions;
+}


Test patch:
diff --git a/src/test/test_bulk.ts b/src/test/test_bulk.ts
index 11f63dbf7b..dbf1a3e3bd 100644
--- a/src/test/test_bulk.ts
+++ b/src/test/test_bulk.ts
@@ -3,7 +3,7 @@ import { expect } from 'chai';
 import * as IORedis from 'ioredis';
 import { beforeEach, describe, it } from 'mocha';
 import { v4 } from 'uuid';
-import { removeAllQueueData } from '../utils';
+import { removeAllQueueData, delay } from '../utils';
 
 describe('bulk jobs', () => {
   let queue: Queue;
@@ -22,7 +22,7 @@ describe('bulk jobs', () => {
   it('should process jobs', async () => {
     const name = 'test';
     let processor;
-    const processing = new Promise(resolve => [
+    const processing = new Promise<void>(resolve => [
       (processor = async (job: Job) => {
         if (job.data.idx === 0) {
           expect(job.data.foo).to.be.equal('bar');
@@ -48,7 +48,38 @@ describe('bulk jobs', () => {
     expect(jobs[1].data.foo).to.be.eql('baz');
 
     await processing;
+    await worker.close();
+  });
 
+  it('should process jobs with custom ids', async () => {
+    const name = 'test';
+    let processor;
+    const processing = new Promise<void>(resolve => [
+      (processor = async (job: Job) => {
+        if (job.data.idx === 0) {
+          expect(job.data.foo).to.be.equal('bar');
+        } else {
+          expect(job.data.idx).to.be.equal(1);
+          expect(job.data.foo).to.be.equal('baz');
+          resolve();
+        }
+      }),
+    ]);
+    const worker = new Worker(queueName, processor);
+    await worker.waitUntilReady();
+
+    const jobs = await queue.addBulk([
+      { name, data: { idx: 0, foo: 'bar' }, opts: { jobId: 'test1' } },
+      { name, data: { idx: 1, foo: 'baz' }, opts: { jobId: 'test2' } },
+    ]);
+    expect(jobs).to.have.length(2);
+
+    expect(jobs[0].id).to.be.eql('test1');
+    expect(jobs[0].data.foo).to.be.eql('bar');
+    expect(jobs[1].id).to.be.eql('test2');
+    expect(jobs[1].data.foo).to.be.eql('baz');
+
+    await processing;
     await worker.close();
   });
 });
diff --git a/src/test/test_flow.ts b/src/test/test_flow.ts
new file mode 100644
index 0000000000..e4885798ac
--- /dev/null
+++ b/src/test/test_flow.ts
@@ -0,0 +1,326 @@
+import { Queue, Worker, Job, Flow } from '../classes';
+import { expect } from 'chai';
+import * as IORedis from 'ioredis';
+import { beforeEach, describe, it } from 'mocha';
+import { v4 } from 'uuid';
+import { removeAllQueueData, delay } from '../utils';
+
+describe('flows', () => {
+  let queue: Queue;
+  let queueName: string;
+
+  beforeEach(async function() {
+    queueName = 'test-' + v4();
+    queue = new Queue(queueName);
+  });
+
+  afterEach(async function() {
+    await queue.close();
+    await removeAllQueueData(new IORedis(), queueName);
+  });
+
+  it('should process children before the parent', async () => {
+    const name = 'child-job';
+    const values = [
+      { bar: 'something' },
+      { baz: 'something' },
+      { qux: 'something' },
+    ];
+
+    const parentQueueName = 'parent-queue';
+
+    let childrenProcessor,
+      parentProcessor,
+      processedChildren = 0;
+    const processingChildren = new Promise<void>(resolve => [
+      (childrenProcessor = async (job: Job) => {
+        processedChildren++;
+
+        if (processedChildren == values.length) {
+          resolve();
+        }
+        return values[job.data.idx];
+      }),
+    ]);
+
+    const processingParent = new Promise<void>((resolve, reject) => [
+      (parentProcessor = async (job: Job) => {
+        try {
+          expect(processedChildren).to.be.equal(3);
+
+          const childrenValues = await job.getChildrenValues();
+
+          for (let i = 0; i < values.length; i++) {
+            const jobKey = queue.toKey(tree.children[i].job.id);
+            expect(childrenValues[jobKey]).to.be.deep.equal(values[i]);
+          }
+          resolve();
+        } catch (err) {
+          console.error(err);
+          reject(err);
+        }
+      }),
+    ]);
+
+    const parentWorker = new Worker(parentQueueName, parentProcessor);
+    const childrenWorker = new Worker(queueName, childrenProcessor);
+
+    const flow = new Flow();
+    const tree = await flow.add({
+      name: 'parent-job',
+      queueName: parentQueueName,
+      data: {},
+      children: [
+        { name, data: { idx: 0, foo: 'bar' }, queueName },
+        { name, data: { idx: 1, foo: 'baz' }, queueName },
+        { name, data: { idx: 2, foo: 'qux' }, queueName },
+      ],
+    });
+
+    expect(tree).to.have.property('job');
+    expect(tree).to.have.property('children');
+
+    const { children } = tree;
+
+    expect(children).to.have.length(3);
+
+    expect(children[0].job.id).to.be.ok;
+    expect(children[0].job.data.foo).to.be.eql('bar');
+    expect(children[1].job.id).to.be.ok;
+    expect(children[1].job.data.foo).to.be.eql('baz');
+    expect(children[2].job.id).to.be.ok;
+    expect(children[2].job.data.foo).to.be.eql('qux');
+
+    await processingChildren;
+    await childrenWorker.close();
+
+    await processingParent;
+    await parentWorker.close();
+
+    await flow.close();
+
+    await removeAllQueueData(new IORedis(), parentQueueName);
+  });
+
+  it('should process a chain of jobs', async () => {
+    const name = 'child-job';
+    const values = [
+      { idx: 0, bar: 'something' },
+      { idx: 1, baz: 'something' },
+      { idx: 2, qux: 'something' },
+    ];
+
+    const topQueueName = 'top-queue';
+
+    let childrenProcessor,
+      parentProcessor,
+      processedChildren = 0;
+    const processingChildren = new Promise<void>((resolve, reject) => [
+      (childrenProcessor = async (job: Job) => {
+        try {
+          const childrenValues = await job.getChildrenValues();
+
+          expect(job.data.idx).to.be.eql(values.length - 1 - processedChildren);
+          switch (job.data.idx) {
+            case 0:
+              {
+                const jobKey = queue.toKey(tree.children[0].children[0].job.id);
+                expect(childrenValues[jobKey]).to.be.deep.equal(values[1]);
+              }
+              break;
+            case 1:
+              {
+                const jobKey = queue.toKey(
+                  tree.children[0].children[0].children[0].job.id,
+                );
+                expect(childrenValues[jobKey]).to.be.deep.equal(values[2]);
+              }
+              break;
+          }
+
+          processedChildren++;
+          if (processedChildren == values.length) {
+            resolve();
+          }
+          return values[job.data.idx];
+        } catch (err) {
+          reject(err);
+        }
+      }),
+    ]);
+
+    const processingTop = new Promise<void>((resolve, reject) => [
+      (parentProcessor = async (job: Job) => {
+        try {
+          expect(processedChildren).to.be.equal(3);
+
+          const childrenValues = await job.getChildrenValues();
+
+          const jobKey = queue.toKey(tree.children[0].job.id);
+          expect(childrenValues[jobKey]).to.be.deep.equal(values[0]);
+
+          resolve();
+        } catch (err) {
+          console.error(err);
+          reject(err);
+        }
+      }),
+    ]);
+
+    const parentWorker = new Worker(topQueueName, parentProcessor);
+    const childrenWorker = new Worker(queueName, childrenProcessor);
+
+    const flow = new Flow();
+    const tree = await flow.add({
+      name: 'root-job',
+      queueName: topQueueName,
+      data: {},
+      children: [
+        {
+          name,
+          data: { idx: 0, foo: 'bar' },
+          queueName,
+          children: [
+            {
+              name,
+              data: { idx: 1, foo: 'baz' },
+              queueName,
+              children: [{ name, data: { idx: 2, foo: 'qux' }, queueName }],
+            },
+          ],
+        },
+      ],
+    });
+
+    expect(tree).to.have.property('job');
+    expect(tree).to.have.property('children');
+
+    const { children } = tree;
+
+    expect(children).to.have.length(1);
+
+    expect(children[0].job.id).to.be.ok;
+    expect(children[0].job.data.foo).to.be.eql('bar');
+    expect(children[0].children).to.have.length(1);
+
+    expect(children[0].children[0].job.id).to.be.ok;
+    expect(children[0].children[0].job.data.foo).to.be.eql('baz');
+
+    expect(children[0].children[0].children[0].job.id).to.be.ok;
+    expect(children[0].children[0].children[0].job.data.foo).to.be.eql('qux');
+
+    await processingChildren;
+    await childrenWorker.close();
+
+    await processingTop;
+    await parentWorker.close();
+
+    await flow.close();
+
+    await removeAllQueueData(new IORedis(), topQueueName);
+  });
+
+  it('should not process parent if child fails', async () => {
+    const name = 'child-job';
+
+    const parentQueueName = 'parent-queue';
+
+    let childrenProcessor;
+    const processingChildren = new Promise<void>(resolve => [
+      (childrenProcessor = async (job: Job) => {
+        resolve();
+        throw new Error('failed job');
+      }),
+    ]);
+
+    const childrenWorker = new Worker(queueName, childrenProcessor);
+
+    const flow = new Flow();
+    const tree = await flow.add({
+      name: 'parent-job',
+      queueName: parentQueueName,
+      data: {},
+      children: [{ name, data: { idx: 0, foo: 'bar' }, queueName }],
+    });
+
+    expect(tree).to.have.property('job');
+    expect(tree).to.have.property('children');
+
+    const { children } = tree;
+
+    expect(children).to.have.length(1);
+
+    expect(children[0].job.id).to.be.ok;
+    expect(children[0].job.data.foo).to.be.eql('bar');
+
+    await processingChildren;
+    await childrenWorker.close();
+
+    const parentQueue = new Queue(parentQueueName);
+    const numJobs = await parentQueue.getWaitingCount();
+    expect(numJobs).to.be.equal(0);
+
+    await removeAllQueueData(new IORedis(), parentQueueName);
+  });
+
+  it('should not process parent until queue is unpaused', async () => {
+    const name = 'child-job';
+    const parentQueueName = 'parent-queue';
+
+    let childrenProcessor, parentProcessor;
+    const processingChildren = new Promise<void>(resolve => [
+      (childrenProcessor = async (job: Job) => {
+        resolve();
+      }),
+    ]);
+
+    const childrenWorker = new Worker(queueName, childrenProcessor);
+
+    const processingParent = new Promise<void>(resolve => [
+      (parentProcessor = async (job: Job) => {
+        resolve();
+      }),
+    ]);
+
+    const parentWorker = new Worker(parentQueueName, parentProcessor);
+
+    const parentQueue = new Queue(parentQueueName);
+    await parentQueue.pause();
+
+    const flow = new Flow();
+    const tree = await flow.add({
+      name: 'parent-job',
+      queueName: parentQueueName,
+      data: {},
+      children: [{ name, data: { idx: 0, foo: 'bar' }, queueName }],
+    });
+
+    expect(tree).to.have.property('job');
+    expect(tree).to.have.property('children');
+
+    const { children } = tree;
+
+    expect(children).to.have.length(1);
+
+    expect(children[0].job.id).to.be.ok;
+    expect(children[0].job.data.foo).to.be.eql('bar');
+
+    await processingChildren;
+    await childrenWorker.close();
+
+    await delay(500);
+
+    let numJobs = await parentQueue.getWaitingCount();
+    expect(numJobs).to.be.equal(1);
+
+    await parentQueue.resume();
+
+    await processingParent;
+    await parentWorker.close();
+
+    numJobs = await parentQueue.getWaitingCount();
+    expect(numJobs).to.be.equal(0);
+
+    await removeAllQueueData(new IORedis(), parentQueueName);
+  });
+});
diff --git a/src/test/test_job.ts b/src/test/test_job.ts
index 48c0febbf6..2d0320aa35 100644
--- a/src/test/test_job.ts
+++ b/src/test/test_job.ts
@@ -12,6 +12,8 @@ import { after } from 'lodash';
 import { afterEach, beforeEach, describe, it } from 'mocha';
 import { v4 } from 'uuid';
 
+const sinon = require('sinon');
+
 describe('Job', function() {
   let queue: Queue;
   let queueName: string;
@@ -387,6 +389,68 @@ describe('Job', function() {
     });
   });
 
+  describe('.getState', () => {
+    describe('when redisVersion is less than 6.0.6', () => {
+      it('should get job actual state', async () => {
+        const redisVersionStub = sinon
+          .stub(queue, 'redisVersion')
+          .get(() => '6.0.5');
+        const job = await queue.add('job', { foo: 'bar' }, { delay: 1 });
+        const delayedState = await job.getState();
+
+        expect(delayedState).to.be.equal('delayed');
+
+        await queue.pause();
+        await job.promote();
+        await queue.resume();
+        const waitingState = await job.getState();
+
+        expect(waitingState).to.be.equal('waiting');
+
+        await job.moveToFailed(new Error('test error'), '0', true);
+        const failedState = await job.getState();
+
+        expect(failedState).to.be.equal('failed');
+
+        await job.moveToCompleted('succeeded', '0', true);
+        const completedState = await job.getState();
+
+        expect(completedState).to.be.equal('completed');
+        redisVersionStub.restore();
+      });
+    });
+
+    describe('when redisVersion is greater or equal than 6.0.6', () => {
+      it('should get job actual state', async () => {
+        const redisVersionStub = sinon
+          .stub(queue, 'redisVersion')
+          .get(() => '6.0.6');
+        const job = await queue.add('job', { foo: 'bar' }, { delay: 1 });
+        const delayedState = await job.getState();
+
+        expect(delayedState).to.be.equal('delayed');
+
+        await queue.pause();
+        await job.promote();
+        await queue.resume();
+        const waitingState = await job.getState();
+
+        expect(waitingState).to.be.equal('waiting');
+
+        await job.moveToFailed(new Error('test error'), '0', true);
+        const failedState = await job.getState();
+
+        expect(failedState).to.be.equal('failed');
+
+        await job.moveToCompleted('succeeded', '0', true);
+        const completedState = await job.getState();
+
+        expect(completedState).to.be.equal('completed');
+        redisVersionStub.restore();
+      });
+    });
+  });
+
   // TODO:
   // Divide into several tests
   //
diff --git a/src/test/test_worker.ts b/src/test/test_worker.ts
index 92a05c0131..dd2ac568d5 100644
--- a/src/test/test_worker.ts
+++ b/src/test/test_worker.ts
@@ -1632,24 +1632,56 @@ describe('workers', function() {
   });
 
   describe('Manually process jobs', () => {
-    it('should allow to complete jobs manually', async () => {
-      const worker = new Worker(queueName);
-      const token = 'my-token';
+    describe('when redisVersion is less than 6.0.6', () => {
+      it('should allow to complete jobs manually', async () => {
+        const redisVersionStub = sinon
+          .stub(queue, 'redisVersion')
+          .get(() => '6.0.5');
+        const worker = new Worker(queueName);
+        const token = 'my-token';
 
-      await queue.add('test', { foo: 'bar' });
+        await queue.add('test', { foo: 'bar' });
 
-      const job = (await worker.getNextJob(token)) as Job;
+        const job = (await worker.getNextJob(token)) as Job;
 
-      const isActive = await job.isActive();
-      expect(isActive).to.be.equal(true);
+        const isActive = await job.isActive();
+        expect(isActive).to.be.equal(true);
 
-      await job.moveToCompleted('return value', token);
+        await job.moveToCompleted('return value', token);
 
-      const isCompleted = await job.isCompleted();
+        const isCompleted = await job.isCompleted();
 
-      expect(isCompleted).to.be.equal(true);
+        expect(isCompleted).to.be.equal(true);
 
-      await worker.close();
+        await worker.close();
+        redisVersionStub.restore();
+      });
+    });
+
+    describe('when redisVersion is greater or equal than 6.0.6', () => {
+      it('should allow to complete jobs manually', async () => {
+        const redisVersionStub = sinon
+          .stub(queue, 'redisVersion')
+          .get(() => '6.0.6');
+        const worker = new Worker(queueName);
+        const token = 'my-token';
+
+        await queue.add('test', { foo: 'bar' });
+
+        const job = (await worker.getNextJob(token)) as Job;
+
+        const isActive = await job.isActive();
+        expect(isActive).to.be.equal(true);
+
+        await job.moveToCompleted('return value', token);
+
+        const isCompleted = await job.isCompleted();
+
+        expect(isCompleted).to.be.equal(true);
+
+        await worker.close();
+        redisVersionStub.restore();
+      });
     });
 
     it('should allow to fail jobs manually', async () => {