fix(job): set delay value on current job instance when it is retried (#…

…2266) (python)

python/bullmq/job.py

@@ -91,6 +91,7 @@ async def moveToFailed(self, err, token:str, fetchNext:bool = False):
 
         move_to_failed = False
         finished_on = 0
+        delay = 0
         command = 'moveToFailed'
 
         async with self.queue.redisConnection.conn.pipeline(transaction=True) as pipe:
@@ -137,6 +138,9 @@ async def moveToFailed(self, err, token:str, fetchNext:bool = False):
         if finished_on and type(finished_on) == int:
             self.finishedOn = finished_on
 
+        if delay and type(delay) == int:
+            self.delay = delay
+
     async def saveStacktrace(self, pipe, err:str):
         stacktrace = traceback.format_exc()
         stackTraceLimit = self.opts.get("stackTraceLimit")

src/classes/job.ts

@@ -618,7 +618,7 @@ export class Job<
     // Check if an automatic retry should be performed
     //
     let moveToFailed = false;
-    let finishedOn;
+    let finishedOn, delay;
     if (
       this.attemptsMade < this.opts.attempts &&
       !this.discarded &&
@@ -627,7 +627,7 @@ export class Job<
       const opts = queue.opts as WorkerOptions;
 
       // Check if backoff is needed
-      const delay = await Backoffs.calculate(
+      delay = await Backoffs.calculate(
         <BackoffOptions>this.opts.backoff,
         this.attemptsMade,
         err,
@@ -687,6 +687,10 @@ export class Job<
     if (finishedOn && typeof finishedOn === 'number') {
       this.finishedOn = finishedOn;
     }
+
+    if (delay && typeof delay === 'number') {
+      this.delay = delay;
+    }
   }
 
   /**

tests/test_worker.ts

@@ -2004,15 +2004,12 @@ describe('workers', function () {
       );
       await worker.waitUntilReady();
 
-      const job = await queue.add(
-        'test',
-        { bar: 'baz' },
-        { attempts: 3, backoff: 300 },
-      );
+      await queue.add('test', { bar: 'baz' }, { attempts: 3, backoff: 300 });
 
       const failed = new Promise<void>((resolve, reject) => {
-        worker.once('failed', async () => {
+        worker.once('failed', async job => {
           try {
+            expect(job?.delay).to.be.eql(300);
             const gotJob = await queue.getJob(job.id!);
             expect(gotJob!.delay).to.be.eql(300);
             resolve();
@@ -2073,6 +2070,52 @@ describe('workers', function () {
       await worker.close();
     });
 
+    describe('when backoff type is exponential', () => {
+      it("updates job's delay property if it fails and backoff is set", async () => {
+        const worker = new Worker(
+          queueName,
+          async () => {
+            await delay(100);
+            throw new Error('error');
+          },
+          { connection, prefix },
+        );
+        await worker.waitUntilReady();
+
+        await queue.add(
+          'test',
+          { bar: 'baz' },
+          {
+            attempts: 3,
+            backoff: {
+              type: 'exponential',
+              delay: 200,
+            },
+          },
+        );
+
+        const failed = new Promise<void>((resolve, reject) => {
+          worker.on('failed', async job => {
+            try {
+              const attemptsMade = job?.attemptsMade;
+              if (attemptsMade! > 2) {
+                expect(job!.delay).to.be.eql(0);
+                const gotJob = await queue.getJob(job.id!);
+                expect(gotJob!.delay).to.be.eql(0);
+                resolve();
+              }
+            } catch (err) {
+              reject(err);
+            }
+          });
+        });
+
+        await failed;
+
+        await worker.close();
+      });
+    });
+
     describe('when attempts is 1 and job fails', () => {
       it('should execute job only once and emits retries-exhausted event', async () => {
         const worker = new Worker(