guide · quickstart

Working queue in a minute.

Create a queue, add jobs, process them with a Worker, and turn on persistence. On Bun everything runs embedded in a single process with zero configuration; from any other language, start the server once and connect.

The smallest working queue

Install the client for your runtime (see the SDK guide), save the snippet as a file, run it. On Bun the queue runs embedded in your process; in every other language, start the server once with bunx bunqueue start first.

Bun

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

// The queue: where you put jobs
const queue = new Queue('emails', { embedded: true });

// The worker: pulls jobs and runs your function on each one
const worker = new Worker('emails', async (job) => {
  console.log(`Sending "${job.data.subject}" to ${job.data.to}`);
  return { sent: true };
}, { embedded: true });

// Add a job
await queue.add('welcome', { to: 'user@example.com', subject: 'Welcome!' });

Node.js / Deno

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

// The queue: where you put jobs (connects to localhost:6789 by default)
const queue = new Queue('emails');

// The worker: pulls jobs and runs your function on each one
const worker = new Worker('emails', async (job) => {
  console.log(`Sending "${job.data.subject}" to ${job.data.to}`);
  return { sent: true };
});

// Add a job
await queue.add('welcome', { to: 'user@example.com', subject: 'Welcome!' });

Python

from bunqueue import Queue, Worker

# The queue: where you put jobs (connects to localhost:6789 by default)
queue = Queue("emails")
queue.add("welcome", {"to": "user@example.com", "subject": "Welcome!"})

# The worker: pulls jobs and runs your function on each one
def process(job):
    print(f"Sending {job.data['subject']} to {job.data['to']}")
    return {"sent": True}

Worker("emails", process).run()

PHP

use Bunqueue\Queue;
use Bunqueue\Worker;

// The queue: where you put jobs (connects to localhost:6789 by default)
$queue = new Queue('emails');
$queue->add('welcome', ['to' => 'user@example.com', 'subject' => 'Welcome!']);

// The worker: pulls jobs and runs your function on each one
$worker = new Worker('emails', function (Bunqueue\Job $job) {
    $data = $job->data();
    echo "Sending {$data['subject']} to {$data['to']}\n";
    return ['sent' => true];
});
$worker->run();

Go

// The queue: where you put jobs (connects to localhost:6789 by default)
queue := bunqueue.NewQueue("emails", bunqueue.Options{})
defer queue.Close()

queue.Add("welcome", map[string]any{
    "to": "user@example.com", "subject": "Welcome!",
}, nil)

// The worker: pulls jobs and runs your function on each one
worker := bunqueue.NewWorker("emails", func(job *bunqueue.Job) (any, error) {
    data := job.Data()
    fmt.Printf("Sending %v to %v\n", data["subject"], data["to"])
    return map[string]any{"sent": true}, nil
}, bunqueue.WorkerOptions{})
worker.Run()

Rust

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

// The queue: where you put jobs (connects to localhost:6789 by default)
let queue = Queue::new("emails", ConnectionOptions::default());
let data = Value::Map(vec![
    (Value::from("to"), Value::from("user@example.com")),
    (Value::from("subject"), Value::from("Welcome!")),
]);
queue.add("welcome", data, JobOptions::default())?;

// The worker: pulls jobs and runs your function on each one
let worker = Worker::new(
    "emails",
    |job| {
        println!("Sending {:?}", job.data());
        Ok(Value::from(true))
    },
    WorkerOptions::default(),
);
worker.run()?;

Elixir

# The queue: where you put jobs (connects to localhost:6789 by default)
queue = Bunqueue.queue("emails")

{:ok, _job} =
  Bunqueue.Queue.add(queue, "welcome", %{to: "user@example.com", subject: "Welcome!"})

# The worker: pulls jobs and runs your function on each one
worker =
  Bunqueue.Worker.new("emails", fn job ->
    IO.puts("Sending #{job.data["subject"]} to #{job.data["to"]}")
    {:ok, %{sent: true}}
  end)

