ceejbot / fivebeans Goto Github PK

After reserving a job, I cannot reserve again until I delete the job. That means I can't do any jobs in parallel.

What if I queue a job that takes 40 seconds due to some asynchronous call out to the internet. I can't handle any other jobs in the mean time, until that call returns and then I call destroy.

What I've been doing is calling reserve right after a reserve returns:

client.reserve(function handleJob(error, jobId, payload) {
    setImmediate(function() {
        client.reserve(handleJob);
    });

    someCallThatTakesALongTimeToReturn(function() {
       client.destroy(jobId, ...);
    });
});

and for whatever reason, that does not work. I have to do

client.reserve(function handleJob(error, jobId, payload) {
    someCallThatTakesALongTimeToReturn(function() {
       client.destroy(jobId, function() {
            client.reserve(handleJob);
       });
    });
});

Is there a particular reason it has to work this way? Can beanstalkd not handle parallel requests?

Is WATCH not allowed while in RESERVE call? Is this the way beanstalkd works or is it fivebeans?

In the below code, first reserve command is run and then watch command after 2 secs, but watch command callback is never run.

consumer.js

var fivebeans = require('fivebeans');

var client = new fivebeans.client('127.0.0.1', 11300);
client
    .on('connect', function () {
        console.log('Connected');
    })
    .on('error', function (err) {
        // connection failure
    })
    .on('close', function () {
        // underlying connection has closed
    })
    .connect();

client.reserve(function (err, jobid, payload) {
    if (err) { console.log(err) }
    console.log(`Job Id ${jobid}`);
    console.log(`Job data ${payload.toString()}`);
    client.destroy(jobid, function (err) {
        if (err) { console.log(err) }
        console.log(`Job deleted ${jobid}`);
    });
});

setTimeout(() => {
    console.log('going to watch')
    client.watch('low-volume', (err, tubecount) => {
        console.log(err);
        console.log(tubecount);
    })
}, 2000)    

Output

Connetced
going to watch

hi mate,
I try to put jobs with loop as below, but it seem need to exit process.
Is there any better idea?

thanks you.

var fivebeans = require('fivebeans');
var host = '127.0.0.1';
var port = 11300;
var tube = 'my_tube';

for (var i=0;i<3;i++){
  var stalker=new fivebeans.client(host, port);
  stalker.connect(function(err){
    stalker.use(tube, function(err, tubename){
      stalker.put(0,0,60,"foo", function(err, jobid){
        console.log(jobid);
        // process.exit(0);
      });
    });
  });
}

Is there a way to limit number of workers that can be run at the same time?

Is there an is connected method available on client object?

Whats the best way to make sure client is connected to beanstalk?

it adds 17MB to an install of fivebeans, which is 85% of the installed size

This is not really an issue, but more like a feature idea.

So the handler's work() method has access to the payload, but it doesn't know anything about the job's details. Using this info, the work method could make better decisions about how to handle the job.

Example:

The jobstat - among other things - includes the number of tries, buries and time-left. These could be used to implement some kinda functionality that buries / deletes jobs after X tries and runs each retry with decreasing frequency.

But this is only one use-case. I guess having access to the job data (and the beans client) would make the handler more powerful.

So something like this would be cool:

.work(jobid, payload, callback)

or

.work(stats-job, payload, callback)

In the first case, the handler should have access to the beanstalkd client (created by worker).

In second case, by "jobs-stat" I meant the most important info about the job and its history, not necessarily the whole stats-job() response.

What do you think? :)

emitkeys: require('./emitkeyshandler')
needs to be
emitkeys: require('./emitkeyshandler')()

Not a great experience when the only example in the documentation is incorrect!

If i launch say 20 reserves at the same time, then doing release/destroy for each of them doesn't get called. If I lower the number of parallel requests to say, 9, it works fine.

Is there some type of edge case I'm hitting?

example: i have 1 thing in the queue.
I spin up 20 requests to beantalk. The reserve works, but the destroy inside the reserve doesn't.
If I only spin up 1 request, i can reserve/destroy just fine.

Hello everyone, this is not really an issue, but I'm having some troubles with the handlers.

I have one worker and two handlers doing completely different tasks.

I want to use Scribe-js to log what my worker are doing.

If I 'require' Scribe-js where my worker is, it will not be available in the handlers.

If I 'require' Scribe-js in both handlers I got an error because it looks like it's two separate instances of Scribe-js trying to write on the same files (did not work when I configure two different paths for the logs either).

Why I can't use a variable in my handler that has been declared where I instanciate my worker ?

Thank you for any help.

On line 167, the body length is calculated via this.body.length, which breaks because data is not guaranteed to be ascii.

The fix is to use Buffer.byteLength( this.body ).

Repo case in the gist: https://gist.github.com/stdarg/7362928

C J, if you know what this is or have an easy fix, great! If not, I'll dig into it, but I'm short on time myself. :)

fivebeans/lib/client.js:103

We should check if the callback is undefined before calling it.

I am pushing a job to the queue as such :

Queue.use(Config.beanstalkd.tube, function(err, tname){
    Queue.put(0,15,60,JSON.stringify(['aggregator', job1]),onJobPushedToQueue);
});

I have a simple worker like so :

module.exports = function() {
    function aggregatorWorker() {
        this.type = 'aggregator';
    }


    aggregatorWorker.prototype.work = function (payload, callback) {
        console.log('payload', payload);
        callback('success');
    };

    var handler = new aggregatorWorker();
    return handler;
};

