How to stop bull queue from automatically restarting jobs when restarting the server

Question

I am using the bull queue to process jobs.

Let's say a job is running with a status of active when I restart my dev server. When the worker script starts up again, the process is still set to active in the queue, so bull decides to start up the worker process again.

This become disruptive very quickly, as the script is often restarting during development, so many processes end up running and making a mess of things. All I want is bull to NOT restart these jobs when the server starts.

Things I've tried:

• setting attempts: 1 when creating a job (also tried 0) https://github.com/OptimalBits/bull/blob/HEAD/REFERENCE.md#queueadd
• using queue.clean() (https://github.com/OptimalBits/bull/blob/HEAD/REFERENCE.md#queueclean)
• Writing a script to remove active jobs (throws Error: Missing lock for job xx failed):

    let active_jobs = await queue.getJobs(['active']);
    active_jobs.forEach(async (active_job) => {
      await active_job.discard()
      await active_job.moveToFailed(new Error("Auto-killed during dev server restart"))
    })

None of this works. Anyone have a solution to achieve this?

Answer 1

It actually easy if you have the needed functions in place. You could implement some code to completely clean the specific queues. That code has to live in a place thats called one time once your server starts up. So a good place would be the constructor of your class (of the producer-server). That way you always start with a clean queue with zero entries for development purposes. So best is to wrap the function call to empty the queues into some statement that prooves for development conditions.

You can use something like the following:

const getKeys = async (q) => {
  const multi = q.multi();
  multi.keys('*');
  const keys = await multi.exec();
  return keys[0][1]
}

const filterQueueKeys = (q, keys) => {
  const prefix = `${q.keyPrefix}:${q.name}`;
  return keys.filter(k => k.includes(prefix));
}

const deleteKeys = async (q, keys) => {
  const multi = q.multi();
  keys.forEach(k => multi.del(k));
  await multi.exec();
}

const emptyQueue = async (q) => {  
  const keys = await getKeys(q);
  const queueKeys = filterQueueKeys(q, keys);
  await deleteKeys(q, queueKeys);
}

if (process.env === 'development') {
    emptyQueue(this.workerQueue).then(() => {
      console.log('QUEUE EMPTY!')
    })
}