Repository: taskforcesh/bullmq
Pull request #569 (2021-05-30 00:07:28)

Fix moveStalledJobsToWait: send failedReason to queueEvents


Patch:
diff --git a/src/commands/moveStalledJobsToWait-8.lua b/src/commands/moveStalledJobsToWait-8.lua
index 02852019fe..8f61f8e7c9 100644
--- a/src/commands/moveStalledJobsToWait-8.lua
+++ b/src/commands/moveStalledJobsToWait-8.lua
@@ -20,7 +20,8 @@
 
     Events:
       'stalled' with stalled job id.
-]] local rcall = redis.call
+]]
+local rcall = redis.call
 
 -- Check if we need to check for stalled jobs now.
 if rcall("EXISTS", KEYS[5]) == 1 then return {{}, {}} end
@@ -59,11 +60,13 @@ if (#stalling > 0) then
                 local stalledCount = rcall("HINCRBY", jobKey, "stalledCounter",
                                            1)
                 if (stalledCount > MAX_STALLED_JOB_COUNT) then
+                    local failedReason = "job stalled more than allowable limit" 
                     rcall("ZADD", KEYS[4], ARGV[3], jobId)
                     rcall("HSET", jobKey, "failedReason",
-                          "job stalled more than allowable limit")
+                          failedReason)
                     rcall("XADD", KEYS[8], "*", "event", "failed", "jobId",
-                          jobId, 'prev', 'active')
+                          jobId, 'prev', 'active', 'failedReason',
+                          failedReason)
                     table.insert(failed, jobId)
                 else
                     -- Move the job back to the wait queue, to immediately be picked up by a waiting worker.


Test patch:
diff --git a/src/test/test_getters.ts b/src/test/test_getters.ts
index 8cb22cff6e..babafdd4cd 100644
--- a/src/test/test_getters.ts
+++ b/src/test/test_getters.ts
@@ -2,14 +2,14 @@
 /* tslint:disable: no-floating-promises */
 'use strict';
 
-import { Queue, Job } from '@src/classes';
+import { Queue, Job } from '../classes';
 import { describe, beforeEach, it } from 'mocha';
 import { expect } from 'chai';
 import * as IORedis from 'ioredis';
 import { v4 } from 'uuid';
-import { Worker } from '@src/classes/worker';
+import { Worker } from '../classes/worker';
 import { after } from 'lodash';
-import { removeAllQueueData } from '@src/utils';
+import { removeAllQueueData } from '../utils';
 
 describe('Jobs getters', function() {
   this.timeout(4000);
@@ -168,9 +168,7 @@ describe('Jobs getters', function() {
       after(3, async function() {
         try {
           const jobs = await queue.getJobs('completed');
-          expect(jobs)
-            .to.be.an('array')
-            .that.have.length(3);
+          expect(jobs).to.be.an('array').that.have.length(3);
           expect(jobs[0]).to.have.property('finishedOn');
           expect(jobs[1]).to.have.property('finishedOn');
           expect(jobs[2]).to.have.property('finishedOn');
@@ -204,9 +202,7 @@ describe('Jobs getters', function() {
         try {
           queue;
           const jobs = await queue.getJobs('failed');
-          expect(jobs)
-            .to.be.an('array')
-            .that.has.length(3);
+          expect(jobs).to.be.an('array').that.has.length(3);
           expect(jobs[0]).to.have.property('finishedOn');
           expect(jobs[1]).to.have.property('finishedOn');
           expect(jobs[2]).to.have.property('finishedOn');
@@ -235,9 +231,7 @@ describe('Jobs getters', function() {
       after(3, async function() {
         try {
           const jobs = await queue.getJobs('completed', 1, 2, true);
-          expect(jobs)
-            .to.be.an('array')
-            .that.has.length(2);
+          expect(jobs).to.be.an('array').that.has.length(2);
           expect(jobs[0].data.foo).to.be.eql(2);
           expect(jobs[1].data.foo).to.be.eql(3);
           expect(jobs[0]).to.have.property('finishedOn');
@@ -265,9 +259,7 @@ describe('Jobs getters', function() {
       after(3, async function() {
         try {
           const jobs = await queue.getJobs('completed', -3, -1, true);
-          expect(jobs)
-            .to.be.an('array')
-            .that.has.length(3);
+          expect(jobs).to.be.an('array').that.has.length(3);
           expect(jobs[0].data.foo).to.be.equal(1);
           expect(jobs[1].data.foo).to.be.eql(2);
           expect(jobs[2].data.foo).to.be.eql(3);
@@ -292,9 +284,7 @@ describe('Jobs getters', function() {
       after(3, async function() {
         try {
           const jobs = await queue.getJobs('completed', -300, 99999, true);
-          expect(jobs)
-            .to.be.an('array')
-            .that.has.length(3);
+          expect(jobs).to.be.an('array').that.has.length(3);
           expect(jobs[0].data.foo).to.be.equal(1);
           expect(jobs[1].data.foo).to.be.eql(2);
           expect(jobs[2].data.foo).to.be.eql(3);
diff --git a/src/test/test_stalled_jobs.ts b/src/test/test_stalled_jobs.ts
index a32cbdda26..8ac0cce6a5 100644
--- a/src/test/test_stalled_jobs.ts
+++ b/src/test/test_stalled_jobs.ts
@@ -45,7 +45,7 @@ describe('stalled jobs', function() {
 
     await worker.waitUntilReady();
 
-    const jobs = await Promise.all([
+    await Promise.all([
       queue.add('test', { bar: 'baz' }),
       queue.add('test', { bar1: 'baz1' }),
       queue.add('test', { bar2: 'baz2' }),
@@ -89,11 +89,14 @@ describe('stalled jobs', function() {
   it('fail stalled jobs that stall more than allowable stalled limit', async function() {
     this.timeout(6000);
 
+    const queueEvents = new QueueEvents(queueName);
+    await queueEvents.waitUntilReady();
+
     const concurrency = 4;
 
     const worker = new Worker(
       queueName,
-      async job => {
+      async () => {
         return delay(10000);
       },
       {
@@ -108,7 +111,7 @@ describe('stalled jobs', function() {
 
     await worker.waitUntilReady();
 
-    const jobs = await Promise.all([
+    await Promise.all([
       queue.add('test', { bar: 'baz' }),
       queue.add('test', { bar1: 'baz1' }),
       queue.add('test', { bar2: 'baz2' }),
@@ -125,11 +128,26 @@ describe('stalled jobs', function() {
 
     await worker.close(true);
 
+    const errorMessage = 'job stalled more than allowable limit';
     const allFailed = new Promise(resolve => {
-      queueScheduler.on('failed', after(concurrency, resolve));
+      queueScheduler.on(
+        'failed',
+        after(concurrency, (jobId, failedReason) => {
+          expect(failedReason.message).to.be.equal(errorMessage);
+          resolve();
+        }),
+      );
+    });
+
+    const globalAllFailed = new Promise(resolve => {
+      queueEvents.on('failed', ({ failedReason }) => {
+        expect(failedReason).to.be.equal(errorMessage);
+        resolve();
+      });
     });
 
     await allFailed;
+    await globalAllFailed;
 
     await queueScheduler.close();
   });