[Bug]: workers stopped consuming job, sometime event loss

[theyashwantsoni]

Version

1.84.0

Platform

NodeJS

What happened?

Sometime experiencing few jobs are not getting processed and sometime no jobs are getting processed, seems happening because of memory issue.

Happening just after the memory spike.

How to reproduce.

No response

Relevant log output

No response

Code of Conduct

• I agree to follow this project's Code of Conduct

[manast]

Is it version 1.84.0 ? thats pretty old... also we will need some code to reproduce the issue.

[theyashwantsoni]

i have more than 200 type of jobs running in my system, and everything was fine even with the older version. This started happening recently and I haven't made any changes in the code.

Earlier I thought, due to unhandled exception node process is killed, but tried to reproduce and no luck.

[theyashwantsoni]

import bullRedisConnection from "@/config/bullmq/redisConnection";
import { Job, Queue, QueueScheduler, Worker } from "bullmq";
import JobProcessor from "./jobProcessor";

export default class BullWorker{
    queueName: string
    worker: Worker
    constructor(priority:TaskPriority){
        this.queueName = priority
        this.initWorker()
    }

    private async initWorker(){
        const _ = new QueueScheduler(this.queueName, {connection: bullRedisConnection.getConnection()});
        this.worker = new Worker(this.queueName,async (job:Job) => {
                let jobProcessor = await new JobProcessor(job)
                let result = await jobProcessor.execute()
                return result
        },
        { 
            autorun: false,
            connection: bullRedisConnection.getConnection() ,
        })
    }

    public run(){
        this.worker.run()
    }
}

[theyashwantsoni]

import { TaskState } from "@/enums/taskResult/taskState";
import { Job } from "bullmq";
import AdapterFactory from "./adapterFactory";
import JobProgressUpdateInterface from "./jobProgressUpdateInterface";

export default class JobProcessor implements JobProgressUpdateInterface{
    job: Job
    adapterFactory: AdapterFactory
    constructor(job:Job){
        this.job = job
        this.adapterFactory = new AdapterFactory(job.name, this)
    }

    async execute():Promise<TaskResultOutput> {
        try{
            let result = await this.adapterFactory.getAdapter().processTask(this.job.data)
            return JSON.stringify(result)
        }   
        catch(err){
            throw(err)
        }
        
            
        
    }


}

[manast]

Unfortunatelly this information is not enough for us to be able to perform any further action. Maybe you could start by checking in which statuses the jobs are in right now, verify that all the workers are indeed connected to Redis and so on.