Slowness and possible race condition with locally pausing queue (with code demonstration)

Created by: geoffreak

Version 3.3.0

There's something quite wrong with the queue pause functionality that will cause it to take a significant amount of time to pause a queue. This has been causing huge problems as I've been trying to use Bull and test my processor functions.

After pulling out my hair all day on this, I finally have an example of the problem. You can see that simply having some sort of delay after the job completes (I had a database lookup in my code) causes the pause to be significantly delayed. For me, it take an additional 5 seconds to pause when there is a delay (mocha tests fail after 2 seconds by default).

Here is the Mocha test script:

var Queue = require('bull');

describe('Pause', function() {
  it('does stuff slowly', function() {
    var queue;
    this.timeout(10000);
    queue = new Queue('test', {
      prefix: '{bull}'
    });
    return queue.add({
      foo: 'bar'
    }, {
      removeOnComplete: true,
      removeOnFail: false
    }).then(function() {
      var p;
      p = new Promise(function(resolve, reject) {
        queue.on('completed', function(job, value) {
          return resolve({job, value});
        });
        return queue.on('failed', function(job, value) {
          return reject(new Error(job.stacktrace));
        });
      });
      queue.process(function(job) {
        return Promise.resolve();
      });
      return p.then(function() {
        var delay;
        // 100ms delay to simulate other actions being taken
        delay = new Promise(function(resolve) {
          return setTimeout(resolve, 100);
        });
        return delay.then(function() {
          var start;
          console.log('this is reached quickly');
          start = new Date().getTime();
          return queue.pause(true).finally(function() {
            var finish;
            finish = new Date().getTime();
            return console.log('this takes a really really long time:', finish - start, 'ms');
          });
        });
      });
    });
  });
  it('does stuff quickly', function() {
    var queue;
    this.timeout(10000);
    queue = new Queue('test', {
      prefix: '{bull}'
    });
    return queue.add({
      foo: 'bar'
    }, {
      removeOnComplete: true,
      removeOnFail: false
    }).then(function() {
      var p;
      p = new Promise(function(resolve, reject) {
        queue.on('completed', function(job, value) {
          return resolve({job, value});
        });
        return queue.on('failed', function(job, value) {
          return reject(new Error(job.stacktrace));
        });
      });
      queue.process(function(job) {
        return Promise.resolve();
      });
      return p.then(function() {
        var start;
        // No 100ms delay this time
        console.log('this is reached quickly');
        start = new Date().getTime();
        return queue.pause(true).finally(function() {
          var finish;
          finish = new Date().getTime();
          return console.log('this also is reached quickly:', finish - start, 'ms');
        });
      });
    });
  });
});

Output:

  Pause
this is reached quickly
this takes a really really long time: 5043 ms
    ✓ does stuff slowly (5186ms)
this is reached quickly
this also is reached quickly: 1 ms
    ✓ does stuff quickly


  2 passing (5s)