fix(dynamic-rate-limit): validate job existence

Author: roggervalf

src/classes/scripts.ts

@@ -1387,13 +1387,11 @@ export class Scripts {
    */
   async moveJobFromActiveToWait(jobId: string, token: string) {
     const client = await this.queue.client;
-    const lockKey = `${this.queue.toKey(jobId)}:lock`;
 
     const keys: (string | number)[] = [
       this.queue.keys.active,
       this.queue.keys.wait,
       this.queue.keys.stalled,
-      lockKey,
       this.queue.keys.paused,
       this.queue.keys.meta,
       this.queue.keys.limiter,
@@ -1404,13 +1402,22 @@ export class Scripts {
 
     const args = [jobId, token, this.queue.toKey(jobId)];
 
-    const pttl = await this.execCommand(
+    const result = await this.execCommand(
       client,
       'moveJobFromActiveToWait',
       keys.concat(args),
     );
 
-    return pttl < 0 ? 0 : pttl;
+    if (result < 0) {
+      throw this.finishedErrors({
+        code: result,
+        jobId,
+        command: 'moveJobFromActiveToWait',
+        state: 'active',
+      });
+    }
+
+    return result;
   }
 
   async obliterate(opts: { force: boolean; count: number }): Promise<number> {

src/commands/moveJobFromActiveToWait-10.lua renamed to src/commands/moveJobFromActiveToWait-9.lua

@@ -5,13 +5,12 @@
     KEYS[2]  wait key
     
     KEYS[3]  stalled key
-    KEYS[4]  job lock key
-    KEYS[5]  paused key
-    KEYS[6]  meta key
-    KEYS[7]  limiter key
-    KEYS[8]  prioritized key
-    KEYS[9]  marker key
-    KEYS[10] event key
+    KEYS[4]  paused key
+    KEYS[5]  meta key
+    KEYS[6]  limiter key
+    KEYS[7]  prioritized key
+    KEYS[8]  marker key
+    KEYS[9] event key
 
     ARGV[1] job id
     ARGV[2] lock token
@@ -24,37 +23,45 @@ local rcall = redis.call
 --- @include "includes/pushBackJobWithPriority"
 --- @include "includes/getOrSetMaxEvents"
 --- @include "includes/getTargetQueueList"
+--- @include "includes/removeLock"
 
 local jobId = ARGV[1]
 local token = ARGV[2]
-local lockKey = KEYS[4]
+local jobKey = ARGV[3]
 
-local lockToken = rcall("GET", lockKey)
-local pttl = rcall("PTTL", KEYS[7])
-if lockToken == token then
-  local metaKey = KEYS[6]
+if rcall("EXISTS", jobKey) == 1 then
+  local errorCode = removeLock(jobKey, KEYS[3], token, jobId)
+  if errorCode < 0 then
+    return errorCode
+  end
+
+  local metaKey = KEYS[5]
   local removed = rcall("LREM", KEYS[1], 1, jobId)
   if removed > 0 then
-    local target, isPausedOrMaxed = getTargetQueueList(metaKey, KEYS[1], KEYS[2], KEYS[5])
-
-    rcall("SREM", KEYS[3], jobId)
+    local target, isPausedOrMaxed = getTargetQueueList(metaKey, KEYS[1], KEYS[2], KEYS[4])
 
     local priority = tonumber(rcall("HGET", ARGV[3], "priority")) or 0
 
     if priority > 0 then
-      pushBackJobWithPriority(KEYS[8], priority, jobId)
+      pushBackJobWithPriority(KEYS[7], priority, jobId)
     else
-      addJobInTargetList(target, KEYS[9], "RPUSH", isPausedOrMaxed, jobId)
+      addJobInTargetList(target, KEYS[8], "RPUSH", isPausedOrMaxed, jobId)
     end
 
-    rcall("DEL", lockKey)
-
     local maxEvents = getOrSetMaxEvents(metaKey)
 
     -- Emit waiting event
-    rcall("XADD", KEYS[10], "MAXLEN", "~", maxEvents, "*", "event", "waiting",
+    rcall("XADD", KEYS[9], "MAXLEN", "~", maxEvents, "*", "event", "waiting",
       "jobId", jobId)
   end
+else
+  return -1
 end
 
-return pttl
+local pttl = rcall("PTTL", KEYS[6])
+
+if pttl > 0 then
+  return pttl
+else
+  return 0
+end

tests/test_rate_limiter.ts

@@ -423,6 +423,53 @@ describe('Rate Limiter', function () {
       await worker.close();
     });
 
+    describe('when job does not exist', () => {
+      it('should fail with job existence error', async () => {
+        const dynamicLimit = 250;
+        const duration = 100;
+
+        const worker = new Worker(
+          queueName,
+          async job => {
+            if (job.attemptsStarted === 1) {
+              await queue.rateLimit(dynamicLimit);
+              await queue.obliterate({ force: true });
+              throw Worker.RateLimitError();
+            }
+          },
+          {
+            autorun: false,
+            concurrency: 10,
+            drainDelay: 10, // If test hangs, 10 seconds here helps to fail quicker.
+            limiter: {
+              max: 2,
+              duration,
+            },
+            connection,
+            prefix,
+          },
+        );
+
+        await worker.waitUntilReady();
+
+        const failing = new Promise<void>(resolve => {
+          worker.on('error', err => {
+            expect(err.message).to.be.equal(
+              `Missing key for job ${job.id}. moveJobFromActiveToWait`,
+            );
+            resolve();
+          });
+        });
+
+        const job = await queue.add('test', { foo: 'bar' });
+
+        worker.run();
+
+        await failing;
+        await worker.close();
+      }).timeout(4000);
+    });
+
     describe('when rate limit is too low', () => {
       it('should move job to wait anyway', async function () {
         this.timeout(4000);