reference · examples

Code examples you copy and ship.

Short recipes for the tasks you hit first: retries, schedules, dedup, events, shutdown and workflows. Each one links to the guide that covers it in depth.

This page starts with one local job, adds reliability and operational controls, then finishes with workflows and a tested PostgreSQL multi-broker deployment. For domain scenarios such as email, webhooks, and payments, see use cases.

Learning path

Follow the stages in order on a first read. Each stage links directly to the relevant recipe, so you can return later and use the page as a reference.

1. 1
   Run one job

   Start with one queue and one worker in the same process.

2. 2
   Make failure predictable

   Follow the lifecycle, then add retries, schedules, and idempotency.

   • Retries
   • Schedules
   • Deduplication
3. 3
   Split into processes

   Move to TCP, observe events, and shut workers down safely.

4. 4
   Orchestrate workflows

   Coordinate compensation and human decisions after queue fundamentals are clear.

5. 5
   Scale to multiple brokers

   Finish with the tested PostgreSQL topology and its engineering report.

Persistence

The Bun tabs use embedded mode, meaning the queue runs inside your process with no separate server. Pass dataPath so jobs are saved to a SQLite file, otherwise everything is in-memory and lost on restart:

const queue = new Queue('tasks', { embedded: true, dataPath: './data/bunq.db' });

The other languages connect to a bunqueue server (default localhost:6789), where persistence is configured server-side with --data-path.

Minimal queue and worker

The smallest complete setup: add a job, process it in the background.

Bun

import { Queue, Worker } from 'bunqueue/client';

const queue = new Queue('tasks', { embedded: true, dataPath: './data/bunq.db' });

const worker = new Worker(
  'tasks',
  async (job) => {
    console.log('processing', job.data);
    return { done: true };
  },
  { embedded: true, concurrency: 5 }
);

await queue.add('hello', { message: 'world' });

Node.js / Deno

import { Queue, Worker } from 'bunqueue-client';

const queue = new Queue('tasks'); // connects to localhost:6789

const worker = new Worker(
  'tasks',
  async (job) => {
    console.log('processing', job.data);
    return { done: true };
  },
  { concurrency: 5 }
);

await queue.add('hello', { message: 'world' });

Python

from bunqueue import Queue, Worker

queue = Queue("tasks")  # connects to localhost:6789

def process(job):
    print("processing", job.data)
    return {"done": True}

worker = Worker("tasks", process, concurrency=5)

queue.add("hello", {"message": "world"})
worker.run()  # blocks while the background loop processes jobs

PHP

use Bunqueue\Queue;
use Bunqueue\Worker;

$queue = new Queue('tasks'); // connects to localhost:6789
$queue->add('hello', ['message' => 'world']);

$worker = new Worker('tasks', function (Bunqueue\Job $job) {
    print_r($job->data());
    return ['done' => true];
});
$worker->run(); // blocking loop

Go

queue := bunqueue.NewQueue("tasks", bunqueue.Options{}) // localhost:6789
defer queue.Close()

queue.Add("hello", map[string]any{"message": "world"}, nil)

worker := bunqueue.NewWorker("tasks", func(job *bunqueue.Job) (any, error) {
    fmt.Println("processing", job.Data())
    return map[string]any{"done": true}, nil
}, bunqueue.WorkerOptions{Concurrency: 5})

worker.Run() // blocking pull loop

Rust

use bunqueue_client::{ConnectionOptions, JobOptions, Queue, Value, Worker, WorkerOptions};

let queue = Queue::new("tasks", ConnectionOptions::default()); // localhost:6789
let data = Value::Map(vec![(Value::from("message"), Value::from("world"))]);
queue.add("hello", data, JobOptions::default())?;

let worker = Worker::new(
    "tasks",
    |job| {
        println!("processing {:?}", job.data());
        Ok(Value::from(true))
    },
    WorkerOptions { concurrency: 5, ..Default::default() },
);
worker.run()?;

