Skip to content
bunqueue bunqueue
Get started
Get started
Simple Mode: Queue + Worker in One Object
guide · simple mode

Queue and worker, one object.

Simple Mode gives you a Queue and a Worker in a single object. Add jobs, process them, add middleware, schedule crons, all from one place, one thing to close on shutdown.

If your producer and consumer live in the same process, creating a Queue and a Worker separately is boilerplate. Bunqueue wraps both:

Simple Mode ships in the Bun package, the TypeScript client (bunqueue-client) and the Python SDK; in PHP, Go, Rust and Elixir compose Queue + Worker directly.

import { Bunqueue } from 'bunqueue/client';


const app = new Bunqueue<{ to: string }>('emails', {
  embedded: true,
  processor: async (job) => {
    console.log(`Sending to ${job.data.to}`);
    return { sent: true };
  },
});


await app.add('send', { to: 'alice@example.com' });

Under the hood, Bunqueue is exactly new Queue() + new Worker() plus optional subsystems. Each job flows through: circuit breaker check → TTL check → cancellation setup → retry wrapper → middleware → your processor. Every subsystem is off until you configure it.

Routes

Section titled “Routes”

Route jobs to different handlers by name:

const app = new Bunqueue<{ to: string }>('notifications', {
  embedded: true,
  routes: {
    'send-email': async (job) => {
      await sendEmail(job.data.to);
      return { channel: 'email' };
    },
    'send-sms': async (job) => {
      await sendSMS(job.data.to);
      return { channel: 'sms' };
    },
  },
});


await app.add('send-email', { to: 'alice' });
await app.add('send-sms', { to: 'bob' });

Middleware

Section titled “Middleware”

Wraps every job execution, like middleware in a web framework. Each middleware receives the job and a next() function:

// Timing middleware
app.use(async (job, next) => {
  const start = Date.now();
  const result = await next();
  console.log(`${job.name}: ${Date.now() - start}ms`);
  return result;
});


// Error recovery middleware
app.use(async (job, next) => {
  try {
    return await next();
  } catch (err) {
    return { recovered: true, error: err.message };
  }
});

Execution order is onion-style: mw1 → mw2 → processor → mw2 → mw1. With no middleware added, there is zero overhead.

Batch processing

Section titled “Batch processing”

Accumulate N jobs and process them together, ideal for bulk database inserts:

const app = new Bunqueue('db-inserts', {
  embedded: true,
  batch: {
    size: 50,        // flush every 50 jobs
    timeout: 2000,   // or every 2 seconds, whichever comes first
    processor: async (jobs) => {
      const rows = jobs.map(j => j.data.row);
      await db.insertMany('table', rows);
      return jobs.map(() => ({ inserted: true }));
    },
  },
});

On close(), remaining buffered jobs are flushed.

Advanced retry

Section titled “Advanced retry”

Five backoff strategies (how long to wait between retry attempts) plus a predicate to decide what is worth retrying:

const app = new Bunqueue('api-calls', {
  embedded: true,
  processor: async (job) => {
    const res = await fetch(job.data.url);
    if (!res.ok) throw new Error(`HTTP ${res.status}`);
    return { status: res.status };
  },
  retry: {
    maxAttempts: 5,
    delay: 1000,
    strategy: 'jitter',   // 'fixed' | 'exponential' | 'jitter' | 'fibonacci' | 'custom'
    retryIf: (error) => error.message.includes('503'),  // only retry on 503
  },
});
StrategyFormulaUse case
fixeddelay every timeRate-limited APIs
exponentialdelay × 2^(attempt-1)General purpose
jitterdelay × 2^(attempt-1) × random(0.5-1.5)Avoid retry storms
fibonaccidelay × fib(attempt) (1x, 2x, 3x, 5x, 8x, …)Gradual backoff
customcustomBackoff(attempt, error) → msAnything

This is in-process retry: the job stays active while retrying. Different from core attempts/backoff, which re-queues the job. Synchronous throws and rejected Promises follow the same retry policy. The pending backoff is tied to the job’s cancellation signal, so cancel() or close() clears it and cannot invoke the processor again after shutdown.

Graceful cancellation

Section titled “Graceful cancellation”

Cancel running jobs via an AbortController signal (the standard way to tell async code to stop):

