A light-weight job scheduling library for Node.js

Agenda offers

• Minimal overhead. Agenda aims to keep its code base small.
• Mongo backed persistence layer.
• Promises based API.
• Scheduling with configurable priority, concurrency, and repeating.
• Scheduling via cron or human readable syntax.
• Event backed job queue that you can hook into.
• Agendash: optional standalone web-interface.
• Agenda-rest: optional standalone REST API.
• inversify-agenda - Some utilities for the development of agenda workers with Inversify

Feature Comparison

Since there are a few job queue solutions, here a table comparing them to help you use the one that better suits your needs.

Agenda is great if you need a MongoDB job scheduler, but try Bree if you need something simpler (built by a previous maintainer).

| Feature | Bull | Bee | Agenda | | :------------- |:-------------:|:---:|:------:| | Backend | redis |redis| mongo | | Priorities | ✓ | | ✓ | | Concurrency | ✓ | ✓ | ✓ | | Delayed jobs | ✓ | | ✓ | | Global events | ✓ | | | | Rate Limiter | ✓ | | | | Pause/Resume | ✓ | | | | Sandboxed worker| ✓ | | | | Repeatable jobs | ✓ | | ✓ | | Atomic ops | ✓ | ✓ | | | Persistence | ✓ | ✓ | ✓ | | UI | ✓ | | ✓ | | REST API | | | ✓ | | Optimized for | Jobs / Messages | Messages | Jobs |

Kudos for making the comparison chart goes to Bull maintainers.

Installation

Install via NPM

npm install agenda

You will also need a working Mongo database (v3) to point it to.

Example Usage

const mongoConnectionString = 'mongodb://127.0.0.1/agenda';

const agenda = new Agenda({db: {address: mongoConnectionString}});
// Or override the default collection name:
// const agenda = new Agenda({db: {address: mongoConnectionString, collection: 'jobCollectionName'}});
// or pass additional connection options:
// const agenda = new Agenda({db: {address: mongoConnectionString, collection: 'jobCollectionName', options: {ssl: true}}});
// or pass in an existing mongodb-native MongoClient instance
// const agenda = new Agenda({mongo: myMongoClient});
agenda.define('delete old users', async job => {
  await User.remove({lastLogIn: {$lt: twoDaysAgo}});
});
(async function() { // IIFE to give access to async/await
  await agenda.start();
  await agenda.every('3 minutes', 'delete old users');
  // Alternatively, you could also do:
  await agenda.every('*/3 * * * *', 'delete old users');
})();

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

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

(async function() {
  const weeklyReport = agenda.create('send email report', {to: '[email protected]'});
  await agenda.start();
  await weeklyReport.repeatEvery('1 week').save();
})();

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
• Agenda Events
• Defining job processors
• Creating jobs
• Managing jobs
• Starting the job processor
• Multiple job processors
• Manually working with jobs
• Job Queue Events
• Frequently asked questions
• Example Project structure
• Known Issues
• Debugging Issues
• Acknowledgements

Configuring an agenda

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

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

Agenda uses Human Interval for specifying the intervals. It supports the following units:

seconds

minutes

hours

days

weeks

months

years

More sophisticated examples

agenda.processEvery('one minute');
agenda.processEvery('1.5 minutes');
agenda.processEvery('3 days and 4 hours');
agenda.processEvery('3 days, 4 hours and 36 seconds');

database(url, [collectionName])

Specifies the database at the

url

agendaJobs

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

You can also specify it during instantiation.

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

Agenda will emit a

ready

agenda.start()

db

database

ready

mongo(dbInstance)

Use an existing mongodb-native MongoClient/Db instance. This can help consolidate connections to a database. You can instead use

.database

You can also specify it during instantiation:

const agenda = new Agenda({mongo: mongoClientInstance.db('agenda-test')});

Note that MongoClient.connect() returns a mongoClientInstance since node-mongodb-native 3.0.0, while it used to return a dbInstance that could then be directly passed to agenda.

name(name)

Takes a string

name

lastModifiedBy

agenda.name(os.hostname + '-' + process.pid);

You can also specify it during instantiation

const agenda = new Agenda({name: 'test queue'});

processEvery(interval)

Takes a string

interval

3 minutes

Specifies the frequency at which agenda will query the database looking for jobs that need to be processed. Agenda internally uses

setTimeout

Decreasing the frequency will result in fewer database queries, but more jobs being stored in memory.

Also worth noting is that if the job queue is shutdown, any jobs stored in memory that haven't run will still be locked, meaning that you may have to wait for the lock to expire. By default it is

