Agenda

Agenda is a light-weight job scheduling library for Node.js.

It offers:

• Minimal overhead. Agenda aims to keep its code base small.
• Mongo backed persistance layer.
• Scheduling with configurable priority, concurrency, and repeating
• Scheduling via cron or human readable syntax.
• Event backed job queue that you can hook into.

Installation

Install via NPM

npm install agenda

You will also need a working mongo database (2.4+) to point it to.

Example Usage

var agenda = new Agenda({db: { address: 'localhost:27017/agenda-example'}});

agenda.define('delete old users', function(job, done) {
  User.remove({lastLogIn: { $lt: twoDaysAgo }}, done);
});

agenda.every('3 minutes', 'delete old users');

// Alternatively, you could also do:

agenda.every('*/3 * * * *', 'delete old users');

agenda.start();

agenda.define('send email report', {priority: 'high', concurrency: 10}, function(job, done) {
  var data = job.attrs.data;
  emailClient.send({
    to: data.to,
    from: '[email protected]',
    subject: 'Email Report',
    body: '...'
  }, done);
});

agenda.schedule('in 20 minutes', 'send email report', {to: '[email protected]'});
agenda.start();

var weeklyReport = agenda.schedule('Saturday at noon', 'send email report', {to: '[email protected]'});
weeklyReport.repeatEvery('1 week').save();
agenda.start();

Full documentation

Agenda's basic control structure is an instance of an agenda. Agenda's are mapped to a database collection and load the jobs from within.

Table of Contents

• Configuring an agenda
• Defining job processors
• Creating jobs
• Managing jobs
• Starting the job processor
• Multiple job processors
• Manually working with jobs
• Job Queue Events
• Frequently asked questions

Configuring an agenda

All configuration methods are chainable, meaning you can do something like:

var agenda = new Agenda();
agenda
  .database(...)
  .processEvery('3 minutes')
  ...;

database(url, [collectionName])

Specifies the database at the url specified. If no collection name is give, agendaJobs is used.

agenda.database('localhost:27017/agenda-test', 'agendaJobs');

You can also specify it during instantiation.

var agenda = new Agenda({db: { address: 'localhost:27017/agenda-test', collection: 'agendaJobs' }});

mongo(mongoSkinInstance)

Use an existing mongoskin instance. This can help consolidate connections to a database. You can instead use .database to have agenda handle connecting for you.

You can also specify it during instantiation.

var agenda = new Agenda({mongo: mongoSkinInstance});

processEvery(interval)

Takes a string interval which can be either a traditional javascript number, or a string such as 3 minutes

Specifies the frequency at which agenda will query the database looking for jobs that need to be processed. If your jobs are time sensitive, you will want to specify a low value.

agenda.processEvery('1 minute');

You can also specify it during instantiation

var agenda = new Agenda({processEvery: '30 seconds'});

maxConcurrency(number)

Takes a number which specifies the max number of jobs that can be running at any given moment. By default it is 20.

agenda.maxConcurrency(20);

You can also specify it during instantiation

var agenda = new Agenda({maxConcurrency: 20});

defaultConcurrency(number)

Takes a number which specifies the default number of a specific that can be running at any given moment. By default it is 5.

agenda.defaultConcurrency(5);

You can also specify it during instantiation

var agenda = new Agenda({defaultConcurrency: 5});

Defining Job Processors

Before you can use a job, you must define its processing behavior.

define(jobName, [options], fn)

Defines a job with the name of jobName. When a job of job name gets run, it will be passed to fn(job, done). To maintain asynchronous behavior, you must call done() when you are processing the job. If your function is synchronous, you may omit done from the signature.

options is an optional argument which can overwrite the defaults. It can take the following:

• concurrency: number maxinum number of that job that can be running at once (per instance of agenda)
• lockLifetime: number interval in ms of how long the job stays locked for (see multiple job processors for more info)
• priority: (lowest|low|normal|high|highest|number) specifies the priority of the job. Higher priority jobs will run first. See the priority mapping below