Bunqueue.Worker.run(worker)

On Bun, embedded: true means the queue runs inside your process, no server needed. Every other client talks TCP to the server (bunx bunqueue start) and defaults to localhost:6789.

The one mistake everyone makes

On Bun, Queue and Worker must use the same mode. If one has embedded: true and the other doesn’t, the other tries to connect to a TCP server that isn’t running and fails with a “Command timeout” error.

// ✅ Both embedded
const queue = new Queue('tasks', { embedded: true });
const worker = new Worker('tasks', handler, { embedded: true });

// ❌ Mixed modes = timeout error
const queue = new Queue('tasks', { embedded: true });
const worker = new Worker('tasks', handler);  // Missing embedded: true!

Add jobs with options

Bun

// Typed queue: job.data is type-checked
interface EmailJob {
  to: string;
  subject: string;
}
const emailQueue = new Queue<EmailJob>('emails', { embedded: true });

// Priority, delay, retries
await emailQueue.add('send-email', { to: 'a@test.com', subject: 'Hi' }, {
  priority: 10,   // Higher = processed first
  delay: 5000,    // Wait 5 seconds before processing
  attempts: 3,    // Retry up to 3 times if the processor throws
  backoff: 1000,  // Wait 1 second between retries (grows on each attempt)
});

// Many jobs at once (one optimized batch)
await emailQueue.addBulk([
  { name: 'send-email', data: { to: 'a@test.com', subject: 'Hi' } },
  { name: 'send-email', data: { to: 'b@test.com', subject: 'Hi' } },
]);

Node.js / Deno

// Typed queue: job.data is type-checked
interface EmailJob {
  to: string;
  subject: string;
}
const emailQueue = new Queue<EmailJob>('emails');

// Priority, delay, retries
await emailQueue.add('send-email', { to: 'a@test.com', subject: 'Hi' }, {
  priority: 10,   // Higher = processed first
  delay: 5000,    // Wait 5 seconds before processing
  attempts: 3,    // Retry up to 3 times if the processor throws
  backoff: 1000,  // Wait 1 second between retries (grows on each attempt)
});

// Many jobs at once (one optimized batch)
await emailQueue.addBulk([
  { name: 'send-email', data: { to: 'a@test.com', subject: 'Hi' } },
  { name: 'send-email', data: { to: 'b@test.com', subject: 'Hi' } },
]);

Python

# Priority, delay, retries
queue.add("send-email", {"to": "a@test.com", "subject": "Hi"},
    priority=10,   # Higher = processed first
    delay=5000,    # Wait 5 seconds before processing
    attempts=3,    # Retry up to 3 times if the processor raises
    backoff=1000)  # Wait 1 second between retries (grows on each attempt)

# Many jobs at once (one optimized batch)
queue.add_bulk([
    {"name": "send-email", "data": {"to": "a@test.com", "subject": "Hi"}},
    {"name": "send-email", "data": {"to": "b@test.com", "subject": "Hi"}},
])

PHP

// Priority, delay, retries
$queue->add('send-email', ['to' => 'a@test.com', 'subject' => 'Hi'], [
    'priority' => 10,   // Higher = processed first
    'delay' => 5000,    // Wait 5 seconds before processing
    'attempts' => 3,    // Retry up to 3 times if the processor throws
    'backoff' => 1000,  // Wait 1 second between retries (grows on each attempt)
]);

// Many jobs at once (one optimized batch)
$queue->addBulk([
    ['name' => 'send-email', 'data' => ['to' => 'a@test.com', 'subject' => 'Hi']],
    ['name' => 'send-email', 'data' => ['to' => 'b@test.com', 'subject' => 'Hi']],
]);

Go

