Workflow Pipelines

Table of Contents

• FlowProducer - Parent-Child Job Trees
• Reading Child Results
• DAG Workflows - Multiple Parents
• moveToWaitingChildren - Dynamic Children
• chain - Sequential Pipeline
• group - Parallel Execution
• chord - Parallel + Callback
• Suspend/Resume as a Workflow Primitive
• Budget on Flows
• AI Workflow Patterns
• Broadcast

---

FlowProducer

FlowProducer enqueues a tree of parent and child jobs. Each level (a parent and its leaf children) is created in one atomic call. Nested sub-flows are separate calls made bottom-up, and in cluster mode leaf children in another queue are created and wired to the parent separately, so a failure part way can leave the lower levels created. A parent job only becomes runnable once all of its children have successfully completed; failed or dead-lettered children do not unblock the parent.

import { FlowProducer } from 'glide-mq';

const flow = new FlowProducer({ connection });

const { job: parent } = await flow.add({
  name: 'aggregate',
  queueName: 'reports',
  data: { month: '2025-01' },
  children: [
    { name: 'fetch-sales', queueName: 'data', data: { region: 'eu' } },
    { name: 'fetch-returns', queueName: 'data', data: { region: 'eu' } },
    {
      name: 'fetch-inventory',
      queueName: 'data',
      data: {},
      // Nested: child can itself have children
      children: [
        { name: 'load-warehouse-a', queueName: 'data', data: {} },
        { name: 'load-warehouse-b', queueName: 'data', data: {} },
      ],
    },
  ],
});

console.log('Parent job ID:', parent.id);

await flow.close();

Bulk flows

const nodes = await flow.addBulk([
  {
    name: 'report-jan',
    queueName: 'reports',
    data: {},
    children: [{ name: 'data-jan', queueName: 'data', data: {} }],
  },
  {
    name: 'report-feb',
    queueName: 'reports',
    data: {},
    children: [{ name: 'data-feb', queueName: 'data', data: {} }],
  },
]);

---

Reading Child Results

In the parent processor, call job.getChildrenValues() to retrieve the return values of all direct children. The keys are internal dependency identifiers (implementation detail - prefer Object.values() when you only need the results).

const worker = new Worker(
  'reports',
  async (job) => {
    // Runs only after all children have completed
    const childValues = await job.getChildrenValues();
    // Keys are opaque internal identifiers; use Object.values() for the results:
    const results = Object.values(childValues);
    // [ { sales: 42000 }, { returns: 300 } ]

    const totalSales = results.reduce((s, v) => s + (v.sales ?? 0), 0);
    return { totalSales };
  },
  { connection },
);

---

DAG Workflows - Multiple Parents

FlowProducer.addDAG() lets you define arbitrary DAG (Directed Acyclic Graph) topologies where any job can have multiple parent dependencies. A job only becomes runnable once all of its dependencies have successfully completed.

Use cases

• Fan-in merge: Multiple parallel data sources converge into a single aggregation job
• Diamond dependencies: Job D depends on both B and C, which both depend on A
• Multi-stage pipelines: Complex workflows where certain jobs must wait for multiple upstream tasks

API

import { FlowProducer, dag } from 'glide-mq';

const flow = new FlowProducer({ connection });

// Submit a DAG using the helper function
const jobs = await dag(
  [
    { name: 'A', queueName: 'tasks', data: { step: 1 } },
    { name: 'B', queueName: 'tasks', data: { step: 2 }, deps: ['A'] },
    { name: 'C', queueName: 'tasks', data: { step: 3 }, deps: ['A'] },
    { name: 'D', queueName: 'tasks', data: { step: 4 }, deps: ['B', 'C'] },
  ],
  connection,
);

// Or use FlowProducer.addDAG() directly
const jobs = await flow.addDAG({
  nodes: [
    { name: 'A', queueName: 'tasks', data: { step: 1 } },
    { name: 'B', queueName: 'tasks', data: { step: 2 }, deps: ['A'] },
    { name: 'C', queueName: 'tasks', data: { step: 3 }, deps: ['A'] },
    { name: 'D', queueName: 'tasks', data: { step: 4 }, deps: ['B', 'C'] },
  ],
});
// Returns Map<string, Job> keyed by node name

Each DAGNode has:

• name - unique identifier within this DAG (used in deps arrays)
• queueName - queue to submit this job to
• data - job payload
• opts? - job options (delay, priority, attempts, etc.)
• deps? - array of node names that must complete before this job runs

Example: Fan-in merge

import { dag } from 'glide-mq';

// Three parallel data fetches, then one merge job
const jobs = await dag(
  [
    { name: 'fetch-sales', queueName: 'data', data: { source: 'sales-db' } },
    { name: 'fetch-inventory', queueName: 'data', data: { source: 'warehouse-db' } },
    { name: 'fetch-returns', queueName: 'data', data: { source: 'returns-db' } },
    {
      name: 'merge-reports',
      queueName: 'data',
      data: { reportId: 'Q1-2025' },
      deps: ['fetch-sales', 'fetch-inventory', 'fetch-returns'],
    },
  ],
  connection,
);

