Cancel Job

Description

For any number of reasons you may need to cancel a job currently in the queue. There are two ways you can do this; using Queue.cancelJob, or by calling next() passing a customized Error object.

This document explains the differences and when to use them.

Queue.cancelJob

When to use?

When a job is no longer valid and has not been completed, cancelled, or terminated. See the Job Status for more detail on these status values.

Scope

In a distributed processing environment this will cancel the job for all Queue objects connected to the queue. If another Queue object is currently processing this job, depending on the user defined onCancel(callback) function, it will gracefully stop the job processing.

What does it do?

The Queue.cancelJob method can be called while a job is in the queue with a waiting, failed, or active status. By calling the Queue.cancelJob method you will cause the job to change status to cancelled which will permanently prevent the job from being picked up for processing.

Note: You can reanimate jobs. See the Job Editing document for more detail.

If you are using rethinkdb-job-queue in a distributed environment and you have the change feed option enabled for your worker Queue objects, calling Queue.cancelJob will be received by all Queue objects. If you cancel the job from a Queue object that is not processing the job, and the job is being processed by another Queue object, a Queue.process callback called onCancel will be invoked. This will allow the Queue object processing the job to gracefully stop the processing of the job.

Cancel Job with next(err)

When to use?

The next() function is presented to your job processing code when you call Queue.process. Cancel the job by calling next(err) with a customized Error object if the job processing fails and will never be able to be retried.

Scope

This method for cancelling a job is called whilst the job is being processed. It will change the job status to cancelled in the queue, permanently preventing further processing.

What does it do?

Calling Queue.cancelJob(job) above is a great way of cancelling a job however it is achieved from outside the processing function for the job.

If you have a need to cancel a job from within the processing handler function, then you can pass an Error object to the next() function with a custom property on the Error object.

Here is how you would construct the Error object.

let err = new Error('Something went wrong')
err.cancelJob = 'Processing error occurred'
next(err)

The next() function checks for the Error.cancelJob property and if it exists will cancel the job in the queue. You would place this code within your jobs processing error handling function.

For documentation purposes, the value of the Error.cancelJob property being a string, number, or object, will be logged against the job in the database. See Job.log for more detail.

Note: The onCancel function does not get invoked if you cancel a job from within the processing code.

Example

This example shows cancelling a job in the queue using Queue.cancelJob. Also within this code you can see there is a catch block which will call next(err) with the cancelJob property.

const warpEngine = require('warpengine')
const Queue = require('rethinkdb-job-queue')
const q = new Queue()

let job = q.createJob()
job.warpEngineParameters = [254,387,984]

q.addJob(job).catch(err => console.error(err))

q.process((job, next, onCancel) => {
  // Process your job here
  warpEngine.engage(job.warpEngineParameters).then(() => {
    // Call next() when done passing null for the error
    next(null, 'Warp factor 9')
  }).catch(err => {
    // This will cancel the job preventing retries
    err.cancelJob = 'Warp engine failure'
    return next(err)
  })
  // The following callback will only be invoked
  // if the job is cancelled from outside this function
  onCancel(job, () => {
    // Gracefully stop your job here
    warpEngine.shutdown()
  })
})

// Elsewhere in your code, cancel jobs.
// This can be on another process or host as long
// as the changeFeed option is enabled.
q.cancelJob(job).catch(err => console.error(err))

Related Documents

• Queue.reanimateJob
• Queue.removeJob