JoSk

Simple package with similar API to native setTimeout and setInterval methods, but synced between all running Node.js instances via MongoDB Collection.

Multi-instance task manager for Node.js. This package has the support of clusters, multi-server and multi-threaded Node.js instances. This package goal is to make sure that the only single process of each task (job/cron) is running across multi-server (multi-thread/multi-instance) setup.

This is a server-only package.

ToC:

• Install as NPM package
• Install Meteor as Atmosphere package
• API
  • Constructor new JoSk()
  • JoSk#setInterval()
  • JoSk#setTimeout()
  • JoSk#setImmediate()
  • JoSk#clearInterval()
  • JoSk#clearTimeout()
  • JoSk#destroy()
• ~90% tests coverage

Main features:

• 👨‍🔬 ~90% tests coverage;
• 📦 Zero dependencies, written from scratch for top performance;
• 🏢 Synchronize single task across multiple servers;
• 💪 Bulletproof design, built-in retries, and "zombie" task recovery 🧟🔫.

Install:

# for node@>=8.9.0
npm install josk --save

# for node@<8.9.0
npm install josk@=1.1.0 --save

1const JoSk = require('josk');
2
3//ES6 Style:
4import JoSk from 'josk';

Install Meteor:

meteor add ostrio:cron-jobs

1import JoSk from 'meteor/ostrio:cron-jobs';

Known Meteor Issues:

Error: Can't wait without a fiber

Can be easily solved via "bounding to Fiber":

1const bound = Meteor.bindEnvironment((callback) => {
2  callback();
3});
4
5const db  = Collection.rawDatabase();
6const job = new JoSk({db: db});
7
8const task = (ready) => {
9  bound(() => { // <-- use "bound" inside of a task
10    ready();
11  });
12};
13
14job.setInterval(task, 60 * 60 * 1000, 'task');

Notes:

• This package is perfect when you have multiple servers for load-balancing, durability, an array of micro-services or any other solution with multiple running copies of code when you need to run repeating tasks, and you need to run it only once per app, not per server.
• Limitation — task must be run not often than once per two seconds (from 2 to ∞ seconds). Example tasks: Email, SMS queue, Long-polling requests, Periodical application logic operations or Periodical data fetch and etc.
• Accuracy — Delay of each task depends on MongoDB and "de-synchronization delay". Trusted time-range of execution period is task_delay ± (256 + MongoDB_Connection_And_Request_Delay). That means this package won't fit when you need to run a task with very certain delays. For other cases, if ±256 ms delays are acceptable - this package is the great solution.
• Use opts.minRevolvingDelay and opts.maxRevolvingDelay to set the range for random delays between executions. Revolving range acts as a safety control to make sure different servers not picking the same task at the same time. Default values (32 and 256) are the best for 3-server setup (the most common topology). Tune these options to match needs of your project. Higher opts.minRevolvingDelay will reduce load on MongoDB.
• To avoid "DB locks" — it's recommended to use separate DB from "main" application DB (same MongoDB server can have multiple DBs)

API:

new JoSk({opts}):

• opts.db {Object} - [Required] Connection to MongoDB, like returned as argument from MongoClient.connect()
• opts.prefix {String} - [Optional] use to create multiple named instances
• opts.autoClear {Boolean} - [Optional] Remove (Clear) obsolete tasks (any tasks which are not found in the instance memory (runtime), but exists in the database). Obsolete tasks may appear in cases when it wasn't cleared from the database on process shutdown, and/or was removed/renamed in the app. Obsolete tasks may appear if multiple app instances running different codebase within the same database, and the task may not exist on one of the instances. Default: false
• opts.resetOnInit {Boolean} - [Optional] make sure all old tasks is completed before set new one. Useful when you run only one instance of app, or multiple app instances on one machine, in case machine was reloaded during running task and task is unfinished
• opts.zombieTime {Number} - [Optional] time in milliseconds, after this time - task will be interpreted as "zombie". This parameter allows to rescue task from "zombie mode" in case when: ready() wasn't called, exception during runtime was thrown, or caused by bad logic. While resetOnInit option helps to make sure tasks are done on startup, zombieTime option helps to solve same issue, but during runtime. Default value is 900000 (15 minutes). It's not recommended to set this value to less than a minute (60000ms)
• opts.minRevolvingDelay {Number} - [Optional] Minimum revolving delay — the minimum delay between tasks executions in milliseconds. Default: 32
• opts.maxRevolvingDelay {Number} - [Optional] Maximum revolving delay — the maximum delay between tasks executions in milliseconds. Default: 256
• opts.onError {Function} - [Optional] Informational hook, called instead of throwing exceptions. Default: false. Called with two arguments:
  • title {String}
  • details {Object}
  • details.description {String}
  • details.error {Mix}
  • details.uid {String} - Internal uid, suitable for .clearInterval() and .clearTimeout()
