[!IMPORTANT] This is a customized fork of BreeJS the allows use of multi-process support as well as worker threads. You should use the original project if you only need worker thread support.

---

Table of Contents

• Foreword
• Install
• Upgrading
• Usage and Examples
  • ECMAScript modules (ESM)
  • CommonJS (CJS)
• Instance Options
• Job Options
• Job Interval and Timeout Values
• Listening for events
• Custom error/message handling
• Cancellation, Retries, Stalled Jobs, and Graceful Reloading
• Interval, Timeout, Date, and Cron Validation
• Writing jobs with Promises and async-await
• Callbacks, Done, and Completion States
• Long-running jobs
• Complex timeouts and intervals
• Custom Worker Options
• Using functions for jobs
• Typescript and Usage with Bundlers
• Concurrency
• Plugins
  • Available Plugins
  • Creating plugins for Bree
• Contributors
• License

Foreword

Bree was created to give you fine-grained control with simplicity, and has built-in support for workers, sandboxed processes, graceful reloading, cron jobs, dates, human-friendly time representations, and much more.

We recommend you to query a persistent database in your jobs, to prevent specific operations from running more than once.

Bree does not force you to use an additional database layer of [Redis][] or [MongoDB][] to manage job state.

In doing so, you should manage boolean job states yourself using queries. For instance, if you have to send a welcome email to users, only send a welcome email to users that do not have a Date value set yet for welcome_email_sent_at.

Install

[npm][]:

1npm install @mintplex-labs/bree

[yarn][]:

1yarn add @mintplex-labs/bree

Upgrading

To see details about upgrading from the last major version, please see UPGRADING.md.

IMPORTANT: Bree v9.0.0 has several breaking changes, please see UPGRADING.md for more insight.

NOTE: Bree v6.5.0 is the last version to support Node v10 and browsers.

Usage and Examples

The example below assumes that you have a directory jobs in the root of the directory from which you run this example. For example, if the example below is at /path/to/script.js, then /path/to/jobs/ must also exist as a directory. If you wish to disable this feature, then pass root: false as an option.

Inside this jobs directory are individual scripts which are run using [Workers][] per optional timeouts, and additionally, an optional interval or cron expression. The example below contains comments, which help to clarify how this works.

The option jobs passed to a new instance of Bree (as shown below) is an Array. It contains values which can either be a String (name of a job in the jobs directory, which is run on boot) OR it can be an Object with name, path, timeout, and interval properties. If you do not supply a path, then the path is created using the root directory (defaults to jobs) in combination with the name. If you do not supply values for timeout and/nor interval, then these values are defaulted to 0 (which is the default for both, see index.js for more insight into configurable default options).

We have also documented all Instance Options and Job Options in this README below. Be sure to read those sections so you have a complete understanding of how Bree works.

ECMAScript modules (ESM)

1// app.mjs
2
3import Bree from '@mintplex-labs/bree';
4
5const bree = new Bree({
6  // ... (see below) ...
7});
8
9// top-level await supported in Node v14.8+
10await bree.start();
11
12// ... (see below) ...

Please reference the #CommonJS example below for more insight and options.

CommonJS (CJS)

