Skip / Update Job if it has waited too long in the queue

[vin-ni]

Description

Hello,
I'm trying to discard jobs, which have spent too much time in the queue, as a self regulating measure.

Basically the queue fills up and once there are too many jobs, the ones that have been waiting over 10 seconds, should be skipped or fast forwarded through.

I'm using express and returning the result as a response. I'm aware this is not the absolute standard use case, but I'm trying to implement a first come, first served logic.

My route:

const imgM = require('../imageManipulation');
const manipulator = new imgM();
var imageQueue = new Queue('image processing');
imageQueue.process(cpuCount, __dirname + '/../processor.js');

router.post('/', async (req, res, next) => {
  let err = false;

  let job;
  job = await imageQueue.add({ imgData: req.body.imgData }, { removeOnComplete: true, removeOnFail: true }); //https://github.com/OptimalBits/bull/blob/develop/REFERENCE.md
  const result = await job.finished().catch(() => {
    console.log('That did not go well.');
    err = true;
  });

  if (err) {
    res.status(503).send('Service unavailable. Please retry.');
  } else {
    res.status(200).json({ message: 'Created images', imgData: result });
  }
});

And a simplified version of my image processor:

const imgM = require('./imageManipulation');
const manipulator = new imgM();

const manipulation = async (job) => {

    // cropping images
  const image1Cropped = await manipulator.cropImage(job.data.imgData);

  //building compositions
  const composition1 = await manipulator.compositeImage(image1Cropped);

  const addedCounter1 = await manipulator.addCounter(composition1);

  return Promise.resolve(addedCounter1);
};

module.exports = manipulation;

Maybe someone could point me in the right direction on how to res a 503 in case a job is waiting longer than 5 seconds in the queue.

I tried to pass cancellation tokens, to skip the async awaits, but didn't manage to do it.

I also tried adding a setTimeout in the route and then calling job.update() with some cancellation token, but this also didn't work.

Bull version

3.13.0

Thank you!

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.