Lesson 5 — Job Flows (Parent / Child)

Tied to mission: a single job can't express "do these N things in parallel, then aggregate." Flows do exactly that — the building block for image processing and webhook fan-out in your mission. Builds on Lesson 1, Lesson 2, Lesson 4.

The mental model

A flow is a tree of jobs. A parent job sits in a new state, waiting-children, and is not moved to waiting (where a worker can grab it) until all its children have completed successfully. Parent and child jobs are otherwise ordinary jobs.

Trees can be arbitrarily deep — a child can itself have children.

The FlowProducer — your one entry point

You don't use Queue.add() for flows. You use the FlowProducer class, whose add() inserts the entire tree atomically (all-or-nothing):

import { FlowProducer } from 'bullmq';
const flow = new FlowProducer({ connection });

const tree = await flow.add({
  name: 'finalize',
  queueName: 'pipeline',
  data: { album: 'vacation' },
  children: [
    { name: 'resize', data: { size: 200 }, queueName: 'images' },
    { name: 'resize', data: { size: 400 }, queueName: 'images' },
    { name: 'resize', data: { size: 800 }, queueName: 'images' },
  ],
});
// → adds 4 jobs atomically: 1 to 'pipeline', 3 to 'images'

Each node carries its own queue

Notice every node has its own queueName. Children can live in a different queue than the parent — and different from each other. This is flows' superpower: cross-queue dependencies. Put CPU-heavy image work on the images queue (sandboxed workers) and the fast aggregation on pipeline.

Options on a flow node — what's NOT allowed

The opts on a flow node is a normal JobsOptions minus repeat, debounce, deduplication (and parent). So:

Inspecting a flow

// parent worker aggregates children's results
new Worker('pipeline', async (job) => {
  const results = await job.getChildrenValues();  // { '<childKey>': 200url, ... }
  const urls = Object.values(results);
  await sendEmail({ album: job.data.album, urls });
}, { connection });

getFlow takes bounds so deep trees don't explode your memory:

const shallow = await flow.getFlow({
  id: parentJob.id,
  queueName: 'pipeline',
  depth: 1,         // only first level of children
  maxChildren: 50,  // cap per node
});

The failure gotcha — orphans, by default

Policy 1 — continueParentOnFailure (best-effort)

Set on a child: if that child fails, the parent starts processing immediately instead of waiting for siblings. Pair with removeUnprocessedChildren() to cancel the rest, and getFailedChildrenValues() to detect whether you're here due to a failure.

children: [
  {
    name: 'resize', data: { size: 200 }, queueName: 'images',
    opts: { continueParentOnFailure: true },  // parent proceeds if THIS child fails
  },
  ...
],
// in parent processor:
await job.removeUnprocessedChildren();   // cancel siblings still pending
const failed = await job.getFailedChildrenValues();
if (Object.keys(failed).length) console.warn('partial success');

Policy 2 — failParentOnFailure (fail-fast)

Set on a child: if that child fails, the parent is moved to failed immediately. It's selective (only children with the flag trigger it) and recursive — if ancestors also carry the flag, the failure propagates up the tree.

children: [
  {
    name: 'validate', data: { file }, queueName: 'checks',
    opts: { failParentOnFailure: true },  // parent fails if validation fails
  },
  ...
],

Removal cascades

Removing a job inside a flow is not independent. BullMQ cascades:

1. Remove parent → all children removed too.
2. Remove a child → its link to the parent is cut; if it was the last child, the parent completes (no worker runs it).
3. A node that's both parent and child triggers both rules.
4. If any affected job is currently locked (active), nothing is removed and an exception throws.

Full picture — fan-out resize, fan-in email

import { FlowProducer, Worker, Queue } from 'bullmq';
import IORedis from 'ioredis';
const connection = new IORedis({ maxRetriesPerRequest: null });
const flow = new FlowProducer({ connection });

type Img = { album: string; size: number };

// Fan-out + fan-in as a single atomic tree
await flow.add({
  name: 'finalize',
  queueName: 'pipeline',
  data: { album: 'vacation' },
  children: [200, 400, 800].map((size) => ({
    name: 'resize',
    data: { album: 'vacation', size } as Img,
    queueName: 'images',
    opts: { attempts: 3, continueParentOnFailure: true, removeOnComplete: 100 },
  })),
});

// Child worker (CPU-bound — use sandboxed processors in prod)
new Worker<Img>('images', async (job) => {
  return await resize(job.data);   // returnvalue surfaces in getChildrenValues()
}, { connection });

// Parent worker — only fires when children resolve
new Worker('pipeline', async (job) => {
  const urls = Object.values(await job.getChildrenValues());
  await sendEmail({ album: job.data.album, urls });
}, { connection });

What you just learned

• A flow is a job tree; the parent sits in waiting-children until all children complete.
• Use FlowProducer.add() (not Queue.add) — inserts the whole tree atomically.
• Each node has its own queueName → cross-queue dependencies, arbitrary depth.
• Flow nodes can't be repeat/debounce/deduplicate.
• Default failure behavior orphans the parent. Always pick a policy: continueParentOnFailure or failParentOnFailure.
• Inspect with getFlow(), getChildrenValues(), getDependenciesCount(), parentKey.
• Removal cascades; locked jobs block the whole removal.

🧠 Quiz — 10 questions

📚 Primary sources

📎 Reference

💬 Ask me next

• "Show me a 3-level flow (resize → watermark → upload → email)."
• "How do I retry just one failed child without re-running the whole tree?"
• "My parent is stuck in waiting-children — how do I find the culprit child?"
• "How do flows interact with rate limiting across queues?"

Or say "next" for Lesson 6: metrics, stalled jobs & queue cleanup.