toad-scheduler

In-memory Node.js and browser job scheduler

CRONQueue
kibertoad/toad-schedulertoad-scheduler

README

toad-scheduler


In-memory TypeScript job scheduler that repeatedly executes given tasks within specified intervals of time (e. g. "each 20 seconds"). Cron syntax is also supported in case you need it.

Node.js 12+ and modern browsers are supported

Getting started


First install the package:

  1. ```bash
  2. npm i toad-scheduler
  3. ```

Next, set up your jobs:

  1. ```js
  2. const { ToadScheduler, SimpleIntervalJob, Task } = require('toad-scheduler')

  4. const scheduler = new ToadScheduler()

  6. const task = new Task('simple task', () => { counter++ })
  7. const job = new SimpleIntervalJob({ seconds: 20, }, task)

  9. scheduler.addSimpleIntervalJob(job)

  11. // when stopping your app
  12. scheduler.stop()
  13. ```


Usage with async tasks


In order to avoid unhandled rejections, make sure to use AsyncTask if your task is asynchronous:

  1. ```js
  2. const { ToadScheduler, SimpleIntervalJob, AsyncTask } = require('toad-scheduler')

  4. const scheduler = new ToadScheduler()

  6. const task = new AsyncTask(
  7.     'simple task',
  8.     () => { return db.pollForSomeData().then((result) => { /* continue the promise chain */ }) },
  9.     (err: Error) => { /* handle error here */ }
  10. )
  11. const job = new SimpleIntervalJob({ seconds: 20, }, task)

  13. scheduler.addSimpleIntervalJob(job)

  15. // when stopping your app
  16. scheduler.stop()
  17. ```

Note that in order to avoid memory leaks, it is recommended to use promise chains instead of async/await inside task definition. See talk on common Promise mistakes for more details.

Asynchronous error handling


Note that your error handlers can be asynchronous and return a promise. In such case an additional catch block will be attached to them, and should
there be an error while trying to resolve that promise, and logging error will be logged using the default error handler (console.error).

Preventing task run overruns


In case you want to prevent second instance of a task from being fired up while first one is still executing, you can use preventOverrun options:
  1. ```js
  2. import { ToadScheduler, SimpleIntervalJob, Task } from 'toad-scheduler';

  4. const scheduler = new ToadScheduler();

  6. const task = new Task('simple task', () => {
  7.     // if this task runs long, second one won't be started until this one concludes
  8. console.log('Task triggered');
  9. });

  11. const job = new SimpleIntervalJob(
  12. { seconds: 20, runImmediately: true },
  13. task,
  14.     {
  15.         id: 'id_1',
  16.         preventOverrun: true,
  17.     }
  18. );

  20. //create and start jobs
  21. scheduler.addSimpleIntervalJob(job);
  22. ```

Using IDs and ES6-style imports


You can attach IDs to tasks to identify them later. This is helpful in projects that run a lot of tasks and especially if you want to target some of the tasks specifically (e. g. in order to stop or restart them, or to check their status).

  1. ```js
  2. import { ToadScheduler, SimpleIntervalJob, Task } from 'toad-scheduler';

  4. const scheduler = new ToadScheduler();

  6. const task = new Task('simple task', () => {
  7. console.log('Task triggered');
  8. });

  10. const job1 = new SimpleIntervalJob(
  11. { seconds: 20, runImmediately: true },
  12. task,
  13.     { id: 'id_1' }
  14. );

  16. const job2 = new SimpleIntervalJob(
  17. { seconds: 15, runImmediately: true },
  18. task,
  19.     { id: 'id_2' }
  20. );

  22. //create and start jobs
  23. scheduler.addSimpleIntervalJob(job1);
  24. scheduler.addSimpleIntervalJob(job2);

  26. // stop job with ID: id_2
  27. scheduler.stopById('id_2');

  29. // remove job with ID: id_1
  30. scheduler.removeById('id_1');

  32. // check status of jobs
  33. console.log(scheduler.getById('id_1').getStatus()); // returns Error (job not found)

  35. console.log(scheduler.getById('id_2').getStatus()); // returns "stopped" and can be started again

  37. ```

Cron support


You can use CronJob instances for handling Cron-style scheduling:
  1. ```ts
  2.       const task = new AsyncTask('simple task', () => {
  3.         // Execute your asynchronous logic here
  4.       })
  5.       const job = new CronJob(
  6.         {
  7.           cronExpression: '*/2 * * * * *',
  8.         },
  9.         task,
  10.         {
  11.           preventOverrun: true,
  12.         }
  13.       )
  14.       scheduler.addCronJob(job)
  15. ```

Note that you need to install "croner" library for this to work. Run npm i croner in order to install this dependency.

Usage in clustered environments


toad-scheduler does not persist its state by design, and has no out-of-the-box concurrency management features. In case it is necessary
to prevent parallel execution of jobs in clustered environment, it is highly recommended to use redis-semaphore in your tasks.

