Kue: Cluster workers not being able to pick up Jobs / double pick up

[JperF]

The following program will create two cluster workers, then one job task. The job task will be marked as failed immediately. Every second the job is marked back as inactive which should be picked up again by a single worker. However occasionally more than one worker is picking up the job cause it to be executed twice.

Is there something wrong with my logic or is this an issue with the cluster implementation in kue?

var kue = require('kue');
var jobs = kue.createQueue();
var cluster = require('cluster');

if(cluster.isMaster){
    cluster.fork();
    cluster.fork();
    jobs.create('test').save();

    setInterval(function () {
        kue.Job.rangeByState( 'failed', 0, -1, 'asc', function( err, jobs ) {
            if(err){
                console.log(err);
            }
            else{
                jobs.forEach( function ( job ) {
                    job.state('inactive').save();
                });
            }
        });
    }, 1000);
}

if(cluster.isWorker){
    jobs.process('test', function ( job, done ) {
        console.log('worker' , cluster.worker.id + ' has control of: ' + job.id + ' | ' +  new Date());
        done('failed');
    });
}
else{
    jobs.on('job failed', function ( errorMessage ) {
        console.log('Job failed');
    })
}

When running the following program the jobs are eventually are not picked up at all.

Note: The done call is never called in this example which causes the state of the jobs to be changed from active to inactive. Instead of failed to inactive like the previous example.

var kue = require('kue');
var jobs = kue.createQueue();
var cluster = require('cluster');

if(cluster.isMaster){
    cluster.fork();
    cluster.fork();
    jobs.create('test').save();

    setInterval(function () {
        kue.Job.rangeByState( 'active', 0, -1, 'asc', function( err, jobs ) {
            if(err){
                console.log(err);
            }
            else{
                jobs.forEach( function ( job ) {
                    job.state('inactive').save();
                });
            }
        });
    }, 1000);
}

if(cluster.isWorker){
    jobs.process('test', function ( job, done ) {
        console.log('worker' , cluster.worker.id + ' has control of: ' + job.id + ' | ' +  new Date());
    });
}
else{
    jobs.on('job failed', function ( errorMessage ) {
        console.log('Job failed');
    })
}

[behrad]

Can you check this against v1 branch also?

[JperF]

@behrad Running the same program in the v1 branch throws the following error:

TypeError:  kue.Job.rangeByState is not a function
    at null._repeat (/Users/test/clusterTest.js:18:13)
    at wrapper [as _onTimeout] (timers.js:275:11)
    at Timer.listOnTimeout (timers.js:92:15)

[behrad]

Sorry I've n't created a migration check list yes, rangeByState is moved to queue object :)

[JperF]

@behrad

Changing to kue.rangeByState(..) throws an error as well.

TypeError: kue.rangeByState is not a function

[behrad]

queue = createQueue()
queue.rangeByState...

[JperF]

Same problem is happening as I initially stated:

However occasionally more than one worker is picking up the job cause it to be executed twice.

Note that only a maximum of 2 workers ever pick up the job regardless of how many there are.

[JperF]

@behrad
Has there been any progress on this issue?

[victusfate]

To be clear, if workers are triggered from a stand alone proc ie node worker.js this is not an issue?

Also is the cluster module up to date with current nodejs , I see it was last updated in 2012

update looks like this is a known issue with kuejs
https://github.com/Automattic/kue/issues/905

[victornikitin]

@JperF Have you found the solution?

[JperF]

Hey @victornikitin,

Unfortunately I have not looked into this for more than 2 years. So I am not sure where Kue's code stands nor if this is still an issue.