fix(rate-limit): move job to wait even if ttl is 0 (#2403)

Author: roggervalf

src/classes/scripts.ts

@@ -222,9 +222,7 @@ export class Scripts {
     return <string>result;
   }
 
-  async pause(pause: boolean): Promise<void> {
-    const client = await this.queue.client;
-
+  protected pauseArgs(pause: boolean): (string | number)[] {
     let src = 'wait',
       dst = 'paused';
     if (!pause) {
@@ -242,7 +240,17 @@ export class Scripts {
       this.queue.keys.marker,
     );
 
-    return (<any>client).pause(keys.concat([pause ? 'paused' : 'resumed']));
+    const args = [pause ? 'paused' : 'resumed'];
+
+    return keys.concat(args);
+  }
+
+  async pause(pause: boolean): Promise<void> {
+    const client = await this.queue.client;
+
+    const args = this.pauseArgs(pause);
+
+    return (<any>client).pause(args);
   }
 
   private removeRepeatableArgs(
@@ -1034,6 +1042,7 @@ export class Scripts {
       this.queue.keys.meta,
       this.queue.keys.limiter,
       this.queue.keys.prioritized,
+      this.queue.keys.marker,
       this.queue.keys.events,
     ];
 

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

@@ -1,16 +1,17 @@
 --[[
   Function to move job from active state to wait.
   Input:
-    KEYS[1] active key
-    KEYS[2] wait key
+    KEYS[1]  active key
+    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] event 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
 
     ARGV[1] job id
     ARGV[2] lock token
@@ -19,7 +20,9 @@
 local rcall = redis.call
 
 -- Includes
+--- @include "includes/addJobInTargetList"
 --- @include "includes/pushBackJobWithPriority"
+--- @include "includes/getOrSetMaxEvents"
 --- @include "includes/getTargetQueueList"
 
 local jobId = ARGV[1]
@@ -28,10 +31,11 @@ local lockKey = KEYS[4]
 
 local lockToken = rcall("GET", lockKey)
 local pttl = rcall("PTTL", KEYS[7])
-if lockToken == token and pttl > 0 then
+if lockToken == token then
+  local metaKey = KEYS[6]
   local removed = rcall("LREM", KEYS[1], 1, jobId)
-  if (removed > 0) then
-    local target = getTargetQueueList(KEYS[6], KEYS[2], KEYS[5])
+  if removed > 0 then
+    local target, isPaused = getTargetQueueList(metaKey, KEYS[2], KEYS[5])
 
     rcall("SREM", KEYS[3], jobId)
 
@@ -40,13 +44,16 @@ if lockToken == token and pttl > 0 then
     if priority > 0 then
       pushBackJobWithPriority(KEYS[8], priority, jobId)
     else
-      rcall("RPUSH", target, jobId)
+      addJobInTargetList(target, KEYS[9], "RPUSH", isPaused, jobId)
     end
 
     rcall("DEL", lockKey)
 
+    local maxEvents = getOrSetMaxEvents(metaKey)
+
     -- Emit waiting event
-    rcall("XADD", KEYS[9], "*", "event", "waiting", "jobId", jobId)
+    rcall("XADD", KEYS[10], "MAXLEN", "~", maxEvents, "*", "event", "waiting",
+      "jobId", jobId)
   end
 end
 

tests/test_rate_limiter.ts

@@ -423,6 +423,69 @@ describe('Rate Limiter', function () {
       await worker.close();
     });
 
+    describe('when rate limit is too low', () => {
+      it('should move job to wait anyway', async function () {
+        this.timeout(4000);
+
+        const numJobs = 10;
+        const dynamicLimit = 1;
+        const duration = 100;
+
+        const ttl = await queue.getRateLimitTtl();
+        expect(ttl).to.be.equal(-2);
+
+        const worker = new Worker(
+          queueName,
+          async job => {
+            if (job.attemptsStarted === 1) {
+              delay(50);
+              await worker.rateLimit(dynamicLimit);
+              const currentTtl = await queue.getRateLimitTtl();
+              expect(currentTtl).to.be.lessThanOrEqual(dynamicLimit);
+              throw Worker.RateLimitError();
+            }
+          },
+          {
+            connection,
+            prefix,
+            maxStalledCount: 0,
+            limiter: {
+              max: 1,
+              duration,
+            },
+          },
+        );
+
+        const result = new Promise<void>((resolve, reject) => {
+          queueEvents.on(
+            'completed',
+            // after every job has been completed
+            after(numJobs, async () => {
+              try {
+                resolve();
+              } catch (err) {
+                reject(err);
+              }
+            }),
+          );
+
+          queueEvents.on('failed', async err => {
+            await worker.close();
+            reject(err);
+          });
+        });
+
+        const jobs = Array.from(Array(numJobs).keys()).map(() => ({
+          name: 'rate test',
+          data: {},
+        }));
+        await queue.addBulk(jobs);
+
+        await result;
+        await worker.close();
+      });
+    });
+
     describe('when reaching max attempts and we want to move the job to failed', () => {
       it('should throw Unrecoverable error', async function () {
         const dynamicLimit = 550;