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

Bun

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 retries if the processor throws (default: 3)
  backoff: 2000,          // Wait between retries in ms (default: 1000, jitter applied)
  // 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
});

Node.js / Deno

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 retries if the processor throws (default: 3)
  backoff: 2000,          // Wait between retries in ms (default: 1000)
  // 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 retries if the processor raises (default: 3)
    backoff=2000,             # Wait between retries 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 retries (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 retries (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 retries (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 retries (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.

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:

Bun

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

Node.js / Deno

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

addBulk is ordered and uses accepted-prefix semantics; it is not an all-or-nothing flow transaction. If a later entry is rejected, earlier entries that were already accepted remain in the queue and later entries are not evaluated. The rejected entry itself is never left as a hidden in-memory job. Use FlowProducer when the whole graph must commit atomically.

Repeat on a schedule

Bun

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

Node.js / Deno

// 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):

Bun

const job = await queue.add('sync', { endpoint: '/api/v1' }, { repeat: { every: 60000 } });
await job.updateData({ endpoint: '/api/v2' });  // Next run uses /api/v2

Node.js / Deno

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 and the Cron guide.

Durable jobs (no data loss)

By default bunqueue batches writes to disk every 10ms for speed (~100k jobs/sec). A crash inside that window can lose the not-yet-flushed jobs. For jobs where that is unacceptable, durable: true writes to disk before add() returns:

Bun

await queue.add('process-payment', { orderId: '123', amount: 99.99 }, {
  durable: true,
});

Node.js / Deno

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
  )

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 contract applies in Embedded and TCP mode and to durable entries in addBulk.

Mode	Throughput	Data loss window	Use for
Default	~100k jobs/sec	Up to 10ms	Emails, notifications, analytics
Durable	~10k jobs/sec	None	Payments, orders, audit records

Where to go next

Queue API	Create a queue in embedded or TCP mode
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
Namespaces, Auto-Batching and Store-and-Forward	Namespaces, auto-batching, store-and-forward
JobOptions Reference	Every JobOptions field, with defaults