Skip to content
bunqueue bunqueue
Get started
Get started
Flow Patterns: Chains, Fan-In and Trees
View Markdown
guide · flow producer

Chain it, fan it out, merge it back.

Four shapes cover nearly every pipeline: one after another, many at once into a single parent, a deeper tree, and the plumbing that lets a parent read what its children produced.

Common Tasks

Section titled “Common Tasks”

Run jobs one after another (chain)

Section titled “Run jobs one after another (chain)”

addChain executes jobs in order: each starts only when the previous one completes.

// fetch → process → store
const { jobIds } = await flow.addChain([
  { name: 'fetch', queueName: 'pipeline', data: { url: 'https://api.example.com' } },
  { name: 'process', queueName: 'pipeline', data: {} },
  { name: 'store', queueName: 'pipeline', data: {} },
]);

Run jobs in parallel, then merge (fan-in)

Section titled “Run jobs in parallel, then merge (fan-in)”

addBulkThen runs a batch concurrently and fires a final job after all of them complete.

//   fetch-api-1 ──┐
//   fetch-api-2 ──┼──→ merge-results
//   fetch-api-3 ──┘
const { parallelIds, finalId } = await flow.addBulkThen(
  [
    { name: 'fetch-api-1', queueName: 'parallel', data: { source: 'api1' } },
    { name: 'fetch-api-2', queueName: 'parallel', data: { source: 'api2' } },
    { name: 'fetch-api-3', queueName: 'parallel', data: { source: 'api3' } },
  ],
  { name: 'merge-results', queueName: 'parallel', data: {} }
);

addBulkThen is available in the Bun package, the TypeScript SDK and the Python SDK. In PHP, Go, Rust and Elixir build the same shape with add: a parent whose children are the parallel jobs (children complete first, then the parent runs).

Build a tree

Section titled “Build a tree”

addTree creates a hierarchy where children depend on their parent (the parent runs first, then its children), within the creation limits:

const { jobIds } = await flow.addTree({
  name: 'root',
  queueName: 'tree',
  data: { level: 0 },
  children: [
    {
      name: 'branch-1', queueName: 'tree', data: { level: 1 },
      children: [
        { name: 'leaf-1a', queueName: 'tree', data: { level: 2 } },
        { name: 'leaf-1b', queueName: 'tree', data: { level: 2 } },
      ],
    },
    { name: 'branch-2', queueName: 'tree', data: { level: 1 } },
  ],
});

addTree (parent-first hierarchies) is available in the Bun package and the TypeScript SDK only for now. Every SDK’s add builds the inverse tree, where children complete before their parent (Quick Start).

Read results from earlier jobs

Section titled “Read results from earlier jobs”

In flow.add() flows, the parent calls await job.getChildrenValues() (shown in the Quick Start).

In addChain / addBulkThen / addTree flows, bunqueue injects parent IDs into the job data, and FlowProducer can look up their results in embedded or TCP mode:

const runtimeOptions = { embedded: false, connection: { host: '127.0.0.1', port: 6789 } };
const flow = new FlowProducer(runtimeOptions);


const worker = new Worker('pipeline', async (job) => {
  if (job.data.__flowParentId) {           // chain: one parent
    const parentResult = await flow.getParentResult(job.data.__flowParentId);
  }
  if (job.data.__flowParentIds) {          // merge: many parents
    const results = await flow.getParentResults(job.data.__flowParentIds);
  }
  return { processed: true };
}, runtimeOptions);

getParentResult / getParentResults are helpers of the Bun package and the TypeScript SDK, for both embedded and TCP runtimes. Await them in transport-neutral code: embedded keeps the historical synchronous return, while TCP performs GetResult round trips. The other SDKs read a parent with getResult on the queue (get_result in Python, Rust and Elixir, GetResult in Go).

Injected fields: __flowParentId, __flowParentIds, plus the BullMQ-compatible __parentId, __parentQueue, and __childrenIds. A non-root addChain or addTree step receives __flowParentId, __parentId, and __parentQueue for its exact predecessor, including the predecessor’s real queue in cross-queue flows. Fan-in jobs retain their existing parent/children fields. They are typed via the FlowJobData interface, exposed by Worker and Queue reads, persisted in the selected SQLite or PostgreSQL backend, and survive restarts when persistence is configured.

Keys beginning with __ are reserved on flow input. job.updateData(userData) preserves the engine-owned topology fields and rejects attempts to forge them; the user payload, including a user name key, remains separate from the job’s own name.

Set per-job and per-queue options

Section titled “Set per-job and per-queue options”

Each step accepts normal job options via opts. With flow.add(), you can also set defaults for every job on a given queue:

await flow.add(
  {
    name: 'report',
    queueName: 'reports',
    children: [
      { name: 'fetch', queueName: 'api', data: {}, opts: { priority: 10 } },
      { name: 'render', queueName: 'cpu', data: {} },
    ],
  },
  {
    queuesOptions: {
      api: { attempts: 5, backoff: 2000 },  // defaults for all 'api' jobs
      cpu: { timeout: 60000 },              // defaults for all 'cpu' jobs
    },
  }
);

Per-queue defaults (queuesOptions) are supported in the Bun package, the TypeScript SDK and the Python SDK; in the other SDKs set per-step opts on each node instead (available everywhere).

Per-job opts override queuesOptions defaults. Identity is the exception: set jobId (or Python job_id) in the individual node’s opts. It is rejected inside queuesOptions / queues_options, where one default could otherwise assign the same ID to multiple nodes. Note that delay on a chained step sets its earliest run time, but the step still waits for its dependency to complete first.

The Bun producer also carries group: { id, priority?, maxSize? } from each node into the atomic graph. Group priority uses 0 first and then ascending values. If a maxSize capacity check fails, no node in that add() or addBulk() transaction is admitted. PostgreSQL serializes the capacity check across brokers; SQLite performs it in the same local admission transaction.

Inspect an existing graph

Section titled “Inspect an existing graph”
const tree = await flow.getFlow({
  id: node.job.id,
  queueName: 'reports',
  depth: 2,
  maxChildren: 10,
});


if (!tree) console.log('root not found, or queue name did not match');

depth and maxChildren accept non-negative integers (or Infinity); maxChildren: 0 returns only the requested node. A missing root or queue mismatch returns null. A missing descendant, malformed topology, cycle, or real TCP/server error throws instead of returning a misleading partial tree. Each returned descendant must also point back to the parent being traversed; corrupt or cross-linked ownership is rejected. A child that no longer exists, such as a completed removeOnComplete child or one cancelled by removeUnprocessedChildren, counts as a missing descendant: getFlow throws Flow child job <id> not found, so bound depth above it. The Python, PHP and Go SDKs expose getFlow(jobId, depth) without queueName or maxChildren. Dependency keys returned by getDependencies() use each child’s actual queue, so cross-queue keys are childQueue:childId.

Where to go next

Section titled “Where to go next”
Flow ProducerYour first parent/child graph, and what is guaranteed
Flow Failure HandlingWhat a parent does when a child dies for good
Flow Producer ReferenceEvery producer method, job helper and step field