Priority mapping:

{
  highest: 20,
  high: 10,
  default: 0,
  low: -10,
  lowest: -20
}

Async Job:

agenda.define('some long running job', function(job, done) {
  doSomelengthyTask(function(data) {
    formatThatData(data);
    sendThatData(data);
    done();
  });
});

Sync Job:

agenda.define('say hello', function(job) {
  console.log("Hello!");
});

Creating Jobs

every(interval, name, [data])

Runs job name at the given interval. Optionally, data can be passed in. Every creates a job of type single, which means that it will only create one job in the database, even if that line is run multiple times. This lets you put it in a file that may get run multiple times, such as webserver.js which may reboot from time to time.

interval can be a human-readable format String, a cron format String, or a Number.

data is an optional argument that will be passed to the processing function under job.attrs.data.

Returns the job.

agenda.define('printAnalyticsReport', function(job, done) {
  User.doSomethingReallyIntensive(function(err, users) {
    processUserData();
    console.log("I print a report!");
    done();
  });
});

agenda.every('15 minutes', 'printAnalyticsReport');

schedule(when, name, data)

Schedules a job to run name once at a given time. when can be a Date or a String such as tomorrow at 5pm.

data is an optional argument that will be passed to the processing function under job.data.

Returns the job.

agenda.schedule('tomorrow at noon', 'printAnalyticsReport', {userCount: 100});

now(name, data)

Schedules a job to run name once immediately.

data is an optional argument that will be passed to the processing function under job.data.

Returns the job.

agenda.now('do the hokey pokey');

create(jobName, data)

Returns an instance of a jobName with data. This does NOT save the job in the database. See below to learn how to manually work with jobs.

var job = agenda.create('printAnalyticsReport', {userCount: 100});
job.save(function(err) {
  console.log("Job successfully saved");
});

Managing Jobs

jobs(mongoskin query)

Lets you query all of the jobs in the agenda job's database. This is a full mongoskin find query. See mongoskin's documentation for details.

agenda.jobs({name: 'printAnalyticsReport'}, function(err, jobs) {
  // Work with jobs (see below)
});

purge(cb)

Removes all jobs in the database without defined behaviors. Useful if you change a definition name and want to remove old jobs.

IMPORTANT: Do not run this before you finish defining all of your jobs. If you do, you will nuke your database of jobs.

agenda.purge(function(err, numRemoved) {
});

Starting the job processor

To get agenda to start processing jobs from the database you must start it. This will schedule an interval (based on processEvery) to check for new jobs and run them. You can also stop the queue.

start

Starts the job queue processing, checking processEvery time to see if there are new jobs.

stop

Stops the job queue processing.

Multiple job processors

Sometimes you may want to have multiple node instances / machines process from the same queue. Agenda supports a locking mechanism to ensure that multiple queues don't process the same job.

You can configure the locking mechanism by specifying lockLifetime as an interval when defining the job.

agenda.define('someJob', {lockLifetime: 10000}, function(job, cb) {
  //Do something in 10 seconds or less...
});

This will ensure that no other job processor (this one included) attempts to run the job again for the next 10 seconds. If you have a particularly long running job, you will want to specify a longer lockLifetime.

By default it is 10 minutes. Typically you shouldn't have a job that runs for 10 minutes, so this is really insurance should the job queue crash before the job is unlocked.

Manually working with a job

A job instance has many instance methods. All mutating methods must be followed with a call to job.save() in order to persist the changes to the database.

repeatEvery(interval)

Specifies an interval on which the job should repeat.

interval can be a human-readable format String, a cron format String, or a Number.

job.repeatEvery('10 minutes');
job.save();

schedule(time)

Specifies the next time at which the job should repeat.

job.schedule('tomorrow at 6pm');
job.save();

priority(priority)

Specifies the priority weighting of the job. Can be a number or a string from the above priority table.

job.priority('low');
job.save();