Elixir

queue = Bunqueue.queue("tasks")  # connects to localhost:6789
{:ok, _job} = Bunqueue.Queue.add(queue, "hello", %{message: "world"})

worker =
  Bunqueue.Worker.new("tasks", fn job ->
    IO.inspect(job.data, label: "processing")
    {:ok, %{done: true}}
  end, concurrency: 5)

Bunqueue.Worker.run(worker)

More in the quickstart.

Understand the job lifecycle

Every job starts with a producer, waits until it is eligible, and is claimed by one worker. A successful acknowledgement completes it. A failure either schedules another attempt after backoff or moves the job to the dead letter queue when no attempt remains.

Use the controls to compare the success, retry, and terminal-failure routes one transition at a time. The same state rules apply in embedded, SQLite, and PostgreSQL deployments.

Pick an outcome, then step through the job's states.

Succeeds first time Fails once, then succeeds Exhausts every attempt

1. 1 Producer waiting
2. 2 Ready queue waiting
3. 3 Worker active
4. 4 Completed completed

Succeeds first time. Step 1 of 4: Producer, state waiting. queue.add() persists the job and returns its ID.

Producer waiting queue.add() persists the job and returns its ID.

Ready queue waiting The job is eligible and waits in scheduling order.

Worker active One worker claims the job and owns its active attempt.

Completed completed The ACK saves the result and releases the concurrency slot.

Producer waiting queue.add() persists the job with attempts and backoff.

Ready queue waiting The first attempt becomes eligible for a worker.

Attempt 1 active ✕ fails The worker throws, so bunqueue records a failed attempt.

Retry delay delayed Backoff keeps the job ineligible until its retry time.

Ready again waiting Once the backoff elapses, the job is eligible again in scheduling order.

Attempt 2 active A worker claims the next legal attempt.

Completed completed The successful ACK stores the result exactly once.

Producer waiting queue.add() persists the job with a finite attempt budget.

Ready queue waiting The job becomes eligible for its first attempt.

Attempt 1 active ✕ fails Processing fails and consumes one attempt.

Retry delay delayed Backoff prevents an immediate hot retry.

Final attempt active ✕ fails The worker fails after the remaining attempt is claimed.

Dead letter queue failed The terminal failure stays available for inspection or replay.

← Previous state Step 1 of 4 Next state →

Retries and the dead letter queue

A thrown error retries the job with backoff, a growing delay between attempts. Jobs that run out of attempts land in the dead letter queue (DLQ), a holding area you can inspect and retry.

Bun

await queue.add(
  'flaky-call',
  { url: 'https://api.example.com' },
  {
    attempts: 5, // try up to 5 times
    backoff: 2000, // wait ~4s, ~8s, ~16s... between tries
  }
);

// After all attempts fail:
const failed = queue.getDlq(); // inspect what died and why
queue.retryDlq(); // send everything back for another run

Node.js / Deno

await queue.add(
  'flaky-call',
  { url: 'https://api.example.com' },
  {
    attempts: 5, // try up to 5 times
    backoff: 2000, // wait ~4s, ~8s, ~16s... between tries
  }
);

// After all attempts fail:
const failed = await queue.getDlqAsync(); // inspect what died and why
await queue.retryDlqAsync(); // send everything back for another run

Python

queue.add("flaky-call", {"url": "https://api.example.com"},
          attempts=5,    # try up to 5 times
          backoff=2000)  # wait ~4s, ~8s, ~16s... between tries

# After all attempts fail:
failed = queue.get_dlq()   # inspect what died and why
queue.retry_dlq()          # send everything back for another run

PHP

$queue->add('flaky-call', ['url' => 'https://api.example.com'], [
    'attempts' => 5,   // try up to 5 times
    'backoff' => 2000, // wait ~4s, ~8s, ~16s... between tries
]);