// Priority, delay, retries
queue.Add("send-email", map[string]any{"to": "a@test.com", "subject": "Hi"},
    bunqueue.JobOptions{
        "priority": 10,   // Higher = processed first
        "delay":    5000, // Wait 5 seconds before processing
        "attempts": 3,    // Retry up to 3 times if the processor errors
        "backoff":  1000, // Wait 1 second between retries (grows on each attempt)
    })

// Many jobs at once (one optimized batch)
ids, err := queue.AddBulk([]bunqueue.BulkEntry{
    {Name: "send-email", Data: map[string]any{"to": "a@test.com", "subject": "Hi"}},
    {Name: "send-email", Data: map[string]any{"to": "b@test.com", "subject": "Hi"}},
})

Rust

use bunqueue_client::{Backoff, BulkEntry, JobOptions, Value};

let data = Value::Map(vec![
    (Value::from("to"), Value::from("a@test.com")),
    (Value::from("subject"), Value::from("Hi")),
]);

// Priority, delay, retries
queue.add("send-email", data.clone(), JobOptions {
    priority: Some(10),                         // Higher = processed first
    delay: Some(5000),                          // Wait 5 seconds before processing
    attempts: Some(3),                          // Retry up to 3 times on failure
    backoff: Some(Backoff::Milliseconds(1000)), // Wait 1 second between retries
    ..Default::default()
})?;

// Many jobs at once (one optimized batch)
queue.add_bulk(vec![
    BulkEntry { name: "send-email".into(), data: data.clone(), options: JobOptions::default() },
    BulkEntry { name: "send-email".into(), data, options: JobOptions::default() },
])?;

Elixir

# Priority, delay, retries
{:ok, _job} =
  Bunqueue.Queue.add(queue, "send-email", %{to: "a@test.com", subject: "Hi"},
    # Higher = processed first; wait 5s; retry 3 times; 1s between retries
    priority: 10,
    delay: 5000,
    attempts: 3,
    backoff: 1000
  )

# Many jobs at once (one optimized batch)
{:ok, _ids} =
  Bunqueue.Queue.add_bulk(queue, [
    %{name: "send-email", data: %{to: "a@test.com", subject: "Hi"}},
    %{name: "send-email", data: %{to: "b@test.com", subject: "Hi"}}
  ])

All options are in the Queue guide.

Do more inside the processor

Bun

const worker = new Worker<EmailJob>('emails', async (job) => {
  await job.updateProgress(50, 'Sending email...');  // Report progress
  await sendEmail(job.data);                          // Do the work
  await job.log('Email sent successfully');           // Attach a log line
  return { sent: true, timestamp: Date.now() };       // Result, stored and queryable
}, {
  embedded: true,
  concurrency: 5,  // Process 5 jobs in parallel
});

Node.js / Deno

const worker = new Worker<EmailJob>('emails', async (job) => {
  await job.updateProgress(50, 'Sending email...');  // Report progress
  await sendEmail(job.data);                          // Do the work
  await job.log('Email sent successfully');           // Attach a log line
  return { sent: true, timestamp: Date.now() };       // Result, stored and queryable
}, {
  concurrency: 5,  // Process 5 jobs in parallel
});

Python

def process(job):
    job.update_progress(50, "Sending email...")  # Report progress
    send_email(job.data)                          # Do the work
    job.log("Email sent successfully")            # Attach a log line
    return {"sent": True}                         # Result, stored and queryable

Worker("emails", process, concurrency=5).run()   # Process 5 jobs in parallel

PHP

$worker = new Worker('emails', function (Bunqueue\Job $job) {
    $job->updateProgress(50, 'Sending email...');  // Report progress
    sendEmail($job->data());                        // Do the work
    $job->log('Email sent successfully');           // Attach a log line
    return ['sent' => true];                        // Result, stored and queryable
});
$worker->run();  // The PHP worker is sequential by design (one job at a time)

Go