• opts.onExecuted {Function} - [Optional] Informational hook, called when task is finished. Default: false. Called with two arguments:
  • uid {String} - uid passed into .setImmediate(), .setTimeout(), or setInterval() methods
  • details {Object}
  • details.uid {String} - Internal uid, suitable for .clearInterval() and .clearTimeout()
  • details.date {Date} - Execution timestamp as JS {Date}
  • details.delay {Number} - Execution delay (e.g. interval for .setInterval())
  • details.timestamp {Number} - Execution timestamp as unix {Number}

Initialization:

1MongoClient.connect('url', (error, client) => {
2  // To avoid "DB locks" — it's a good idea to use separate DB from "main" application DB
3  const db = client.db('dbName');
4  const job = new JoSk({db: db});
5});

Initialization in Meteor:

1// Meteor.users.rawDatabase() is available in most Meteor setups
2// If this is not your case, you can use `rawDatabase` form any other collection
3const db  = Meteor.users.rawDatabase();
4const job = new JoSk({db: db});

Note: This library relies on job ID, so you can not pass same job (with the same ID). Always use different uid, even for the same task:

1const task = function (ready) {
2  //...some code here
3  ready();
4};
5
6job.setInterval(task, 60 * 60 * 1000, 'task-1000');
7job.setInterval(task, 60 * 60 * 2000, 'task-2000');

Passing arguments (not really fancy solution, sorry):

1const job = new JoSk({db: db});
2let globalVar = 'Some top level or env.variable (can be changed during runtime)';
3
4const task = function (arg1, arg2, ready) {
5  //...some code here
6  ready();
7};
8
9const taskB = function (ready) {
10  task(globalVar, 'b', ready);
11};
12
13const task1 = function (ready) {
14  task(1, globalVar, ready);
15};
16
17job.setInterval(taskB, 60 * 60 * 1000, 'taskB');
18job.setInterval(task1, 60 * 60 * 1000, 'task1');

Note: To clean up old tasks via MongoDB use next query pattern:

1// Run directly in MongoDB console:
2db.getCollection('__JobTasks__').remove({});
3// If you're using multiple JoSk instances with prefix:
4db.getCollection('__JobTasks__PrefixHere').remove({});

setInterval(func, delay, uid)

• func {Function} - Function to call on schedule
• delay {Number} - Delay for first run and interval between further executions in milliseconds
• uid {String} - Unique app-wide task id

Set task into interval execution loop. ready() is passed as the first argument into task function.

In this example, next task will not be scheduled until the current is ready:

1const syncTask = function (ready) {
2  //...run sync code
3  ready();
4};
5const asyncTask = function (ready) {
6  asyncCall(function () {
7    //...run more async code
8    ready();
9  });
10};
11
12job.setInterval(syncTask, 60 * 60 * 1000, 'syncTask');
13job.setInterval(asyncTask, 60 * 60 * 1000, 'asyncTask');

In this example, next task will not wait for the current task to finish:

1const syncTask = function (ready) {
2  ready();
3  //...run sync code
4};
5const asyncTask = function (ready) {
6  ready();
7  asyncCall(function () {
8    //...run more async code
9  });
10};
11
12job.setInterval(syncTask, 60 * 60 * 1000, 'syncTask');
13job.setInterval(asyncTask, 60 * 60 * 1000, 'asyncTask');

In this example, we're assuming to have long running task, executed in a loop without delay, but after full execution:

1const longRunningAsyncTask = function (ready) {
2  asyncCall((error, result) => {
3    if (error) {
4      ready(); // <-- Always run `ready()`, even if call was unsuccessful
5    } else {
6      anotherCall(result.data, ['param'], (error, response) => {
7        waitForSomethingElse(response, () => {
8          ready(); // <-- End of full execution
9        });
10      });
11    }
12  });
13};
14
15job.setInterval(longRunningAsyncTask, 0, 'longRunningAsyncTask');

setTimeout(func, delay, uid)

• func {Function} - Function to call on schedule
• delay {Number} - Delay in milliseconds
• uid {String} - Unique app-wide task id

Set task into timeout execution. setTimeout is useful for cluster - when you need to make sure task was executed only once. ready() is passed as the first argument into task function.

1const syncTask = function (ready) {
2  //...run sync code
3  ready();
4};
5const asyncTask = function (ready) {
6  asyncCall(function () {
7    //...run more async code
8    ready();
9  });
10};
11
12job.setTimeout(syncTask, 60 * 60 * 1000, 'syncTask');
13job.setTimeout(asyncTask, 60 * 60 * 1000, 'asyncTask');

setImmediate(func, uid)

• func {Function} - Function to execute
• uid {String} - Unique app-wide task id

Immediate execute the function, and only once. setImmediate is useful for cluster - when you need to execute function immediately and only once across all servers. ready() is passed as the first argument into the task function.

1const syncTask = function (ready) {
2  //...run sync code
3  ready();
4};
5const asyncTask = function (ready) {
6  asyncCall(function () {
7    //...run more async code
8    ready();
9  });
10};
11
12job.setImmediate(syncTask, 'syncTask');
13job.setImmediate(asyncTask, 'asyncTask');

clearInterval(timer)

Cancel (abort) current interval timer. Must be called in a separate event loop from setInterval.

1const timer = job.setInterval(func, 34789, 'unique-taskid');
2job.clearInterval(timer);

clearTimeout(timer)

Cancel (abort) current timeout timer. Should be called in a separate event loop from setTimeout.

1const timer = job.setTimeout(func, 34789, 'unique-taskid');
2job.clearTimeout(timer);

destroy()

Destroy JoSk instance. This method shouldn't be called in normal circumstances. Stop internal interval timer. After JoSk is destroyed — calling public methods would end up logged to std or if onError hook was passed to JoSk it would receive an error. Only permitted methods are clearTimeout and clearInterval.

1const job = new JoSk({db: db});
2
3job.destroy();

Running Tests

1. Clone this package
2. In Terminal (Console) go to directory where package is cloned
3. Then run:

# Before run tests make sure NODE_ENV === development
# Install NPM dependencies
npm install --save-dev

# Before run tests you need to have running MongoDB
MONGO_URL="mongodb://127.0.0.1:27017/npm-josk-test-001" npm test

# Be patient, tests are taking around 2 mins

Running Tests in Meteor environment

# Default
meteor test-packages ./ --driver-package=meteortesting:mocha

# With custom port
meteor test-packages ./ --driver-package=meteortesting:mocha --port 8888

# With local MongoDB and custom port
MONGO_URL="mongodb://127.0.0.1:27017/meteor-josk-test-001" meteor test-packages ./ --driver-package=meteortesting:mocha --port 8888

# Be patient, tests are taking around 2 mins

Why JoSk?

JoSk is Job-Task - Is randomly generated name by "uniq" project

Support our open source contribution:

• Support via PayPal — support my open source contributions once or on regular basis
• Use ostr.io — Monitoring, Analytics, WebSec, Web-CRON and Pre-rendering for a website