// After all attempts fail:
$failed = $queue->getDlq();  // inspect what died and why
$queue->retryDlq();          // send everything back for another run

Go

queue.Add("flaky-call", map[string]any{"url": "https://api.example.com"}, bunqueue.JobOptions{
    "attempts": 5,    // try up to 5 times
    "backoff":  2000, // wait ~4s, ~8s, ~16s... between tries
})

// After all attempts fail:
failed, _ := queue.GetDlq(0)  // inspect what died and why (0 = server default count)
queue.RetryDlq("", 0)         // send everything back for another run

Rust

use bunqueue_client::{Backoff, JobOptions};

queue.add("flaky-call", data, JobOptions {
    attempts: Some(5),                          // try up to 5 times
    backoff: Some(Backoff::Milliseconds(2000)), // wait ~4s, ~8s, ~16s... between tries
    ..Default::default()
})?;

// After all attempts fail:
let failed = queue.get_dlq(None)?;  // inspect what died and why
queue.retry_dlq(None, None)?;       // send everything back for another run

Elixir

{:ok, _job} =
  Bunqueue.Queue.add(queue, "flaky-call", %{url: "https://api.example.com"},
    attempts: 5,   # try up to 5 times
    backoff: 2000  # wait ~4s, ~8s, ~16s... between tries
  )

# After all attempts fail:
{:ok, failed} = Bunqueue.Queue.dlq(queue)        # inspect what died and why
{:ok, _count} = Bunqueue.Queue.retry_dlq(queue)  # send everything back for another run

Details and auto-retry config in the DLQ guide.

Scheduled and repeating jobs

Attach a repeat option, or use upsertJobScheduler() for named schedules. Both persist in the selected durable backend and survive restarts; PostgreSQL mode coordinates named schedules across brokers.

Bun

// Cron expression: every day at 6 AM
await queue.add(
  'daily-report',
  { type: 'sales' },
  {
    repeat: { pattern: '0 6 * * *' },
  }
);

// Plain interval: every 30 minutes
await queue.add(
  'health-check',
  {},
  {
    repeat: { every: 1_800_000 },
  }
);

// Named, updatable schedule
await queue.upsertJobScheduler(
  'cleanup',
  { pattern: '0 3 * * *' },
  {
    data: { olderThanDays: 30 },
  }
);

Node.js / Deno

// Cron expression: every day at 6 AM
await queue.add(
  'daily-report',
  { type: 'sales' },
  {
    repeat: { pattern: '0 6 * * *' },
  }
);

// Plain interval: every 30 minutes
await queue.add(
  'health-check',
  {},
  {
    repeat: { every: 1_800_000 },
  }
);

// Named, updatable schedule
await queue.upsertJobScheduler(
  'cleanup',
  { pattern: '0 3 * * *' },
  {
    data: { olderThanDays: 30 },
  }
);

Python

# Cron expression: every day at 6 AM
queue.add("daily-report", {"type": "sales"}, repeat={"pattern": "0 6 * * *"})

# Plain interval: every 30 minutes
queue.add("health-check", {}, repeat={"every": 1_800_000})

# Named, updatable schedule
queue.upsert_job_scheduler("cleanup", {"pattern": "0 3 * * *"},
                           {"data": {"olderThanDays": 30}})

PHP

// Cron expression: every day at 6 AM
$queue->add('daily-report', ['type' => 'sales'], ['repeat' => ['pattern' => '0 6 * * *']]);

// Plain interval: every 30 minutes
$queue->add('health-check', [], ['repeat' => ['every' => 1800000]]);

// Named, updatable schedule
$queue->upsertJobScheduler('cleanup',
    ['pattern' => '0 3 * * *'],
    ['data' => ['olderThanDays' => 30]],
);

Go

// Cron expression: every day at 6 AM
queue.Add("daily-report", map[string]any{"type": "sales"},
    bunqueue.JobOptions{"repeat": map[string]any{"pattern": "0 6 * * *"}})

