# FlowProducer: Job Pipelines and Dependency Trees

Create job pipelines and dependency trees with bunqueue's FlowProducer. Parent-child workflows, fan-out patterns, and chain processing.

Canonical: https://bunqueue.dev/blog/job-pipelines-flows/

---

import { Aside } from '@astrojs/starlight/components';

<div class="bq-wrap bq-hero">
  <span class="bq-eyebrow">blog · flows</span>
  <h1 class="bq-hero-h1 bq-bench-h1">Job flows that wait for their <em>children.</em></h1>
  <p class="bq-hero-sub">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.</p>
</div>

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

## The FlowProducer API

```typescript
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.

```typescript
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:

```typescript
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:

```typescript
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:

```typescript
// 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:

```typescript
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:

```typescript
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:

```typescript
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
      },
    },
  ],
});
```

<Aside type="tip">
  Use `failParentOnFailure` for critical steps and `ignoreDependencyOnFailure` for optional enrichment steps. This gives you fine-grained control over pipeline behavior.
</Aside>

## Real-World Example: Image Processing Pipeline

```typescript
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](/guide/benchmarks/) for
hardware-qualified measurements; example code does not embed unrepeatable
latency numbers.