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

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

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

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

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

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
  )

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