1// app.js
2
3const path = require('path');
4
5// optional
6const ms = require('ms');
7const dayjs = require('dayjs');
8const Graceful = require('@ladjs/graceful');
9const Cabin = require('cabin');
10
11// required
12const Bree = require('@mintplex-labs/bree');
13
14//
15// NOTE: see the "Instance Options" section below in this README
16// for the complete list of options and their defaults
17//
18const bree = new Bree({
19  //
20  // NOTE: by default the `logger` is set to `console`
21  // however we recommend you to use CabinJS as it
22  // will automatically add application and worker metadata
23  // to your log output, and also masks sensitive data for you
24  // <https://cabinjs.com>
25  //
26  // NOTE: You can also pass `false` as `logger: false` to disable logging
27  //
28  logger: new Cabin(),
29
30  //
31  // NOTE: instead of passing this Array as an option
32  // you can create a `./jobs/index.js` file, exporting
33  // this exact same array as `module.exports = [ ... ]`
34  // doing so will allow you to keep your job configuration and the jobs
35  // themselves all in the same folder and very organized
36  //
37  // See the "Job Options" section below in this README
38  // for the complete list of job options and configurations
39  //
40  jobs: [
41    // runs `./jobs/foo.js` on start
42    'foo',
43
44    // runs `./jobs/foo-bar.js` on start
45    {
46      name: 'foo-bar'
47    },
48
49    // runs `./jobs/some-other-path.js` on start
50    {
51      name: 'beep',
52      path: path.join(__dirname, 'jobs', 'some-other-path')
53    },
54
55    // runs `./jobs/worker-1.js` on the last day of the month
56    {
57      name: 'worker-1',
58      interval: 'on the last day of the month'
59    },
60
61    // runs `./jobs/worker-2.js` every other day
62    {
63      name: 'worker-2',
64      interval: 'every 2 days'
65    },
66
67    // runs `./jobs/worker-3.js` at 10:15am and 5:15pm every day except on Tuesday
68    {
69      name: 'worker-3',
70      interval: 'at 10:15 am also at 5:15pm except on Tuesday'
71    },
72
73    // runs `./jobs/worker-4.js` at 10:15am every weekday
74    {
75      name: 'worker-4',
76      cron: '15 10 ? * *',
77      cronValidate: {
78        override: {
79          useBlankDay: true
80        }
81      }
82    },
83
84    // runs `./jobs/worker-5.js` on after 10 minutes have elapsed
85    {
86      name: 'worker-5',
87      timeout: '10m'
88    },
89
90    // runs `./jobs/worker-6.js` after 1 minute and every 5 minutes thereafter
91    {
92      name: 'worker-6',
93      timeout: '1m',
94      interval: '5m'
95      // this is unnecessary but shows you can pass a Number (ms)
96      // interval: ms('5m')
97    },
98
99    // runs `./jobs/worker-7.js` after 3 days and 4 hours
100    {
101      name: 'worker-7',
102      // this example uses `human-interval` parsing
103      timeout: '3 days and 4 hours'
104    },
105
106    // runs `./jobs/worker-8.js` at midnight (once)
107    {
108      name: 'worker-8',
109      timeout: 'at 12:00 am'
110    },
111
112    // runs `./jobs/worker-9.js` every day at midnight
113    {
114      name: 'worker-9',
115      interval: 'at 12:00 am'
116    },
117
118    // runs `./jobs/worker-10.js` at midnight on the 1st of every month
119    {
120      name: 'worker-10',
121      cron: '0 0 1 * *'
122    },
123
124    // runs `./jobs/worker-11.js` at midnight on the last day of month
125    {
126      name: 'worker-11',
127      cron: '0 0 L * *',
128      cronValidate: {
129        useLastDayOfMonth: true
130      }
131    },
132
133    // runs `./jobs/worker-12.js` at a specific Date (e.g. in 3 days)
134    {
135      name: 'worker-12',
136      // <https://github.com/iamkun/dayjs>
137      date: dayjs().add(3, 'days').toDate()
138      // you can also use momentjs
139      // <https://momentjs.com/>
140      // date: moment('1/1/20', 'M/D/YY').toDate()
141      // you can pass Date instances (if it's in the past it will not get run)
142      // date: new Date()
143    },
144
145    // runs `./jobs/worker-13.js` on start and every 2 minutes
146    {
147      name: 'worker-13',
148      interval: '2m'
149    },
150
151    // runs `./jobs/worker-14.js` on start with custom `new Worker` options (see below)
152    {
153      name: 'worker-14',
154      // <https://nodejs.org/api/worker_threads.html#worker_threads_new_worker_filename_options>
155      worker: {
156        workerData: {
157          foo: 'bar',
158          beep: 'boop'
159        }
160      }
161    },
162
163    // runs `./jobs/worker-15.js` **NOT** on start, but every 2 minutes
164    {
165      name: 'worker-15',
166      timeout: false, // <-- specify `false` here to prevent default timeout (e.g. on start)
167      interval: '2m'
168    },
169
170    // runs `./jobs/worker-16.js` on January 1st, 2022
171    // and at midnight on the 1st of every month thereafter
172    {
173      name: 'worker-16',
174      date: dayjs('1-1-2022', 'M-D-YYYY').toDate(),
175      cron: '0 0 1 * *'
176    },
177
178    // runs `./jobs/worker-13.js` in a new process as opposed to a worker thread
179    {
180      name: 'worker-13',
181      interval: '2m',
182      runAs: 'process',
183    },
184  ]
185});
186
187// handle graceful reloads, pm2 support, and events like SIGHUP, SIGINT, etc.
188const graceful = new Graceful({ brees: [bree] });
189graceful.listen();
190
191// start all jobs (this is the equivalent of reloading a crontab):
192(async () => {
193  await bree.start();
194})();
195
196/*
197// start only a specific job:
198(async () => {
199  await bree.start('foo');
200})();
201
202// stop all jobs
203bree.stop();
204
205// stop only a specific job:
206bree.stop('beep');
207
208// run all jobs (this does not abide by timeout/interval/cron and spawns workers immediately)
209bree.run();
210
211// run a specific job (...)
212bree.run('beep');
213
214(async () => {
215  // add a job array after initialization:
216  const added = await bree.add(['boop']); // will return array of added jobs
217  // this must then be started using one of the above methods
218
219  // add a job after initialization:
220  await bree.add('boop');
221  // this must then be started using one of the above methods
222})();
223
224
225// remove a job after initialization:
226bree.remove('boop');
227*/

