FlowProducer: Job Pipelines and Dependency Trees

blog · flows

Job flows that wait for their children.

Some tasks are not a single job, they are a pipeline: resize an image, generate thumbnails, update the CDN. FlowProducer expresses these dependencies naturally, children run first and the parent runs only when all of them complete.

bunqueue’s FlowProducer lets you build parent-child dependency trees, fan-out patterns, and processing chains, all covered below.

The FlowProducer API

import { FlowProducer, Worker } from 'bunqueue/client';

const flow = new FlowProducer({ embedded: true });

FlowProducer supports the BullMQ v5 API for defining job trees with parent-child relationships.

Basic Flow: Parent Waits for Children

The most common pattern: a parent job that depends on multiple child jobs.

const result = await flow.add({
  name: 'generate-report',
  queueName: 'reports',
  data: { reportId: 'q4-2024' },
  children: [
    {
      name: 'fetch-sales',
      queueName: 'data-fetch',
      data: { source: 'sales-db', quarter: 'Q4' },
    },
    {
      name: 'fetch-expenses',
      queueName: 'data-fetch',
      data: { source: 'expense-db', quarter: 'Q4' },
    },
    {
      name: 'fetch-metrics',
      queueName: 'data-fetch',
      data: { source: 'analytics', quarter: 'Q4' },
    },
  ],
});

console.log(result.job.id);                // Parent job ID
console.log(result.children?.length);       // 3 children

The execution order:

1. All three fetch-* children start processing (in parallel)
2. When all children complete, generate-report becomes available
3. A worker picks up generate-report and can access children’s results

Accessing Children’s Results

The parent job can retrieve the results of its children:

const reportWorker = new Worker('reports', async (job) => {
  // Get all children's return values
  const childResults = await job.getChildrenValues();

  // childResults is a Record<string, unknown>
  // Keys are "{queueName}:{jobId}" format
  const salesData = Object.values(childResults)[0];
  const expenseData = Object.values(childResults)[1];
  const metricsData = Object.values(childResults)[2];

  return generateReport(salesData, expenseData, metricsData);
}, { embedded: true });

Nested Flows (Multi-Level Trees)

Children can have their own children, creating deep dependency trees:

await flow.add({
  name: 'deploy',
  queueName: 'deployment',
  data: { version: '2.1.0' },
  children: [
    {
      name: 'build',
      queueName: 'ci',
      data: { step: 'build' },
      children: [
        {
          name: 'lint',
          queueName: 'ci',
          data: { step: 'lint' },
        },
        {
          name: 'test',
          queueName: 'ci',
          data: { step: 'test' },
        },
      ],
    },
    {
      name: 'migrate-db',
      queueName: 'db',
      data: { migration: '045_add_index' },
    },
  ],
});

Execution order:

1. lint and test run in parallel
2. When both complete, build runs
3. migrate-db also runs in parallel with build
4. When both build and migrate-db complete, deploy runs

Chain Pattern: Sequential Steps

For strictly sequential pipelines:

// Using nested children for a chain
await flow.add({
  name: 'step-3-notify',
  queueName: 'pipeline',
  data: { step: 3 },
  children: [
    {
      name: 'step-2-process',
      queueName: 'pipeline',
      data: { step: 2 },
      children: [
        {
          name: 'step-1-fetch',
          queueName: 'pipeline',
          data: { step: 1 },
        },
      ],
    },
  ],
});
// Executes: step-1 → step-2 → step-3

Bulk Flows

Add multiple independent flows at once:

const results = await flow.addBulk([
  {
    name: 'process-order',
    queueName: 'orders',
    data: { orderId: 'A001' },
    children: [
      { name: 'validate', queueName: 'validation', data: { orderId: 'A001' } },
      { name: 'check-stock', queueName: 'inventory', data: { orderId: 'A001' } },
    ],
  },
  {
    name: 'process-order',
    queueName: 'orders',
    data: { orderId: 'A002' },
    children: [
      { name: 'validate', queueName: 'validation', data: { orderId: 'A002' } },
      { name: 'check-stock', queueName: 'inventory', data: { orderId: 'A002' } },
    ],
  },
]);

Retrieving Flow State

Inspect a flow tree and its current state:

const tree = await flow.getFlow({
  id: parentJobId,
  queueName: 'reports',
  depth: 3,           // How deep to traverse
  maxChildren: 100,   // Max children per level
});

// tree.job - the parent job details
// tree.children - array of child nodes (recursive)

Error Handling in Flows

Control how child failures affect the parent:

await flow.add({
  name: 'parent',
  queueName: 'main',
  data: {},
  children: [
    {
      name: 'critical-child',
      queueName: 'tasks',
      data: {},
      opts: {
        failParentOnFailure: true,  // Parent fails if this child fails
      },
    },
    {
      name: 'optional-child',
      queueName: 'tasks',
      data: {},
      opts: {
        ignoreDependencyOnFailure: true,  // Parent proceeds even if this fails
      },
    },
  ],
});

Tip

Use failParentOnFailure for critical steps and ignoreDependencyOnFailure for optional enrichment steps. This gives you fine-grained control over pipeline behavior.

Real-World Example: Image Processing Pipeline

import { FlowProducer, Worker } from 'bunqueue/client';

const flow = new FlowProducer({ embedded: true });

// Define the pipeline
async function processImage(imageUrl: string) {
  return await flow.add({
    name: 'update-cdn',
    queueName: 'cdn',
    data: { imageUrl },
    children: [
      {
        name: 'generate-thumbnails',
        queueName: 'images',
        data: { imageUrl, sizes: [100, 300, 800] },
        children: [
          {
            name: 'download-original',
            queueName: 'images',
            data: { imageUrl },
          },
        ],
      },
      {
        name: 'extract-metadata',
        queueName: 'images',
        data: { imageUrl },
        opts: { ignoreDependencyOnFailure: true },
      },
    ],
  });
}

// Workers for each queue
new Worker('images', async (job) => {
  switch (job.name) {
    case 'download-original':
      return await downloadImage(job.data.imageUrl);
    case 'generate-thumbnails':
      const original = await job.getChildrenValues();
      return await createThumbnails(original, job.data.sizes);
    case 'extract-metadata':
      return await extractEXIF(job.data.imageUrl);
  }
}, { embedded: true });

new Worker('cdn', async (job) => {
  const results = await job.getChildrenValues();
  await uploadToCDN(results);
  return { published: true };
}, { embedded: true });

Dependency Resolution Performance

bunqueue uses event-driven dependency resolution via microtask coalescing. When a child job completes, the parent’s dependency state is updated immediately rather than on a polling interval:

Pattern	Old Polling (100ms)	Event-Driven (p50)
Simple parent-child	~100ms latency	~14 microseconds
Chain (4 levels)	~300ms	~31 microseconds
Fan-out (1 to 5)	~100ms	~28 microseconds

This makes complex flow trees execute with minimal overhead between steps.