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
const queue = new Queue('emails', { embedded: false });

// 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: false }
);

// 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 = Worker("emails", process)
worker.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)

The snippets above are the body of a program, not a whole file: drop each one into your project’s entry point, then run it. The worker keeps running until you stop it with Ctrl-C:

bun run app.ts       # Bun
node app.js          # Node.js (needs "type": "module" in package.json)
deno run -A app.ts   # Deno
python app.py        # Python
php app.php          # PHP
go run .             # Go
cargo run            # Rust
mix run --no-halt    # Elixir

The worker prints one line per job:

Sending "Welcome!" to user@example.com    # Bun, Node.js / Deno
Sending Welcome! to user@example.com      # Python, PHP, Go, Elixir
Sending Map([(String(Utf8String { s: Ok("to") }), ...)])    # Rust: `{:?}` on the decoded msgpack value

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', { embedded: false });

// 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
  },
  {
    embedded: false,
    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

// The Rust worker has no event emitter: handle each outcome in the processor,
// and use the connection telemetry callback for transport lifecycle events.
let worker = Worker::new(
    "emails",
    |job| match send_email(job.data()) {
        Ok(_) => {
            println!("Job {} completed", job.id());
            Ok(Value::from(true))
        }
        Err(e) => {
            eprintln!("Job {} failed: {e}", job.id());
            Err(ProcessError::retryable(e.to_string()))
        }
    },
    WorkerOptions::default(),
);

Elixir

# The Elixir worker has no event emitter: handle each outcome in the handler,
# and use the connection `:event_handler` callback for transport lifecycle events.
worker =
  Bunqueue.Worker.new("emails", fn job ->
    case send_email(job.data) do
      :ok ->
        IO.puts("Job #{job.id} completed")
        {:ok, %{sent: true}}

      {:error, reason} ->
        IO.puts("Job #{job.id} failed: #{inspect(reason)}")
        {:error, reason}
    end
  end)

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
// BUNQUEUE_DATA_PATH=./data/bunqueue.db bun run app.ts

Every embedded Queue and Worker in the process shares one database. Naming the same dataPath again is fine; naming a different one throws instead of silently opening a second database.

In server mode persistence is configured once on the server, and no client in any language changes:

bunx bunqueue start --data-path ./data/bunq.db

dataPath is an embedded (Bun) option only. BUNQUEUE_DATA_PATH is the canonical variable; BQ_DATA_PATH, DATA_PATH and SQLITE_PATH are still read, in that order, as fallbacks.

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

import { shutdownManager } from 'bunqueue-client';

process.on('SIGINT', async () => {
  await worker.close(); // Finish active jobs
  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:

Comparison	Embedded mode	Server mode
Best for	Single-process apps, serverless	Multi-process, microservices
Setup	embedded: true	Run bunx 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. The server uses memory/SQLite by default; for several active brokers sharing one queue, configure the PostgreSQL 15–18 backend; 18.6 is recommended.

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', {
  embedded: false,
  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"})

PHP

Simple Mode ships in the TypeScript and Python SDKs only. In PHP, compose Queue and Worker directly, and route on $job->name() inside the processor.

Go

Simple Mode ships in the TypeScript and Python SDKs only. In Go, compose Queue and Worker directly, and route on job.Name() inside the processor.

Rust

Simple Mode ships in the TypeScript and Python SDKs only. In Rust, compose Queue and Worker directly, and route on job.name() inside the processor.

Elixir

Simple Mode ships in the TypeScript and Python SDKs only. In Elixir, compose Queue and Worker directly, and route on job.name inside the handler.

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 -g bunqueue                     # provides the bunqueue-mcp binary
bun add -g @modelcontextprotocol/sdk    # required by the MCP server only
claude mcp add bunqueue -- bunx --package=bunqueue 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