flow doesn't work with worker's backoffStrategies

[koiszzz]

running code below, when children failed, it won't retry again.

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

const ioRedis = require('ioredis');
const connection = new ioRedis({
    host: '11.24.122.177',
    password: 'jjrcb@123',
    db: 1
});

const timeoutPro = () => {
    const num = Math.random();
    return new Promise((resolve, reject) => {
        if (num < 0.5) {
            reject(new Error(`${num} too small`))
        }
        setTimeout(() => {
            resolve(Math.random())
        }, 10 * Math.random() * 1000)
    })
}
const work1 = new Worker('work1', async (job) => {
    console.log('work1', job.data.name);
    return await timeoutPro();
}, {
    connection,
    settings: {
        backoffStrategies: {
            jitter(attempts) {
                console.log('work1 try ' + attempts)
                return attempts * 1000;
            }
        }
    }
});

work1.on('failed', (job, error) => {
    console.log(job.name, job.id, error);
})
work1.on('completed', (job) => {
    console.log(job.name, job.id, job.returnvalue);
})

const work2 = new Worker('work2', async (job) => {
    console.log('work2', job.data.name);
    return await timeoutPro();
}, {
    connection,
    concurrency: 2,
    settings: {
        backoffStrategies: {
            jitter(attempts) {
                console.log('work2 try ' + attempts)
                return attempts * 1000;
            }
        }
    }
});

work2.on('failed', (job, error) => {
    console.log(job.name, job.id, error);
})
work2.on('completed', (job) => {
    console.log(job.name, job.id, job.returnvalue);
})

const queue1 = new Queue('work1', {
    connection,
    defaultJobOptions: {
        removeOnComplete: true,
        attempts: 5,
        backoff: {
            type: 'jitter'
        }
    }
})

const queue2 = new Queue('work2', {
    connection,
    defaultJobOptions: {
        removeOnComplete: true,
        attempts: 5,
        backoff: {
            type: 'jitter'
        }
    }
})

const flowPro = new FlowProducer({
    connection,
});

(async () => {
    await queue1.clean(0, 10000, 'completed');
    await queue2.clean(0, 10000, 'completed');
    await flowPro.add({
        name: 'parent',
        queueName: 'work1',
        data: {
            name: 'parent',
        },
        opts: {
            attempts: 3,
            removeOnComplete: true,
            // backoff: {
            //     type: 'jitter'
            // }
        },
        children: [1, 2].map(() => {
            return {
                name: 'children',
                queueName: 'work2',
                data: {
                    name: 'child',
                },
                opts: {
                    attempts: 5,
                    removeOnComplete: true,
                    backoff: {
                        type: 'jitter'
                    }
                },
            }
        })
    });

    await flowPro.add({
        name: 'single-work',
        queueName: 'work1',
        data: {
            name: 'single',
        },
        opts: {
            attempts: 3,
            removeOnComplete: true,
        },
    });

    const work1Count = await queue1.getJobCounts('completed', 'failed', 'delayed', 'waiting', 'active', 'waiting-children')
    const work2Count = await queue2.getJobCounts('completed', 'failed', 'delayed', 'waiting', 'active', 'waiting-children')

    console.log(work1Count, work2Count)
})();

[roggervalf]

hi @koiszzz, I'm seeing that in your example, QueueScheduler is required but there is no instance of it, you need to create a QueueScheduler to use the backoff strategy

[koiszzz]

Thanks. After adding scheduler, stratefy works.