JoSk
"JoSk" is a Node.js task manager for horizontally scaled apps and apps that would need to scale horizontally quickly at some point of growth.
"JoSk" mimics the native API of setTimeout
and setInterval
and supports CRON expressions. All queued tasks are synced between all running application instances via Redis, MongoDB, or a custom adapter.
The "JoSk" package is made for a variety of horizontally scaled apps, such as clusters, multi-servers, and multi-threaded Node.js instances, that are running either on the same or different machines or even different data centers. "JoSk ensures that the only single execution of each task occurs across all running instances of the application.
"JoSk" is not just for multi-instance apps. It seamlessly integrates with single-instance applications as well, showcasing its versatility and adaptability.
Note: JoSk is the server-only package.
ToC
- Main features
- Prerequisites
- Install as NPM package
- API
- Examples
- Important notes
- ~99% tests coverage
- Why it's named "JoSk"
- Support Section
Main features
- 🏢 Synchronize single task across multiple servers;
- 🔏 Read locking to avoid simultaneous task executions across complex infrastructure;
- 📦 Zero dependencies, written from scratch for top performance;
- 👨🔬 ~99% tests coverage;
- 💪 Bulletproof design, built-in retries, and "zombie" task recovery 🧟🔫.
Prerequisites
redis-server@>=5.0.0
— Redis Server Version (if used with RedisAdapter)mongod@>=4.0.0
— MongoDB Server Version (if used with MongoAdapter)node@>=14.20.0
— Node.js version
Older releases compatibility
mongod@<4.0.0
— usejosk@=1.1.0
node@<14.20.0
— usejosk@=3.0.2
node@<8.9.0
— usejosk@=1.1.0
Install:
npm install josk --save
1// ES Module Style 2import { JoSk, RedisAdapter, MongoAdapter } from 'josk'; 3 4// CommonJS 5const { JoSk, RedisAdapter, MongoAdapter } = require('josk');
API:
Constructor options for JoSK, MongoAdapter, and RedisAdapter
new JoSk(opts)
opts.adapter
{RedisAdapter|MongoAdapter} - [Required] Instance ofRedisAdapter
orMongoAdapter
or custom adapteropts.debug
{Boolean} - [Optional] Enable debugging messages, useful during developmentopts.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.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. WhileresetOnInit
option helps to make sure tasks aredone
on startup,zombieTime
option helps to solve same issue, but during runtime. Default value is900000
(15 minutes). It's not recommended to set this value to below60000
(one minute)opts.minRevolvingDelay
{Number} - [Optional] Minimum revolving delay — the minimum delay between tasks executions in milliseconds. Default:128
opts.maxRevolvingDelay
{Number} - [Optional] Maximum revolving delay — the maximum delay between tasks executions in milliseconds. Default:768
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} - Internaluid
, 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()
, orsetInterval()
methodsdetails
{Object}details.uid
{String} - Internaluid
, suitable for.clearInterval()
and.clearTimeout()
details.date
{Date} - Execution timestamp as JS {Date}details.delay
{Number} - Executiondelay
(e.g.interval
for.setInterval()
)details.timestamp
{Number} - Execution timestamp as unix {Number}
new RedisAdapter(opts)
Since v5.0.0
opts.client
{RedisClient} - [Required]RedisClient
instance, like one returned fromawait redis.createClient().connect()
methodopts.prefix
{String} - [Optional] use to create multiple named instancesopts.resetOnInit
{Boolean} - [Optional] (use with caution) make sure all old tasks are completed during initialization. Useful for single-instance apps to clean up unfinished that occurred due to intermediate shutdown, reboot, or exception. Default:false
new MongoAdapter(opts)
Since v5.0.0
opts.db
{Db} - [Required] Mongo'sDb
instance, like one returned fromMongoClient#db()
methodopts.prefix
{String} - [Optional] use to create multiple named instancesopts.lockCollectionName
{String} - [Optional] By default all JoSk instances use the same__JobTasks__.lock
collection for lockingopts.resetOnInit
{Boolean} - [Optional] (use with caution) make sure all old tasks are completed during initialization. Useful for single-instance apps to clean up unfinished that occurred due to intermediate shutdown, reboot, or exception. Default:false
Initialization
JoSk is storage-agnostic (since v4.0.0
). It's shipped with Redis and MongoDB "adapters" out of the box, with option to extend its capabilities by creating and passing a custom adapter
Redis Adapter
JoSk has no dependencies, hence make sure redis
NPM package is installed in order to support Redis Storage Adapter. RedisAdapter
utilize basic set of commands SET
, GET
, DEL
, EXISTS
, HSET
, HGETALL
, and SCAN
. RedisAdapter
is compatible with all Redis-alike databases, and was well-tested with Redis and KeyDB
1import { JoSk, RedisAdapter } from 'josk'; 2import { createClient } from 'redis'; 3 4const redisClient = await createClient({ 5 url: 'redis://127.0.0.1:6379' 6}).connect(); 7 8const jobs = new JoSk({ 9 adapter: new RedisAdapter({ 10 client: redisClient, 11 prefix: 'app-scheduler', 12 }), 13 onError(reason, details) { 14 // Use onError hook to catch runtime exceptions 15 // thrown inside scheduled tasks 16 console.log(reason, details.error); 17 } 18});
MongoDB Adapter
JoSk has no dependencies, hence make sure mongodb
NPM package is installed in order to support MongoDB Storage Adapter. Note: this package will add two new MongoDB collections per each new JoSk()
. One collection for tasks and second for "Read Locking" with .lock
suffix
1import { JoSk, MongoAdapter } from 'josk'; 2import { MongoClient } from 'mongodb'; 3 4const client = new MongoClient('mongodb://127.0.0.1:27017'); 5// To avoid "DB locks" — it's a good idea to use separate DB from the "main" DB 6const mongoDb = client.db('joskdb'); 7const jobs = new JoSk({ 8 adapter: new MongoAdapter({ 9 db: mongoDb, 10 prefix: 'cluster-scheduler', 11 }), 12 onError(reason, details) { 13 // Use onError hook to catch runtime exceptions 14 // thrown inside scheduled tasks 15 console.log(reason, details.error); 16 } 17});
Create the first task
After JoSk initialized simply call JoSk#setInterval
to create recurring task
1const jobs = new JoSk({ /*...*/ }); 2 3jobs.setInterval((ready) => { 4 /* ...code here... */ 5 ready(); 6}, 60 * 60000, 'task1h'); // every hour 7 8jobs.setInterval((ready) => { 9 /* ...code here... */ 10 asyncCall(() => { 11 /* ...more code here...*/ 12 ready(); 13 }); 14}, 15 * 60000, 'asyncTask15m'); // every 15 mins 15 16/** 17 * no need to call ready() inside async function 18 */ 19jobs.setInterval(async () => { 20 try { 21 await asyncMethod(); 22 } catch (err) { 23 console.log(err) 24 } 25}, 30 * 60000, 'asyncAwaitTask30m'); // every 30 mins 26 27/** 28 * no need to call ready() when call returns Promise 29 */ 30jobs.setInterval(() => { 31 return asyncMethod(); // <-- returns Promise 32}, 2 * 60 * 60000, 'asyncAwaitTask2h'); // every two hours
Note: This library relies on job ID. Always use different uid
, even for the same task:
1const task = function (ready) { 2 //... code here 3 ready(); 4}; 5 6jobs.setInterval(task, 60000, 'task-1m'); // every minute 7jobs.setInterval(task, 2 * 60000, 'task-2m'); // every two minutes
setInterval(func, delay, uid)
func
{Function} - Function to call on scheduledelay
{Number} - Delay for the first run and interval between further executions in millisecondsuid
{String} - Unique app-wide task id- Returns: {Promise}
Set task into interval execution loop. ready()
callback is passed as the first argument into a task function.
In the example below, the next task will not be scheduled until the current is ready:
1jobs.setInterval(function (ready) { 2 /* ...run sync code... */ 3 ready(); 4}, 60 * 60000, 'syncTask1h'); // will execute every hour + time to execute the task 5 6jobs.setInterval(async function () { 7 try { 8 await asyncMethod(); 9 } catch (err) { 10 console.log(err) 11 } 12}, 60 * 60000, 'asyncAwaitTask1h'); // will execute every hour + time to execute the task
In the example below, the next task will not wait for the current task to finish:
1jobs.setInterval(function (ready) { 2 ready(); 3 /* ...run sync code... */ 4}, 60 * 60000, 'syncTask1h'); // will execute every hour 5 6jobs.setInterval(async function () { 7 /* ...task re-scheduled instantly here... */ 8 process.nextTick(async () => { 9 await asyncMethod(); 10 }); 11}, 60 * 60000, 'asyncAwaitTask1h'); // will execute every hour
In the next example, a long running task is executed in a loop without delay after the full execution:
1jobs.setInterval(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 if (error) { 8 ready(); // <-- Always run `ready()`, even if call was unsuccessful 9 return; 10 } 11 12 waitForSomethingElse(response, () => { 13 ready(); // <-- End of the full execution 14 }); 15 }); 16 } 17 }); 18}, 0, 'longRunningAsyncTask'); // run in a loop as soon as previous run is finished
Same task combining await
/async
and callbacks
1jobs.setInterval(function (ready) { 2 process.nextTick(async () => { 3 try { 4 const result = await asyncCall(); 5 const response = await anotherCall(result.data, ['param']); 6 7 waitForSomethingElse(response, () => { 8 ready(); // <-- End of the full execution 9 }); 10 } catch (err) { 11 console.log(err) 12 ready(); // <-- Always run `ready()`, even if call was unsuccessful 13 } 14 }); 15}, 0, 'longRunningAsyncTask'); // run in a loop as soon as previous run is finished
setTimeout(func, delay, uid)
func
{Function} - Function to call afterdelay
delay
{Number} - Delay in millisecondsuid
{String} - Unique app-wide task id- Returns: {Promise}
Run a task after delay in ms. setTimeout
is useful for cluster - when you need to make sure task executed only once. ready()
callback is passed as the first argument into a task function.
1jobs.setTimeout(function (ready) { 2 /* ...run sync code... */ 3 ready(); 4}, 60000, 'syncTaskIn1m'); // will run only once across the cluster in a minute 5 6jobs.setTimeout(function (ready) { 7 asyncCall(function () { 8 /* ...run async code... */ 9 ready(); 10 }); 11}, 60000, 'asyncTaskIn1m'); // will run only once across the cluster in a minute 12 13jobs.setTimeout(async function () { 14 try { 15 /* ...code here... */ 16 await asyncMethod(); 17 /* ...more code here...*/ 18 } catch (err) { 19 console.log(err) 20 } 21}, 60000, 'asyncAwaitTaskIn1m'); // will run only once across the cluster in a minute
setImmediate(func, uid)
func
{Function} - Function to executeuid
{String} - Unique app-wide task id- Returns: {Promise}
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.
1jobs.setImmediate(function (ready) { 2 //...run sync code 3 ready(); 4}, 'syncTask'); // will run immediately and only once across the cluster 5 6jobs.setImmediate(function (ready) { 7 asyncCall(function () { 8 //...run more async code 9 ready(); 10 }); 11}, 'asyncTask'); // will run immediately and only once across the cluster 12 13jobs.setImmediate(async function () { 14 try { 15 /* ...code here... */ 16 await asyncMethod(); 17 } catch (err) { 18 console.log(err) 19 } 20}, 'asyncTask'); // will run immediately and only once across the cluster
clearInterval(timerId)
timerId
{String|Promise} — Timer id returned fromJoSk#setInterval()
method- Returns: {Promise}
true
when task is successfully cleared, orfalse
when task was not found
Cancel current interval timer.
1const timer = await jobs.setInterval(func, 34789, 'unique-taskid'); 2await jobs.clearInterval(timer);
clearTimeout(timerId)
timerId
{String|Promise} — Timer id returned fromJoSk#setTimeout()
method- Returns: {Promise}
true
when task is successfully cleared, orfalse
when task was not found
Cancel current timeout timer.
1const timer = await jobs.setTimeout(func, 34789, 'unique-taskid'); 2await jobs.clearTimeout(timer);
destroy()
- Returns: {boolean}
true
if instance successfully destroyed,false
if instance already destroyed
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 stdout
or if onError
hook was passed to JoSk it would receive an error. Only permitted methods are clearTimeout
and clearInterval
.
1// EXAMPLE: DESTROY JoSk INSTANCE UPON SERVER PROCESS TERMINATION 2const jobs = new JoSk({ /* ... */ }); 3 4const cleanUpBeforeTermination = function () { 5 /* ...CLEAN UP AND STOP OTHER THINGS HERE... */ 6 jobs.destroy(); 7 process.exit(1); 8}; 9 10process.stdin.resume(); 11process.on('uncaughtException', cleanUpBeforeTermination); 12process.on('exit', cleanUpBeforeTermination); 13process.on('SIGHUP', cleanUpBeforeTermination);
ping()
- Returns: {Promise}
Ping JoSk instance. Check scheduler readiness and its connection to the "storage adapter"
1// EXAMPLE: DESTROY JoSk INSTANCE UPON SERVER PROCESS TERMINATION 2const jobs = new JoSk({ /* ... */ }); 3 4const pingResult = await jobs.ping(); 5console.log(pingResult) 6/** 7In case of the successful response 8{ 9 status: 'OK', 10 code: 200, 11 statusCode: 200, 12} 13 14Failed response 15{ 16 status: 'Error reason', 17 code: 500, 18 statusCode: 500, 19 error: ErrorObject 20} 21*/
Examples
Use cases and usage examples
CRON
Use JoSk to invoke synchronized tasks by CRON schedule, and cron-parser
package to parse CRON expressions. To simplify CRON scheduling — grab and use setCron
function below:
1import parser from 'cron-parser'; 2 3const jobsCron = new JoSk({ 4 adapter: new RedisAdapter({ 5 client: await createClient({ url: 'redis://127.0.0.1:6379' }).connect(), 6 prefix: 'cron-scheduler' 7 }), 8 minRevolvingDelay: 512, // Adjust revolving delays to higher values 9 maxRevolvingDelay: 1000, // as CRON schedule defined to seconds 10}); 11 12// CRON HELPER FUNCTION 13const setCron = async (uniqueName, cronTask, task) => { 14 const nextTimestamp = +parser.parseExpression(cronTask).next().toDate(); 15 16 return await jobsCron.setInterval(function (done) { 17 done(parser.parseExpression(cronTask).next().toDate()); 18 task(); 19 }, nextTimestamp - Date.now(), uniqueName); 20}; 21 22setCron('Run every two seconds cron', '*/2 * * * * *', function () { 23 console.log(new Date); 24});
Pass arguments
Passing arguments can be done via wrapper function
1const jobs = new JoSk({ /* ... */ }); 2const myVar = { key: 'value' }; 3let myLet = 'Some top level or env.variable (can get changed during runtime)'; 4 5const task = function (arg1, arg2, ready) { 6 //... code here 7 ready(); 8}; 9 10jobs.setInterval((ready) => { 11 task(myVar, myLet, ready); 12}, 60 * 60000, 'taskA'); 13 14jobs.setInterval((ready) => { 15 task({ otherKey: 'Another Value' }, 'Some other string', ready); 16}, 60 * 60000, 'taskB');
Async/Await with ready() callback
For long-running async tasks, or with callback-apis it might be needed to call ready()
explicitly. Wrap task's body into process.nextTick
to enjoy await
/async
combined with classic callback-apis
1jobs.setInterval((ready) => { 2 process.nextTick(async () => { 3 try { 4 const result = await asyncCall(); 5 waitForSomethingElse(async (error, data) => { 6 if (error) { 7 ready(); // <-- Always run `ready()`, even if call was unsuccessful 8 return; 9 } 10 11 await saveCollectedData(result, [data]); 12 ready(); // <-- End of the full execution 13 }); 14 } catch (err) { 15 console.log(err) 16 ready(); // <-- Always run `ready()`, even if call was unsuccessful 17 } 18 }); 19}, 60 * 60000, 'longRunningTask1h'); // once every hour
Clean up old tasks
During development and tests you may want to clean up Adapter's Storage
Clean up Redis
To clean up old tasks via Redis CLI use the next query pattern:
redis-cli --no-auth-warning KEYS "josk:default:*" | xargs redis-cli --raw --no-auth-warning DEL # If you're using multiple JoSk instances with prefix: redis-cli --no-auth-warning KEYS "josk:prefix:*" | xargs redis-cli --raw --no-auth-warning DEL
Clean up MongoDB
To clean up old tasks via MongoDB use the 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({});
MongoDB connection fine tuning
1// Recommended MongoDB connection options 2// When used with ReplicaSet 3const options = { 4 writeConcern: { 5 j: true, 6 w: 'majority', 7 wtimeout: 30000 8 }, 9 readConcern: { 10 level: 'majority' 11 }, 12 readPreference: 'primary' 13}; 14 15MongoClient.connect('mongodb://url', options, (error, client) => { 16 // To avoid "DB locks" — it's a good idea to use separate DB from "main" application DB 17 const db = client.db('dbName'); 18 const jobs = new JoSk({ 19 adapter: new MongoAdapter({ 20 db: db, 21 }) 22 }); 23});
Notes
- This package is perfect when you have multiple horizontally scaled servers for load-balancing, durability, an array of micro-services or any other solution with multiple running copies of code running repeating tasks that needs to run only once per application/cluster, not per server/instance;
- 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, sync, and etc;
- Accuracy — Delay of each task depends on storage and "de-synchronization delay". Trusted time-range of execution period is
task_delay ± (256 + Storage_Request_Delay)
. That means this package won't fit when you need to run a task with very precise delays. For other cases, if±256 ms
delays are acceptable - this package is the great solution; - Use
opts.minRevolvingDelay
andopts.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 (128
and768
) are the best for 3-server setup (the most common topology). Tune these options to match needs of your project. Higheropts.minRevolvingDelay
will reduce storage read/writes; - This package implements "Read Locking" via "RedLock" for Redis and dedicated
.lock
collection for MongoDB.
Running Tests
- Clone this package
- In Terminal (Console) go to directory where package is cloned
- Then run:
# Before running tests make sure NODE_ENV === development # Install NPM dependencies npm install --save-dev # Before running tests you need # to have access to MongoDB and Redis servers REDIS_URL="redis://127.0.0.1:6379" MONGO_URL="mongodb://127.0.0.1:27017/npm-josk-test-001" npm test # If previous run has errors — add "debug" to output extra logs DEBUG=true REDIS_URL="redis://127.0.0.1:6379" MONGO_URL="mongodb://127.0.0.1:27017/npm-josk-test-001" npm test # Be patient, tests are taking around 6 mins
Run Redis tests only
Run Redis-related tests only
# Before running Redis tests you need to have Redis server installed and running REDIS_URL="redis://127.0.0.1:6379" npm run test-redis # Be patient, tests are taking around 3 mins
Run MongoDB tests only
Run MongoDB-related tests only
# Before running Mongo tests you need to have MongoDB server installed and running MONGO_URL="mongodb://127.0.0.1:27017/npm-josk-test-001" npm run test-mongo # Be patient, tests are taking around 3 mins
Why JoSk?
JoSk
is Job-Task - Is randomly generated name by "uniq" project
Support our open source contribution:
- Upload and share files using ☄️ meteor-files.com — Continue interrupted file uploads without losing any progress. There is nothing that will stop Meteor from delivering your file to the desired destination
- Use ▲ ostr.io for Server Monitoring, Web Analytics, WebSec, Web-CRON and SEO Pre-rendering of a website
- Star on GitHub
- Star on NPM
- Star on Atmosphere
- Sponsor via GitHub
- Support via PayPal