# Namespaces, Auto-Batching and Store-and-Forward

Advanced bunqueue queue behaviour: prefixKey namespace isolation for multi-tenant servers, transparent TCP auto-batching, and forwarding local jobs to a central server.

Canonical: https://bunqueue.dev/guide/queue/advanced/

---

import { Tabs, TabItem } from '@astrojs/starlight/components';

<div class="bq-wrap bq-hero">
  <span class="bq-eyebrow">guide · queue</span>
  <h1 class="bq-hero-h1 bq-bench-h1">Sharing a server, <em>and leaving it.</em></h1>
  <p class="bq-hero-sub">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.</p>
</div>

## 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.

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
// 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
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
// 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
```

</TabItem>
<TabItem label="Python">

```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
```

</TabItem>
<TabItem label="PHP">

```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
```

</TabItem>
<TabItem label="Go">

```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
```

</TabItem>
<TabItem label="Rust">

```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
```

</TabItem>
<TabItem label="Elixir">

```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
```

</TabItem>
</Tabs>

_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:

<Tabs syncKey="lang">
<TabItem label="Bun">

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

</TabItem>
<TabItem label="Node.js / Deno">

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

</TabItem>
<TabItem label="Python">

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

</TabItem>
<TabItem label="PHP">

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

</TabItem>
<TabItem label="Go">

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

</TabItem>
<TabItem label="Rust">

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

</TabItem>
<TabItem label="Elixir">

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

</TabItem>
</Tabs>

_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](/guide/benchmarks/) instead
of treating an older point measurement as a universal rate.

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
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)
  },
});
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
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)
  },
});
```

</TabItem>
<TabItem label="Python">

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

</TabItem>
<TabItem label="PHP">

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

</TabItem>
<TabItem label="Go">

```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}},
})
```

</TabItem>
<TabItem label="Rust">

```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() },
])?;
```

</TabItem>
<TabItem label="Elixir">

```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}}
  ])
```

</TabItem>
</Tabs>

_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._

:::caution[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:

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
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();
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
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();
```

</TabItem>
<TabItem label="Python">

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

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

</TabItem>
<TabItem label="PHP">

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

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

</TabItem>
<TabItem label="Go">

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

```go
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})
```

</TabItem>
<TabItem label="Rust">

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

```rust
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() })?;
```

</TabItem>
<TabItem label="Elixir">

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

```elixir
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)
```

</TabItem>
</Tabs>

_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](/guide/iot-edge/).

## Where to go next

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