Improvements for Queue#processStalledJobs

README.md

@@ -317,7 +317,7 @@ listened by some other service that stores the results in a database.
 ## Reference
 
 <a name="queue"/>
-###Queue(queueName, redisPort, redisHost, [redisOpts])
+###Queue(queueName, redisPort, redisHost, [redisOpts], [queueOpts])
 
 This is the Queue constructor. It creates a new Queue that is persisted in
 Redis. Everytime the same queue is instantiated it tries to process all the
@@ -330,6 +330,9 @@ __Arguments__
     redisPort {Number} A port where redis server is running.
     redisHost {String} A host specified as IP or domain where redis is running.
     redisOptions {Object} Options to pass to the redis client. https://github.com/mranney/node_redis
+    queueOpts {Object} Options to drive the Queue behavior
+    queueOpts.processStalledJobs {Boolean} Automatically process potentially jobs stuck in active
+    state (default: true)
 ```
 
 ---------------------------------------

lib/queue.js

@@ -47,7 +47,7 @@ var LOCK_RENEW_TIME = 5000; // 5 seconds is the renew time.
 var CLIENT_CLOSE_TIMEOUT_MS = 5000;
 var POLLING_INTERVAL = 5000;
 
-var Queue = function Queue(name, redisPort, redisHost, redisOptions){
+var Queue = function Queue(name, redisPort, redisHost, redisOptions, queueOptions){
   if(!(this instanceof Queue)){
     return new Queue(name, redisPort, redisHost, redisOptions);
   }
@@ -57,7 +57,7 @@ var Queue = function Queue(name, redisPort, redisHost, redisOptions){
     var redisOpts = opts.redis || {};
     redisPort = redisOpts.port;
     redisHost = redisOpts.host;
-    redisOptions = redisOpts.opts || {};
+    redisOptions = redisOpts.opts || {};
     redisOptions.db = redisOpts.DB;
   }
 
@@ -77,6 +77,13 @@ var Queue = function Queue(name, redisPort, redisHost, redisOptions){
   redisPort = redisPort || 6379;
   redisHost = redisHost || '127.0.0.1';
 
+  queueOptions = _.pick(queueOptions, ['processStalledJobs']);
+  queueOptions = _.defaults(queueOptions, {
+    processStalledJobs: true
+  });
+
+  this.opts = queueOptions;
+
   var _this = this;
 
   this.name = name;
@@ -423,7 +430,10 @@ Queue.prototype.run = function(concurrency){
   var promises = [];
   var _this = this;
 
-  return this.processStalledJobs().then(function(){
+  // In case of connection loss, running `processStalledJobs` will repick jobs here.
+  var start = this.opts.processStalledJobs ? this.processStalledJobs() : Promise.resolve();
+
+  return start.then(function(){
 
     while(concurrency--){
       promises.push(new Promise(_this.processJobs));
@@ -433,8 +443,11 @@ Queue.prototype.run = function(concurrency){
     // Set process Stalled jobs intervall
     //
     clearInterval(_this.stalledJobsInterval);
-    _this.stalledJobsInterval =
-      setInterval(_this.processStalledJobs, _this.LOCK_RENEW_TIME);
+    if(_this.opts.processStalledJobs) {
+      _this.stalledJobsInterval =
+        setInterval(_this.processStalledJobs, _this.LOCK_RENEW_TIME);
+
+    }
 
     return Promise.all(promises);
   });
@@ -475,15 +488,19 @@ Queue.prototype.updateDelayTimer = function(newDelayedTimestamp){
 };
 
 /**
-  Process jobs that have been added to the active list but are not being
-  processed properly.
+ * Process jobs that have been added to the active list but are not being
+ * processed properly.
+ *
+ * @param {Number?} limit Only process this many number of jobs. Greater than 1, otherwise -1
 */
-Queue.prototype.processStalledJobs = function(){
+Queue.prototype.processStalledJobs = function(limit){
   var _this = this;
+  limit = limit > 0 ? limit - 1 : -1;
+
   if(this.closing){
     return this.closing;
   } else{
-    return this.client.lrangeAsync(this.toKey('active'), 0, -1).then(function(jobs){
+    return this.client.lrangeAsync(this.toKey('active'), 0, limit).then(function(jobs){
       return Promise.each(jobs, function(jobId) {
         return Job.fromId(_this, jobId).then(_this.processStalledJob);
       });

test/test_queue.js

@@ -574,6 +574,178 @@ describe('Queue', function () {
       }).catch(done);
     });
 
+    it('only process a limited amount of stalled jobs', function(done) {
+      this.timeout(12000);
+
+      var queue2 = utils.buildQueue('limited-stalled-job-processing' + uuid(), {
+        processStalledJobs: false
+      });
+
+      var completed = _.after(5, function(){
+        queue2.removeListener('completed', completed);
+        var client = redis.createClient();
+        var movingPromises = [];
+        // This simulates all 5 jobs to be in a stalled state.
+        for(var i = 1; i <= 5; i++) {
+          movingPromises.push(client.multi()
+                              .srem(queue2.toKey('completed'), i)
+                              .lpush(queue2.toKey('active'), i)
+                              .execAsync());
+        }
+
+        Promise.all(movingPromises).then(function() {
+          return queue2.getActiveCount().then(function(count) {
+            expect(count).to.equal(5);
+          });
+        }).then(function() {
+          return queue2.processStalledJobs(3);
+        }).then(function() {
+          return queue2.getActiveCount().then(function(count) {
+            expect(count).to.equal(2);
+          });
+        }).then(function() {
+          return queue2.getCompletedCount().then(function(count) {
+            expect(count).to.equal(3);
+          });
+        }).catch(function(err) {
+          expect(err).to.be(null);
+        }).finally(function() {
+          queue2.close(true).then(done);
+        });
+      });
+
+      queue2.on('completed', completed);
+
+      for(var i = 1; i <= 5; i++) {
+        queue2.add({ foo: 'bar' });
+      }
+
+      queue2.process(function (job, jobDone) {
+        expect(job.data.foo).to.be.equal('bar');
+        jobDone();
+      });
+    });
+
+    it('should not process stalled jobs if disabled', function(done) {
+      this.timeout(12000);
+
+      var queue2 = utils.buildQueue('limited-stalled-job-processing' + uuid(), {
+        processStalledJobs: false
+      });
+
+      queue2.LOCK_RENEW_TIME = 100;
+
+      var completed = _.after(5, function(){
+        queue2.removeListener('completed', completed);
+        var client = redis.createClient();
+        var movingPromises = [];
+        // This simulates all 5 jobs to be in a stalled state.
+        for(var i = 1; i <= 5; i++) {
+          movingPromises.push(client.multi()
+                              .srem(queue2.toKey('completed'), i)
+                              .lpush(queue2.toKey('active'), i)
+                              .execAsync());
+        }
+
+        Promise.all(movingPromises).then(function() {
+          return queue2.getActiveCount().then(function(count) {
+            expect(count).to.equal(5);
+          });
+        }).delay(200).then(function() {
+          return queue2.getActiveCount().then(function(count) {
+            expect(count).to.equal(5);
+          });
+        }).catch(function(err) {
+          expect(err).to.be(null);
+        }).finally(function() {
+          queue2.close(true).then(done);
+        });
+      });
+
+      queue2.on('completed', completed);
+
+      for(var i = 1; i <= 5; i++) {
+        queue2.add({ foo: 'bar' });
+      }
+
+      queue2.process(function (job, jobDone) {
+        expect(job.data.foo).to.be.equal('bar');
+        jobDone();
+      });
+    });
+
+    it('should not process stalled jobs on a reconnection', function(done) {
+      this.timeout(12000);
+      var queueName = 'dont-process-stalled-jobs-on-reconnection' + uuid();
+
+      var queue2 = utils.buildQueue(queueName, {
+        processStalledJobs: false
+      });
+
+      var queue3;
+
+      var jobHandler = function(job, jobDone) {
+        jobDone();
+      };
+
+      queue2.LOCK_RENEW_TIME = 30;
+
+      var completed = _.after(5, function(){
+        queue2.removeListener('completed', completed);
+        var client = redis.createClient();
+        var movingPromises = [];
+        // This simulates all 5 jobs to be in a stalled state.
+        for(var i = 1; i <= 5; i++) {
+          movingPromises.push(client.multi()
+                              .srem(queue2.toKey('completed'), i)
+                              .lpush(queue2.toKey('active'), i)
+                              .execAsync());
+        }
+
+        Promise.all(movingPromises).then(function() {
+          return queue2.getActiveCount().then(function(count) {
+            expect(count).to.equal(5);
+          });
+        }).delay(100).then(function() {
+          return queue2.getActiveCount().then(function(count) {
+            expect(count).to.equal(5);
+          });
+        }).then(function() {
+          queue3 = utils.buildQueue(queueName, {
+            processStalledJobs: false
+          });
+
+          return new Promise(function(resolve) {
+            queue3.on('ready', function() {
+              queue3.process(jobHandler);
+              resolve();
+            });
+          });
+        }).delay(100).then(function() {
+          return queue3.getActiveCount().then(function(count) {
+            expect(count).to.equal(5);
+          });
+        }).catch(function(err) {
+          expect(err).to.be(null);
+        }).finally(function() {
+          var closing = [queue2.close(true)];
+          if (queue3) closing.push(queue3.close(true));
+
+          Promise.all(closing).then(function() {
+            done();
+          });
+        });
+      });
+
+      queue2.on('completed', completed);
+
+      for(var i = 1; i <= 5; i++) {
+        queue2.add({ foo: 'bar' });
+      }
+
+      queue2.process(jobHandler);
+    });
+
     it('process a job that fails', function (done) {
       var jobError = new Error('Job Failed');
 

test/utils.js

@@ -13,8 +13,8 @@ function simulateDisconnect(queue){
   queue.eclient.stream.end();
 }
 
-function buildQueue(name) {
-  var queue = new Queue(name || STD_QUEUE_NAME, 6379, '127.0.0.1');
+function buildQueue(name, queueOptions) {
+  var queue = new Queue(name || STD_QUEUE_NAME, 6379, '127.0.0.1', {}, queueOptions);
   queues.push(queue);
   return queue;
 }