Warden

Warden is a module designed to schedule and repeat jobs with a persistence layer backed by SQL (currently only MySQL is configured).

This module is in its very early stages and will be updated to add more functionality such as compatibility with other SQL database types.

The ability for the module to retry a job has been added in version 2.0.0. To migrate to this new version, a new column must be added to the database title "numberOfRetries" with a datatype of INT(11) and a default value of 0. This is a breaking change and it will not work with previous database tables unless they are migrated to the new schema.

Installation

This package can be installed through NPM.

npm install task-warden

Usage

The SQL connection is made possible through the module Sequelize.

// ES5 via require statement
const { Warden, Process } = require("task-warden");

// ES6 via import statement
import { Warden, Process } from "task-warden";

// Create instance of Warden
const warden = new Warden({
  db: {
    name: "test",
    host: "123.45.67.89",
    username: "foo",
    password: "bar",
    port: 3306,
    logging: false,
  },
  frequency: 300000,
  maxConcurrent: 10,
});

// Create a new Process via warden.createProcess(processName, function, options)
let process1 = warden.createProcess("test1", () => console.log("test1"), {
  maxWorkers: 5,
  lockLifetime: 60000,
  maxRetries: 3,
});

// Start Warden
// Frequency can be defined either here or when creating a new Warden instance.
await warden.start({ frequency: 300000 });

// Schedule the process to run at specific time
await warden.schedule(process1, {}, { runAt: new Date() });

// Update a Job
await warden.updateJob(123, { data: "test1" });

// Cancel a Job
await warden.cancelJob(123);

//Stop Warden from executing any further tasks.
await warden.stop();

NOTE: To see more verbose logging, you may add the env var WARDEN_LOG_LEVEL and assign it one of the Winston log levels. Otherwise it will default to info.

Warden Constructor Options

The constructor will automatically initiate a database connection and create a job table if one does not exist. Once initiated, it will emit a db-initialized event, which the start() event automatically handles and waits for. You may listen for this event and others on warden.emitter.

Database

db.name : Name of the database to be used.
db.host : IP Address of the database or localhost.
db.username : User to sign into the database with.
db.password : Password to sign into database with.
db.port : Port number of the database. Defaults to 3306. Optional
db.logging : Whether or not Sequelize should log database calls. Defaults to false. Optional

Other Options

frequency : How often Warden should check the database for new tasks. Optional
maxConcurrent : How many jobs should be allowed to run at the same time. Optional timezone : Timezone to use for parsing cron expressions. Defaults to "UTC". Optional

Controlling Warden

start(options)

The start() method must be called before any process can be scheduled or executed. Have the warden start processing jobs base on the frequency entered when the warden was created, the value passed in the options here, or the default of 300000(ms).

options.frequency : Number of ms between each fetch from the database.

stop()

When called, Warden will no longer fetch new jobs and all jobs are removed from the queue. Any jobs that are currently running will still be completed.

Defining Processes

You may define a process either directly on the warden object with createProcess() or by creating a process with new Process and using the assign() method to assign it the warden.

It is worth noting that if you would like an instance of warden to only schedule tasks in the database but not run them, you can do so by passing false to the isActive option.

createProcess(name, function, options)

name : Custom name for the process. Used to identify jobs for that process in the database.
function : The function to execute when a job for this process runs.
options.maxWorkers : Maximum number of workers to execute tasks for a process. Defaults to 1. Optional
options.maxRetries : Maximum number of times to retry a process. Defaults to 0. _Optional
options.isActive : Whether or not the process is active and accepting jobs. Defaults to true. _Optional
options.lockLifetime : How long warden should wait to try to re-fetch a job and try again. Defaults to 60000(ms). Optional

Returns a process object.

new Process(name, function, options)

The parameters and options are the same as createProcess() above.

Returns a process object.

assign(process)

process : A process object either created with new Process or the createProcess() method.
name : Custom name for the process. Used to identify jobs for that process in the database.
function : The function to execute when a job for this process runs.
options.maxWorkers : Maximum number of workers to execute tasks for a process. Defaults to 5. Optional
options.lockLifetime : How long warden should wait to try to re-fetch a job and try again. Defaults to 60000(ms). Optional

Scheduling Jobs

NOTE: If no options are passed for scheduling, the task will execute immediately.

schedule(process, data, options)

It is possible to set both the cron and runAt values. If done, the cron info will be used for calculating future jobs and the runAt value when be when the first job runs. This can be useful for scheduling weekly or monthly tasks that you would like to execute the first job early on.

process : A process object returned from the createProcess function.
data : Any data to be passed to the function defined in the process to be executed.
options.runAt : A JavaScript date that the job should run at. Optional
options.cron : A cron string defining how when the job should run. Optional

Fetching Jobs

If you would like to retrieve jobs from the database, you can use the below method.

getJobs(options)

By default, all jobs that are either created, pending, or running are returned. The below filters can be used as well.

options.jobId : Id of the job you would like returned. Optional
options.processName : Name of the process you would like jobs returned for. Optional
options.status : Job status(es) you would like returned jobs to match. Defaults to ["created", "pending", "running"] Optional

Returns an array of job objects.

Updating Jobs

updateJob(jobId, options)

NOTE: If both cron and nextRunAt options are passed, the job will run next at the date specified by nextRunAt, and then all future executions will be determined by the cron string.

jobId : Id of the job you would like to cancel. options.cron : A cron string for recurring jobs. Optional options.data: Data that the job will pass to the process being executed. Optional options.nextRunAt : A date object representing when the next job should run at. Optional

Cancelling Jobs

cancelJob(jobId)

Removes the job from the queue and changes the status of the job in the database to cancelled.

jobId : Id of the job you would like to cancel.

Contributing

Contribution is welcome. Please create an issue before any pull requests are added.

License

MIT