// Plain interval: every 30 minutes
queue.Add("health-check", nil,
    bunqueue.JobOptions{"repeat": map[string]any{"every": 1800000}})

// Named, updatable schedule
queue.UpsertJobScheduler("cleanup",
    bunqueue.SchedulerRepeat{Pattern: "0 3 * * *"},
    bunqueue.SchedulerTemplate{Data: map[string]any{"olderThanDays": 30}},
)

Rust

use bunqueue_client::{JobOptions, SchedulerRepeat, SchedulerTemplate, Value};

// Cron expression: every day at 6 AM
let repeat = Value::Map(vec![(Value::from("pattern"), Value::from("0 6 * * *"))]);
queue.add("daily-report", data, JobOptions { repeat: Some(repeat), ..Default::default() })?;

// Plain interval: every 30 minutes
let repeat = Value::Map(vec![(Value::from("every"), Value::from(1_800_000))]);
queue.add("health-check", Value::Nil, JobOptions { repeat: Some(repeat), ..Default::default() })?;

// Named, updatable schedule
queue.upsert_job_scheduler(
    "cleanup",
    SchedulerRepeat { pattern: Some("0 3 * * *".into()), ..Default::default() },
    SchedulerTemplate {
        data: Value::Map(vec![(Value::from("olderThanDays"), Value::from(30))]),
        ..Default::default()
    },
)?;

Elixir

# Cron expression: every day at 6 AM
{:ok, _} =
  Bunqueue.Queue.add(queue, "daily-report", %{type: "sales"},
    repeat: %{pattern: "0 6 * * *"}
  )

# Plain interval: every 30 minutes
{:ok, _} = Bunqueue.Queue.add(queue, "health-check", %{}, repeat: %{every: 1_800_000})

# Named, updatable schedule
:ok =
  Bunqueue.Queue.upsert_scheduler(queue, "cleanup",
    %{pattern: "0 3 * * *"},
    %{data: %{olderThanDays: 30}}
  )

Timezones and schedule management in the cron guide.

Deduplicate jobs with jobId

Adding a job with the jobId of an unfinished job (waiting, delayed, waiting for children, or active) returns the existing job instead of creating a duplicate, so re-runs after a restart do not queue the same work twice. Once that job has completed or moved to the DLQ, the same jobId creates a fresh job, so it does not by itself guarantee “exactly one welcome email per user”.

Bun

const a = await queue.add('notify', { userId: 'u1' }, { jobId: 'welcome-u1' });
const b = await queue.add('notify', { userId: 'u1' }, { jobId: 'welcome-u1' });

console.log(a.id === b.id); // true, same job

Node.js / Deno

const a = await queue.add('notify', { userId: 'u1' }, { jobId: 'welcome-u1' });
const b = await queue.add('notify', { userId: 'u1' }, { jobId: 'welcome-u1' });

console.log(a.id === b.id); // true, same job

Python

a = queue.add("notify", {"user_id": "u1"}, job_id="welcome-u1")
b = queue.add("notify", {"user_id": "u1"}, job_id="welcome-u1")

print(a.id == b.id)  # True, same job

PHP

$a = $queue->add('notify', ['userId' => 'u1'], ['jobId' => 'welcome-u1']);
$b = $queue->add('notify', ['userId' => 'u1'], ['jobId' => 'welcome-u1']);

var_dump($a->id() === $b->id()); // true, same job

Go

a, _ := queue.Add("notify", map[string]any{"userId": "u1"}, bunqueue.JobOptions{"jobId": "welcome-u1"})
b, _ := queue.Add("notify", map[string]any{"userId": "u1"}, bunqueue.JobOptions{"jobId": "welcome-u1"})

fmt.Println(a.ID() == b.ID()) // true, same job

Rust

