Creating a workflow with dynamic child dependencies

[cg1972]

I was wanting to use BullMQ to implement workflows that may contain forks / joins and switch statements. I wanted to take advantage of using FlowProducer where I can but I understand that all dependencies must be defined when the flow is added. If I had a simple branch scenario, for example:

A -> B (fork)
B -> C & D1
D1 -> D2
C -> E
D2 -> E

Would it be possible to define 2 flows here. The first one would be
E
-> C
-> (Dependency Job 1)
-> A. (this is a child of either C or dependency job 1)

and the Dependency Job 1 would be:
D2
-> D1

The idea being that in the worker when it hits the (Dependency Job1) you would do something like:

1. Add the Dependency Job 1 to the flow producer setting the opt.parent to the job id from flow 1.
2. Call job.moveToWaitingChildren(token) on the Dependency Job 1 from flow 1.

The idea being that the dependency job 1 gets put on the wait queue while D1 & D2 get processed and then the results get sent back to the Dependency Job 1 once they have completed.

I have seen in the documentation that this is what the moveToWaitingChildren() is for, however, I have not been able to get it to work correctly when using a flow producer job. Even though the call to job.moveToWaitingChildren(token) returns true for the Dependency Job 1, the flow producer for D1 and D2 does not trigger and the workflow stalls.

Any suggestions would be appreciated.

[roggervalf]

hi @cg1972, for creating dynamic children we have this pattern https://docs.bullmq.io/patterns/process-step-jobs#waiting-children. From your processor you can add new child dependencies and/or create new flows as well

[cg1972]

Thanks roggervalf for the information. I have tried this out and it is now working. I think part of my problem was not setting the lock_on_hashtags command line argument required when running with Dragonfly!