guide · queue

Sharing a server, and leaving it.

Namespacing so several environments can share one broker, batching that makes concurrent adds an order of magnitude faster, and draining an edge queue into a central one.

Namespace Isolation (prefixKey)

prefixKey namespaces queue membership, crons, stats, pause state, DLQ, and rate limits on a shared broker. The client prefixes the broker queue key; Queue.name keeps reporting the logical name. Custom jobId values remain broker-wide, so include your tenant/environment in each custom ID when it must be isolated too.

Bun

// Same server, fully isolated namespaces
const devQueue = new Queue('emails', { prefixKey: 'dev:' });
const prodQueue = new Queue('emails', { prefixKey: 'prod:' });

await devQueue.add('send', { to: 'tester@example.com' });
await prodQueue.getJobCountsAsync(); // never sees dev jobs

Node.js / Deno

// Same server, fully isolated namespaces
const devQueue = new Queue('emails', { prefixKey: 'dev:' });
const prodQueue = new Queue('emails', { prefixKey: 'prod:' });

await devQueue.add('send', { to: 'tester@example.com' });
await prodQueue.getJobCountsAsync(); // never sees dev jobs

Python

# Same server, fully isolated namespaces
dev_queue = Queue("emails", prefix_key="dev:")
prod_queue = Queue("emails", prefix_key="prod:")

dev_queue.add("send", {"to": "tester@example.com"})
prod_queue.get_job_counts()  # never sees dev jobs

PHP

// Prefix the broker queue name directly.
$devQueue = new Queue('dev:emails');
$prodQueue = new Queue('prod:emails');

$devQueue->add('send', ['to' => 'tester@example.com']);
$prodQueue->getJobCounts(); // never sees dev jobs

Go

// Prefix the broker queue name directly.
devQueue := bunqueue.NewQueue("dev:emails", bunqueue.Options{})
prodQueue := bunqueue.NewQueue("prod:emails", bunqueue.Options{})

devQueue.Add("send", map[string]any{"to": "tester@example.com"}, nil)
prodQueue.GetJobCounts() // never sees dev jobs

Rust

// Prefix the broker queue name directly.
let dev_queue = Queue::new("dev:emails", ConnectionOptions::default());
let prod_queue = Queue::new("prod:emails", ConnectionOptions::default());

dev_queue.add("send", data, JobOptions::default())?;
let counts = prod_queue.get_job_counts()?; // never sees dev jobs

Elixir

# Prefix the broker queue name directly.
dev_queue = Bunqueue.queue("dev:emails")
prod_queue = Bunqueue.queue("prod:emails")

{:ok, _job} = Bunqueue.Queue.add(dev_queue, "send", %{to: "tester@example.com"})
{:ok, _counts} = Bunqueue.Queue.get_job_counts(prod_queue) # never sees dev jobs

A prefixKey option exists in both TypeScript packages and the Python SDK. In the other SDKs, prefix the queue name directly (e.g. new Queue('dev:emails')). Custom job IDs still need an explicit namespace.

A Worker must use the same prefixKey to consume the prefixed queue:

Bun

const devWorker = new Worker('emails', processor, { prefixKey: 'dev:' });

Node.js / Deno

const devWorker = new Worker('emails', processor, { prefixKey: 'dev:' });

Python

dev_worker = Worker("dev:emails", process)

PHP

$devWorker = new Worker('dev:emails', $processor);

Go

devWorker := bunqueue.NewWorker("dev:emails", processor, bunqueue.WorkerOptions{})

Rust

let dev_worker = Worker::new("dev:emails", processor, WorkerOptions::default());

Elixir

dev_worker = Bunqueue.Worker.new("dev:emails", processor)

In the other SDKs, give the Worker the prefixed name (e.g. new Worker('dev:emails', processor)).

Common patterns: dev: / staging: / prod: on one server, tenant-${id}: per customer, per-service prefixes in a monorepo, test-${runId}: for parallel test isolation.

Notes:

• Queue membership, worker locks, counts, pause/drain/obliterate, rate limits, and cron schedulers are scoped by the prefixed queue key (two prefixes can reuse the same schedulerId).
• Custom jobId ownership is broker-wide: dev:order-123 and prod:order-123 are distinct; two queues using plain order-123 refer to the same live identity. A prefix is naming isolation, not an authorization boundary.
• Backward compatible: without prefixKey, behavior is unchanged. Works in embedded and TCP modes.
• The only user-visible side effect: Job.queueName inside processors shows the prefixed key (e.g. dev:emails).

Auto-batching (TCP mode)

In TCP mode, concurrent queue.add() calls are transparently combined into single bulk commands. It is enabled by default with no code changes: sequential await add() sends immediately, while concurrent adds (Promise.all) can share one round trip. Throughput depends on batch shape, durability, database size, and backend; use the current benchmark workloads instead of treating an older point measurement as a universal rate.

Bun

const queue = new Queue('tasks', {
  autoBatch: {
    enabled: true, // default
    maxSize: 50, // flush when the buffer reaches this size (default: 50)
    maxDelayMs: 5, // max wait before flushing (default: 5)
  },
});

Node.js / Deno

const queue = new Queue('tasks', {
  autoBatch: {
    enabled: true, // default
    maxSize: 50, // flush when the buffer reaches this size (default: 50)
    maxDelayMs: 5, // max wait before flushing (default: 5)
  },
});

Python

# The network SDK sends this batch in one round-trip.
queue.add_bulk([
    {"name": "task", "data": {"id": 1}},
    {"name": "task", "data": {"id": 2}},
])