const app = new Bunqueue('encoding', {
  embedded: true,
  processor: async (job) => {
    const signal = app.getSignal(job.id);
    for (const chunk of chunks) {
      if (signal?.aborted) throw new Error('Cancelled');
      await encode(chunk);
    }
    return { done: true };
  },
});


const job = await app.add('video', { file: 'big.mp4' });
app.cancel(job.id);        // cancel immediately
app.cancel(job.id, 5000);  // cancel after 5s grace period

In TypeScript the signal is a standard AbortSignal, so it works with fetch too: await fetch(url, { signal }). Python’s CancelSignal exposes the same cooperative aborted flag.

Cancellation applies once the job is running and its controller has been registered. If code adds and immediately cancels a job, wait for the Worker’s active event first; sleeping for a fixed interval races worker polling. Calling cancel() for an unknown, queued, or finished id is a no-op. Repeated graceful calls keep the earliest requested deadline: a shorter grace period advances cancellation, while a longer one cannot postpone it. An immediate call supersedes and clears the pending grace timer. Job completion and close() also clear owned cancellation timers, so they do not keep the process alive after the job or app has finished. This cleanup also runs when a processor or middleware throws synchronously before returning a Promise, and it remains guaranteed if a user circuit-breaker callback throws.

Circuit breaker

Section titled “Circuit breaker”

When a downstream service is down, retrying every job just burns attempts. A circuit breaker pauses the worker after too many consecutive failures, then probes periodically until the service recovers:

const app = new Bunqueue('payments', {
  embedded: true,
  processor: async (job) => paymentGateway.charge(job.data),
  circuitBreaker: {
    threshold: 5,         // open (pause) after 5 consecutive failures
    resetTimeout: 30000,  // try again after 30s
    onOpen: () => alert('Gateway down!'),
    onClose: () => alert('Gateway recovered'),
  },
});


app.getCircuitState();  // 'closed' | 'open' | 'half-open'
app.resetCircuit();     // force close + resume worker

When both retry and circuit breaker are active: one job exhausting all its retries counts as one circuit breaker failure. close() terminally destroys the breaker, so aborting a pending retry during shutdown cannot call hooks or arm another reset timer. Explicit cancellation keeps the normal cooperative outcome: a processor that ignores the signal and completes still reports success, while a cancellation-induced rejection reports failure.

Event triggers

Section titled “Event triggers”

Create follow-up jobs automatically when a job completes or fails:

const app = new Bunqueue('orders', {
  embedded: true,
  routes: {
    'place-order': async (job) => ({ orderId: job.data.id, total: 99 }),
    'send-receipt': async (job) => ({ sent: true }),
    'fraud-alert': async (job) => ({ alerted: true }),
  },
});


// On complete → create follow-up
app.trigger({
  on: 'place-order',
  create: 'send-receipt',
  data: (result, job) => ({ id: job.data.id }),
});


// Conditional trigger; `result` is typed as unknown, cast it
app.trigger({
  on: 'place-order',
  create: 'fraud-alert',
  data: (result) => ({ amount: (result as { total: number }).total }),
  condition: (result) => (result as { total: number }).total > 1000,
});

Triggers chain: step-1 → step-2 → step-3. For anything more complex, use the Workflow Engine.

Job TTL

Section titled “Job TTL”

Expire jobs that waited too long, checked when the worker picks the job up:

const app = new Bunqueue('otp', {
  embedded: true,
  processor: async (job) => verifyOTP(job.data.code),
  ttl: {
    defaultTtl: 300000,           // 5 minutes for all jobs
    perName: {
      'verify-otp': 60000,       // 1 minute for OTP
      'daily-report': 0,         // never expires
    },
  },
});


// Update at runtime
app.setDefaultTtl(120000);
app.setNameTtl('flash-sale', 30000);

Resolution order: perName[job.name] → defaultTtl → 0 (no TTL).

Priority aging

Section titled “Priority aging”

Low-priority jobs can starve behind a stream of high-priority ones. Priority aging automatically boosts jobs the longer they wait:

const app = new Bunqueue('tasks', {
  embedded: true,
  processor: async (job) => ({ done: true }),
  priorityAging: {
    interval: 60000,    // check every 60s
    minAge: 300000,     // start boosting after 5 minutes
    boost: 2,           // +2 priority per tick
    maxPriority: 100,   // cap
    maxScan: 200,       // max jobs per tick
  },
});

