fix(retry-job): consider paused queue

src/classes/job.ts

@@ -521,7 +521,9 @@ export class Job<
         command = 'delayed';
       } else {
         // Retry immediately
-        (<any>multi).retryJob(this.scripts.retryJobArgs(this));
+        (<any>multi).retryJob(
+          this.scripts.retryJobArgs(this.id, this.opts.lifo, token),
+        );
         command = 'retry';
       }
     } else {

src/classes/scripts.ts

@@ -605,20 +605,16 @@ export class Scripts {
     ]);
   }
 
-  retryJobArgs<T = any, R = any, N extends string = string>(
-    job: Job<T, R, N>,
-  ): string[] {
-    const jobId = job.id;
-
-    const keys = ['active', 'wait', jobId].map(name => {
+  retryJobArgs(jobId: string, lifo: boolean, token: string): string[] {
+    const keys = ['active', 'wait', 'paused', jobId, 'meta'].map(name => {
       return this.queue.toKey(name);
     });
 
     keys.push(this.queue.keys.events);
 
-    const pushCmd = (job.opts.lifo ? 'R' : 'L') + 'PUSH';
+    const pushCmd = (lifo ? 'R' : 'L') + 'PUSH';
 
-    return keys.concat([pushCmd, jobId]);
+    return keys.concat([pushCmd, jobId, token]);
   }
 
   protected retryJobsArgs(
@@ -694,7 +690,7 @@ export class Scripts {
     }
   }
 
-  async moveToActive<T, R, N extends string>(token: string, jobId?: string) {
+  async moveToActive(token: string, jobId?: string) {
     const client = await this.queue.client;
     const opts = this.queue.opts as WorkerOptions;
 

src/commands/retryJob-6.lua

@@ -0,0 +1,53 @@
+--[[
+  Retries a failed job by moving it back to the wait queue.
+
+    Input:
+      KEYS[1] 'active',
+      KEYS[2] 'wait'
+      KEYS[3] 'paused'
+      KEYS[4] job key
+      KEYS[5] 'meta'
+      KEYS[6] events stream
+
+      ARGV[1]  pushCmd
+      ARGV[2]  jobId
+      ARGV[3]  token
+
+    Events:
+      'waiting'
+
+    Output:
+     0  - OK
+     -1 - Missing key
+     -2 - Missing lock
+]]
+local rcall = redis.call
+
+if rcall("EXISTS", KEYS[4]) == 1 then
+
+  if ARGV[3] ~= "0" then
+    local lockKey = KEYS[4] .. ':lock'
+    if rcall("GET", lockKey) == ARGV[3] then
+      rcall("DEL", lockKey)
+    else
+      return -2
+    end
+  end
+
+  local target
+  if rcall("HEXISTS", KEYS[5], "paused") ~= 1 then
+    target = KEYS[2]
+  else
+    target = KEYS[3]
+  end
+
+  rcall("LREM", KEYS[1], 0, ARGV[2])
+  rcall(ARGV[1], target, ARGV[2])
+
+  -- Emit waiting event
+  rcall("XADD", KEYS[6], "*", "event", "waiting", "jobId", ARGV[2], "prev", "failed");
+
+  return 0
+else
+  return -1
+end

tests/test_pause.ts

@@ -152,7 +152,7 @@ describe('Pause', function () {
     let counter = 2;
     let process;
     const processPromise = new Promise<void>(resolve => {
-      process = async (job: Job) => {
+      process = async () => {
         expect(worker.isPaused()).to.be.eql(false);
         counter--;
         if (counter === 0) {
@@ -335,7 +335,7 @@ describe('Pause', function () {
   it('should pause and resume worker without error', async function () {
     const worker = new Worker(
       queueName,
-      async job => {
+      async () => {
         await delay(100);
       },
       { connection },
@@ -352,4 +352,45 @@ describe('Pause', function () {
 
     return worker.close();
   });
+
+  describe('when backoff is 0', () => {
+    it('moves job into paused queue', async () => {
+      await queue.add('test', { foo: 'bar' }, { attempts: 2, backoff: 0 });
+
+      let worker: Worker;
+      const processing = new Promise<void>(resolve => {
+        worker = new Worker(
+          queueName,
+          async job => {
+            await delay(10);
+            if (job.attemptsMade == 1) {
+              await queue.pause();
+              throw new Error('Not yet!');
+            }
+
+            resolve();
+          },
+          {
+            connection,
+          },
+        );
+      });
+
+      const waitingEvent = new Promise<void>(resolve => {
+        queueEvents.on('waiting', async ({ prev }) => {
+          if (prev === 'failed') {
+            const count = await queue.getJobCountByTypes('paused');
+            expect(count).to.be.equal(1);
+            await queue.resume();
+            resolve();
+          }
+        });
+      });
+
+      await waitingEvent;
+      await processing;
+
+      await worker.close();
+    });
+  });
 });