PHP

// The network SDK sends this batch in one round-trip.
$queue->addBulk([
    ['name' => 'task', 'data' => ['id' => 1]],
    ['name' => 'task', 'data' => ['id' => 2]],
]);

Go

// The network SDK sends this batch in one round-trip.
ids, err := queue.AddBulk([]bunqueue.BulkEntry{
    {Name: "task", Data: map[string]any{"id": 1}},
    {Name: "task", Data: map[string]any{"id": 2}},
})

Rust

// The network SDK sends this batch in one round-trip.
let ids = queue.add_bulk(vec![
    BulkEntry { name: "task".into(), data: Value::from(1), options: JobOptions::default() },
    BulkEntry { name: "task".into(), data: Value::from(2), options: JobOptions::default() },
])?;

Elixir

# The network SDK sends this batch in one round-trip.
{:ok, ids} =
  Bunqueue.Queue.add_bulk(queue, [
    %{name: "task", data: %{id: 1}},
    %{name: "task", data: %{id: 2}}
  ])

Auto-batching is available in both TypeScript packages (bunqueue/client and bunqueue-client); in the other SDKs, use addBulk to batch producer traffic into one round-trip.

Durable jobs bypass the batcher

Jobs with durable: true are always sent individually rather than through the client batcher. On a SQLite server they also bypass its write buffer; PostgreSQL admission is already transactional.

Store-and-forward: queue.forward()

Drain a source queue to a remote bunqueue server. The usual edge/IoT pattern uses an embedded SQLite queue as the offline buffer and a central server as the destination; the same API also supports a TCP source broker:

Bun

const forwarder = queue.forward({
  to: { host: 'queue.example.com', port: 6789, tls: true, token: process.env.BQ_TOKEN },
  queue: 'central-name', // optional remote queue name (default: same)
  concurrency: 4, // parallel forwards (default: 4)
  durable: true, // push remotely with durable: true (default: false)
});

forwarder.on('forwarded', ({ id, remoteId, name }) => {});
forwarder.on('error', (err) => {});
await forwarder.close();

Node.js / Deno

const forwarder = queue.forward({
  to: { host: 'queue.example.com', port: 6789, tls: true, token: process.env.BQ_TOKEN },
  queue: 'central-name', // optional remote queue name (default: same)
  concurrency: 4, // parallel forwards (default: 4)
  durable: true, // push remotely with durable: true (default: false)
});

forwarder.on('forwarded', ({ id, remoteId, name }) => {});
forwarder.on('error', (err) => {});
await forwarder.close();

Python

The network SDK has no local embedded buffer. When offline buffering is not required, write directly to the central broker:

central = Queue(
    "central-name", host="queue.example.com", port=6789,
    token=os.environ["BQ_TOKEN"],
)
central.add("event", data, durable=True)

PHP

The network SDK has no local embedded buffer. When offline buffering is not required, write directly to the central broker:

$central = new Queue('central-name', [
    'host' => 'queue.example.com', 'port' => 6789, 'token' => getenv('BQ_TOKEN'),
]);
$central->add('event', $data, ['durable' => true]);

Go

The network SDK has no local embedded buffer. When offline buffering is not required, write directly to the central broker:

central := bunqueue.NewQueue("central-name", bunqueue.Options{
    Host: "queue.example.com", Port: 6789, Token: os.Getenv("BQ_TOKEN"),
})
central.Add("event", data, bunqueue.JobOptions{"durable": true})

Rust

The network SDK has no local embedded buffer. When offline buffering is not required, write directly to the central broker:

let central = Queue::new("central-name", ConnectionOptions {
    host: "queue.example.com".into(),
    port: 6789,
    token: std::env::var("BQ_TOKEN").ok(),
    ..Default::default()
});
central.add("event", data, JobOptions { durable: Some(true), ..Default::default() })?;

Elixir

The network SDK has no local embedded buffer. When offline buffering is not required, write directly to the central broker:

central =
  Bunqueue.queue("central-name",
    host: "queue.example.com", port: 6789, token: System.fetch_env!("BQ_TOKEN")
  )

{:ok, _job} = Bunqueue.Queue.add(central, "event", data, durable: true)

Only the Bun-runtime bunqueue package provides the forward() drain loop. Its source may be embedded or TCP, but only an embedded source provides the in-process SQLite offline buffer. Direct network writes do not retain jobs locally while the central broker is unavailable.

If the remote is down, locally persisted jobs stay in the source queue (retry, then DLQ) while that process and volume survive. Use durable: true locally if SQLite’s 10ms hard-crash window is unacceptable. Full guide: IoT & Edge.

Where to go next

Guide	What it covers
Queue API	Create a queue in embedded or TCP mode
Adding Jobs	add, addBulk, priorities, delays, durability
Deduplication and Idempotent Job Adds	Idempotent adds, dedup keys, custom job ids
Querying Jobs	Fetch jobs, states, counts and results
Queue Control and Maintenance	Pause, drain, obliterate, clean and repair
Progress, Job Logs and Dependencies	Progress, per-job logs and dependencies
Queue Rate Limiting and Global Concurrency	Rate limits and global concurrency caps
Job Schedulers from the Queue	Named repeatable schedules from the queue
DLQ Operations from the Queue Object	Failed-job operations from the Queue object
Workers, Stats and Metrics from the Queue	Registered workers, stats and metrics windows
JobOptions Reference	Every JobOptions field, with defaults