fail(reason)

Sets job.attrs.failedAt to now, and sets job.attrs.failReason to reason.

Optionally, reason can be an error, in which case job.attrs.failReason will be set to error.message

job.fail('insuficient disk space');
// or
job.fail(new Error('insufficient disk space'));
job.save();

run(callback)

Runs the given job and calls callback(err, job) upon completion. Normally you never need to call this manually.

job.run(function(err, job) {
  console.log("I don't know why you would need to do this...");
});

save(callback)

Saves the job.attrs into the database.

job.save(function(err) {
    if(!err) console.log("Successfully saved job to collection");
})

remove(callback)

Removes the job from the database.

job.save(function(err) {
    if(!err) console.log("Successfully saved job to collection");
})

Job Queue Events

An instance of an agenda will emit the following events:

• start - called just before a job starts
• start:job name - called just before the specified job starts

agenda.on('start', function(job) {
  console.log("Job %s starting", job.attrs.name);
});

• complete - called when a job finishes, regardless of if it succeeds or fails
• complete:job name - called when a job finishes, regardless of if it succeeds or fails

agenda.on('complete', function(job) {
  console.log("Job %s finished", job.attrs.name);
});

• success - called when a job finishes successfully
• success:job name - called when a job finishes successfully

agenda.once('success:send email', function(job) {
  console.log("Sent Email Successfully to: %s", job.attrs.data.to);
});

• fail - called when a job throws an error
• fail:job name - called when a job throws an error

agenda.on('fail:send email', function(err, job) {
  console.log("Job failed with error: %s", err.message);
});

Frequently Asked Questions

Mongo vs Redis

The decision to use Mongo instead of Redis is intentional. Redis is often used for non-essential data (such as sessions) and without configuration doesn't guarantee the same level of persistence as Mongo (should the server need to be restarted/crash).

Agenda decides to focus on persistence without requiring special configuration of Redis (thereby degrading the performance of the Redis server on non-critical data, such as sessions).

Ultimately if enough people want a Redis driver instead of Mongo, I will write one. (Please open an issue requesting it). For now, Agenda decided to focus on guaranteed persistence.

Spawning / forking processes.

Ultimately Agenda can work from a single job queue across multiple machines, node processes, or forks. If you are interested in having more than one worker, Bars3s has written up a fantastic example of how one might do it:

var cluster = require('cluster'),
    cpuCount = require('os').cpus().length,
    jobWorkers = [],
    webWorkers = [];

if (cluster.isMaster) {

    // Create a worker for each CPU
    for (var i = 0; i < cpuCount; i += 1) {
        addJobWorker();
        addWebWorker();
    }

    cluster.on('exit', function (worker, code, signal) {

        if (jobWorkers.indexOf(worker.id) != -1) {
            console.log('job worker ' + worker.process.pid + ' died. Trying to respawn...');
            removeJobWorker(worker.id);
            addJobWorker();
        }

        if (webWorkers.indexOf(worker.id) != -1) {
            console.log('http worker ' + worker.process.pid + ' died. Trying to respawn...');
            removeWebWorker(worker.id);
            addWebWorker();
        }
    });

} else {
    if (process.env.web) {
        console.log('start http server: ' + cluster.worker.id);
        require('./app/web-http');//initialize the http server here
    }

    if (process.env.job) {
        console.log('start job server: ' + cluster.worker.id);
        require('./app/job-worker');//initialize the agenda here 
    }
}

function addWebWorker() {
    webWorkers.push(cluster.fork({web: 1}).id);
}

function addJobWorker() {
    jobWorkers.push(cluster.fork({job: 1}).id);
}

function removeWebWorker(id) {
    webWorkers.splice(webWorkers.indexOf(id), 1);
}

function removeJobWorker(id) {
    jobWorkers.splice(jobWorkers.indexOf(id), 1);
}

License

(The MIT License)

Copyright (c) 2013 Ryan Schmukler [email protected]

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the 'Software'), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.