Rate limit in flows by parent queue

[natilusky]

Hi there,
I have departments, each department has its own queue and I need to set the rate limit for each department.
Each department receives series of tasks, and each such series is a child queue of that department,
I'm trying to set the rate limit at the parent level and can't,
I would appreciate the community's help

i use bullmq 3.5.1

my code

import { FlowProducer, Worker, Queue } from 'bullmq';
import { connection } from './redis-connection.js'


const option = {
    attempts: 4,
    backoff: {
        type: 'exponential',
        delay: 20000,
    }
}

const departmentId = 'bc59e36e-0b3f-408b-9cfd-91b6e36fe870'
const distributionIdA = '25'
const distributionIdB = '26'

const NewflowerTest = async() => {
    try {

        const parentQueue = new Queue(departmentId, {
            connection,
            limiter: {
              groupKey: 'dpartmentId',
            },
          });

        const flowProducerA = new FlowProducer({ connection });
        const flowProducerB = new FlowProducer({ connection });

        const flowA = await flowProducerA.add({
            name: 'test',
            queueName: parentQueue.name, // parent queue name
            children: [
                { name: distributionIdA, data: { place: 'A', departmentId }, queueName: distributionIdA, opts: option },
                { name: distributionIdA, data: { place: 'A', departmentId }, queueName: distributionIdA, opts: option },
                { name: distributionIdA, data: { place: 'A', departmentId }, queueName: distributionIdA, opts: option },
            ],
        });

        const flowB = await flowProducerB.add({
            name: 'test',
            queueName: parentQueue.name, // parent queue name
            children: [
                { name: distributionIdB, data: { place: 'B', departmentId }, queueName: distributionIdB, opts: option },
                { name: distributionIdB, data: { place: 'B', departmentId }, queueName: distributionIdB, opts: option },
                { name: distributionIdB, data: { place: 'B', departmentId }, queueName: distributionIdB, opts: option },
            ],
        });
        
        const workerA = new Worker(distributionIdA, async (job) => {
            console.log(`${new Date().toUTCString()} ${job.name} - ${JSON.stringify(job.data)}`);
          },{
            connection,
            limiter: {
                max: 1,
                duration: 5000,
                groupKey: 'dpartmentId',
            },
            concurrency: 1
        });

        const workerB = new Worker(distributionIdB, async (job) => {
            console.log(`${new Date().toUTCString()} ${job.name} - ${JSON.stringify(job.data)}`);
          },{
            connection,
            limiter: {
                max: 1,
                duration: 5000,
                groupKey: 'dpartmentId',
            },
            concurrency: 1
        });

    } catch (error) {
        console.log(error);
    }
}

this is the log:
Mon, 26 Dec 2022 08:52:36 GMT 25 - {"place":"A","departmentId":"bc59e36e-0b3f-408b-9cfd-91b6e36fe870"}
Mon, 26 Dec 2022 08:52:36 GMT 26 - {"place":"B","departmentId":"bc59e36e-0b3f-408b-9cfd-91b6e36fe870"}
Mon, 26 Dec 2022 08:52:41 GMT 26 - {"place":"B","departmentId":"bc59e36e-0b3f-408b-9cfd-91b6e36fe870"}
Mon, 26 Dec 2022 08:52:41 GMT 25 - {"place":"A","departmentId":"bc59e36e-0b3f-408b-9cfd-91b6e36fe870"}
Mon, 26 Dec 2022 08:52:46 GMT 26 - {"place":"B","departmentId":"bc59e36e-0b3f-408b-9cfd-91b6e36fe870"}
Mon, 26 Dec 2022 08:52:46 GMT 25 - {"place":"A","departmentId":"bc59e36e-0b3f-408b-9cfd-91b6e36fe870"}

[manast]

Which version of BullMQ are you using? group keys have been deprecated in version 3.0 and above: https://github.com/taskforcesh/bullmq/blob/master/docs/gitbook/changelog.md#300-2022-10-25

[natilusky]

Yes I see, but I couldnt understand what is the alternative?

1. I tried to run the 2 workers on the child queue but it sent them at the same time.
2. I tried to run the 2 workers on the parent queue but it did not receive jobs from it.

2 workers on the child queue example

const departmentId = 'bc59e36e-0b3f-408b-9cfd-91b6e36fe870'
const distributionIdA = '25'
const distributionIdB = '26'

const NewflowerTest = async () => {
        const parentQueue = new Queue(departmentId, { connection});

        const flowProducerA = new FlowProducer({ connection });
        const flowProducerB = new FlowProducer({ connection });

        const flowA = await flowProducerA.add({
            name: 'test',
            queueName: parentQueue.name, // parent queue name
            children: [
                { name: distributionIdA, data: { place: 'A', departmentId }, queueName: distributionIdA, opts: option },
                { name: distributionIdA, data: { place: 'A', departmentId }, queueName: distributionIdA, opts: option }
            ],
        });

        const flowB = await flowProducerB.add({
            name: 'test',
            queueName: parentQueue.name, // parent queue name
            children: [
                { name: distributionIdB, data: { place: 'B', departmentId }, queueName: distributionIdB, opts: option },
                { name: distributionIdB, data: { place: 'B', departmentId }, queueName: distributionIdB, opts: option },
            ],
        });

        const workerA = new Worker(distributionIdA, async (job) => {
            console.log(`${new Date().toUTCString()} ${job.name} - ${JSON.stringify(job.data)}`);
        }, {
            connection,
            limiter: {
                max: 1,
                duration: 5000,
            },
            concurrency: 1
        });

        const workerB = new Worker(distributionIdB, async (job) => {
            console.log(`${new Date().toUTCString()} ${job.name} - ${JSON.stringify(job.data)}`);
        }, {
            connection,
            limiter: {
                max: 1,
                duration: 5000,
            },
            concurrency: 1
        });
}

logs:
Tue, 27 Dec 2022 09:37:43 GMT 25 - {"place":"A","departmentId":"bc59e36e-0b3f-408b-9cfd-91b6e36fe870"}
Tue, 27 Dec 2022 09:37:43 GMT 26 - {"place":"B","departmentId":"bc59e36e-0b3f-408b-9cfd-91b6e36fe870"}
Tue, 27 Dec 2022 09:37:48 GMT 26 - {"place":"B","departmentId":"bc59e36e-0b3f-408b-9cfd-91b6e36fe870"}
Tue, 27 Dec 2022 09:37:48 GMT 25 - {"place":"A","departmentId":"bc59e36e-0b3f-408b-9cfd-91b6e36fe870"}

=========================================================================

2 workers on the parent queue example

        const workerA = new Worker(parentQueue.name, async (job) => {
            console.log(`${new Date().toUTCString()} ${job.name} - ${JSON.stringify(job.data)}`);
        }, {
            connection,
            limiter: {
                max: 1,
                duration: 5000,
            },
            concurrency: 1
        });

        const workerB = new Worker(parentQueue.name, async (job) => {
            console.log(`${new Date().toUTCString()} ${job.name} - ${JSON.stringify(job.data)}`);
        }, {
            connection,
            limiter: {
                max: 1,
                duration: 5000,
            },
            concurrency: 1
        });

[natilusky]

@manast Will BullMQ pro solve my problem?

[manast]

I need to study your case better, let me come back to you during the day...