// All three fetches run in parallel.
// 'merge-reports' runs only after all three complete.

Example: Diamond dependency

import { dag } from 'glide-mq';

// Job topology:
//       A
//      / \
//     B   C
//      \ /
//       D

const jobs = await dag(
  [
    { name: 'A', queueName: 'tasks', data: { step: 'root' } },
    { name: 'B', queueName: 'tasks', data: { step: 'left' }, deps: ['A'] },
    { name: 'C', queueName: 'tasks', data: { step: 'right' }, deps: ['A'] },
    { name: 'D', queueName: 'tasks', data: { step: 'converge' }, deps: ['B', 'C'] },
  ],
  connection,
);

// A runs first, then B and C in parallel, then D after both complete.

Implementation notes:

• DAG validation runs automatically - cycles are detected and rejected with CycleError.
• Jobs are submitted level by level in reverse-topological order (dependents first, prerequisites last) so each node's BullMQ-parents already exist by the time we wire it; all jobs within a level are pipelined in a single batch, so submission cost is O(levels) round trips rather than O(N). Before the leaf level is added, each leaf is registered in all of its dependents' deps sets, so a parent is never released by a fast sibling while another of its deps is still being wired.
• If any parent fails or is dead-lettered, dependent jobs remain blocked indefinitely (manual cleanup required).
• Cross-queue dependencies are supported - each node can specify its own queueName. A child that completes before its registration reaches the parent is parked on the parent (depearly) and counted once the registration lands. A producer on a library before 126 registers with a plain SADD, which cannot count a parked completion; the scheduler tick's glidemq_healEarlyDeps counts it and releases the parent, so a mixed-version rollout does not leave a parent waiting for children that already finished.

Reading results from multiple parents

Use job.getParents() to list all parent references, then fetch each parent job to read its result:

const worker = new Worker(
  'tasks',
  async (job) => {
    if (job.name === 'D') {
      const parentRefs = await job.getParents();
      // parentRefs: Array<{ queue: string; id: string }>
      // Fetch each parent job to read its return value:
      const results = [];
      for (const ref of parentRefs) {
        const parentJob = await queue.getJob(ref.id);
        if (parentJob) results.push(parentJob.returnvalue);
      }
      return { merged: results };
    }
  },
  { connection },
);

---

moveToWaitingChildren - Dynamic Children

FlowProducer and addDAG() define the job graph up front before any processing begins. Sometimes a parent processor needs to spawn children dynamically based on runtime data - for example, splitting a file into N chunks where N is unknown until the file is read.

job.moveToWaitingChildren() handles this. It pauses the parent job (transitions it back to waiting-children) until all dynamically-added children complete. When the last child finishes, the parent processor re-executes from the top.

How it works

1. The parent processor runs and decides it needs child jobs.
2. It creates children via queue.add() (or FlowProducer) with a parent option pointing back to the current job.
3. It calls await job.moveToWaitingChildren().
4. This throws a WaitingChildrenError internally - the worker framework catches it and moves the parent to waiting-children state.
5. When all children complete, the parent processor is invoked again from the top.
6. On re-entry, call job.getChildrenValues() to collect results and return the final value.

Example: dynamic fan-out

import { Queue, Worker } from 'glide-mq';

const connection = { addresses: [{ host: 'localhost', port: 6379 }] };
const queue = new Queue('processing', { connection });

const worker = new Worker(
  'processing',
  async (job) => {
    // Check if children have already completed (re-entry after waiting)
    const existing = await job.getChildrenValues();
    if (Object.keys(existing).length > 0) {
      // All children done  -  aggregate and return
      const results = Object.values(existing);
      return { total: results.reduce((sum, r) => sum + r.count, 0) };
    }

    // First execution: inspect data and spawn children dynamically
    const { urls } = job.data;

    for (const url of urls) {
      await queue.add(
        'fetch-url',
        { url },
        {
          parent: { id: job.id!, queue: 'processing' }, // plain queue name of the parent
        },
      );
    }

    // Pause until all children complete  -  throws WaitingChildrenError
    await job.moveToWaitingChildren();
  },
  { connection },
);

Key points

• moveToWaitingChildren() always throws (WaitingChildrenError). Do not put code after it - it will not execute.
• The processor re-runs from the top when children complete. Use getChildrenValues() or job.data to detect re-entry.
• You can call moveToWaitingChildren() multiple times across re-entries to create multi-round fan-out patterns.
• Children must reference the parent via opts.parent: { id, queue } so the dependency tracking works.

---

chain

Execute a list of jobs sequentially, specified in reverse execution order (the last element in the array runs first). Each step can read the previous step's result via getChildrenValues().

import { chain } from 'glide-mq';

// Execution order: download → parse → transform → upload
await chain(
  'pipeline',
  [
    { name: 'upload', data: { bucket: 'my-bucket' } }, // runs last  (root)
    { name: 'transform', data: {} },
    { name: 'parse', data: {} },
    { name: 'download', data: { url: 'https://example.com/file.csv' } }, // runs first (leaf)
  ],
  connection,
);

