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 .
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.
01 Run one job Start with one queue and one worker in the same process.
02 03 04 05
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.
The smallest complete setup: add a job, process it in the background.
import { Queue , Worker } from 'bunqueue/client' ;
const queue = new Queue ( 'tasks' , { embedded : true , dataPath : './data/bunq.db' } ) ;
const worker = new Worker (
console . log ( 'processing' , job . data) ;
{ embedded : true , concurrency : 5 }
await queue . add ( 'hello' , { message : 'world' } ) ;
import { Queue , Worker } from 'bunqueue-client' ;
const queue = new Queue ( 'tasks' ) ; // connects to localhost:6789
const worker = new Worker (
console . log ( 'processing' , job . data) ;
await queue . add ( 'hello' , { message : 'world' } ) ;
from bunqueue import Queue , Worker
queue = Queue ( "tasks" ) # connects to localhost:6789
print ( "processing" , job . data )
worker = Worker ( "tasks" , process , concurrency = 5 )
queue . add ( "hello" , { "message" : "world" })
$queue = new Queue ( 'tasks' ); // connects to localhost:6789
$queue -> add ( 'hello' , [ 'message' => 'world' ]);
$worker = new Worker ( 'tasks' , function ( Bunqueue \ Job $ job ) {
$worker -> run (); // blocking loop
queue := bunqueue . NewQueue ( "tasks" , bunqueue . Options {}) // localhost:6789
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
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 (
println! ( "processing { :? } " , job . data ());
WorkerOptions { concurrency : 5 , .. Default :: default () },
queue = Bunqueue . queue ( "tasks" ) # connects to localhost:6789
{: ok , _job } = Bunqueue . Queue . add ( queue , "hello" , % { message : "world" })
Bunqueue . Worker . new ( "tasks" , fn job ->
IO . inspect ( job . data , label : "processing" )
Bunqueue . Worker . run ( worker )
More in the quickstart .
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.
Interactive lifecycle
Follow one job, one state at a time Select an outcome, then advance the state. Succeeds first time Fails once, then succeeds Exhausts every attempt
Producer queue.add() persists the job and returns its ID. Ready queue The job is eligible and waits in scheduling order. Worker One worker claims the job and owns its active attempt. Completed The ACK saves the result and releases the concurrency slot. Producer queue.add() persists the job with attempts and backoff. Ready queue The first attempt becomes eligible for a worker. Attempt 1 The worker throws, so bunqueue records a failed attempt. Retry delay Backoff keeps the job ineligible until its retry time. Ready again The delayed job is promoted back into scheduling order. Attempt 2 A worker claims the next legal attempt. Completed The successful ACK stores the result exactly once. Producer queue.add() persists the job with a finite attempt budget. Ready queue The job becomes eligible for its first attempt. Attempt 1 Processing fails and consumes one attempt. Retry delay Backoff prevents an immediate hot retry. Final attempt The worker fails after the remaining attempt is claimed. Dead letter queue The terminal failure stays available for inspection or replay. Previous state 1 / 4 Next state
Succeeds first time. Step 1 of 4: queue.add() persists the job and returns its ID.
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.
{ url : 'https://api.example.com' },
attempts : 5 , // try up to 5 times
backoff : 2000 , // wait 2s, 4s, 8s... between tries
// After all attempts fail:
const failed = queue . getDlq () ; // inspect what died and why
queue . retryDlq () ; // send everything back for another run
{ url : 'https://api.example.com' },
attempts : 5 , // try up to 5 times
backoff : 2000 , // wait 2s, 4s, 8s... between tries
// After all attempts fail:
const failed = await queue . getDlq () ; // inspect what died and why
await queue . retryDlq () ; // send everything back for another run
queue . add ( "flaky-call" , { "url" : "https://api.example.com" },
attempts = 5 , # try up to 5 times
backoff = 2000 ) # wait 2s, 4s, 8s... between tries
# After all attempts fail:
failed = queue . get_dlq () # inspect what died and why
queue . retry_dlq () # send everything back for another run
$queue -> add ( 'flaky-call' , [ 'url' => 'https://api.example.com' ], [
'attempts' => 5 , // try up to 5 times
'backoff' => 2000 , // wait 2s, 4s, 8s... between tries
// After all attempts fail:
$failed = $queue -> getDlq (); // inspect what died and why
$queue -> retryDlq (); // send everything back for another run
queue . Add ( "flaky-call" , map [ string ] any { "url" : "https://api.example.com" }, bunqueue . JobOptions {
"attempts" : 5 , // try up to 5 times
"backoff" : 2000 , // wait 2s, 4s, 8s... 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
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 2s, 4s, 8s... between tries
// 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
Bunqueue . Queue . add ( queue , "flaky-call" , % { url : "https://api.example.com" },
attempts : 5 , # try up to 5 times
backoff : 2000 # wait 2s, 4s, 8s... 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 .
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.
// Cron expression: every day at 6 AM
repeat : { pattern : '0 6 * * *' },
// Plain interval: every 30 minutes
repeat : { every : 1_800_000 },
// Named, updatable schedule
await queue . upsertJobScheduler (
{ pattern : '0 3 * * *' },
data : { olderThanDays : 30 },
// Cron expression: every day at 6 AM
repeat : { pattern : '0 6 * * *' },
// Plain interval: every 30 minutes
repeat : { every : 1_800_000 },
// Named, updatable schedule
await queue . upsertJobScheduler (
{ pattern : '0 3 * * *' },
data : { olderThanDays : 30 },
# 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 }})
// 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 ]],
// 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 }},
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 (
SchedulerRepeat { pattern : Some ( "0 3 * * *" . into ()), .. Default :: default () },
data : Value :: Map ( vec! [( Value :: from ( "olderThanDays" ), Value :: from ( 30 ))]),
# Cron expression: every day at 6 AM
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
Bunqueue . Queue . upsert_scheduler ( queue , "cleanup" ,
% { data : % { olderThanDays : 30 }}
Timezones and schedule management in the cron guide .
Adding a job with a jobId that already exists returns the existing job instead of creating a duplicate. Useful for “exactly one welcome email per user” and safe re-runs after a restart.
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
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
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
$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
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
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
{: 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
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.
Interactive topology
Change only the boundary you need The Queue and Worker API stays familiar. Embedded TCP broker PostgreSQL multi-broker
Bun application Producer Queue runtime Worker
direct calls
Optional persistence SQLite
Best fit The smallest deployment, local services, and edge processes.
Durability owner Memory or one local SQLite file. Client processes Producer Worker A Worker B
TCP
Queue service bunqueue broker
local storage
Best fit Several processes, several languages, or one central queue service.
Durability owner The broker owns memory or one SQLite file. Client processes Producers Workers QueueEvents
TCP through a load balancer
N active brokers Broker A Broker B Broker C
transactional coordination
Shared persistence PostgreSQL
Best fit Horizontal broker scale, failover, and shared limits across hosts.
Durability owner PostgreSQL is authoritative for every broker. Embedded: The producer, queue runtime, and worker share one Bun process.
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
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' ;
return { processed : job . data . id };
{ connection : { host : 'localhost' , port : 6789 }, concurrency : 50 }
import { Queue } from 'bunqueue-client' ;
const queue = new Queue ( 'tasks' , { 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' ;
return { processed : job . data . id };
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 ()
$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' ]]);
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 })
// worker (run as many copies as you want)
worker := bunqueue . NewWorker ( "tasks" , func ( job * bunqueue . Job ) ( any , error ) {
return map [ string ] any { "processed" : job . Data ()[ "id" ]}, nil
}, bunqueue . WorkerOptions { Concurrency : 50 })
use bunqueue_client :: { BulkEntry , ConnectionOptions , JobOptions , Queue , Worker , WorkerOptions };
let queue = Queue :: new ( "tasks" , ConnectionOptions :: default ());
. map ( | data | BulkEntry { name : "process" . into (), data , options : JobOptions :: default () })
let ids = queue . add_bulk ( entries ) ? ;
// worker (run as many copies as you want)
let worker = Worker :: new ( "tasks" , | job | process ( job ), WorkerOptions {
queue = Bunqueue . queue ( "tasks" , host : "localhost" , port : 6789 )
Bunqueue . Queue . add_bulk ( queue , Enum . map ( items , &% { name : "process" , data : &1 }))
# worker (run as many copies as you want)
Bunqueue . Worker . new ( "tasks" , fn job ->
{: ok , % { processed : job . data [ "id" ]}}
Bunqueue . Worker . run ( worker )
Server setup, auth and TLS in the server guide .
QueueEvents streams lifecycle events for a queue, and workers emit their own events.
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)) ;
// Worker-side events (QueueEvents streaming is a Bun bunqueue feature)
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
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 ))
$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 " ));
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 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:? } " ))),
# 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 only; see the SDK guide .
Dashboards, metrics and Prometheus in the monitoring guide .
On SIGTERM, stop pulling new jobs, let active ones finish, then close.
async function shutdown () {
worker . pause () ; // stop accepting new jobs
await worker . close () ; // wait for active jobs (worker.close(true) forces a stop)
process . on ( 'SIGTERM' , shutdown) ;
process . on ( 'SIGINT' , shutdown) ;
async function shutdown () {
await worker . close () ; // stop pulling, flush batched ACKs, drain in-flight jobs
process . on ( 'SIGTERM' , shutdown) ;
process . on ( 'SIGINT' , shutdown) ;
except KeyboardInterrupt :
worker . close () # wait for in-flight jobs to drain
$worker -> installSignalHandlers (); // SIGTERM / SIGINT -> graceful stop
$worker -> run (); // returns after the in-flight job finishes
$worker -> close (); // unregister and close the connection
sig := make ( chan os . Signal , 1 )
signal . Notify ( sig , syscall . SIGINT , syscall . SIGTERM )
worker . Stop () // stop pulling; in-flight jobs finish
worker . Close () // unregister and close the connection
// 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
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 .
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' )
await inventory . reserve ((ctx . input as { orderId : string } ) . orderId) ;
return { reserved : true };
compensate : async () => {
await inventory . release () ;
}, // runs if a later step fails
const txId = await stripe . charge ((ctx . input as { amount : number } ) . amount) ;
compensate : async () => {
. step ( 'confirm' , async ( ctx ) => {
const { txId } = ctx . steps[ 'charge' ] as { txId : string };
await mailer . send ( 'order-confirm' , { txId } ) ;
const engine = new Engine ( { embedded : true } ) ;
engine . register (orderFlow) ;
await engine . start ( 'order' , { orderId : 'ORD-1' , amount : 99.99 } ) ;
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 .
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 →