The aging scheduler owns one interval. Shutdown invalidates callbacks that were already queued or waiting on a job query before clearing that interval, so an old tick cannot modify priorities after close().

Deduplication defaults

Section titled “Deduplication defaults”

Prevent duplicate jobs automatically: jobs with the same name + data get the same dedup ID within the TTL window:

const app = new Bunqueue('webhooks', {
  embedded: true,
  processor: async (job) => processWebhook(job.data),
  deduplication: {
    ttl: 60000,       // dedup window: 60 seconds
  },
});


await app.add('hook', { event: 'user.created', userId: '123' });
await app.add('hook', { event: 'user.created', userId: '123' }); // deduplicated!
await app.add('hook', { event: 'user.updated', userId: '123' }); // different data → new job

Override per job: await app.add('task', data, { deduplication: { id: 'my-id', ttl: 5000 } }) (Python: app.add("task", data, deduplication={"id": "my-id", "ttl": 5000})). Strategies (extend, replace) are explained in Queue → Deduplication.

Debounce

Section titled “Debounce”

The debounce: { ttl } option attaches a default debounce id (the job name) to every job. It is BullMQ-compatible metadata, visible via job.opts.debounce, but it does not suppress duplicates by itself in the current engine. To actually coalesce rapid duplicates, use deduplication with replace: true (last write wins).

Rate limiting

Section titled “Rate limiting”

Control processing speed:

const app = new Bunqueue('api', {
  embedded: true,
  processor: async (job) => callExternalAPI(job.data),
  rateLimit: { max: 100, duration: 1000 },  // max 100 jobs per second
});


// Per-group limiting (e.g. per customer). With groupKey set, `max` becomes
// a per-group concurrency cap (max active jobs per group) and duration is ignored.
const app2 = new Bunqueue('api', {
  embedded: true,
  processor: async (job) => callAPI(job.data),
  rateLimit: { max: 10, duration: 1000, groupKey: 'customerId' },
});


// Runtime updates
app.setGlobalRateLimit(50, 1000);
app.removeGlobalRateLimit();

DLQ (Dead Letter Queue)

Section titled “DLQ (Dead Letter Queue)”

The DLQ collects jobs that failed permanently. Simple Mode can auto-retry and prune it:

const app = new Bunqueue('critical', {
  embedded: true,
  processor: async (job) => riskyOperation(job.data),
  dlq: {
    autoRetry: true,            // re-queue failed jobs periodically
    autoRetryInterval: 3600000, // every hour
    maxAutoRetries: 3,
    maxAge: 604800000,          // purge entries older than 7 days
    maxEntries: 10000,
  },
});


// Query
const entries = app.getDlq();
const stats = app.getDlqStats();                    // { total, byReason, ... }
const timeouts = app.getDlq({ reason: 'timeout' });


// Act
app.retryDlq();           // retry all
app.retryDlq('job-id');   // retry one
app.purgeDlq();           // clear all
app.setDlqConfig({ autoRetry: false });

Failure reasons tracked: explicit_fail, max_attempts_exceeded, timeout, stalled, ttl_expired, worker_lost, plus unknown as a fallback.

Cron jobs

Section titled “Cron jobs”
await app.cron('daily-report', '0 9 * * *', { type: 'report' });
await app.cron('eu-digest', '0 8 * * 1', { type: 'weekly' }, { timezone: 'Europe/Rome' });
await app.every('healthcheck', 30000, { type: 'ping' });


await app.listCrons();
await app.removeCron('healthcheck');

See the Cron guide for advanced options.

Events, control, direct access

Section titled “Events, control, direct access”
// Events (same as Worker)
app.on('completed', (job, result) => { });
app.on('failed', (job, error) => { });
// also: active, progress, stalled, error, ready, drained, closed


// Control
app.pause();           // pause queue + worker
app.resume();          // resume both
await app.close();     // graceful shutdown
await app.close(true); // force shutdown
app.isRunning(); app.isPaused(); app.isClosed();


// Escape hatch: the underlying Queue and Worker are yours
app.queue.setStallConfig({ stallInterval: 30000 });
app.worker.concurrency = 20;

