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 lets multiple environments, tenants, or services share one server without their jobs, crons, stats, pause state, DLQ, or rate limits overlapping. The prefix is added to the queue name server-side; Queue.name keeps reporting the logical name.

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

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

await devQueue.add('send', { to: 'tester@example.com' });
await prodQueue.getJobCounts(); // 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 the Bun bunqueue package and the Python SDK; in the other SDKs, prefix the queue name directly (e.g. new Queue('dev:emails')), the isolation is identical.

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('dev:emails', processor);

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:

• Everything is isolated per prefix: jobs, worker locks, counts, pause/drain/obliterate, rate limits, and cron schedulers (two prefixes can reuse the same schedulerId).
• 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. Enabled by default, no code changes: sequential await add() sends immediately with no penalty (~10k ops/s), while concurrent adds (Promise.all) batch into one round-trip (~145k ops/s).

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

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

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 the Bun bunqueue package only; 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 so they hit the disk immediately, they are never batched.

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

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

const central = new Queue('central-name', {
  host: 'queue.example.com', port: 6789, token: process.env.BQ_TOKEN,
});
await central.add('event', data, { durable: true });

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, jobs stay local (retry, then DLQ), nothing is lost. Full guide: IoT & Edge.

Where to go next

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