let opts = || JobOptions { job_id: Some("welcome-u1".into()), ..Default::default() };
let a = queue.add("notify", data.clone(), opts())?;
let b = queue.add("notify", data, opts())?;

assert_eq!(a.id(), b.id()); // same job

Elixir

{:ok, a} = Bunqueue.Queue.add(queue, "notify", %{user_id: "u1"}, jobId: "welcome-u1")
{:ok, b} = Bunqueue.Queue.add(queue, "notify", %{user_id: "u1"}, jobId: "welcome-u1")

a.id == b.id  # true, same job

Choose a deployment topology

Start embedded while one process is the right boundary. Introduce a TCP broker when producers and workers need separate processes or different languages. Add PostgreSQL and multiple active brokers only when broker failover, horizontal scale, or shared cross-host limits justify the extra moving parts.

Pick a topology. The Queue and Worker API stays familiar.

Embedded TCP broker PostgreSQL multi-broker

The producer, queue runtime, and worker share one Bun process.

Bun application

ProducerQueue runtimeWorker

direct calls

Optional persistence

SQLite

Best fit

The smallest deployment, local services, and edge processes.

Durability owner

Memory or one local SQLite file.

Embedded: The producer, queue runtime, and worker share one Bun process.

Distributed mode (server + TCP)

Run one bunqueue server, connect producers and workers from any number of processes or machines, in any language.

bunqueue start --tcp-port 6789 --data-path ./data/tasks.db

Bun

producer.ts

import { Queue } from 'bunqueue/client';
const queue = new Queue('tasks', { connection: { host: 'localhost', port: 6789 } });
await queue.addBulk(items.map((i) => ({ name: 'process', data: i })));

// worker.ts (run as many copies as you want)
import { Worker } from 'bunqueue/client';
new Worker(
  'tasks',
  async (job) => {
    return { processed: job.data.id };
  },
  { connection: { host: 'localhost', port: 6789 }, concurrency: 50 }
);

Node.js / Deno

producer.ts

import { Queue } from 'bunqueue-client';
const queue = new Queue('tasks', { connection: { host: 'localhost', port: 6789 } });
await queue.addBulk(items.map((i) => ({ name: 'process', data: i })));

// worker.ts (run as many copies as you want)
import { Worker } from 'bunqueue-client';
new Worker(
  'tasks',
  async (job) => {
    return { processed: job.data.id };
  },
  { concurrency: 50 }
);

Python

producer.py

from bunqueue import Queue
queue = Queue("tasks", host="localhost", port=6789)
queue.add_bulk([{"name": "process", "data": i} for i in items])

# worker.py (run as many copies as you want)
from bunqueue import Worker
Worker("tasks", lambda job: {"processed": job.data["id"]}, concurrency=50).run()

PHP

producer.php

$queue = new Bunqueue\Queue('tasks', ['host' => 'localhost', 'port' => 6789]);
$queue->addBulk(array_map(fn ($i) => ['name' => 'process', 'data' => $i], $items));

// worker.php (run as many copies as you want)
$worker = new Bunqueue\Worker('tasks', fn (Bunqueue\Job $job) => ['processed' => $job->data()['id']]);
$worker->run();

Go

// producer
queue := bunqueue.NewQueue("tasks", bunqueue.Options{Host: "localhost", Port: 6789})
entries := make([]bunqueue.BulkEntry, 0, len(items))
for _, item := range items {
    entries = append(entries, bunqueue.BulkEntry{Name: "process", Data: item})
}
queue.AddBulk(entries)

// worker (run as many copies as you want)
worker := bunqueue.NewWorker("tasks", func(job *bunqueue.Job) (any, error) {
    return map[string]any{"processed": any(job.Data()).(map[string]any)["id"]}, nil
}, bunqueue.WorkerOptions{Concurrency: 50})
worker.Run()

Rust

use bunqueue_client::{BulkEntry, ConnectionOptions, JobOptions, Queue, Worker, WorkerOptions};

