Rate limiting with flows

[aravindabose06]

Hello,

I have been using the flows for a while, and recently updated the package version to the v3.4.1.
Now I'm having trouble using rate limiter with flows.

Here is the worker options for the child queue.

{
    concurrency: 600,
    limiter: {
        max: 300,
        duration: 1000,
    },
}

Even though it is set to process 300 jobs per second, it is not processing that way. It is processing the jobs like the limiter is dynamically increasing starting from 2.
Eg., it is taking 2 jobs initially, once they gets completed, the next 3 jobs are processed, and then 4 and so on.

And we are not using the worker's rateLimit function too. For the simplicity to test this. I just added a sleep in the processor and nothing else.
Am I missing something here?. Is something else need to be configured?.

Any advice on this would be really helpful!

Thanks :)

[manast]

You are not providing a lot of code so it is quite difficult to give any reasonable answer 🤔

[aravindabose06]

Hi @manast, thank you for your response!

Sorry about the lack of information there 😅

Here is a reproducible code,

const { FlowProducer, Worker, Queue } = require('bullmq');

const connection = {
    host: "localhost",
    port: 6379
}

const parentQueue = new Queue('parent-queue', { connection });

const childQueue = new Queue('child-queue', {
    connection,
});

const flowProducer = new FlowProducer({ connection });

const childJobs = Array(20).fill("").map((_, index) => {
    return {
        name: "child-job-a",
        data: { id: `id-${index}` },
        queueName: "child-queue",
    }
})


const addJobsByFlow = async () => {
    const flow = await flowProducer.add({
        name: 'parent-job',
        queueName: 'parent-queue',
        children: childJobs,
    }, {
        queuesOptions: {
            'child-queue': {
                connection,
                concurrency: 600,
                limiter: {
                    max: 100,
                    duration: 5000,
                },
            }
        }
    });

    return
}

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

const childWorker = new Worker('child-queue', async (job) => {
    console.count("child-worker")
    console.log(`Job entered: ${job?.data?.id}`)

    await sleep(5000)

    console.log(`Job completed: ${job?.data?.id}`)

    return "success"
}, {
    connection,
    concurrency: 600,
    limiter: {
        max: 300,
        duration: 1000,
    },
});

addJobsByFlow()

[manast]

We have a PR open that we think will solve this issue: #1597