feat: add remove support for flows

Author: manast

src/classes/flow.ts

@@ -8,8 +8,9 @@ import { RedisConnection } from './redis-connection';
 import { KeysMap, QueueKeys } from './queue-keys';
 import { FlowJob } from '../interfaces/flow-job';
 import { Job } from './job';
+import { ParentOpts } from './scripts';
 
-interface JobNode {
+export interface JobNode {
   job: Job;
   children?: JobNode[];
 }
@@ -123,6 +124,8 @@ export class Flow extends EventEmitter {
       node.opts?.jobId || parentId,
     );
 
+    const parentKey = getParentKey(parent?.parentOpts);
+
     if (node.children) {
       // Create parent job, will be a job in status "waiting-children".
       const queueKeysParent = new QueueKeys(node.prefix);
@@ -134,6 +137,7 @@ export class Flow extends EventEmitter {
       job.addJob(<Redis>(multi as unknown), {
         parentDependenciesKey: parent?.parentDependenciesKey,
         waitChildrenKey,
+        parentKey,
       });
 
       const parentDependenciesKey = `${queueKeysParent.toKey(
@@ -153,6 +157,7 @@ export class Flow extends EventEmitter {
     } else {
       job.addJob(<Redis>(multi as unknown), {
         parentDependenciesKey: parent?.parentDependenciesKey,
+        parentKey,
       });
 
       return { job };
@@ -208,3 +213,9 @@ export class Flow extends EventEmitter {
     return this.connection.disconnect();
   }
 }
+
+function getParentKey(opts: { id: string; queue: string }) {
+  if (opts) {
+    return `${opts.queue}:${opts.id}`;
+  }
+}

src/classes/queue.ts

@@ -142,6 +142,18 @@ export class Queue<
     return (await this.repeat).removeRepeatableByKey(key);
   }
 
+  /**
+   * Removes the given job from the queue as well as all its
+   * dependencies.
+   *
+   * @param jobId The if of the job to remove
+   * @returns 1 if it managed to remove the job or -1 if the job or
+   * any of its dependencies was locked.
+   */
+  async remove(jobId: string) {
+    return Scripts.remove(this, jobId);
+  }
+
   /**
    * Drains the queue, i.e., removes all jobs that are waiting
    * or delayed, but not active, completed or failed.

src/classes/scripts.ts

@@ -33,6 +33,7 @@ export type MinimalQueue = Pick<
 export type ParentOpts = {
   waitChildrenKey?: string;
   parentDependenciesKey?: string;
+  parentKey?: string;
 };
 
 export class Scripts {
@@ -60,6 +61,7 @@ export class Scripts {
     parentOpts: ParentOpts = {
       waitChildrenKey: null,
       parentDependenciesKey: null,
+      parentKey: null,
     },
   ) {
     const queueKeys = queue.keys;
@@ -87,6 +89,7 @@ export class Scripts {
       opts.delay ? job.timestamp + opts.delay : 0,
       opts.priority || 0,
       opts.lifo ? 'RPUSH' : 'LPUSH',
+      parentOpts.parentKey,
     ];
 
     keys = keys.concat(<string[]>args);
@@ -114,18 +117,8 @@ export class Scripts {
   static async remove(queue: MinimalQueue, jobId: string) {
     const client = await queue.client;
 
-    const keys = [
-      'active',
-      'wait',
-      'delayed',
-      'paused',
-      'completed',
-      'failed',
-      'priority',
-      jobId,
-      `${jobId}:logs`,
-    ].map(name => queue.toKey(name));
-    return (<any>client).removeJob(keys.concat([queue.keys.events, jobId]));
+    const keys = [jobId].map(name => queue.toKey(name));
+    return (<any>client).removeJob(keys.concat([jobId]));
   }
 
   static async extendLock(

src/commands/addJob-10.lua

@@ -36,6 +36,7 @@
       ARGV[8]  delayedTimestamp
       ARGV[9]  priority
       ARGV[10] LIFO
+      ARGV[11] parentKey?
 ]]
 local jobId
 local jobIdKey
@@ -56,7 +57,7 @@ end
 
 -- Store the job.
 rcall("HMSET", jobIdKey, "name", ARGV[3], "data", ARGV[4], "opts", ARGV[5],
-      "timestamp", ARGV[6], "delay", ARGV[7], "priority", ARGV[9])
+      "timestamp", ARGV[6], "delay", ARGV[7], "priority", ARGV[9], "parentKey", ARGV[11])
 
 -- Check if job is delayed
 local delayedTimestamp = tonumber(ARGV[8])

src/commands/extendLock-2.lua

@@ -14,6 +14,7 @@
 ]]
 local rcall = redis.call
 if rcall("GET", KEYS[1]) == ARGV[1] then
+  --   if rcall("SET", KEYS[1], ARGV[1], "PX", ARGV[2], "XX") then
   if rcall("SET", KEYS[1], ARGV[1], "PX", ARGV[2]) then
     rcall("SREM", KEYS[2], ARGV[3])
     return 1

src/commands/removeJob-1.lua

@@ -0,0 +1,119 @@
+--[[
+    Remove a job from all the queues it may be in as well as all its data.
+    In order to be able to remove a job, it cannot be active.
+
+     Input:
+      KEYS[1] jobId
+      ARGV[1]  jobId
+
+     Events:
+      'removed'
+]]
+
+local rcall = redis.call
+
+local getJobIdFromKey = function (jobKey)
+    return string.match(jobKey, ".*:(.*)")
+end
+
+local getJobKeyPrefix = function (jobKey, jobId)
+    return string.sub(jobKey, 0, #jobKey - #jobId)
+end
+
+-- recursively check if there are no locks on the
+-- jobs to be removed.
+local function isLocked( prefix, jobId)
+    local jobKey = prefix .. jobId;
+
+    -- Check if this job is locked
+    local lockKey = jobKey .. ':lock'
+    local lock = rcall("GET", lockKey)
+    if not lock then
+        local dependencies = rcall("SMEMBERS", jobKey .. ":dependencies")
+        if (#dependencies > 0) then
+            for i, childJobKey in ipairs(dependencies) do
+                -- We need to get the jobId for this job.
+                local childJobId = getJobIdFromKey(childJobKey)
+                local childJobPrefix = getJobKeyPrefix(childJobKey, childJobId)
+                local result = isLocked( childJobPrefix, childJobId )
+                if result then
+                    return true
+                end
+            end
+        end
+        return false
+    end
+    return true
+end
+
+local function removeJob( prefix, jobId)
+    local jobKey = prefix .. jobId;
+
+    -- Check if this job has a parent. If so we will just remove it from
+    -- the parent child list, but if it is the last child we should move the parent to "wait/paused" 
+    -- which requires code from "moveToFinished"
+    local parentKey = rcall("HGET", jobKey, "parentKey")
+    if( (type(parentKey) == "string") and (rcall("EXISTS", parentKey) == 1)) then
+        local parentDependenciesKey = parentKey .. ":dependencies"
+        local result = rcall("SREM", parentDependenciesKey, jobKey)
+        if rcall("SCARD", parentDependenciesKey) == 0 then 
+            local parentId = getJobIdFromKey(parentKey)
+            local parentPrefix = getJobKeyPrefix(parentKey, parentId)
+
+            rcall("ZREM", parentPrefix .. "waiting-children", parentId)
+
+            if rcall("HEXISTS", parentPrefix .. "meta", "paused") ~= 1 then
+                rcall("RPUSH", parentPrefix .. "wait", parentId)
+            else
+                rcall("RPUSH", parentPrefix .. "parentPrefixpaused", parentId)
+            end
+
+            local parentEventStream = parentPrefix .. "events"
+            rcall("XADD", parentEventStream, "*", "event", "active", "jobId", parentId, "prev", "waiting-children")
+        end
+    end
+
+    rcall("LREM", prefix .. "active", 0, jobId)
+    rcall("LREM", prefix .. "wait", 0, jobId)
+    rcall("ZREM", prefix .. "delayed", jobId)
+    rcall("LREM", prefix .. "paused", 0, jobId)
+    rcall("ZREM", prefix .. "completed", jobId)
+    rcall("ZREM", prefix .. "failed", jobId)
+    rcall("ZREM", prefix .. "priority", jobId)
+    rcall("ZREM", prefix .. "waiting-children", jobId)
+    rcall("DEL", jobKey)
+    rcall("DEL", jobKey .. ":logs")
+
+    -- Check if this job has children
+    -- If so, we are going to try to remove the children recursively in deep first way because
+    -- if some job is locked we must exit with and error.
+    local dependencies = rcall("SMEMBERS", jobKey .. ":dependencies")
+    if (#dependencies > 0) then
+        for i, childJobKey in ipairs(dependencies) do
+            -- We need to get the jobId for this job.
+            local childJobId = getJobIdFromKey(childJobKey)
+            local childJobPrefix = getJobKeyPrefix(childJobKey, childJobId)
+            removeJob( childJobPrefix, childJobId )
+        end
+    end
+
+    rcall("DEL", jobKey .. ":dependencies")
+
+    -- -- delete keys related to rate limiter
+        -- local limiterIndexTable = KEYS[10] .. ":index"
+        -- local limitedSetKey = rcall("HGET", limiterIndexTable, jobId)
+        -- if limitedSetKey then
+        --     rcall("SREM", limitedSetKey, jobId)
+        --     rcall("HDEL", limiterIndexTable, jobId)
+    -- end
+
+    rcall("XADD", prefix .. "events", "*", "event", "removed", "jobId", jobId, "prev", "unknown");
+end
+
+local prefix = getJobKeyPrefix(KEYS[1], ARGV[1])
+
+if not isLocked(prefix, ARGV[1]) then
+    removeJob(prefix, ARGV[1])
+    return 1
+end
+return 0

src/commands/removeJob-10.lua

@@ -17,7 +17,8 @@
 
      Events:
       'waiting'
-]] local rcall = redis.call
+]]
+local rcall = redis.call
 
 -- Try to get as much as 1000 jobs at once
 local jobs = rcall("ZRANGEBYSCORE", KEYS[1], 0, tonumber(ARGV[2]) * 0x1000,