Here is an example:

  1. ```ts
  2. import { randomUUID } from 'node:crypto'

  4. import type { Redis } from 'ioredis'
  5. import type { LockOptions } from 'redis-semaphore'
  6. import { Mutex } from 'redis-semaphore'
  7. import { AsyncTask } from 'toad-scheduler';

  9. export type BackgroundJobConfiguration = {
  10.     jobId: string
  11. }

  13. export type LockConfiguration = {
  14.     lockName?: string
  15.     refreshInterval?: number
  16.     lockTimeout: number
  17. }

  19. // Abstract Job
  20. export abstract class AbstractBackgroundJob {
  21.     public readonly jobId: string
  22.     protected readonly redis: Redis

  24.     protected constructor(
  25.         options: BackgroundJobConfiguration,
  26.         redis: Redis,
  27.     ) {
  28.         this.jobId = options.jobId
  29.         this.redis = redis
  30.     }

  32.     protected abstract processInternal(executionUuid: string): Promise<void>

  34.     async process() {
  35.         const uuid = randomUUID()

  37.         try {
  38.             await this.processInternal(uuid)
  39.         } catch (err) {
  40.             console.error(logObject)
  41.         }
  42.     }

  44.     protected getJobMutex(key: string, options: LockOptions) {
  45.         return new Mutex(this.redis, this.getJobLockName(key), options)
  46.     }

  48.     protected async tryAcquireExclusiveLock(lockConfiguration: LockConfiguration) {
  49.         const mutex = this.getJobMutex(lockConfiguration.lockName ?? 'exclusive', {
  50.             acquireAttemptsLimit: 1,
  51.             refreshInterval: lockConfiguration.refreshInterval,
  52.             lockTimeout: lockConfiguration.lockTimeout,
  53.         })

  55.         const lock = await mutex.tryAcquire()
  56.         // If someone else already has this lock, skip
  57.         if (!lock) {
  58.             return
  59.         }

  61.         return mutex
  62.     }

  64.     protected getJobLockName(key: string) {
  65.         return `${this.jobId}:locks:${key}`
  66.     }
  67. }

  69. // Job example

  71. const LOCK_TIMEOUT_IN_MSECS = 60 * 1000
  72. const LOCK_REFRESH_IN_MSECS = 10 * 1000

  74. export class SampleJob extends AbstractBackgroundJob {
  75.   constructor(redis: Redis) {
  76.     super(
  77.       {
  78.         jobId: 'SampleJob',
  79.       },
  80.       redis,
  81.     )
  82.   }

  84.   protected async processInternal(executionUuid: string): Promise<void> {
  85.     // We only want a single instance of this job running in entire cluster, let's see if someone else is already processing it
  86.     const lock = await this.tryAcquireExclusiveLock({
  87.       lockTimeout: LOCK_TIMEOUT_IN_MSECS,
  88.       refreshInterval: LOCK_REFRESH_IN_MSECS,
  89.     })

  91.     // Job is already running, skip
  92.     if (!lock) {
  93.       this.logger.debug(`Job already running in another node, skipping (${executionUuid})`)
  94.       return
  95.     }

  97.     try {
  98.       // Process job logic here
  99.       await this.sampleJobLogic()
  100.     } finally {
  101.       await lock.release()
  102.     }
  103.   }

  105.   private async sampleJobLogic() {
  106.     // dummy processing logic
  107.     return Promise.resolve()
  108.   }


  111. // Job registration
  112. function createTask(job: AbstractBackgroundJob): AsyncTask {
  113.     return new AsyncTask(
  114.         job.jobId,
  115.         () => {
  116.             return job.process()
  117.         },
  118.     )
  119. }
  120. ```

API for schedule


days?: number - how many days to wait before executing the job for the next time;
hours?: number - how many hours to wait before executing the job for the next time;
minutes?: number - how many minutes to wait before executing the job for the next time;
seconds?: number - how many seconds to wait before executing the job for the next time;
milliseconds?: number - how many milliseconds to wait before executing the job for the next time;
runImmediately?: boolean - if set to true, in addition to being executed on a given interval, job will also be executed immediately when added or restarted.

API for jobs


start(): void - starts, or restarts (if it's already running) the job;
stop(): void - stops the job. Can be restarted again with start command;
getStatus(): JobStatus - returns the status of the job, which is one of: running, stopped.

API for scheduler


addSimpleIntervalJob(job: SimpleIntervalJob): void - registers and starts a new job;
addLongIntervalJob(job: SimpleIntervalJob): void - registers and starts a new job with support for intervals longer than 24.85 days;
addIntervalJob(job: SimpleIntervalJob | LongIntervalJob): void - registers and starts new interval-based job;
stop(): void - stops all jobs, registered in the scheduler;
getById(id: string): Job - returns the job with a given id.
existsById(id: string): boolean - returns true if job with given id exists, false otherwise.
getAllJobs(): Job[] - returns all registered jobs.
getAllJobsByStatus(status: JobStatus): Job[] - returns all registered jobs with a given status.
stopById(id: string): void - stops the job with a given id.
removeById(id: string): Job | undefined - stops the job with a given id and removes it from the scheduler. If no such job exists, returns undefined, otherwise returns the job.
startById(id: string): void - starts, or restarts (if it's already running) the job with a given id.

[npm-image]: https://img.shields.io/npm/v/toad-scheduler.svg
[npm-url]: https://npmjs.org/package/toad-scheduler
[downloads-image]: https://img.shields.io/npm/dm/toad-scheduler.svg
[downloads-url]: https://npmjs.org/package/toad-scheduler