BULLMQ job stalled more than allowable limit

[Hunglmc]

Hi guys, I have a problem with the code below, I tried to add 10 jobs at the same time but it didn't work, I don't know where I'm going wrong and how I need to fix it. Below is my code

export class CitadGetwayWorkerService implements OnApplicationBootstrap {
    private readonly logger = new Logger(CitadGetwayWorkerService.name);
    private readonly currentTime: string;
    constructor(
        private readonly bullmqConfigService: BullmqConfigService,
        // private readonly transactionService: TransactionServices,
        // private readonly oracleConnectSevice: OracleDBService
    ) {
        this.currentTime = moment().format('DD-MM-YYYY HH:mm:ss');
    }

    onApplicationBootstrap() {
        this.createWorker(QueueConstants.SYNC_CITAD);
    }
    private createWorker(queueName: string) {
        const worker = new Worker(
            queueName,
            path.join(__dirname, 'sync.citad.processor.js'),
            {
                connection: this.bullmqConfigService.getBullmqConfig(),
                concurrency: 10,
            },
        );

        worker.on('completed', (job: Job) => {
            this.logger.log(`${queueName} job completed at ${this.currentTime} with result ${JSON.stringify(job.returnvalue)}`);
        });

        worker.on('failed', (job, err) => {
            this.logger.error(`${queueName} job failed at ${this.currentTime} with reason ${err}`);
        });
        worker.on('active', (job: Job) => {
            this.logger.log(`${queueName} job active at ${this.currentTime}`);
        })
    }
}

and

export default async function (job: SandboxedJob) {
   
    const logger = new Logger("Sanbox_citad");
    try {
        logger.log(`Job ${job.id} started at ${moment().format('DD-MM-YYYY HH:mm:ss')}`);
        logger.log(`Job data: ${JSON.stringify(job.data)}`);

        // Simulate an image optimization process
        await optimizeImage('dummy-image', job);

        // TODO: Add job-specific business logic here...
        logger.log(`Job ${job.id} successfully completed.`);
    } catch (error) {
        logger.error(`Error processing job ${job.id}: ${error.message}`);
        throw error;
    }
};

/**
 * Simulates an image optimization process asynchronously.
 *
 * @param image Image to be optimized
 * @returns A promise resolving to the optimized image
 */
const optimizeImage = async (image: string, Job: SandboxedJob): Promise<string> => {
    for (let index = 0; index < 10e5; index++) {
        const progress = ((index * 100) / 10e5).toFixed(2);
        Job.updateProgress(Number(progress))
        console.log(` optimizeImage ${progress}%`);
    }
    return await new Promise((resolve) =>
        setTimeout(() => resolve(image), 1000),
    );
};

[manast]

You are not awaiting the this call Job.updateProgress(Number(progress)) so I cannot guess what effects that could have on the processing. Await that call and let us know how it behaves.

[Hunglmc]

Thank you for your support, the Job.updateProgress function simply updates the percentage of job processing progress. By the way, let me ask you if in nest js I can declare a class without using export default, will it be effective with using sandbox, because I want to call some more services to handle the logic, but I don't understand how to do this. According to bullmq documentation, sandbox is always called with export default.