Flows

A flow is a tree of jobs where a parent waits for its children. FlowProducer creates the whole graph in one call; NextMQ runs each node as an ordinary job.

Creating a flow#

jobs/checkout.ts

import { FlowProducer } from '@nextmq/sdk'

const flow = new FlowProducer()

await flow.add({
  name: 'finalize-order',
  queueName: 'orders',
  data: { orderId },
  children: [
    { name: 'charge', queueName: 'payments', data: { orderId } },
    { name: 'reserve-stock', queueName: 'inventory', data: { orderId } },
  ],
})

The parent finalize-orderonly runs once both children complete. Children run in parallel, subject to each queue's concurrency and rate limits.

orders:finalize-order
├─ payments:charge
└─ inventory:reserve-stock

Reading child results#

In the parent processor, gather the children's return values with job.getChildrenValues().

export const orderWorker = new Worker('orders', async (job) => {
  const children = await job.getChildrenValues()
  // keyed by each child's job key → its return value:
  // { 'payments:42': { receiptId }, 'inventory:43': { reservationId } }
  return { finalized: true, children }
})

Failure semantics#

Control how a failing child affects its parent with these flags, passed in a child's opts:

Flag	Effect when the child fails
`failParentOnFailure`	Fail the parent immediately.
`continueParentOnFailure`	Let the parent proceed even though this child failed.
`ignoreDependencyOnFailure`	Don't block the parent on this child's result.
`removeDependencyOnFailure`	Drop this child from the parent's dependency set.

await flow.add({
  name: 'finalize-order',
  queueName: 'orders',
  data: { orderId },
  children: [
    {
      name: 'charge',
      queueName: 'payments',
      data: { orderId },
      opts: { attempts: 3, failParentOnFailure: true },
    },
    {
      name: 'send-analytics',
      queueName: 'analytics',
      data: { orderId },
      opts: { continueParentOnFailure: true },
    },
  ],
})

Inspecting a flow#

const tree = await flow.getFlow({ id: parentJobId, queueName: 'orders' })
// addBulk() creates multiple independent flows in one call:
await flow.addBulk([flowA, flowB])

The FlowProducer methods and child options are listed in the SDK reference.

PreviousSchedulers & repeatables NextLocal development