For more examples - including setting up bree with TypeScript, ESModules, and implementing an Email Queue, see the examples folder.

For a more complete demo using express see: Bree Express Demo

Instance Options

Here is the full list of options and their defaults. See src/index.js for more insight if necessary.

Property	Type	Default Value	Description
logger	Object	console	This is the default logger. We recommend using [Cabin][cabin] instead of using console as your default logger. Set this value to false to disable logging entirely (uses noop function)
root	String	path.resolve('jobs')	Resolves a jobs folder relative to where the project is ran (the directory you call node in). Set this value to false to prevent requiring a root directory of jobs (e.g. if your jobs are not all in one directory). Set this to path.join(__dirname, 'jobs') to keep your jobs directory relative to the file where Bree is set up.
silenceRootCheckError	Boolean	false	Silences errors from requiring the root folder. Set this to false if you do not want to see errors from this operation
doRootCheck	Boolean	true	Attempts to require the root directory, when jobs is empty or null. Set this to false to prevent requiring the root directory
removeCompleted	Boolean	false	Removes job upon completion. Set this to true in order to remove jobs from the array upon completion.
timeout	Number	0	Default timeout for jobs (e.g. a value of 0 means that jobs will start on boot by default unless a job has a property of timeout or interval defined. Set this to false if you do not wish for a default value to be set for jobs. This value does not apply to jobs with a property of date.
interval	Number	0	Default interval for jobs (e.g. a value of 0 means that there is no interval, and a value greater than zero indicates a default interval will be set with this value). This value does not apply to jobs with a property of cron.
jobs	Array	[]	Defaults to an empty Array, but if the root directory has a index.js file, then it will be used. This allows you to keep your jobs and job definition index in the same place. See Job Options below, and Usage and Examples above for more insight.
hasSeconds	Boolean	false	This value is passed to later for parsing jobs, and can be overridden on a per job basis. See later cron parsing documentation for more insight. Note that setting this to true will automatically set cronValidate defaults to have { preset: 'default', override: { useSeconds: true } }
cronValidate	Object	{}	This value is passed to cron-validate for validation of cron expressions. See the cron-validate documentation for more insight.
closeWorkerAfterMs	Number	0	If you set a value greater than 0 here, then it will terminate workers after this specified time (in milliseconds). As of v6.0.0, workers now terminate after they have been signaled as "online" (as opposed to previous versions which did not take this into account and started the timer when jobs were initially "run"). By default there is no termination done, and jobs can run for infinite periods of time.
defaultRootIndex	String	index.js	This value should be the file name inside of the root directory option (if you pass a root directory or use the default root String value (and your index file name is different than index.js).
defaultExtension	String	js	This value can either be js or mjs. The default is js, and is the default extension added to jobs that are simply defined with a name and without a path. For example, if you define a job test, then it will look for /path/to/root/test.js as the file used for workers.
acceptedExtensions	Array	['.js', '.mjs']	This defines all of the accepted extensions for file validation and job creation. Please note if you add to this list you must override the createWorker function to properly handle the new file types.
worker	Object	{}	These are default options to pass when creating a new Worker instance. See the Worker class documentation for more insight.
outputWorkerMetadata	Boolean	false	By default worker metadata is not passed to the second Object argument of logger. However if you set this to true, then logger will be invoked internally with two arguments (e.g. logger.info('...', { worker: ... })). This worker property contains isMainThread (Boolean), resourceLimits (Object), and threadId (String) properties; all of which correspond to [Workers][] metadata. This can be overridden on a per job basis.
errorHandler	Function	null	Set this function to receive a callback when an error is encountered during worker execution (e.g. throws an exception) or when it exits with non-zero code (e.g. process.exit(1)). The callback receives two parameters error and workerMetadata. Important note, when this callback is present default error logging will not be executed.
workerMessageHandler	Function	null	Set this function to receive a callback when a worker sends a message through parentPort.postMessage. The callback receives at least two parameters name (of the worker) and message (coming from postMessage), if outputWorkerMetadata is enabled additional metadata will be sent to this handler.
runJobsAs	String	worker	Run all defined jobs as a worker using NodeJS worker threads or process to run in all new forked processes.

Job Options

See Interval, Timeout, Date, and Cron Validate below for more insight besides this table:

Property	Type	Description
name	String	The name of the job. This should match the base file path (e.g. foo if foo.js is located at /path/to/jobs/foo.js) unless path option is specified. A value of index, index.js, and index.mjs are reserved values and cannot be used here.
path	String	The path of the job or function used for spawning a new [Worker][workers] with. If not specified, then it defaults to the value for name plus the default file extension specified under Instance Options.
timeout	Number, Object, String, or Boolean	Sets the duration in milliseconds before the job starts (it overrides the default inherited timeout as set in Instance Options. A value of 0 indicates it will start immediately. This value can be a Number, String, or a Boolean of false (which indicates it will NOT inherit the default timeout from Instance Options). See Job Interval and Timeout Values below for more insight into how this value is parsed.
interval	Number, Object, or String	Sets the duration in milliseconds for the job to repeat itself, otherwise known as its interval (it overrides the default inherited interval as set in Instance Options). A value of 0 indicates it will not repeat and there will be no interval. If the value is greater than 0 then this value will be used as the interval. See Job Interval and Timeout Values below for more insight into how this value is parsed.
date	Date	This must be a valid JavaScript Date (we use instance of Date for comparison). If this value is in the past, then it is not run when jobs are started (or run manually). We recommend using [dayjs][] for creating this date, and then formatting it using the toDate() method (e.g. dayjs().add('3, 'days').toDate()). You could also use [moment][] or any other JavaScript date library, as long as you convert the value to a Date instance here.
cron	String	A cron expression to use as the job's interval, which is validated against [cron-validate][] and parsed by [later][].
hasSeconds	Boolean	Overrides the Instance Options hasSeconds property if set. Note that setting this to true will automatically set cronValidate defaults to have { preset: 'default', override: { useSeconds: true } }
cronValidate	Object	Overrides the Instance Options cronValidate property if set.
closeWorkerAfterMs	Number	Overrides the Instance Options closeWorkerAfterMs property if set.
worker	Object	Overrides the Instance Options worker property if set.
outputWorkerMetadata	Boolean	Overrides the Instance Options outputWorkerMetadata property if set.
runAs	String	worker	Run specific job as a worker using NodeJS worker threads or process to run in all new forked processes. Will be overridden by BreeConfig runJobsAs if set.

Job Interval and Timeout Values

These values can include Number, Object, and String variable types:

• Number values indicates the number of milliseconds for the timeout or interval
• Object values must be a [later][] schedule object value (e.g. later.parse.cron('15 10 * * ? *')))
• String values can be either a [later][], [human-interval][], or [ms][] String values (e.g. [later][] supports Strings such as every 5 mins, [human-interval][] supports Strings such as 3 days and 4 hours, and [ms][] supports Strings such as 4h for four hours)

Listening for events

Bree extends from EventEmitter and emits two events:

• worker created with an argument of name
• worker deleted with an argument of name

If you'd like to know when your workers are created (or deleted), you can do so through this example:

1bree.on('worker created', (name) => {
2  console.log('worker created', name);
3  console.log(bree.workers.get(name));
4});
5
6bree.on('worker deleted', (name) => {
7  console.log('worker deleted', name);
8  console.log(!bree.worker.has(name));
9});

Custom error/message handling

If you'd like to override default behavior for worker error/message handling, provide a callback function as errorHandler or workerMessageHandler parameter when creating a Bree instance.

NOTE: Any console.log calls, from within the worker, will not be sent to stdout/stderr until the main thread is available. Furthermore, any console.log calls, from within the worker, will not be sent if the process is terminated before the message is printed. You should use parentPort.postMessage() alongside errorHandler or workerMessageHandler to print to stdout/stderr during worker execution. This is a known bug for workers.

An example use-case. If you want to call an external service to record an error (like Honeybadger, Sentry, etc.) along with logging the error internally. You can do so with:

1const logger = ('../path/to/logger');
2const errorService = ('../path/to/error-service');
3
4new Bree({
5  jobs: [
6    {
7      name: 'job that sometimes throws errors',
8      path: jobFunction
9    }
10  ],
11  errorHandler: (error, workerMetadata) => {
12    // workerMetadata will be populated with extended worker information only if
13    // Bree instance is initialized with parameter `workerMetadata: true`
14    if (workerMetadata.threadId) {
15      logger.info(`There was an error while running a worker ${workerMetadata.name} with thread ID: ${workerMetadata.threadId}`)
16    } else {
17      logger.info(`There was an error while running a worker ${workerMetadata.name}`)
18    }
19
20    logger.error(error);
21    errorService.captureException(error);
22  }
23});

Cancellation, Retries, Stalled Jobs, and Graceful Reloading

We recommend that you listen for "cancel" event in your worker paths. Doing so will allow you to handle graceful cancellation of jobs. For example, you could use [p-cancelable][]

Here's a quick example of how to do that (e.g. ./jobs/some-worker.js):

1// <https://nodejs.org/api/worker_threads.html>
2const { parentPort } = require('worker_threads');
3
4// ...
5
6function cancel() {
7  // do cleanup here
8  // (if you're using @ladjs/graceful, the max time this can run by default is 5s)
9
10  // send a message to the parent that we're ready to terminate
11  // (you could do `process.exit(0)` or `process.exit(1)` instead if desired
12  // but this is a bit of a cleaner approach for worker termination
13  if (parentPort) parentPort.postMessage('cancelled');
14  else process.exit(0);
15}
16
17if (parentPort)
18  parentPort.once('message', message => {
19    if (message === 'cancel') return cancel();
20  });

If you'd like jobs to retry, simply wrap your usage of promises with [p-retry][].

We leave it up to you to have as much fine-grained control as you wish.

See [@ladjs/graceful][lad-graceful] for more insight into how this package works.

Interval, Timeout, Date, and Cron Validation

If you need help writing cron expressions, you can reference crontab.guru.

We support [later][], [human-interval][], or [ms][] String values for both timeout and interval.

If you pass a cron property, then it is validated against [cron-validate][].

You can pass a Date as the date property, but you cannot combine both date and timeout.

If you do pass a Date, then it is only run if it is in the future.

See Job Interval and Timeout Values above for more insight.

Writing jobs with Promises and async-await

If jobs are running with Node pre-v14.8.0, which enables top-level async-await support, here is the working alternative:

1const { parentPort } = require('worker_threads');
2
3const delay = require('delay');
4const ms = require('ms');
5
6(async () => {
7  // wait for a promise to finish
8  await delay(ms('10s'));
9
10  // signal to parent that the job is done
11  if (parentPort) parentPort.postMessage('done');
12  else process.exit(0);
13})();

Callbacks, Done, and Completion States

To close out the worker and signal that it is done, you can simply parentPort.postMessage('done'); and/or process.exit(0).

While writing your jobs (which will run in [worker][workers] threads), you should do one of the following:

• Signal to the main thread that the process has completed by sending a "done" message (per the example above in Writing jobs with Promises and async-await)
• Exit the process if there is NOT an error with code 0 (e.g. process.exit(0);)
• Throw an error if an error occurs (this will bubble up to the worker event error listener and terminate it)
• Exit the process if there IS an error with code 1 (e.g. process.exit(1))

Long-running jobs

If a job is already running, a new worker thread will not be spawned, instead logger.error will be invoked with an error message (no error will be thrown, don't worry). This is to prevent bad practices from being used. If you need something to be run more than one time, then make the job itself run the task multiple times. This approach gives you more fine-grained control.

By default, workers run indefinitely and are not closed until they exit (e.g. via process.exit(0) or process.exit(1), OR send to the parent port a "close" message, which will subsequently call worker.close() to close the worker thread.

If you wish to specify a maximum time (in milliseconds) that a worker can run, then pass closeWorkerAfterMs (Number) either as a default option when creating a new Bree() instance (e.g. new Bree({ closeWorkerAfterMs: ms('10s') })) or on a per-job configuration, e.g. { name: 'beep', closeWorkerAfterMs: ms('5m') }.

As of v6.0.0 when you pass closeWorkerAfterMs, the timer will start once the worker is signaled as "online" (as opposed to previous versions which did not take this into account).

Complex timeouts and intervals

Since we use [later][], you can pass an instance of later.parse.recur, later.parse.cron, or later.parse.text as the timeout or interval property values (e.g. if you need to construct something manually).

You can also use [dayjs][] to construct dates (e.g. from now or a certain date) to millisecond differences using dayjs().diff(new Date(), 'milliseconds'). You would then pass that returned Number value as timeout or interval as needed.

Custom Worker Options

You can pass a default worker configuration object as new Bree({ worker: { ... } });.

These options are passed to the options argument when we internally invoke new Worker(path, options).

Additionally, you can pass custom worker options on a per-job basis through a worker property Object on the job definition.

See complete documentation for options (but you usually don't have to modify these).

Using functions for jobs

It is highly recommended to use files instead of functions. However, sometimes it is necessary to use functions.

You can pass a function to be run as a job:

1new Bree({ jobs: [someFunction] });

(or)

1new Bree({
2  jobs: [
3    {
4      name: 'job with function',
5      path: someFunction
6      // runAs: property will always be "worker" when using direct function.
7    }
8  ]
9});

The function will be run as if it's in its own file, therefore no variables or dependencies will be shared from the local context by default.

You should be able to pass data via worker.workerData (see Custom Worker Options).

Note that you cannot pass a built-in nor bound function.

Typescript and Usage with Bundlers

When working with a bundler or a tool that transpiles your code in some form or another, we recommend that your bundler is set up in a way that transforms both your application code and your jobs. Because your jobs are in their own files and are run in their own separate threads, they will not be part of your applications dependency graph and need to be setup as their own entry points. You need to ensure you have configured your tool to bundle your jobs into a jobs folder and keep them properly relative to your entry point folder.

We recommend setting the root instance options to path.join(__dirname,'jobs') so that bree searches for your jobs folder relative to the file being ran. (by default it searches for jobs relative to where node is invoked). We recommend treating each job as an entry point and running all jobs through the same transformations as your app code.

After an example transformation - you should expect the output in your dist folder to look like:

1- dist
2  |-jobs
3    |-job.js
4  |-index.js

For some example TypeScript set ups - see the examples folder.

For another alternative also see the @breejs/ts-worker plugin.

Concurrency

We recommend using the following packages in your workers for handling concurrency:

• https://github.com/sindresorhus/p-all
• https://github.com/sindresorhus/p-limit
• https://github.com/sindresorhus/p-queue
• https://github.com/sindresorhus/p-map

Plugins

Plugins can be added to Bree using a similar method to Day.js

To add a plugin use the following method:

1Bree.extend(plugin, options);

Available Plugins

• API
• TypeScript Worker

Creating plugins for Bree

Plugins should be a function that recieves an options object and the Bree class:

1  const plugin = (options, Bree) => {
2    /* plugin logic */
3  };

Contributors

Name	Website
Mintplex Labs Inc	https://mintplexlabs.com
Nick Baugh	http://niftylettuce.com/
shadowgate15	https://github.com/shadowgate15

License

MIT © Mintplex Labs Inc