feat(queue-events): add retries-exhausted event

src/classes/job.ts

@@ -480,6 +480,9 @@ export class Job<
         this.opts.removeOnFail,
         token,
         fetchNext,
+        this.opts.attempts && this.attemptsMade >= this.opts.attempts
+          ? this.attemptsMade
+          : 0,
       );
       (<any>multi).moveToFinished(args);
       command = 'failed';

src/classes/queue-events.ts

@@ -112,6 +112,16 @@ export interface QueueEventsListener {
    */
   resumed: (args: {}, id: string) => void;
 
+  /**
+   * Listen to 'retries-exhausted' event.
+   *
+   * This event is triggered when a job has retried the maximum attempts.
+   */
+  'retries-exhausted': (
+    args: { jobId: string; attemptsMade: string },
+    id: string,
+  ) => void;
+
   /**
    * Listen to 'stalled' event.
    *

src/classes/scripts.ts

@@ -232,6 +232,7 @@ export class Scripts {
     target: string,
     token: string,
     fetchNext = true,
+    retriesExhausted = 0,
   ) {
     const queueKeys = queue.keys;
     const opts: WorkerOptions = <WorkerOptions>queue.opts;
@@ -269,6 +270,7 @@ export class Scripts {
       job.opts?.parent?.id,
       job.opts?.parent?.queue,
       job.parentKey,
+      retriesExhausted,
     ];
 
     return keys.concat(args);
@@ -379,6 +381,7 @@ export class Scripts {
     removeOnFailed: boolean | number,
     token: string,
     fetchNext = false,
+    retriesExhausted = 0,
   ) {
     return this.moveToFinishedArgs(
       queue,
@@ -389,6 +392,7 @@ export class Scripts {
       'failed',
       token,
       fetchNext,
+      retriesExhausted,
     );
   }
 

src/commands/moveToFinished-8.lua

@@ -28,6 +28,7 @@
       ARGV[12] parentId
       ARGV[13] parentQueue
       ARGV[14] parentKey
+      ARGV[14] retriesExhausted
 
     Output:
       0 OK
@@ -131,6 +132,10 @@ if rcall("EXISTS",jobIdKey) == 1 then -- // Make sure job exists
     rcall("XADD", KEYS[6], "*", "event", ARGV[5], "jobId", jobId, ARGV[3],
           ARGV[4])
 
+    if ARGV[15] > 0 then
+      rcall("XADD", KEYS[6], "*", "event", "retries-exhausted", "jobId", jobId, "attemptsMade", ARGV[15])
+    end
+
     -- Try to get next job to avoid an extra roundtrip if the queue is not closing,
     -- and not rate limited.
     if (ARGV[8] == "1") then

tests/test_worker.ts

@@ -1427,6 +1427,41 @@ describe('workers', function () {
   });
 
   describe('Retries and backoffs', () => {
+    describe('when attempts is 1', () => {
+      it('should execute job only once and emits retries-exhausted event', async () => {
+        const worker = new Worker(
+          queueName,
+          async () => {
+            throw new Error('failed');
+          },
+          { connection },
+        );
+
+        await worker.waitUntilReady();
+
+        const job = await queue.add(
+          'test',
+          { foo: 'bar' },
+          {
+            attempts: 1,
+          },
+        );
+
+        await new Promise<void>(resolve => {
+          queueEvents.on(
+            'retries-exhausted',
+            async ({ jobId, attemptsMade }) => {
+              expect(jobId).to.eql(job.id);
+              expect(1).to.eql(Number(attemptsMade));
+              resolve();
+            },
+          );
+        });
+
+        await worker.close();
+      });
+    });
+
     it('should not retry a job if it has been marked as unrecoverable', async () => {
       let tries = 0;
 
@@ -1507,7 +1542,7 @@ describe('workers', function () {
 
       await worker.waitUntilReady();
 
-      await queue.add(
+      const job = await queue.add(
         'test',
         { foo: 'bar' },
         {
@@ -1519,10 +1554,10 @@ describe('workers', function () {
         worker.on('completed', () => {
           reject(new Error('Failed job was retried more than it should be!'));
         });
-        worker.on('failed', () => {
-          if (tries === 3) {
-            resolve();
-          }
+        queueEvents.on('retries-exhausted', async ({ jobId, attemptsMade }) => {
+          expect(jobId).to.eql(job.id);
+          expect(3).to.eql(Number(attemptsMade));
+          resolve();
         });
       });
 
@@ -1892,7 +1927,7 @@ describe('workers', function () {
 
       await worker.waitUntilReady();
 
-      const failing = new Promise<void>((resolve, reject) => {
+      const failing = new Promise<void>(resolve => {
         worker.on('failed', async (job, err) => {
           expect(job.data.foo).to.equal('bar');
           expect(err).to.equal(failedError);
@@ -1902,7 +1937,7 @@ describe('workers', function () {
         });
       });
 
-      const completing = new Promise<void>((resolve, _reject) => {
+      const completing = new Promise<void>(resolve => {
         worker.on('completed', async () => {
           resolve();
         });
@@ -1949,7 +1984,7 @@ describe('workers', function () {
 
       await worker.waitUntilReady();
 
-      const failing = new Promise<void>((resolve, reject) => {
+      const failing = new Promise<void>(resolve => {
         worker.on('failed', async (job, err) => {
           expect(job.data.foo).to.equal('bar');
           expect(err).to.equal(failedError);
@@ -1958,7 +1993,7 @@ describe('workers', function () {
         });
       });
 
-      const completing = new Promise<void>((resolve, _reject) => {
+      const completing = new Promise<void>(resolve => {
         worker.on('completed', async () => {
           resolve();
         });