'5 seconds'

agenda.processEvery('1 minute');

You can also specify it during instantiation

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

maxConcurrency(number)

Takes a

number

20

agenda.maxConcurrency(20);

You can also specify it during instantiation

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

defaultConcurrency(number)

Takes a

number

5

agenda.defaultConcurrency(5);

You can also specify it during instantiation

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

lockLimit(number)

Takes a

number

0

agenda.lockLimit(0);

You can also specify it during instantiation

const agenda = new Agenda({lockLimit: 0});

defaultLockLimit(number)

Takes a

number

0

agenda.defaultLockLimit(0);

You can also specify it during instantiation

const agenda = new Agenda({defaultLockLimit: 0});

defaultLockLifetime(number)

Takes a

number

lockLifetime

A job will unlock if it is finished (ie. the returned Promise resolves/rejects or

done

done()

lockLifetime

agenda.defaultLockLifetime(10000);

You can also specify it during instantiation

const agenda = new Agenda({defaultLockLifetime: 10000});

sort(query)

Takes a

query

By default it is

{ nextRunAt: 1, priority: -1 }

Agenda Events

An instance of an agenda will emit the following events:

• ready

  - called when Agenda mongo connection is successfully opened and indices created. If you're passing agenda an existing connection, you shouldn't need to listen for this, as

  agenda.start()

  will not resolve until indices have been created. If you're using the

  db

  options, or call

  database

  , then you may still need to listen for the

  ready

  event before saving jobs.

  agenda.start()

  will still wait for the connection to be opened.

• error

  - called when Agenda mongo connection process has thrown an error

await agenda.start();

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

jobName

fn(job, done)

fn

done

fn

done

done()

done

options

• concurrency

  :

  number

  maximum number of that job that can be running at once (per instance of agenda)

• lockLimit

  :

  number

  maximum number of that job that can be locked 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). A job will automatically unlock once a returned promise resolves/rejects (or if

  done

  is specified in the signature and

  done()

  is called).

• 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,
  normal: 0,
  low: -10,
  lowest: -20
}

Async Job:

js
agenda.define('some long running job', async job => {
  const data = await doSomelengthyTask();
  await formatThatData(data);
  await sendThatData(data);
});

