fix: remove stalled job when finishing fixes OptimalBits#1600

Author: manast

lib/commands/moveToFinished-7.lua lib/commands/moveToFinished-8.lua

@@ -14,6 +14,7 @@
       KEYS[6] active event key
 
       KEYS[7] delayed key
+      KEYS[8] stalled key
 
       ARGV[1]  jobId
       ARGV[2]  timestamp
@@ -40,6 +41,7 @@ if rcall("EXISTS", KEYS[3]) == 1 then -- // Make sure job exists
     local lockKey = KEYS[3] .. ':lock'
     if rcall("GET", lockKey) == ARGV[5] then
       rcall("DEL", lockKey)
+      rcall("SREM", KEYS[8], ARGV[1])
     else
       return -2
     end

lib/getters.js

@@ -90,6 +90,16 @@ module.exports = function(Queue) {
     return this.getJobCountByTypes('wait', 'paused');
   };
 
+  /**
+   * 
+   * @returns the potential stalled jobs. Only useful for tests.
+   */
+  Queue.prototype.getStalledCount = function() {
+    const key = this.toKey('stalled');
+    return this.client.scard(key);
+  };
+
+
   // TO BE DEPRECATED --->
   Queue.prototype.getPausedCount = function() {
     return this.getJobCountByTypes('paused');

lib/scripts.js

@@ -125,7 +125,8 @@ const scripts = {
       queueKeys.wait,
       queueKeys.priority,
       queueKeys.active + '@' + queue.token,
-      queueKeys.delayed
+      queueKeys.delayed,
+      queueKeys.stalled
     ];
 
     if (typeof shouldRemove === 'boolean') {

test/test_queue.js

@@ -1480,6 +1480,39 @@ describe('Queue', () => {
         .catch(done);
     });
 
+    it('should clear job from stalled set when job completed', (done) => {
+      const queue2 = utils.buildQueue('running-job-' + uuid.v4(), {
+        settings: {
+          stalledInterval: 10
+        }
+      });
+
+      queue2.process(job => {
+        expect(job.data.foo).to.be.equal('bar');
+        return delay(100);
+      });
+
+      queue2.add({ foo: 'bar' }).then(
+        job => {
+          expect(job.id).to.be.ok;
+          expect(job.data.foo).to.be.eql('bar');
+        },
+        err => {
+          done(err);
+        }
+      );
+
+      queue2.once('completed', async () => {
+        const stalled = await queue2.getStalledCount();
+        try {
+          expect(stalled).to.be.equal(0);
+          done();
+        } catch (err) {
+          done(err);
+        }
+      });
+    });
+
     it('process a job that fails', done => {
       const jobError = new Error('Job Failed');