client.stats_job() never responds

[gtamas]

I'm new to Beanstalkd. I'm trying to use the stats_job method in a worker, like this:

worker.client.stats_job(job_info.id, function(err, stats) {
                if(!err) {
                    if(stats.buries !== 0) {
                        worker.client.destroy(stats.id, function(err) {});
                    } else if(stats.releases >= settings.facebook.max_releases) {
                        worker.client.bury(stats.id, stats.pri, function(err) {});
                    }
                }
            });

However, the callback never seem to be called. There is no response at all. And processing of the queue seems to be stuck as well. If I remove this call, everything works fine.

What could be wrong? Thanks in advance! I compiled Beanstalkd from source, I'm on MacOS.

[ceejbot]

Some basic questions about the client: Is it watching a set of tubes, including the tube where the job was queued?

[gtamas]

Hi, thanks for the response.

Yes it is watching tubes. In fact, there is only 1 tube at the moment, and that's where the job is.

Here is the complete source of the worker, it's much like the example in your docs.

var BeanstalkdWorker = require('fivebeans').worker;

module.exports = function(settings) {
    var worker = new BeanstalkdWorker({
        id: 'dashboard_worker',
        host: settings.host,
        port: settings.port,
        handlers:
        {
            fb_friends_initial: require('./handlers/fb_friends_initial.js')(),
            fb_history_initial: require('./handlers/fb_history_initial.js')()
        },
        ignoreDefault: true
    });
    worker.start([settings.facebook.tubename]);

    worker.on('error',function(err){
        // some error handling goes here..
    });
    worker.on('job.handled',function(job_info){
        if(job_info.action === 'release') {
            worker.client.stats_job(job_info.id, function(err, stats) {
                // THE PROBLEM IS HERE: This never runs!
            });
        }
    });
}

BTW: I tried to talk to Beanstalkd over telnet, and that works fine. I can run the same command there, so the server seems to be fine.

Update: If I replace the above worker.client.stats_job(...) call with some other command, for instance worker.client.peek_ready(...), that's good. It works fine. But the stats command doesn't respond.

[ceejbot]

That's very interesting info! Thank you for the code snippet. I'm going to look into this more deeply tonight.

[gtamas]

Hi

OK thanks.

Update:

The stats_job() method seems to be OK. Running it with another client works as expected. The "bsClient" was created by me and it is using the same tube as the other one.

bsClient.stats_job(1, function(err, stats) {
console.log(err,stats) // this is printed!
});

So either there is something wrong with worker's client, or calling the method from that event handler is the issue? Don't get it. I'll try to look more into this later today.

[gtamas]

Update:

Weird. When the bigger code snippet above is executed, there is no warning, info or other event emitted. The only event I get is job.handled, which returns the expected results (job info object).

The job itself remains in the ready queue, even after TTR seconds. Somehow execution seems to be halted. I tested this using telnet.

If I kill node at this point & restart it, the job gets buried due to an error (the job handler tries to read the websocket).

Then, if I add new jobs, the above process is repeated. What am I doing wrong? :)

[gtamas]

One last update: :)

I tried to call stats-job() using a separate client rather than the worker's own, just for testing. This works:

var BeanstalkdWorker = require('fivebeans').worker;
var Beanstalkd = require('fivebeans');

module.exports = function(settings) {

    var BeanstalkdClient = new Beanstalkd.client(settings.host, settings.port);

    BeanstalkdClient.on('connect', function()
    {
    BeanstalkdClient.use(settings.facebook.tubename, function(err, tubename) {
            if(err) {
                // error handling..
            } else {

                var worker = new BeanstalkdWorker({
                    id: 'dashboard_worker',
                    host: settings.host,
                    port: settings.port,
                    handlers:
                    {
                        fb_friends_initial: require('./handlers/fb_friends_initial.js')(),
                        fb_history_initial: require('./handlers/fb_history_initial.js')()
                    },
                    ignoreDefault: true
                });
                worker.start([settings.facebook.tubename]);

                worker.on('error',function(err){
                    // error handling..
                });
                worker.on('warning',function(data){
                    // warning handling..
                });
                worker.on('stopped',function(){
                    // what if stopped...
                });
                worker.on('job.handled',function(job_info){
                    if(job_info.action === 'release') {
                        BeanstalkdClient.stats_job(job_info.id, function(err, stats) {
                            if(!err) {
                                // NOW THIS RUNS!
                                if(stats.releases+1 === settings.facebook.max_release) {
                                    if(stats.buries !== 0) {
                                        BeanstalkdClient.destroy(job_info.id, function(err) {});
                                    } else {
                                        BeanstalkdClient.bury(job_info.id, stats.pri, function(err) {
                                            console.log(err)
                                        });
                                    }
                                }
                            }
                        });
                    }
                });

            } 

        });

    })
    .on('error', function(err)
    {
        // cant connect..
    })
    .on('close', function()
    {
        BeanstalkdClient.quit();
    })
    .connect();

};