// producer
let queue = Queue::new("tasks", ConnectionOptions::default());
let entries = items
    .into_iter()
    .map(|data| BulkEntry { name: "process".into(), data, options: JobOptions::default() })
    .collect::<Vec<_>>();
let ids = queue.add_bulk(entries)?;

// worker (run as many copies as you want)
let worker = Worker::new("tasks", |job| process(job), WorkerOptions {
    concurrency: 50,
    ..Default::default()
});
worker.run()?;

Elixir

# producer
queue = Bunqueue.queue("tasks", host: "localhost", port: 6789)
{:ok, _ids} =
  Bunqueue.Queue.add_bulk(queue, Enum.map(items, &%{name: "process", data: &1}))

# worker (run as many copies as you want)
worker =
  Bunqueue.Worker.new("tasks", fn job ->
    {:ok, %{processed: job.data["id"]}}
  end, concurrency: 50)

Bunqueue.Worker.run(worker)

Server setup, auth and TLS in the server guide.

Watch job events

QueueEvents streams lifecycle events for a queue, and workers emit their own events.

Bun

import { QueueEvents } from 'bunqueue/client';

const events = new QueueEvents('tasks', {
  connection: { host: '127.0.0.1', port: 6789 },
});
await events.waitUntilReady();

events.on('completed', ({ jobId, returnvalue }) => console.log('done', jobId, returnvalue));
events.on('failed', ({ jobId, failedReason }) => console.error('failed', jobId, failedReason));
events.on('progress', ({ jobId, data }) => console.log('progress', jobId, data));

worker.on('completed', (job, result) => console.log('worker finished', job.id));
worker.on('failed', (job, error) => console.error('worker error', error.message));

Node.js / Deno

// Worker-side events (QueueEvents from 'bunqueue-client' streams queue-wide events too)
worker.on('completed', (job, result) => console.log('worker finished', job.id));
worker.on('failed', (job, error) => console.error('worker error', error.message));
worker.on('error', (err) => console.error(err)); // always attach

Python

worker.on("completed", lambda job, result: print("worker finished", job.id))
worker.on("failed", lambda job, err: print("worker error", job.id, err))
worker.on("progress", lambda job, progress: print("progress", job.id, progress))

PHP

$worker->on('completed', fn ($job, $result) => print("worker finished {$job->id()}\n"));
$worker->on('failed', fn ($job, $err) => print("worker error {$job->id()}\n"));
$worker->on('error', fn ($err) => print($err->getMessage() . "\n"));

Go

worker.On("completed", func(args ...any) {
    job := args[0].(*bunqueue.Job)
    log.Printf("worker finished %s", job.ID())
})
worker.On("error", func(args ...any) { log.Println(args[0]) })

Rust

// Rust has no worker event emitter. Per-job outcomes are the processor's return
// value; transport lifecycle arrives on the connection telemetry callback.
let options = ConnectionOptions {
    telemetry: Some(Arc::new(|event| println!("{event:?}"))),
    ..Default::default()
};

Elixir

# Elixir has no worker event emitter. Per-job outcomes are the handler's return
# value; transport lifecycle arrives on the connection `:event_handler` callback.
queue = Bunqueue.queue("emails", event_handler: &IO.inspect/1)

QueueEvents streaming is available in the Bun bunqueue package and in bunqueue-client (Node.js, Deno); the other SDKs have no QueueEvents stream. See the SDK guide.

Dashboards, metrics and Prometheus in the monitoring guide.

Graceful shutdown

On SIGTERM, stop pulling new jobs, let active ones finish, then close.

Bun

async function shutdown() {
  worker.pause(); // stop accepting new jobs
  await worker.close(); // wait for active jobs (worker.close(true) forces a stop)
  await queue.close();
  process.exit(0);
}

process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

Node.js / Deno

