Question about manually processing job status

[ckxddd0324]

Description

I am trying to build a queue with the worker (concurrency = 5). And the job will get updated by a callback(API request) from a 3rd party API. The job is pretty much in idle status and is waiting for the callback to come back and update the job status. It would be either to be completed or failed.

It seems like there is always an issue with missing thread/concurrency count after manually update one of the job with moveToFailed.

Minimal, Working Test code to reproduce the issue.

the endpoint looks something like this

@Patch('completed/:id')
  async markJobCompleted(
    @Param('id') id: string,
    @Body('status') pipelineStatus: string,
  ) {
    console.log(id, 'id');
    const jobStatus = await this.audioQueue.getJob(id);
    if (pipelineStatus) {
      return await jobStatus.moveToCompleted();
    } else {
      return await jobStatus.moveToFailed({ message: 'got failed job!!' });
    }
  }

(An easy to reproduce test case will dramatically decrease the resolution time.)

1. create a queue with 5 concurrency processor
2. for sanity-wise, add a 10min timeout on the process function
3. add 7 jobs to queue
4. 5 of the jobs should be in active state
5. the other 2 of jobs are in waiting state
6. call the endpoint to update one of the job with the jobId
7. the number of active jobs is 4 and the number of the waiting jobs remains 2

the expectation of this case would be 5 and number of waiting job should be 1

Bull version

"@nestjs/bull": "^0.6.3"
"bull": "^4.10.4"

Additional information

building with nestjs

[manast]

Why are you calling moveToCompleted and moveToFailed manually?

[ckxddd0324]

@manast I think I would want the job to move to COMPLETED stage if the webhook request from the 3rd party returned something like pass or completed, else if webhook request sends failed this would trigger the job to failed and lets the queue to handle to rerun with attempts.

[manast]

Just handle those cases inside the processor. Please read this blog: https://blog.taskforce.sh/implementing-a-webhook-system-with-bullmq/

[ckxddd0324]

Thanks. I think what I doing now is, not sure thats the best way. But it seems working to me, it would be nice if the processor does not have to looping/self running.

So what I did is have 2 endpoints, endpoint 1 will add the job to the queue and endpoint 2 will be a webhook called by the 3rd party api and update a custom field in job.data for example something like job.data.jobStatus. Then the worker will be constantly running in a loop checking the job.data.jobStatus for every 20 seconds, if the jobStatus is completed, it returns something else failed will throw an error in the try-catch. This error will make the job failed and this queue should add it back to the queue.

I am curious if there is any thing job event listener or emit when the job.data get updated and this would trigger the process to do something?

hope this is making some sense

[ckxddd0324]

@manast Sorry for tagging you. But I am curious if there is a better way for the worker to get notified if the job.data is updated or some of the events that can trigger the worker to do an action? Thanks