The execution continues. The code in the job.handled event handler generates an error (cannot bury job) though, but that's a different issue.

So it seems the problems I described above only exist if I try to run stats-job() (or stats) using the worker's client.

But of course it doesn't make any sense to create a separate client, the worker already has one. That should be used to execute any command.

[ceejbot]

I just wrote a client test that proves to my satisfaction that it isn't a bug in the underlying client: https://github.com/ceejbot/fivebeans/tree/issue-24

I think you're right that it's a worker problem. Now I shall write more tests. And thank you for the code samples-- always helpful!

[gtamas]

OK thx!

[ceejbot]

Resuming work on this now (sorry, was a very busy month for me). So! Check out this unit test I added earlier today: https://github.com/ceejbot/fivebeans/blob/master/test/test-worker.js#L343

Note that it's a call to stats_job (two calls, in fact) both made using the worker's client object, called on a job the worker has reserved at the time. Both calls succeed & return correct information about the job in progress. This makes me think it isn't a bug inherent to the worker. The question now is: what's different from your code & my test code?

[mopagemo]

I also see this issue. It seems to fail once I try to run things in parallel, and stats_job is called shortly after a job is created.

I managed to create a test case:

var fivebeans = require('fivebeans');
var client = new fivebeans.client('localhost', 11300);
client.connect();

// Allow a second to connect
setTimeout(test, 1000);

function test() {
    client.put(200, 0, 3600, "test", function(err, jobId) {
        console.log('JOB CREATED', err, jobId);

        client.stats_job(jobId, function(err, stats) {
            console.log('STATS', err, stats);
        });
    });

    client.put(200, 0, 3600, "test", function(err, jobId) {
        console.log('JOB CREATED', err, jobId);

        client.stats_job(jobId, function(err, stats) {
            console.log('STATS', err, stats);
        });
    });
}

Output:

JOB CREATED null 224
JOB CREATED null 225

Running only one look-up works:

var fivebeans = require('fivebeans');
var client = new fivebeans.client('localhost', 11300);
client.connect();

// Allow a second to connect
setTimeout(test, 1000);

function test() {
    client.put(200, 0, 3600, "test", function(err, jobId) {
        console.log('JOB CREATED', err, jobId);

        client.stats_job(jobId, function(err, stats) {
            console.log('STATS', err, stats);
        });
    });
}

Output:

JOB CREATED null 228
STATS null { id: 228,
  tube: 'default',
  state: 'ready',
  pri: 200,
  age: 0,
  delay: 0,
  ttr: 3600,
  'time-left': 0,
  file: 0,
  reserves: 0,
  timeouts: 0,
  releases: 0,
  buries: 0,
  kicks: 0 }

[mopagemo]

Digging further into this, this will also trigger it:

var fivebeans = require('fivebeans');
var client = new fivebeans.client('localhost', 11300);
client.connect();

// Allow a second to connect
setTimeout(test, 1000);

function test() {
        client.stats_job(111, function(err, stats) {
            console.log('STATS', err, stats);

        });
        client.stats_job(111, function(err, stats) {
            console.log('STATS', err, stats);
        });
}

So if two stats_job run in "parallel" they both fail to respond. From what I can gather it's because they are both added to the response. parseBody then is confused because the expected length doesn't match.

[mauritsl]

I was able to reproduce this issue. The problem is exactly what mopagemo described; multiple responses are within a single TCP window and are passed to ResponseHandler.parseBody, but that function can only handle a single response.

The pull request fixes this issue and adds a test.

[apmcodes]

Noticing the same issue now. If a client is waiting in a reserve call, then watch, or another command doesn't respond. If reserve is not being executed then watch command works.

The below watch commands callback never gets called. The client seems to hang.
consumer.js

        getNextMessage() {
            logger.info('Waiting for new messages...');
            bsdClientInstance.reserve((err, jobid, payload) => {
                if (err) handleError(err);
                else {
                    logger.info('Got new message...');
                }
            });
        },

work.js

            const onWatch = () => {
                console.log('response', response);
            }
            consumer.watch(id, onWatch)

[apmcodes]

Seems as beanstalkd reserve is a blocking call, other commands such as watch/ ignore are hanging.