[Bug]: Method upsertJobScheduler does not work stably

[marcolubian]

Version

v5.58.5

Platform

NodeJS

What happened?

I'm using method upsertJobScheduler(jobSchedulerId, repeatOpts, jobTemplate) to update the scheduling of an already existing scheduled job, changing its execution interval using attribute every and, for example, changing the scheduling every 8, 4, 9, 2, 3, ... seconds (eg 3 seconds means every set to 3000 ms).

In order to do that, I run the script like this

await queue.upsertJobScheduler(jobId, jobOptions, scheduleData);

where jobId is the identifier (name) of the scheduled job and I put every attribute inside the jobOptions.

What I'm experiencing is that sometimes the update works as expected, so for example if I ask for a scheduling of 3 seconds, actually the job is run each 3 seconds; but sometimes the change is not taken and even if it seems to pass correctly through the above method, the scheduling remains the previous one (e.g. if it was each 3 seconds and I change it to 8 seconds, it remains each 3 seconds...).

If I go in debug, I see that the related method of file job-scheduler

async upsertJobScheduler(jobSchedulerId, repeatOpts, jobName, jobData, opts, { override, producerId }) 

is called multiple times for each scheduled job:

• it's called with override to true when called externally (to actually update it)
• it's called with override to false when called internally on the worker file on this method:

async nextJobFromJobData(jobData, jobId, token) {

the problem is that sometimes there is a de-synchronization between the 2 calls and it happens that upsertJobScheduler is called from the worker with an old repeatOpts and from that moment on it keeps working with the old/wrong options.

I have not understood well how nextJobFromJobData works, but for what I've seen it seems to work with the old copy of the repeatOpts: maybe something that remain in memory / in cache ? If so, isn't it possible to let nextJobFromJobData to retrieve (re-read) its data from the backend (Redis(DB) instead of the in-memory/cached data?

I've seen many issues around upsertJobScheduler topic and some seem very similar to mine, in particular this one that seems not fixed yet (even if marked as closed): #3095

Could you please give priority on this topic? For us it's a big problem if we can't count on the reliability of this method. Consider that we've bought the PRO version and need to go in production soon...

How to reproduce.

No response

Relevant log output

Code of Conduct

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

[manast]

Do you have a small test case that reproduces the issue? that would make it much easier for us to actually find and fix your exact issue. We will in the meanwhile try the code in issue #3095 which somehow I did not notice it was posted after my request.

[marcolubian]

@manast I confirm that you can reuse the code in issue #3095; after a few iteration, the issue happens.
You can change it like this, to change also the repeat interval:

import { Queue, Worker } from 'bullmq';
import { Redis } from 'ioredis';

const REDIS_HOST = 'localhost';
const REDIS_PORT = 6379;
const REDIS_PASSWORD = '';
const REDIS_DB = 11;

const redisClient = new Redis({
    host: REDIS_HOST,
    port: REDIS_PORT,
    password: REDIS_PASSWORD,
    db: REDIS_DB,
    maxRetriesPerRequest: null,
})
    .on('connect', () => console.log('Connected to Redis'))
    .on('error', (e) => console.log(`Failed to connect to Redis: ${e.message}`));

const queueName = 'queue-name';
const jobScheduledId = 'my-schedule';
const jobQueue = new Queue(queueName, { connection: redisClient });

const main = async (iteration) => {
    const jobWorker = new Worker(
        queueName,
        (job) => {
            const elapsedTime = ((Date.now() - start) / 1000).toFixed(1);
            console.log(`Elapsed ${elapsedTime}s:`, JSON.stringify({ name: job.name, data: job.data }));
        },
        { connection: redisClient }
    );
    // create schedule
    await jobQueue.upsertJobScheduler(
        jobScheduledId,
        {
            every: 1000,
        },
        {
            name: 'my-name-1',
            data: { key: 'first' },
        }
    );
    const start = Date.now();
    console.log('--- first upsert ---');

    // wait a couple of executions, then upsert
    await new Promise((resolve) => setTimeout(resolve, 10_000));
    await jobQueue.upsertJobScheduler(
        jobScheduledId,
        {
            every: 2000,
        },
        {
            name: 'my-name-2',
            data: { key: 'second' },
        }
    );
    console.log('--- second upsert ---');

    // wait a couple of executions, then upsert
    await new Promise((resolve) => setTimeout(resolve, 10_000));
    await jobQueue.upsertJobScheduler(
        jobScheduledId,
        {
            every: 3000,
        },
        {
            name: 'my-name-3',
            data: { key: 'third' },
        }
    );
    console.log('--- third upsert ---');

    // keep alive
    await new Promise((resolve) => setTimeout(resolve, 10_000));
    await jobWorker.close();
    console.log(`=== End iteration ${iteration} ===\n`);
};

for (let index = 1; index <= 20; index++) {
    await main(index);
}
await redisClient.quit();

Maybe you launch it 3-4 times and it works, but then suddenly it misses some updates, like these:

[manast]

I have identified the reason for the issue now. Planing for the bug fix.