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.

In the Bun package and every current official external SDK, add and addBulk send one PUSHF command: the complete graph is validated before mutation and becomes visible atomically. With a configured dataPath, every row is also committed in one immediate SQLite transaction before workers can pull a leaf. Embedded mode without a dataPath keeps the atomic visibility guarantee but is intentionally memory-only. Previously published SDKs using PUSH plus UpdateParent remain compatible without retroactively gaining that transaction.

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 completes, a reverse dependency index identifies only the parents that can be affected; readiness is rechecked under the parent shard lock and eligible parents are notified without a polling interval. Resolution is O(completed edges plus affected parents), rather than a scan of every queued job. See the maintained benchmarking contract for hardware-qualified measurements; example code does not embed unrepeatable latency numbers.