Full example

Section titled “Full example”
import { Bunqueue, shutdownManager } from 'bunqueue/client';


const app = new Bunqueue<{ payload: string }>('my-app', {
  embedded: true,
  routes: {
    'process': async (job) => ({ id: job.data.payload, status: 'done' }),
    'notify': async (job) => ({ sent: true }),
    'alert': async (job) => ({ alerted: true }),
  },
  concurrency: 10,


  retry: { maxAttempts: 3, delay: 1000, strategy: 'jitter' },
  circuitBreaker: { threshold: 5, resetTimeout: 30000 },
  ttl: { defaultTtl: 600000, perName: { 'verify-otp': 60000 } },
  priorityAging: { interval: 60000, minAge: 300000, boost: 1 },
  deduplication: { ttl: 5000 },
  rateLimit: { max: 100, duration: 1000 },
  dlq: { autoRetry: true, maxAge: 604800000 },
});


app.use(async (job, next) => {
  const start = Date.now();
  const result = await next();
  console.log(`${job.name}: ${Date.now() - start}ms`);
  return result;
});


app
  .trigger({ on: 'process', create: 'notify', data: (r) => ({ payload: (r as { id: string }).id }) })
  .trigger({ on: 'process', event: 'failed', create: 'alert', data: (_, j) => j.data });


await app.cron('cleanup', '0 2 * * *', { payload: 'nightly' });
await app.add('process', { payload: 'ORD-001' });


process.on('SIGINT', async () => {
  await app.close();
  shutdownManager();
});

API reference

Section titled “API reference”

The tables below describe the TypeScript surface of the Bun package. The TypeScript client (bunqueue-client) exposes the same camelCase surface minus embedded (TCP only; DLQ filter/stats computed client-side). The Python SDK mirrors it in snake_case (get_job_counts, set_default_ttl, priority_aging, …), with these differences: embedded is unavailable (TCP only), DLQ queries and rate-limit updates live on app.queue, and getDlqStats is not available yet.

Constructor options

Section titled “Constructor options”

Processing mode (pick one):

OptionTypeDescription
processor(job) => Promise<R>Single handler
routesRecord<string, Processor>Named handlers
batch{ size, timeout, processor }Batch processing

Worker:

OptionDefaultDescription
concurrency1Parallel jobs
embeddedfalseUse embedded SQLite (BUNQUEUE_EMBEDDED=1 forces it on)
connectionlocalhost:6789TCP server connection
autoruntrueStart worker immediately

Features:

OptionDescription
retry{ maxAttempts, delay, strategy, retryIf, customBackoff }
circuitBreaker{ threshold, resetTimeout, onOpen, onClose, onHalfOpen }
ttl{ defaultTtl, perName }
priorityAging{ interval, minAge, boost, maxPriority, maxScan }
deduplication{ ttl, extend, replace }
debounce{ ttl }
rateLimit{ max, duration, groupKey }
dlq{ autoRetry, autoRetryInterval, maxAutoRetries, maxAge, maxEntries }

Methods

Section titled “Methods”
MethodDescription
add(name, data, opts?)Add a job
addBulk(jobs)Add multiple jobs
getJob(id)Get job by ID
getJobCounts() / count()Job counts
use(middleware)Add middleware
cron(id, pattern, data?, opts?)Schedule cron
every(id, ms, data?, opts?)Schedule interval
removeCron(id) / listCrons()Manage crons
cancel(id, grace?)Cancel running job
isCancelled(id) / getSignal(id)Cancellation state
getCircuitState() / resetCircuit()Circuit breaker
trigger(rule)Register event trigger
setDefaultTtl(ms) / setNameTtl(name, ms)TTL updates
setDlqConfig(config) / getDlqConfig()DLQ config
getDlq(filter?) / getDlqStats()Query DLQ
retryDlq(id?) / purgeDlq()DLQ actions
setGlobalRateLimit(max, duration?)Set rate limit
removeGlobalRateLimit()Remove rate limit
on(event, listener) / once() / off()Events
pause() / resume()Control
close(force?)Shutdown

Properties

Section titled “Properties”
PropertyTypeDescription
namestringQueue name
queueQueue<T>Internal Queue
workerWorker<T, R>Internal Worker