# Adding Jobs: Priorities, Delays and Bulk
Everything about putting work into a bunqueue queue: single and bulk adds, priorities, delays, per-job attempts and timeouts, and durable writes that skip the buffer.
Canonical: https://bunqueue.dev/guide/queue/adding-jobs/
---
import { Tabs, TabItem } from '@astrojs/starlight/components';
guide · queue
Getting work into the queue.
One job or a hundred thousand, ordered by priority, held back by a delay, or written straight to disk when losing it is not an option.
## Add jobs
```typescript
const basicJob = await queue.add('job-name', { key: 'value' });
// With options
const configuredJob = await queue.add('job-name', data, {
priority: 10, // Higher = processed first
delay: 5000, // Wait 5s before processing
attempts: 5, // Max total executions, first run included (default: 3)
backoff: 2000, // Exponential base delay in ms (default: 1000, jitter applied, capped at 1h)
// OR: backoff: { type: 'exponential', delay: 2000 } // 'fixed' | 'exponential'
timeout: 30000, // Fail the job if processing takes longer
jobId: 'custom-id', // Custom ID, makes the add idempotent (see Deduplication)
removeOnComplete: true, // Delete job data after it completes
});
```
```typescript
const basicJob = await queue.add('job-name', { key: 'value' });
// With options
const configuredJob = await queue.add('job-name', data, {
priority: 10, // Higher = processed first
delay: 5000, // Wait 5s before processing
attempts: 5, // Max total executions, first run included (default: 3)
backoff: 2000, // Exponential base delay in ms (default: 1000, jitter applied, capped at 1h)
// OR: backoff: { type: 'exponential', delay: 2000 } // 'fixed' | 'exponential'
timeout: 30000, // Fail the job if processing takes longer
jobId: 'custom-id', // Custom ID, makes the add idempotent (see Deduplication)
removeOnComplete: true, // Delete job data after it completes
});
```
```python
job = queue.add("job-name", {"key": "value"})
# With options
job = queue.add(
"job-name",
data,
priority=10, # Higher = processed first
delay=5000, # Wait 5s before processing
attempts=5, # Max total executions, first run included (default: 3)
backoff=2000, # Exponential base delay in ms (default: 1000)
# OR: backoff={"type": "exponential", "delay": 2000}
timeout=30000, # Fail the job if processing takes longer
job_id="custom-id", # Custom ID, makes the add idempotent
remove_on_complete=True, # Delete job data after it completes
)
```
```php
$job = $queue->add('job-name', ['key' => 'value']);
// With options
$job = $queue->add('job-name', $data, [
'priority' => 10, // Higher = processed first
'delay' => 5000, // Wait 5s before processing
'attempts' => 5, // Max total executions (default: 3)
'backoff' => 2000, // Or ['type' => 'exponential', 'delay' => 2000]
'timeout' => 30000, // Fail the job if processing takes longer
'jobId' => 'custom-id', // Custom ID, makes the add idempotent
'removeOnComplete' => true, // Delete job data after it completes
]);
```
```go
job, err := queue.Add("job-name", map[string]any{"key": "value"}, nil)
// With options
job, err = queue.Add("job-name", data, bunqueue.JobOptions{
"priority": 10, // Higher = processed first
"delay": 5000, // Wait 5s before processing
"attempts": 5, // Max total executions (default: 3)
"backoff": 2000, // Or map[string]any{"type": "exponential", "delay": 2000}
"timeout": 30000, // Fail the job if processing takes longer
"jobId": "custom-id", // Custom ID, makes the add idempotent
"removeOnComplete": true, // Delete job data after it completes
})
```
```rust
use bunqueue_client::{Backoff, JobOptions};
// With options
let job = queue.add("job-name", data, JobOptions {
priority: Some(10), // Higher = processed first
delay: Some(5000), // Wait 5s before processing
attempts: Some(5), // Max total executions (default: 3)
backoff: Some(Backoff::Milliseconds(2000)),
// OR: Backoff::Strategy { kind: "exponential".into(), delay: 2000, max_delay: None }
timeout: Some(30_000), // Fail the job if processing takes longer
job_id: Some("custom-id".into()), // Custom ID, makes the add idempotent
remove_on_complete: Some(true), // Delete job data after it completes
..Default::default()
})?;
```
```elixir
{:ok, job} = Bunqueue.Queue.add(queue, "job-name", %{key: "value"})
# With options
{:ok, job} =
Bunqueue.Queue.add(queue, "job-name", data,
priority: 10, # Higher = processed first
delay: 5000, # Wait 5s before processing
attempts: 5, # Max total executions (default: 3)
backoff: 2000, # Or %{type: "exponential", delay: 2000}
timeout: 30_000, # Fail the job if processing takes longer
jobId: "custom-id", # Custom ID, makes the add idempotent
removeOnComplete: true # Delete job data after it completes
)
```
The full option list is in the [reference table](/guide/queue/options/).
`timeout` starts when the broker marks the job active. The broker tracks the
absolute processing deadline and fails the job with reason `timeout` when it is
reached; it is not rounded to a maintenance sweep interval.
### Add many at once
`addBulk` inserts all jobs in one batch, much faster than a loop of `add`:
```typescript
const jobs = await queue.addBulk([
{ name: 'task-1', data: { id: 1 } },
{ name: 'task-2', data: { id: 2 }, opts: { priority: 10 } },
{ name: 'task-3', data: { id: 3 }, opts: { delay: 5000 } },
]);
```
```typescript
const jobs = await queue.addBulk([
{ name: 'task-1', data: { id: 1 } },
{ name: 'task-2', data: { id: 2 }, opts: { priority: 10 } },
{ name: 'task-3', data: { id: 3 }, opts: { delay: 5000 } },
]);
```
```python
# Each entry: {"name", "data", ...options} with the same names as add()
ids = queue.add_bulk([
{"name": "task-1", "data": {"id": 1}},
{"name": "task-2", "data": {"id": 2}, "priority": 10},
{"name": "task-3", "data": {"id": 3}, "delay": 5000},
])
```
```php
// Each entry: name + data + options, flattened
$ids = $queue->addBulk([
['name' => 'task-1', 'data' => ['id' => 1]],
['name' => 'task-2', 'data' => ['id' => 2], 'priority' => 10],
['name' => 'task-3', 'data' => ['id' => 3], 'delay' => 5000],
]);
```
```go
ids, err := queue.AddBulk([]bunqueue.BulkEntry{
{Name: "task-1", Data: map[string]any{"id": 1}},
{Name: "task-2", Data: map[string]any{"id": 2}, Opts: bunqueue.JobOptions{"priority": 10}},
{Name: "task-3", Data: map[string]any{"id": 3}, Opts: bunqueue.JobOptions{"delay": 5000}},
})
```
```rust
use bunqueue_client::{BulkEntry, JobOptions, Value};
let ids = queue.add_bulk(vec![
BulkEntry {
name: "task-1".into(),
data: Value::Nil,
options: JobOptions::default(),
},
BulkEntry {
name: "task-2".into(),
data: Value::Nil,
options: JobOptions { priority: Some(10), ..Default::default() },
},
BulkEntry {
name: "task-3".into(),
data: Value::Nil,
options: JobOptions { delay: Some(5000), ..Default::default() },
},
])?;
```
```elixir
{:ok, ids} =
Bunqueue.Queue.add_bulk(queue, [
%{name: "task-1", data: %{id: 1}},
%{name: "task-2", data: %{id: 2}, opts: [priority: 10]},
%{name: "task-3", data: %{id: 3}, opts: [delay: 5000]}
])
```
On memory/SQLite brokers, `addBulk` is ordered and uses accepted-prefix
semantics, including over TCP. If a later entry is rejected (for example by a
group `maxSize` limit), earlier accepted entries remain queued and later
entries are not evaluated. The rejected entry is never left as a hidden
in-memory job. PostgreSQL brokers commit `addBulk` in one transaction: an
admission error rolls back the batch. Use `FlowProducer` when the complete
graph must commit atomically across every backend.
### Repeat on a schedule
```typescript
// Every 5 seconds
await queue.add('heartbeat', {}, { repeat: { every: 5000 } });
// Every 24 hours, at most 30 times
await queue.add('daily-report', {}, { repeat: { every: 86400000, limit: 30 } });
// Cron pattern
await queue.add('weekly', {}, { repeat: { pattern: '0 9 * * MON' } });
```
```typescript
// Every 5 seconds
await queue.add('heartbeat', {}, { repeat: { every: 5000 } });
// Every 24 hours, at most 30 times
await queue.add('daily-report', {}, { repeat: { every: 86400000, limit: 30 } });
// Cron pattern
await queue.add('weekly', {}, { repeat: { pattern: '0 9 * * MON' } });
```
```python
# Every 5 seconds
queue.add("heartbeat", {}, repeat={"every": 5000})
# Every 24 hours, at most 30 times
queue.add("daily-report", {}, repeat={"every": 86400000, "limit": 30})
# Cron pattern
queue.add("weekly", {}, repeat={"pattern": "0 9 * * MON"})
```
```php
// Every 5 seconds
$queue->add('heartbeat', [], ['repeat' => ['every' => 5000]]);
// Every 24 hours, at most 30 times
$queue->add('daily-report', [], ['repeat' => ['every' => 86400000, 'limit' => 30]]);
// Cron pattern
$queue->add('weekly', [], ['repeat' => ['pattern' => '0 9 * * MON']]);
```
```go
// Every 5 seconds
queue.Add("heartbeat", nil, bunqueue.JobOptions{"repeat": map[string]any{"every": 5000}})
// Every 24 hours, at most 30 times
queue.Add("daily-report", nil, bunqueue.JobOptions{
"repeat": map[string]any{"every": 86400000, "limit": 30},
})
// Cron pattern
queue.Add("weekly", nil, bunqueue.JobOptions{
"repeat": map[string]any{"pattern": "0 9 * * MON"},
})
```
```rust
use bunqueue_client::{JobOptions, Value};
// Every 5 seconds
let repeat = Value::Map(vec![(Value::from("every"), Value::from(5000))]);
queue.add("heartbeat", Value::Nil, JobOptions { repeat: Some(repeat), ..Default::default() })?;
// Cron pattern
let repeat = Value::Map(vec![(Value::from("pattern"), Value::from("0 9 * * MON"))]);
queue.add("weekly", Value::Nil, JobOptions { repeat: Some(repeat), ..Default::default() })?;
```
```elixir
# Every 5 seconds
{:ok, _} = Bunqueue.Queue.add(queue, "heartbeat", %{}, repeat: %{every: 5000})
# Every 24 hours, at most 30 times
{:ok, _} = Bunqueue.Queue.add(queue, "daily-report", %{}, repeat: %{every: 86_400_000, limit: 30})
# Cron pattern
{:ok, _} = Bunqueue.Queue.add(queue, "weekly", %{}, repeat: %{pattern: "0 9 * * MON"})
```
You can change the data for future runs at any point in the lifecycle with `updateData()`, even after the current run completes (the update follows the repeat chain to the next scheduled execution):
```typescript
const job = await queue.add('sync', { endpoint: '/api/v1' }, { repeat: { every: 60000 } });
await job.updateData({ endpoint: '/api/v2' }); // Next run uses /api/v2
```
```typescript
const job = await queue.add('sync', { endpoint: '/api/v1' }, { repeat: { every: 60000 } });
await job.updateData({ endpoint: '/api/v2' }); // Next run uses /api/v2
```
```python
job = queue.add("sync", {"endpoint": "/api/v1"}, repeat={"every": 60000})
job.update_data({"endpoint": "/api/v2"}) # Next run uses /api/v2
```
```php
$job = $queue->add('sync', ['endpoint' => '/api/v1'], ['repeat' => ['every' => 60000]]);
$queue->updateJobData($job->id(), ['endpoint' => '/api/v2']); // Next run uses /api/v2
```
```go
job, _ := queue.Add("sync", map[string]any{"endpoint": "/api/v1"},
bunqueue.JobOptions{"repeat": map[string]any{"every": 60000}})
queue.UpdateJobData(job.ID(), map[string]any{"endpoint": "/api/v2"}) // Next run uses /api/v2
```
```rust
let repeat = Value::Map(vec![(Value::from("every"), Value::from(60_000))]);
let job = queue.add("sync", data, JobOptions { repeat: Some(repeat), ..Default::default() })?;
let update = Value::Map(vec![(Value::from("endpoint"), Value::from("/api/v2"))]);
queue.update_job_data(&job.id(), update)?; // Next run uses /api/v2
```
```elixir
{:ok, job} =
Bunqueue.Queue.add(queue, "sync", %{endpoint: "/api/v1"}, repeat: %{every: 60_000})
:ok = Bunqueue.Queue.update(queue, job.id, %{endpoint: "/api/v2"}) # Next run uses /api/v2
```
For named, managed schedules, see [Job Schedulers](/guide/queue/schedulers/) and the [Cron guide](/guide/cron/).
### Durable jobs (no SQLite buffer-loss window)
By default SQLite mode batches writes to disk for up to 10 ms. A crash inside
that window can lose the not-yet-flushed jobs. For jobs where that is
unacceptable, `durable: true` bypasses bunqueue's buffer and commits before
`add()` returns. Host, filesystem, and physical-media durability still apply:
```typescript
await queue.add(
'process-payment',
{ orderId: '123', amount: 99.99 },
{
durable: true,
}
);
```
```typescript
await queue.add(
'process-payment',
{ orderId: '123', amount: 99.99 },
{
durable: true,
}
);
```
```python
queue.add("process-payment", {"order_id": "123", "amount": 99.99}, durable=True)
```
```php
$queue->add('process-payment', ['orderId' => '123', 'amount' => 99.99], [
'durable' => true,
]);
```
```go
queue.Add("process-payment", map[string]any{"orderId": "123", "amount": 99.99},
bunqueue.JobOptions{"durable": true})
```
```rust
queue.add("process-payment", data, JobOptions {
durable: Some(true),
..Default::default()
})?;
```
```elixir
{:ok, _job} =
Bunqueue.Queue.add(queue, "process-payment", %{order_id: "123", amount: 99.99},
durable: true
)
```
SQLite durable acceptance is fail closed. `add()` resolves only after SQLite commits
the job and any related custom-ID retirement, dedup replacement, dependency
pin, or parent link. If SQLite rejects the write—for example because the disk
is full—the call rejects and that candidate is not queryable, counted, or
available to a Worker. Reusing a completed or DLQ `jobId` is also atomic: a
failed replacement preserves the previous generation and its result across a
broker restart. The same SQLite contract applies in Embedded and TCP mode and
to durable entries in `addBulk`. PostgreSQL admission is transactional whether
or not the flag is set. Memory-only mode remains ephemeral: `durable: true`
cannot make it survive a process restart.
| SQLite mode | Published native workload median | Data loss window | Use for |
| ----------- | ---------------------------------------: | --------------------------- | ------------------------------- |
| Default | 186,384 jobs/s, public on-disk `addBulk` | Up to 10 ms | Re-creatable work |
| Durable | 60,835 ops/s, sequential Embedded adds | No SQLite buffer-loss window after `add()` resolves | Payments, orders, audit records |
Those figures label different workloads and are not a direct per-operation
speedup ratio. See [Benchmarks](/guide/benchmarks/) for distributions and the
TCP rows. PostgreSQL admissions are already transactional and do not use this
SQLite buffer.
## Where to go next
| Guide | What it covers |
| ------------------------------------------------------------------------- | --------------------------------------------- |
| [Queue API](/guide/queue/) | Create a queue in embedded or TCP mode |
| [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 |
| [Namespaces, Auto-Batching and Store-and-Forward](/guide/queue/advanced/) | Namespaces, auto-batching, store-and-forward |
| [JobOptions Reference](/guide/queue/options/) | Every JobOptions field, with defaults |