Async Job (using

done

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

Sync Job:

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

define()

define(jobName, ...)

define

jobName

Creating Jobs

every(interval, name, [data], [options])

Runs job

name

interval

single

webserver.js

interval

String

String

Number

data

job.attrs.data

options

job.repeatEvery

data

Returns the

job

agenda.define('printAnalyticsReport', async job => {
  const users = await User.doSomethingReallyIntensive();
  processUserData(users);
  console.log('I print a report!');
});

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

Optionally,

name

interval

agenda.every('15 minutes', ['printAnalyticsReport', 'sendNotifications', 'updateUserRecords']);

In this case,

every

jobs

schedule(when, name, [data])

Schedules a job to run

name

when

Date

String

tomorrow at 5pm

data

job.attrs.data

Returns the

job

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

Optionally,

name

every

agenda.schedule('tomorrow at noon', ['printAnalyticsReport', 'sendNotifications', 'updateUserRecords']);

In this case,

schedule

jobs

now(name, [data])

Schedules a job to run

name

data

job.attrs.data

Returns the

job

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

create(jobName, data)

Returns an instance of a

jobName

data

const job = agenda.create('printAnalyticsReport', {userCount: 100});
await job.save();
console.log('Job successfully saved');

Managing Jobs

jobs(mongodb-native query, mongodb-native sort, mongodb-native limit, mongodb-native skip)

Lets you query (then sort, limit and skip the result) all of the jobs in the agenda job's database. These are full mongodb-native

find

sort

limit

skip

const jobs = await agenda.jobs({name: 'printAnalyticsReport'}, {data:-1}, 3, 1);
// Work with jobs (see below)

cancel(mongodb-native query)

Cancels any jobs matching the passed mongodb-native query, and removes them from the database. Returns a Promise resolving to the number of cancelled jobs, or rejecting on error.

const numRemoved = await agenda.cancel({name: 'printAnalyticsReport'});

This functionality can also be achieved by first retrieving all the jobs from the database using

agenda.jobs()

job.remove()

agenda.cancel()

purge()

Removes all jobs in the database without defined behaviors. Useful if you change a definition name and want to remove old jobs. Returns a Promise resolving to the number of removed jobs, or rejecting on error.

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

const numRemoved = await agenda.purge();

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

start

Starts the job queue processing, checking

processEvery

processEvery

every

stop

Stops the job queue processing. Unlocks currently running jobs.

This can be very useful for graceful shutdowns so that currently running/grabbed jobs are abandoned so that other job queues can grab them / they are unlocked should the job queue start again. Here is an example of how to do a graceful shutdown.

async function graceful() {
  await agenda.stop();
  process.exit(0);
}

process.on('SIGTERM', graceful);
process.on('SIGINT' , graceful);

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

agenda.define('someJob', {lockLifetime: 10000}, (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.

When a job is finished (i.e. the returned promise resolves/rejects or

done

done()

Manually working with a job

A job instance has many instance methods. All mutating methods must be followed with a call to

await job.save()

repeatEvery(interval, [options])

Specifies an

interval

interval

String

String

Number

options

options.timezone

options.skipImmediate

true

false

true

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

job.repeatEvery('3 minutes', {
  skipImmediate: true
});
await job.save();

job.repeatEvery('0 6 * * *', {
  timezone: 'America/New_York'
});
await job.save();

repeatAt(time)

Specifies a

time

job.repeatAt('3:30pm');
await job.save();

schedule(time)

Specifies the next

time

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

priority(priority)

Specifies the

priority

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

unique(properties, [options])

Ensure that only one instance of this job exists with the specified properties

options

• insertOnly

  :

  boolean

  will prevent any properties from persisting if the job already exists. Defaults to false.

job.unique({'data.type': 'active', 'data.userId': '123', nextRunAt: date});
await job.save();

IMPORTANT: To avoid high CPU usage by MongoDB, make sure to create an index on the used fields, like

data.type

data.userId

fail(reason)

Sets

job.attrs.failedAt

now

job.attrs.failReason

reason

Optionally,

reason

job.attrs.failReason

error.message

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

run(callback)

Runs the given

job

callback(err, job)

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

save()

Saves the

job.attrs

try {
  await job.save();
  cosole.log('Successfully saved job to collection');
} catch (e) {
  console.error('Error saving job to collection');
}

remove()

Removes the

job

try {
  await job.remove();
  console.log('Successfully removed job from collection');
} catch (e) {
  console.error('Error removing job from collection');
}

disable()

Disables the

job

enable()

Enables the

job

touch()

Resets the lock on the job. Useful to indicate that the job hasn't timed out when you have very long running jobs. The call returns a promise that resolves when the job's lock has been renewed.

agenda.define('super long job', async job => {
  await doSomeLongTask();
  await job.touch();
  await doAnotherLongTask();
  await job.touch();
  await finishOurLongTasks();
});

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', 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', job => {
  console.log(`Job ${job.attrs.name} finished`);
});

• success

  - called when a job finishes successfully

• success:job name

  - called when a job finishes successfully

agenda.on('success:send email', job => {
  console.log(`Sent Email Successfully to ${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', (err, job) => {
  console.log(`Job failed with error: ${err.message}`);
});

Frequently Asked Questions

What is the order in which jobs run?

Jobs are run with priority in a first in first out order (so they will be run in the order they were scheduled AND with respect to highest priority).

For example, if we have two jobs named "send-email" queued (both with the same priority), and the first job is queued at 3:00 PM and second job is queued at 3:05 PM with the same

priority

5

10

The default MongoDB sort object is

{ nextRunAt: 1, priority: -1 }

sort

What is the difference between

lockLimit

and

maxConcurrency

?

Agenda will lock jobs 1 by one, setting the

lockedAt

Job

_lockedJobs

agenda.processEvery

Agenda will also pull jobs from

_lockedJobs

_runningJobs

maxConcurrency

If you have multiple instances of agenda processing the same job definition with a fast repeat time you may find they get unevenly loaded. This is because they will compete to lock as many jobs as possible, even if they don't have enough concurrency to process them. This can be resolved by tweaking the

maxConcurrency

lockLimit

Sample Project Structure?

Agenda doesn't have a preferred project structure and leaves it to the user to choose how they would like to use it. That being said, you can check out the example project structure below.

Can I Donate?

Thanks! I'm flattered, but it's really not necessary. If you really want to, you can find my gittip here.

Web Interface?

Agenda itself does not have a web interface built in but we do offer stand-alone web interface Agendash:

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:

const cluster = require('cluster');
const os = require('os');

const httpServer = require('./app/http-server');
const jobWorker = require('./app/job-worker');
const jobWorkers = [];
const webWorkers = [];
if (cluster.isMaster) {
  const cpuCount = os.cpus().length;
  // Create a worker for each CPU
  for (let i = 0; i < cpuCount; i += 1) {
    addJobWorker();
    addWebWorker();
  }
  cluster.on('exit', (worker, code, signal) => {
    if (jobWorkers.indexOf(worker.id) !== -1) {
      console.log(job worker ${worker.process.pid} exited (signal: ${signal}). Trying to respawn...);
      removeJobWorker(worker.id);
      addJobWorker();
    }
if (webWorkers.indexOf(worker.id) !== -1) {
  console.log(`http worker ${worker.process.pid} exited (signal: ${signal}). Trying to respawn...`);
  removeWebWorker(worker.id);
  addWebWorker();
}
  });
} else {
  if (process.env.web) {
    console.log(start http server: ${cluster.worker.id});
    // Initialize the http server here
    httpServer.start();
  }
  if (process.env.job) {
    console.log(start job server: ${cluster.worker.id});
    // Initialize the Agenda here
    jobWorker.start();
  }
}
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);
}

Recovering lost Mongo connections ("auto_reconnect")

Agenda is configured by default to automatically reconnect indefinitely, emitting an error event when no connection is available on each process tick, allowing you to restore the Mongo instance without having to restart the application.

However, if you are using an existing Mongo client you'll need to configure the

reconnectTries

reconnectInterval

Example Project Structure

Agenda will only process jobs that it has definitions for. This allows you to selectively choose which jobs a given agenda will process.

Consider the following project structure, which allows us to share models with the rest of our code base, and specify which jobs a worker processes, if any at all.

- server.js
- worker.js
lib/
  - agenda.js
  controllers/
    - user-controller.js
  jobs/
    - email.js
    - video-processing.js
    - image-processing.js
   models/
     - user-model.js
     - blog-post.model.js

Sample job processor (eg.

jobs/email.js

let email = require('some-email-lib'),
  User = require('../models/user-model.js');

module.exports = function(agenda) {
  agenda.define('registration email', async job => {
    const user = await User.get(job.attrs.data.userId);
    await email(user.email(), 'Thanks for registering', 'Thanks for registering ' + user.name());
  });
  agenda.define('reset password', async job => {
    // Etc
  });
  // More email related jobs
};

lib/agenda.js ```js const Agenda = require('agenda');

const connectionOpts = {db: {address: 'localhost:27017/agenda-test', collection: 'agendaJobs'}};

const agenda = new Agenda(connectionOpts);

const jobTypes = process.env.JOBTYPES ? process.env.JOBTYPES.split(',') : [];

jobTypes.forEach(type => { require('./jobs/' + type)(agenda); });

if (jobTypes.length) { agenda.start(); // Returns a promise, which should be handled appropriately }

module.exports = agenda; ```

lib/controllers/user-controller.js ```js let app = express(), User = require('../models/user-model'), agenda = require('../worker.js');

app.post('/users', (req, res, next) => { const user = new User(req.body); user.save(err => { if (err) { return next(err); } agenda.now('registration email', {userId: user.primary()}); res.send(201, user.toJson()); }); }); ```

worker.js

js
require('./lib/agenda.js');

Now you can do the following in your project:

bash
node server.js

Fire up an instance with no

JOB_TYPES

bash
JOB_TYPES=email node server.js

Allow your http server to process email jobs.

bash
JOB_TYPES=email node worker.js

Fire up an instance that processes email jobs.

bash
JOB_TYPES=video-processing,image-processing node worker.js

Fire up an instance that processes video-processing/image-processing jobs. Good for a heavy hitting server.

Debugging Issues

If you think you have encountered a bug, please feel free to report it here:

Submit Issue

Please provide us with as much details as possible such as: - Agenda version - Environment (OSX, Linux, Windows, etc) - Small description of what happened - Any relevant stack track - Agenda logs (see below)

To turn on logging, please set your DEBUG env variable like so:

• OSX:

  DEBUG="agenda:*" node index.js

• Linux:

  DEBUG="agenda:*" node index.js

• Windows CMD:

  set DEBUG=agenda:*

• Windows PowerShell:

  $env:DEBUG = "agenda:*"

While not necessary, attaching a text file with this debug information would be extremely useful in debugging certain issues and is encouraged.

Known Issues

"Multiple order-by items are not supported. Please specify a single order-by item."

When running Agenda on Azure cosmosDB, you might run into this issue caused by Agenda's sort query used for finding and locking the next job. To fix this, you can pass custom sort option:

sort: { nextRunAt: 1 }

Acknowledgements

• Agenda was originally created by @rschmukler.
• Agendash was originally created by @joeframbach.
• These days Agenda has a great community of contributors around it. Join us!

License

The MIT License