feat(job): add removeDependencyOnFail option

docs/gitbook/guide/workers/README.md

@@ -1,6 +1,6 @@
 # Workers
 
-Workers are the actual instances that perform some job based on the jobs that are added in the queue. A worker is equivalent to a "message" receiver in a traditional message queue. The worker duty is to complete the job, if it succeeds the job will be moved to the "completed" status. If the worker throws an exception during its processing, the job will automatically be moved to the "failed" status.
+Workers are the actual instances that perform some job based on the jobs that are added in the queue. A worker is equivalent to a "message" receiver in a traditional message queue. The worker duty is to complete the job, if it succeeds, the job will be moved to the "completed" status. If the worker throws an exception during its processing, the job will automatically be moved to the "failed" status.
 
 {% hint style="info" %}
 Failed jobs can be automatically retried, see [Retrying failing jobs](../retrying-failing-jobs.md)
@@ -13,10 +13,10 @@ import { Worker, Job } from 'bullmq';
 
 const worker = new Worker(queueName, async (job: Job) => {
   // Optionally report some progress
-  job.updateProgress(42);
+  await job.updateProgress(42);
 
   // Optionally sending an object as progress
-  job.updateProgress({ foo: 'bar' });
+  await job.updateProgress({ foo: 'bar' });
 
   // Do something with job
   return 'some value';

src/classes/compat.ts

@@ -27,14 +27,14 @@ import { Queue } from './queue';
 import { Worker } from './worker';
 import { QueueEvents } from './queue-events';
 import {
-  JobsOptions,
   QueueOptions,
   RepeatOptions,
   QueueEventsOptions,
   QueueSchedulerOptions,
   WorkerOptions,
   Processor,
 } from '../interfaces';
+import { JobsOptions } from '../types';
 
 type CommonOptions = QueueSchedulerOptions &
   QueueOptions &

src/classes/job.ts

@@ -5,12 +5,16 @@ import {
   BackoffOptions,
   JobJson,
   JobJsonRaw,
-  JobsOptions,
   ParentKeys,
   RedisClient,
   WorkerOptions,
 } from '../interfaces';
-import { JobState, JobJsonSandbox } from '../types';
+import {
+  JobsOptions,
+  JobState,
+  JobJsonSandbox,
+  RedisJobOptions,
+} from '../types';
 import {
   errorObject,
   isEmpty,
@@ -35,6 +39,7 @@ export interface MoveToChildrenOpts {
   };
 }
 
+type ValueOf<T> = T[keyof T];
 export interface DependenciesOpts {
   processed?: {
     cursor?: number;
@@ -233,7 +238,7 @@ export class Job<
     jobId?: string,
   ): Job<T, R, N> {
     const data = JSON.parse(json.data || '{}');
-    const opts = JSON.parse(json.opts || '{}');
+    const opts = Job.optsFromJSON(json.opts);
 
     const job = new this<T, R, N>(
       queue,
@@ -276,6 +281,28 @@ export class Job<
     return job;
   }
 
+  static optsFromJSON(rawOpts?: string): JobsOptions {
+    const opts = JSON.parse(rawOpts || '{}');
+
+    const optionEntries = Object.entries(opts) as Array<
+      [keyof RedisJobOptions, any]
+    >;
+    const options = optionEntries.reduce<Partial<Record<string, any>>>(
+      (acc, item) => {
+        const [attributeName, value] = item;
+        if (attributeName === 'rdof') {
+          acc.removeDependencyOnFail = value;
+        } else {
+          acc[attributeName] = value;
+        }
+        return acc;
+      },
+      {},
+    );
+
+    return options as JobsOptions;
+  }
+
   /**
    * Fetches a Job from the queue given the passed job id.
    *
@@ -311,7 +338,7 @@ export class Job<
       id: this.id,
       name: this.name,
       data: JSON.stringify(typeof this.data === 'undefined' ? {} : this.data),
-      opts: this.opts,
+      opts: this.optsAsJSON(this.opts),
       progress: this.progress,
       attemptsMade: this.attemptsMade,
       finishedOn: this.finishedOn,
@@ -323,6 +350,26 @@ export class Job<
     };
   }
 
+  private optsAsJSON(opts: JobsOptions = {}): RedisJobOptions {
+    const optionEntries = Object.entries(opts) as Array<
+      [keyof JobsOptions, any]
+    >;
+    const options = optionEntries.reduce<Partial<Record<string, any>>>(
+      (acc, item) => {
+        const [attributeName, value] = item;
+        if (attributeName === 'removeDependencyOnFail') {
+          acc.rdof = value;
+        } else {
+          acc[attributeName] = value;
+        }
+        return acc;
+      },
+      {},
+    );
+
+    return options as RedisJobOptions;
+  }
+
   /**
    * Prepares a job to be passed to Sandbox.
    * @returns

src/classes/queue.ts

@@ -1,6 +1,7 @@
 import { get } from 'lodash';
 import { v4 } from 'uuid';
-import { JobsOptions, QueueOptions, RepeatOptions } from '../interfaces';
+import { QueueOptions, RepeatOptions } from '../interfaces';
+import { JobsOptions } from '../types';
 import { isRedisInstance, jobIdForGroup } from '../utils';
 import { BulkJobOptions, Job } from './job';
 import { QueueGetters } from './queue-getters';

src/classes/repeat.ts

@@ -1,5 +1,6 @@
 import { createHash } from 'crypto';
-import { JobsOptions, RepeatOptions } from '../interfaces';
+import { RepeatOptions } from '../interfaces';
+import { JobsOptions } from '../types';
 import { QueueBase } from './queue-base';
 import { Job } from './job';
 import { Scripts } from './scripts';

src/classes/scripts.ts

@@ -17,12 +17,12 @@ import * as semver from 'semver';
 import {
   JobJson,
   JobJsonRaw,
-  JobsOptions,
   QueueSchedulerOptions,
   RedisClient,
   WorkerOptions,
   KeepJobs,
 } from '../interfaces';
+import { JobsOptions } from '../types';
 import { JobState, FinishedStatus, FinishedPropValAttribute } from '../types';
 import { ErrorCode } from '../enums';
 import { array2obj, getParentKey } from '../utils';
@@ -290,6 +290,7 @@ export class Scripts {
       job.parentKey,
       job.opts.attempts,
       job.attemptsMade,
+      job.opts?.removeDependencyOnFail ? '1' : '0',
     ];
 
     return keys.concat(args);

src/commands/includes/updateParentDepsIfNeeded.lua

@@ -1,20 +1,13 @@
 --[[
-  Validate and move or add dependencies to parent.
+  Add processed results, validate and move parent to active if needed.
 ]]
 
+-- Includes
+--- @include "updateParentIfNeeded"
+
 local function updateParentDepsIfNeeded(parentKey, parentQueueKey, parentDependenciesKey,
   parentId, jobIdKey, returnvalue )
   local processedSet = parentKey .. ":processed"
   rcall("HSET", processedSet, jobIdKey, returnvalue)
-  local activeParent = rcall("ZSCORE", parentQueueKey .. ":waiting-children", parentId)
-  if rcall("SCARD", parentDependenciesKey) == 0 and activeParent then 
-    rcall("ZREM", parentQueueKey .. ":waiting-children", parentId)
-    if rcall("HEXISTS", parentQueueKey .. ":meta", "paused") ~= 1 then
-      rcall("RPUSH", parentQueueKey .. ":wait", parentId)
-    else
-      rcall("RPUSH", parentQueueKey .. ":paused", parentId)
-    end
-    local parentEventStream = parentQueueKey .. ":events"
-    rcall("XADD", parentEventStream, "*", "event", "active", "jobId", parentId, "prev", "waiting-children")
-  end
+  updateParentIfNeeded(parentQueueKey, parentDependenciesKey, parentId )
 end

src/commands/includes/updateParentIfNeeded.lua

@@ -0,0 +1,17 @@
+--[[
+  Validate and move parent to active if needed.
+]]
+
+local function updateParentIfNeeded(parentQueueKey, parentDependenciesKey, parentId )
+  local activeParent = rcall("ZSCORE", parentQueueKey .. ":waiting-children", parentId)
+  if rcall("SCARD", parentDependenciesKey) == 0 and activeParent then 
+    rcall("ZREM", parentQueueKey .. ":waiting-children", parentId)
+    if rcall("HEXISTS", parentQueueKey .. ":meta", "paused") ~= 1 then
+      rcall("RPUSH", parentQueueKey .. ":wait", parentId)
+    else
+      rcall("RPUSH", parentQueueKey .. ":paused", parentId)
+    end
+    local parentEventStream = parentQueueKey .. ":events"
+    rcall("XADD", parentEventStream, "*", "event", "active", "jobId", parentId, "prev", "waiting-children")
+  end
+end

src/commands/moveToFinished-8.lua

@@ -30,6 +30,7 @@
       ARGV[14] parentKey
       ARGV[15] max attempts
       ARGV[16] attemptsMade
+      ARGV[17] removeDependencyOnFail
 
     Output:
       0 OK
@@ -44,11 +45,12 @@
 local rcall = redis.call
 
 -- Includes
+--- @include "includes/updateParentDepsIfNeeded"
+--- @include "includes/updateParentIfNeeded"
 --- @include "includes/destructureJobKey"
 --- @include "includes/moveJobFromWaitToActive"
 --- @include "includes/removeJob"
 --- @include "includes/trimEvents"
---- @include "includes/updateParentDepsIfNeeded"
 
 local jobIdKey = KEYS[3]
 if rcall("EXISTS", jobIdKey) == 1 then -- // Make sure job exists
@@ -91,13 +93,17 @@ if rcall("EXISTS", jobIdKey) == 1 then -- // Make sure job exists
         parentId = getJobIdFromKey(ARGV[14])
         parentQueueKey = getJobKeyPrefix(ARGV[14], ":" .. parentId)
     end
-    if parentId ~= "" and ARGV[5] == "completed" then
+    if parentId ~= "" and (ARGV[5] == "completed" or ARGV[17] == "1")then
         local parentKey = parentQueueKey .. ":" .. parentId
         local dependenciesSet = parentKey .. ":dependencies"
         local result = rcall("SREM", dependenciesSet, jobIdKey)
         if result == 1 then
-            updateParentDepsIfNeeded(parentKey, parentQueueKey, dependenciesSet,
-                                     parentId, jobIdKey, ARGV[4])
+            if ARGV[5] == "completed" then
+                updateParentDepsIfNeeded(parentKey, parentQueueKey, dependenciesSet,
+                                        parentId, jobIdKey, ARGV[4])
+            elseif ARGV[17] == "1" then
+                updateParentIfNeeded(parentQueueKey, dependenciesSet, parentId )
+            end
         end
     end
 

src/interfaces/flow-job.ts

@@ -1,4 +1,4 @@
-import { JobsOptions } from './jobs-options';
+import { JobsOptions } from '../types';
 import { QueueOptions } from './queue-options';
 
 export interface FlowJob {

src/interfaces/job-json.ts

@@ -1,10 +1,10 @@
-import { JobsOptions } from './jobs-options';
+import { RedisJobOptions } from '../types';
 
 export interface JobJson {
   id: string;
   name: string;
   data: string;
-  opts: JobsOptions;
+  opts: RedisJobOptions;
   progress: number | object;
   attemptsMade: number;
   finishedOn?: number;

src/interfaces/jobs-options.ts

@@ -1,6 +1,6 @@
 import { RepeatOptions, KeepJobs, BackoffOptions } from './';
 
-export interface JobsOptions {
+export interface JobOptionsBase {
   /**
    * Timestamp when the job was created. Defaults to `Date.now()`.
    */

src/interfaces/parent.ts

@@ -1,4 +1,4 @@
-import { JobsOptions } from './jobs-options';
+import { JobsOptions } from '../types';
 
 /**
  * Describes the parent for a Job.

src/interfaces/queue-options.ts

@@ -1,4 +1,4 @@
-import { JobsOptions } from './jobs-options';
+import { JobsOptions } from '../types';
 import { ConnectionOptions } from './redis-options';
 
 export enum ClientType {

src/interfaces/sandboxed-job.ts

@@ -1,5 +1,5 @@
+import { JobsOptions } from '../types';
 import { JobJson } from './job-json';
-import { JobsOptions } from './jobs-options';
 
 /**
  * @see {@link https://docs.bullmq.io/guide/workers/sandboxed-processors}

src/types/index.ts

@@ -1,3 +1,4 @@
 export * from './finished-status';
 export * from './job-json-sandbox';
+export * from './job-options';
 export * from './job-type';

src/types/job-options.ts

@@ -0,0 +1,15 @@
+import { JobOptionsBase } from '../interfaces';
+
+export type JobsOptions = JobOptionsBase & {
+  /**
+   * If true, removes the job from its parent dependencies when it fails after all attempts.
+   */
+  removeDependencyOnFail?: boolean;
+};
+
+export type RedisJobOptions = JobOptionsBase & {
+  /**
+   * If true, removes the job from its parent dependencies when it fails after all attempts.
+   */
+  rdof?: boolean;
+};

src/utils.ts

@@ -6,11 +6,11 @@ import { v4 } from 'uuid';
 import { get } from 'lodash';
 import {
   RedisClient,
-  JobsOptions,
   QueueOptions,
   ChildMessage,
   ParentMessage,
 } from './interfaces';
+import { JobsOptions } from './types';
 import { ChildProcess } from 'child_process';
 
 export const errorObject: { [index: string]: any } = { value: null };