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

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: {} },
]);

// 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: {} },
]);

# fetch -> process -> store
job_ids = flow.add_chain([
    {"name": "fetch", "queueName": "pipeline", "data": {"url": "https://api.example.com"}},
    {"name": "process", "queueName": "pipeline", "data": {}},
    {"name": "store", "queueName": "pipeline", "data": {}},
])

// fetch -> process -> store
$jobIds = $flow->addChain([
    ['name' => 'fetch', 'queueName' => 'pipeline', 'data' => ['url' => 'https://api.example.com']],
    ['name' => 'process', 'queueName' => 'pipeline'],
    ['name' => 'store', 'queueName' => 'pipeline'],
]);

// fetch -> process -> store
ids, err := flow.AddChain([]bunqueue.ChainStep{
    {Name: "fetch", QueueName: "pipeline", Data: map[string]any{"url": "https://api.example.com"}},
    {Name: "process", QueueName: "pipeline"},
    {Name: "store", QueueName: "pipeline"},
})

use bunqueue_client::{ChainStep, JobOptions, Value};

// fetch -> process -> store
let step = |name: &str| ChainStep {
    name: name.into(),
    queue_name: "pipeline".into(),
    data: Value::Nil,
    options: JobOptions::default(),
};
let ids = flow.add_chain(vec![step("fetch"), step("process"), step("store")])?;

# fetch -> process -> store
{:ok, ids} =
  Bunqueue.FlowProducer.add_chain(flow, [
    %{name: "fetch", queue: "pipeline", data: %{url: "https://api.example.com"}},
    %{name: "process", queue: "pipeline"},
    %{name: "store", queue: "pipeline"}
  ])

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: {} }
);

//   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: {} }
);

#   fetch-api-1 --+
#   fetch-api-2 --+--> merge-results
#   fetch-api-3 --+
result = flow.add_bulk_then(
    [
        {"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": {}},
)
parallel_ids, final_id = result["parallel_ids"], result["final_id"]

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

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 only for now. Every SDK’s add builds the inverse tree, where children complete before their parent (Quick Start).

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 Bun-package helpers 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 external SDKs can read a parent with getResult(parentId) (get_result in Python).

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 to SQLite, and survive restarts.

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

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

Bun
Python

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

flow.add(
    {
        "name": "report",
        "queueName": "reports",
        "children": [
            {"name": "fetch", "queueName": "api", "data": {}, "opts": {"priority": 10}},
            {"name": "render", "queueName": "cpu", "data": {}},
        ],
    },
    {
        "queues_options": {
            "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 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.

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. Dependency keys returned by getDependencies() use each child’s actual queue, so cross-queue keys are childQueue:childId.

Where to go next


Flow Producer	Your first parent/child graph, and what is guaranteed
Flow Failure Handling	What a parent does when a child dies for good
Flow Producer Reference	Every producer method, job helper and step field