worker := bunqueue.NewWorker("emails", func(job *bunqueue.Job) (any, error) {
    job.UpdateProgress(50, "Sending email...") // Report progress
    if err := sendEmail(job.Data()); err != nil { // Do the work
        return nil, err
    }
    job.Log("Email sent successfully", "info") // Attach a log line
    return map[string]any{"sent": true}, nil   // Result, stored and queryable
}, bunqueue.WorkerOptions{Concurrency: 5})     // Process 5 jobs in parallel

Rust

use bunqueue_client::{ProcessError, Value, Worker, WorkerOptions};

let worker = Worker::new(
    "emails",
    |job| {
        let _ = job.update_progress(50.0, Some("Sending email...")); // Report progress
        send_email(job.data())                                        // Do the work
            .map_err(|e| ProcessError::retryable(e.to_string()))?;
        let _ = job.log("Email sent successfully", None);             // Attach a log line
        Ok(Value::from(true))                                         // Result, stored
    },
    WorkerOptions { concurrency: 5, ..Default::default() }, // 5 jobs in parallel
);

Elixir

worker =
  Bunqueue.Worker.new("emails", fn job ->
    Bunqueue.Job.update_progress(job, 50, "Sending email...")  # Report progress
    send_email(job.data)                                        # Do the work
    Bunqueue.Job.log(job, "Email sent successfully")            # Attach a log line
    {:ok, %{sent: true}}                                        # Result, stored
  end, concurrency: 5)                                          # 5 jobs in parallel

React to events

Bun

worker.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed:`, result);
});

worker.on('failed', (job, error) => {
  console.error(`Job ${job.id} failed:`, error.message);
});

worker.on('progress', (job, progress) => {
  console.log(`Job ${job.id} progress: ${progress}%`);
});

Node.js / Deno

worker.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed:`, result);
});

worker.on('failed', (job, error) => {
  console.error(`Job ${job.id} failed:`, error.message);
});

worker.on('progress', (job, progress) => {
  console.log(`Job ${job.id} progress: ${progress}%`);
});

Python

worker.on("completed", lambda job, result: print(f"Job {job.id} completed: {result}"))
worker.on("failed", lambda job, err: print(f"Job {job.id} failed: {err}"))
worker.on("progress", lambda job, progress: print(f"Job {job.id} progress: {progress}%"))

PHP

$worker->on('completed', function ($job, $result) {
    echo "Job {$job->id()} completed\n";
});

$worker->on('failed', function ($job, $err) {
    echo "Job {$job->id()} failed\n";
});

// The PHP worker has no 'progress' event; query progress via the Queue API.

Go

worker.On("completed", func(args ...any) {
    job := args[0].(*bunqueue.Job)
    fmt.Printf("Job %s completed\n", job.ID())
})

worker.On("failed", func(args ...any) {
    job := args[0].(*bunqueue.Job)
    fmt.Printf("Job %s failed\n", job.ID())
})

// The Go worker has no "progress" event; query progress via the Queue API.

Rust and Elixir have no worker event emitter: use the structured telemetry callback for transport lifecycle and normal language control flow for per-job outcomes (see the SDK guide).

The full event list is in the Worker guide.

Turn on persistence

Without a data path, jobs live in memory and disappear on restart. Point bunqueue at a SQLite file to survive restarts:

// Option 1: dataPath option (recommended)
const queue = new Queue('tasks', { embedded: true, dataPath: './data/bunqueue.db' });
const worker = new Worker('tasks', processor, { embedded: true, dataPath: './data/bunqueue.db' });

// Option 2: environment variable
// DATA_PATH=./data/bunqueue.db bun run app.ts

dataPath applies to embedded (Bun) mode only. In server mode, persistence is configured on the server (bunx bunqueue start --data-path ./data/bunq.db); clients in every language need no changes.

Shut down cleanly

Bun

import { shutdownManager } from 'bunqueue/client';

process.on('SIGINT', async () => {
  await worker.close();   // Finish active jobs
  shutdownManager();      // Flush pending writes, close SQLite
  process.exit(0);
});

Node.js / Deno

