fix(scripts): throw error when moving non-active job to delayed (#2740)

lib/commands/includes/removeLock.lua

@@ -0,0 +1,19 @@
+local function removeLock(jobKey, stalledKey, token, jobId)
+  if token ~= "0" then
+    local lockKey = jobKey .. ':lock'
+    local lockToken = rcall("GET", lockKey)
+    if lockToken == token then
+      rcall("DEL", lockKey)
+      rcall("SREM", stalledKey, jobId)
+    else
+      if lockToken then
+        -- Lock exists but token does not match
+        return -6
+      else
+        -- Lock is missing completely
+        return -2
+      end
+    end
+  end
+  return 0
+end

lib/commands/moveToDelayed-4.lua

@@ -21,22 +21,21 @@
 ]]
 local rcall = redis.call
 
+-- Includes
+--- @include "includes/removeLock"
+
 if rcall("EXISTS", KEYS[3]) == 1 then
-  -- Check for job lock
-  if ARGV[3] ~= "0" then
-    local lockKey = KEYS[3] .. ':lock'
-    if rcall("GET", lockKey) == ARGV[3] then
-      rcall("DEL", lockKey)
-      rcall("SREM", KEYS[4], ARGV[2])
-    else
-      return -2
-    end
+  local errorCode = removeLock(KEYS[3], KEYS[4], ARGV[3], ARGV[2])
+  if errorCode < 0 then
+    return errorCode
   end
 
+  local numRemovedElements = rcall("LREM", KEYS[1], -1, ARGV[2])
+  if numRemovedElements < 1 then return -3 end
+
   local score = tonumber(ARGV[1])
   rcall("ZADD", KEYS[2], score, ARGV[2])
   rcall("PUBLISH", KEYS[2], (score / 0x1000))
-  rcall("LREM", KEYS[1], 0, ARGV[2])
 
   return 0
 else

lib/job.js

@@ -339,7 +339,7 @@ Job.prototype.moveToFailed = async function(err, ignoreLock) {
   const results = await multi.exec();
   const code = _.last(results)[1];
   if (code < 0) {
-    throw scripts.finishedErrors(code, this.id, command);
+    throw scripts.finishedErrors(code, this.id, command, 'active');
   }
 };
 

lib/scripts.js

@@ -238,20 +238,24 @@ const scripts = {
     );
     return job.queue.client.moveToFinished(args).then(result => {
       if (result < 0) {
-        throw scripts.finishedErrors(result, job.id, 'finished');
+        throw scripts.finishedErrors(result, job.id, 'finished', 'active');
       } else if (result) {
         return raw2jobData(result);
       }
       return 0;
     });
   },
 
-  finishedErrors(code, jobId, command) {
+  finishedErrors(code, jobId, command, state) {
     switch (code) {
       case -1:
         return new Error('Missing key for job ' + jobId + ' ' + command);
       case -2:
         return new Error('Missing lock for job ' + jobId + ' ' + command);
+      case -3:
+        return new Error(`Job ${jobId} is not in the ${state} state. ${command}`);
+      case -6:
+        return new Error(`Lock mismatch for job ${jobId}. Cmd ${command} from ${state}`);
     }
   },
 

test/test_job.js

@@ -726,6 +726,9 @@ describe('Job', () => {
           .then(isFailed => {
             expect(isFailed).to.be(false);
           })
+          .then(() => {
+            return scripts.moveToActive(queue);
+          })
           .then(() => {
             return job.moveToFailed(new Error('test error'), true);
           })
@@ -893,7 +896,9 @@ describe('Job', () => {
           })
           .then(state => {
             expect(state).to.be('completed');
-            return client.zrem(queue.toKey('completed'), job.id);
+            return client.zrem(queue.toKey('completed'), job.id).then(()=>{
+              return client.lpush(queue.toKey('active'), job.id)
+            });
           })
           .then(() => {
             return job.moveToDelayed(Date.now() + 10000, true);
@@ -907,7 +912,9 @@ describe('Job', () => {
           })
           .then(state => {
             expect(state).to.be('delayed');
-            return client.zrem(queue.toKey('delayed'), job.id);
+            return client.zrem(queue.toKey('delayed'), job.id).then(()=>{
+              return client.lpush(queue.toKey('active'), job.id)
+            });
           })
           .then(() => {
             return job.moveToFailed(new Error('test'), true);