// Pass a shared `client` when you need the returned jobs (getState, waitUntilFinished).
// Without it, chain/group/chord/dag close their owned connection after submit.

• The last element in the array is the leaf - it runs first.
• The first element in the array is the root - it runs last (after all descendants complete).
• Each step's processor can access the prior step's return value via Object.values(job.getChildrenValues())[0].

const worker = new Worker(
  'pipeline',
  async (job) => {
    if (job.name === 'parse') {
      const prev = await job.getChildrenValues();
      const raw = Object.values(prev)[0]; // result from 'download'
      return parse(raw);
    }
    // ...
  },
  { connection },
);

---

group

Execute a list of jobs in parallel. A synthetic __group__ parent waits for all children to complete.

import { group } from 'glide-mq';

await group(
  'tasks',
  [
    { name: 'resize-thumb', data: { imageId: 1, size: 'sm' } },
    { name: 'resize-medium', data: { imageId: 1, size: 'md' } },
    { name: 'resize-large', data: { imageId: 1, size: 'lg' } },
  ],
  connection,
);

The __group__ parent processor (if you define one) can collect results from all children via getChildrenValues().

---

chord

Run a group of jobs in parallel, then execute a callback job once all group members are done. The callback receives the group results.

import { chord } from 'glide-mq';

await chord(
  'tasks',
  // Group (runs in parallel)
  [
    { name: 'score-model-a', data: { modelId: 'a' } },
    { name: 'score-model-b', data: { modelId: 'b' } },
    { name: 'score-model-c', data: { modelId: 'c' } },
  ],
  // Callback (runs after all group members complete)
  { name: 'select-best-model', data: {} },
  connection,
);

In the callback processor:

const worker = new Worker(
  'tasks',
  async (job) => {
    if (job.name === 'select-best-model') {
      const scores = await job.getChildrenValues();
      // Keys are opaque  -  use Object.entries() if you need them, or Object.values():
      const best = Object.entries(scores).sort((a, b) => b[1].score - a[1].score)[0];
      return { score: best[1].score };
    }
    // ... other processors
  },
  { connection },
);

---

---

Suspend/Resume as a Workflow Primitive

job.suspend() can be used within any workflow pattern (chain, group, chord, DAG) to introduce external wait points. This is fundamentally different from moveToDelayed (timer-based) and moveToWaitingChildren (child-completion-based) - suspend waits for an explicit signal from outside the queue system.

Use cases

• Approval gates: A content pipeline generates a draft, suspends for review, then publishes on approval.
• Webhook callbacks: An order flow suspends after sending a payment request, resumes when the webhook arrives.
• Agent loops: An AI agent suspends after presenting options to the user, resumes with the user's choice.

In a DAG, suspending a node does not block sibling branches. Other branches with satisfied dependencies continue executing. The suspended node resumes only when queue.signal() is called.

---

Budget on Flows

FlowProducer.add() accepts an optional budget parameter that creates a shared budget hash for the entire flow. Every job in the flow (parent and children) shares this budget.

Each child job has a budgetKey that points to the shared budget hash. The usage a job reports with reportUsage() is charged when the attempt ends, whether it completes or fails: the worker increments the budget counters via glidemq_recordUsageAndCheckBudget. A retry that reports no new usage is not charged again. Batch workers charge and check budgets the same way, per job. Before each job runs, the worker checks the budget. If limits are exceeded:

• fail: The job that crossed the limit completes normally. Each later job fails with Budget exceeded when it starts (normal retry rules apply).
• pause: Each later job is moved back to delayed when it starts and re-checks the budget every 60 seconds (Worker.BUDGET_PAUSE_RECHECK_MS) until the limits are raised. Raise them with queue.updateFlowBudget(flowId, { maxTotalCost: 2 }): it re-evaluates exceeded against the usage already charged, and the paused jobs run at their next re-check, or right away after job.promote().

---

AI Workflow Patterns

RAG pipeline (chain)

Use chain() to model embed -> retrieve -> generate as a sequential pipeline with budget caps.

Agent loop (suspend/resume cycle)

The processor checks job.signals on re-entry. If empty, it suspends and waits for user input. When the signal arrives, the processor continues with the user's response.

Content pipeline with approval (chain + suspend)

Combine chain() with job.suspend() at the review step. The pipeline halts until an editor approves.

Parallel model comparison (group + budget)

Use FlowProducer.add() with multiple child jobs (one per model) and a shared budget to compare model outputs while enforcing cost limits across all calls.

---

Broadcast

The workflow patterns above (FlowProducer, DAG, chain, group, chord, moveToWaitingChildren) all model dependency graphs - jobs wait for other jobs to complete before running.

glide-mq also supports a Broadcast / BroadcastWorker pub/sub pattern for real-time fan-out where every subscriber receives every message. This is a fundamentally different paradigm: no dependencies between messages - just fire-and-forget delivery to all connected workers.

See USAGE.md for the Broadcast and BroadcastWorker API.