async function shutdown() {
  await worker.close(); // stop pulling, flush batched ACKs, drain in-flight jobs
  queue.close();
  process.exit(0);
}

process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

Python

try:
    worker.run()
except KeyboardInterrupt:
    worker.close()   # wait for in-flight jobs to drain
    queue.close()

PHP

$worker->installSignalHandlers(); // SIGTERM / SIGINT -> graceful stop
$worker->run();                   // returns after the in-flight job finishes
$worker->close();                 // unregister and close the connection

Go

go func() {
    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
    <-sig
    worker.Stop() // stop pulling; in-flight jobs finish
}()
worker.Run()
worker.Close() // unregister and close the connection

Rust

// From a signal handler or another thread:
worker.stop();  // ask the pull loop to exit; run() returns after draining
worker.close(); // unregister and close the connection
queue.close();

Elixir

Bunqueue.Worker.stop(worker)  # drain, unregister and close
Bunqueue.Queue.close(queue)

The full production pattern, including timeouts and the embedded manager, is in the production guide.

Workflow: automatic rollback on failure

The workflow engine runs multi-step processes where each step can declare a compensate function, code that undoes the step if a later one fails. This is the saga pattern: charge succeeded but shipping failed, so the charge is refunded automatically.

The workflow engine ships with the Bun bunqueue package (bunqueue/workflow) and runs embedded. From the other SDKs, use flows for multi-step orchestration against the server.

import { Workflow, Engine } from 'bunqueue/workflow';

const orderFlow = new Workflow('order')
  .step(
    'reserve-stock',
    async (ctx) => {
      await inventory.reserve((ctx.input as { orderId: string }).orderId);
      return { reserved: true };
    },
    {
      compensate: async () => {
        await inventory.release();
      }, // runs if a later step fails
    }
  )
  .step(
    'charge',
    async (ctx) => {
      const txId = await stripe.charge((ctx.input as { amount: number }).amount);
      return { txId };
    },
    {
      compensate: async () => {
        await stripe.refund();
      },
    }
  )
  .step('confirm', async (ctx) => {
    const { txId } = ctx.steps['charge'] as { txId: string };
    await mailer.send('order-confirm', { txId });
    return { done: true };
  });

const engine = new Engine({ embedded: true });
engine.register(orderFlow);
await engine.start('order', { orderId: 'ORD-1', amount: 99.99 });

Workflow: wait for a human decision

waitFor() pauses the workflow until someone calls engine.signal(), hours or days later.

import { Workflow, Engine } from 'bunqueue/workflow';

const expenseFlow = new Workflow('expense')
  .step('submit', async (ctx) => {
    await slack.notify('#approvals', `New expense: ${JSON.stringify(ctx.input)}`);
    return { submitted: true };
  })
  .waitFor('manager-decision')
  .step('process', async (ctx) => {
    const decision = ctx.signals['manager-decision'] as { approved: boolean };
    return { status: decision.approved ? 'paid' : 'rejected' };
  });

const engine = new Engine({ embedded: true });
engine.register(expenseFlow);
const run = await engine.start('expense', { amount: 500 });

// Later, when the manager clicks approve:
await engine.signal(run.id, 'manager-decision', { approved: true });

Branching, parallel steps, loops, sub-workflows and schema validation are all in the workflow guide.

End-to-end example projects

The complete project below combines the earlier concepts. Read it after the single-broker examples if this is your first bunqueue deployment.

PostgreSQL multi-broker

Run PostgreSQL 18.6, three active brokers, multiple queues and workers, authenticated metrics, custom-ID idempotency, retries, DLQ recovery, shared limits, events, and durable flows. Every source is executed in disposable containers and has a published engineering report.

Open the complete example →

Where next

• Use cases, end-to-end patterns for emails, webhooks, images and payments
• Queue guide, every job option explained
• Worker guide, concurrency, heartbeats and batching
• Migration from BullMQ, the API is intentionally close