And the moment i push the job to queue it gets processed. according to the documentation, shouldn't this be executed at least 15sec later?

Running on ubuntu @Version 1.10+1+g86231ba

We're starting to see jobs showing up in the wrong tube in production.

It appears to be a race condition. Our code looks like this:

stalk.use(tube, function(err, tubename) {
  if(err) ...
  stalk.put(...)
}

And of course stalk.use does IO so node schedules before calling the callback, opening up an opportunity for a race condition if another request also calls stalk.use.

I imagine that's how a good number of people have their code structured. To prevent the race condition, we need to either call use & put sequentially without a potential schedule between them, or we need to use a different stalk connection per tube.

A use_and_put function would make the first easier for people, and also make the problem more obvious.

Regardless of solution, the README should probably warn people not to do what I did. :)

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.

Hi there,

was starting to try out using fivebeans worker runner.
I was wondering if its possible to issue a 'touch' command from within the worker handles?

Thanks guys

@ceejbot,

if FiveBeansClient.prototype.connect() called more than once, a callback for an old socket command may exist in self.handlers even after a new socket is created. In this case, the first command with the new socket will stuck, because its callback will not be called (a callback for the old socket will be called instead).

Suggested fix: #132.

An unknown exception in a node.js program may leave your process in an invalid state. The only sane response to an unhandled exception is to exit the process and restart it.

Sending a warning on unhandled exception implies that the process may continue, and is very dangerous.

Hi as mentioned in the title is not an issue, have an php app in which i put some job via https://github.com/pda/pheanstalk/

$pheanstalk
            ->useTube('testtube')
            ->put(json_encode(array('hello' => 'mdjaman')));

then i want fivebeans worker to process the job; unfornatedly i missed something in documentation.
Can someone point me to right track

I am not sure if it's just my inexperience with Fivebeans, but I find that the way the connect callback can be called multiple times makes things kinda weird and breaks the general node.js convention.

Effectively, the callback is being used as a quasi event loop, so it's better to use an actual event.

If I serialize put operation it's painfully slow. I am used to 25-30K operations per second on beanstalkd with php, this one operates at 10 operations per second. When I queue up several thousand operations in async mode then it's moderately fast.

The current version does not seem to support reconnects. Perhaps https://github.com/juliangruber/reconnect-net can be used to do this?

I think polling from multi-servers beanstalkd cluster is a critical feature, any plan to support it? or it already implemented but I failed to find it out?

Looks like [email protected] > [email protected] has security vulnerability.

more about the vulterability here: https://snyk.io/vuln/npm:lodash:20180130

Why is runner a class, since all you can do with it is call go()

In JS we use the Function pattern , ...this smells like a Java "command"

TIP; Infact it should be a class, but it must inherit from EventEmitter and pass on events from Worker....

Hello, the error is:
warn: { message: 'parsing job JSON', ... error: [SyntaxError: Unexpected token P] ...

It throws this error everytime worker get payload with non-latin characters in it.

My solution is to change line 155 of /lib/worker.js:
try { job = JSON.parse(payload.toString('ascii')); }
into:
try { job = JSON.parse(payload.toString()); }

Can you include this bugfix into main brunch?

Hello,

I'm running two workers on the same tube with different ID tho using the fiverunner, but it seems that sometime both worker are trying to process the same job. From what i understood this shouldn't be possible because once a worker reserve the job the other isn't suppose to get the job as well no?

And i tested with different jobs being put on the tube and it seems to be a bit random the fact that both worker are trying to process the same job.

I have got fivebeans running as a part of a small system that uses supervisor to run a node server and the queue runner.
I've noticed that every time a job gets added to the queue a new connection is opened and never closed after the job has been successful or buried.
I thought it might be the server script but that only initialises a singleton of the fivebeans client.
Is there potentially something I could be missing, do I need to manually close the connection somewhere?
I know this is going to be hard to respond to with no code and very little context but unfortunately I can't disclose any code.
Any help or suggestions would be hugely appreciated!

Hello and thanks for your awesome lib.

I was wondering if there was a method to "nuke" a queue altogether?

Cheers
JM

https://github.com/ceejbot/fivebeans/blob/master/lib/worker.js#L141

Any reason for this to be so short?

In version 1.4.1 we used to put host and port inside beanstalk:

var options = {
 		id: this.id,
 		host: config.beanstalkd.host,
 		port: config.beanstalkd.port,
 		handlers: config.handlers,
 		ignoreDefault: config.ignoreDefault
 };

It was changed in this commit. And now in version 1.5.0 we put host and port just inside config:

var options =
{
    id: 'worker_4',
    host: '127.0.0.1',
    port: 11300,
    handlers:
    {
        emitkeys: require('./emitkeyshandler')()
    },
    ignoreDefault: true
}

But documantation about it for yaml configuration hasn't been updated:

beanstalkd:
    host: "127.0.0.1"
    port: 11300
watch:
    - 'circle'
    - 'picadilly'
    - 'northern'
    - 'central'
handlers:
    - "./handlers/holborn.js"
    - "./handlers/greenpark.js"
    - "./handlers/knightsbridge.js"
ignoreDefault: true

Payload is currently a string. stream.write() and Buffer.toString() method is called without encoding specified. Thus, Node.js will encode/decode the payload into/from a UTF-8 Buffer. If the payload is a binary, like a PNG image, or a protocol buffer, it will get messed up, and the payload length is not calculated correctly (you'll see an EXPECTED_CRLF error if you try to send binary data).

Would you like to add an option to specify the encoding?

Thanks.