process.on('SIGINT', async () => {
  await worker.close();   // Stop pulling, drain in-flight jobs
  queue.close();          // Close the connection
  process.exit(0);
});

Python

worker.close()   # Stop pulling, wait for in-flight jobs to drain
queue.close()    # Close the connection

PHP

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

Go

worker.Stop()  // Stop pulling; in-flight jobs finish
worker.Close() // Unregister and close the connection
queue.Close()

Rust

worker.stop();  // Stop pulling; in-flight jobs finish
worker.close(); // Unregister and close the connection
queue.close();

Elixir

# Idempotent drain barrier: waits for active handlers, then unregisters and closes
Bunqueue.Worker.stop(worker)

Need more than one process?

The Bun examples above run in a single process. When multiple services need to share one queue, run bunqueue as a standalone server instead:

	Embedded mode	Server mode
Best for	Single-process apps, serverless	Multi-process, microservices
Setup	embedded: true	Run bunqueue start, drop the option
Clients	Bun only (in process)	Node.js, Deno, Bun, Python, PHP, Go, Rust, Elixir, Cloudflare Workers

See the Server guide. All six official client SDKs speak the same protocol against the same queues, see the SDK guide.

Where to go next

Less boilerplate. Bunqueue (Simple Mode) wraps Queue + Worker in one object with routes, middleware, and cron:

Bun

import { Bunqueue } from 'bunqueue/client';

const app = new Bunqueue('notifications', {
  embedded: true,
  routes: {
    'send-email': async (job) => ({ sent: true }),
    'send-sms': async (job) => ({ sent: true }),
  },
  concurrency: 10,
});

await app.add('send-email', { to: 'alice@example.com' });
await app.cron('daily-report', '0 9 * * *', { type: 'summary' });

Node.js / Deno

import { Bunqueue } from 'bunqueue-client';

const app = new Bunqueue('notifications', {
  routes: {
    'send-email': async (job) => ({ sent: true }),
    'send-sms': async (job) => ({ sent: true }),
  },
  concurrency: 10,
});

await app.add('send-email', { to: 'alice@example.com' });
await app.cron('daily-report', '0 9 * * *', { type: 'summary' });

Python

from bunqueue import Bunqueue

app = Bunqueue(
    "notifications",
    routes={
        "send-email": lambda job: {"sent": True},
        "send-sms": lambda job: {"sent": True},
    },
    concurrency=10,
)

app.add("send-email", {"to": "alice@example.com"})
app.cron("daily-report", "0 9 * * *", {"type": "summary"})

Simple Mode is available in TypeScript and Python. In PHP, Go, Rust and Elixir, compose Queue and Worker directly.

See the Simple Mode guide.

Watch it live. The web dashboard shows queues, jobs, failures, crons, and workers. One command:

bunx bunqueue-dashboard

Try the live demo without installing anything.

Connect AI agents. bunqueue ships an MCP server with 73 tools, so agents like Claude can add jobs, manage crons, and monitor queues via natural language:

bun add bunqueue @modelcontextprotocol/sdk
claude mcp add bunqueue -- bunx bunqueue-mcp

Setup for Claude Desktop, Cursor, and Windsurf is in the MCP guide.

Orchestrate multi-step processes. The built-in workflow engine (Bun runtime) handles branching, parallel steps, rollback on failure, and human approvals:

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

const flow = new Workflow('order')
  .step('validate', async (ctx) => ({ ok: true }))
  .step('charge', async (ctx) => ({ txId: 'tx_123' }))
  .waitFor('manager-approval')  // Pauses until you send a signal
  .step('ship', async (ctx) => ({ shipped: true }));

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

See the Workflow Engine guide.

Next steps

• Queue API, all job options and queue operations
• Worker API, concurrency, events, error handling
• Server Mode, run bunqueue as a standalone server
• Client SDKs, use the queue from Node.js, Deno, Python, PHP, Go, Rust, Elixir
• Code Examples & Recipes, complete examples