How to change job name in process() and immediately retry that job

[zxpectre]

Hi all. As question says, how to change job name inside process() and immediately retry (re-schedule to run as soon as posible on any instance) that job to execute the process() registered for the second name?

I need to replace agenda for bull in my proyect and I need to accomplish this functionality.

Thnx!

[manast]

It is not possible to neither change the name nor to retry a job inside the processor function.

[zxpectre]

Thanks for the answer. I'm using the library for the first time sorry for asking but I really need to solve this.

Maybe a simpler example would be a job that on each run it increments a value (in data,not using progress). Each run occurs as soon as previous ends. This job's execution "jumps" between processes, only one execution of this jobId at a time.

How could I do this?
Thnx.

[manast]

I do not understand what you mean with " "jumps" between processes". BullMQ will only execute one job with a given jobId at a time in any case...

[zxpectre]

I guess i solved it.
This is what I meant, hope somebody else find this usefull as well:

var Queue         = require('bull');
var cluster       = require('cluster');
const cfg         = require('../config');
var redisConnOps  = require("./bullMultiProcessConnection");

//README: 
//https://github.com/OptimalBits/bull/blob/develop/PATTERNS.md#reusing-redis-connections
//https://github.com/OptimalBits/bull/blob/develop/REFERENCE.md
//https://github.com/OptimalBits/bull#separate-processes



const uri=cfg.db.bullRedisdbUri;
const host=cfg.db.bullRedisdbHost;
const port=cfg.db.bullRedisdbPort;
const password=cfg.db.bullRedisdbPass;
const connData={port,host,password};//db:0



function timeoutPromess(delay) {
    return new Promise(resolve => {
      setTimeout(() => {
        resolve();
      }, delay);
    });
  }

var queue = new Queue('video transcoding', {
  ...redisConnOps,
  //redis: connData,
   //settings:{
    //lockDuration: 2000,//number = 30000; // Key expiration time for job locks.
    //lockRenewTime: 1000,//number = 15000; // Interval on which to acquire the job lock
  //}
}); // Specify Redis connection using object


 queue.isReady().then(async () => {
  queue.on('completed', async function(job, result) {
    const pid=process.pid;
    const id=job.id;
    const name=`${pid}:${id}`
    const state=await job.getState();
    console.log(`Job ${name} completed! State:${state}`);
    console.dir({[name]:result});
    await job.remove();
  });
  //WARNING: await queue.obliterate({ force: true }).catch(err=>{});
  clusterify(3,8,20);
  //clusterify(0,8,20);
})
 



const masterCode=async(numJobs)=>{
  const pid=process.pid;
  for(var i=0; i<numJobs; i++){
    // as master, lets create a few jobs for the queue workers
    await queue.add('stage1',{publisher: pid},{attempts:10,backoff:0});      
  };
}

const workerCode=async (numThreads)=>{
  //register stage1 job handler
  await queue.process('stage1',numThreads,async function(job, done){
    const pid=process.pid;
    const id=job.id;
    const name=`${pid}:${id}`
    const data=job.data;
    //console.dir({[name]:id})
    await timeoutPromess(0);
    await job.progress(job.progress()+20);
    job.data.counter=(job.data.counter||0)+1;
    job.data.foo="bar";
    job.data.workers=[...(job.data.workers||[]),pid];
    await job.update(job.data);
    console.dir({[name]:{progress:job.progress()}})
    if(job.progress()>=100){
      done(null,job.data);
    }
    done(new Error ("Need retry"),null);
    //await job.releaseLock();
  }); 
}

  

const clusterify=(numWorkers,numThreadsPerWorker,numJobs)=>{
  const masterPid=process.pid;
  if(numWorkers<=0){
    console.log('Single worker is ' +masterPid );
    masterCode(numJobs);
    workerCode(numThreadsPerWorker);
    return;
  }
  if(cluster.isMaster){
    console.log('Master is ' +masterPid );
    masterCode(numJobs);
    for (var i = 0; i < numWorkers; i++) {
      cluster.fork();//spawn 5 concurrent processes reusing redis connection
    }
    
    cluster.on('online', async function(worker) {
        console.log('Worker ' + worker.process.pid + ' spawned');
    });
    cluster.on('exit', function(worker, code, signal) {
      console.log('worker ' + worker.process.pid + ' died');
    });

  }else{
    workerCode(numThreadsPerWorker);
  }
}