Skip to content
bunqueue bunqueue
Get started
Get started
Simple Mode: Queue + Worker in One Object
View Markdown
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 as Error).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<{ row: unknown }>('db-inserts', {
  embedded: true,
  concurrency: 50,  // batched jobs hold their slot until the batch flushes
  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.

In TypeScript, size is a number (a value below 1 flushes every job, as 1 does, and a fraction rounds up) or Infinity to flush only on timeout (an omitted or NaN size does the same), and timeout a finite number of milliseconds (a negative value or NaN means 0); timeout: Infinity or a non-number throws when the app is created. A timeout longer than about 24.8 days (the runtime’s timer limit) is honoured, not fired at once.

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<{ url: string }>('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.

Every delay is honoured exactly, even past the runtime’s 24.8-day timer limit (with the default 1000 ms base, exponential crosses it at attempt 23), and growth saturates at Number.MAX_SAFE_INTEGER ms instead of overflowing. Unlike core backoff, which is capped at 1 hour by default, these formulas have no cap: bound them in customBackoff, for example (attempt) => Math.min(1000 * 2 ** attempt, 60_000).

In TypeScript, maxAttempts is a number (a value below 1 makes one attempt, as 1 does, and a fraction rounds up), or Infinity (or NaN) to retry until success, cancel() or close() (pair it with a non-zero delay), and delay is a finite number of milliseconds (a negative value or NaN means 0); delay: Infinity or a non-number throws when the app is created. A strategy outside the five listed runs as 'fixed' and logs one warning when the app is created. A falsy retryIf or customBackoff is ignored. customBackoff returns a finite number of milliseconds; a negative value, NaN, undefined or null retries at once and a numeric string is that many milliseconds, as before. Infinity or another non-number fails the job with a TypeError or RangeError whose cause is your processor’s error, instead of retrying.

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.

In TypeScript, the grace period is a finite number of milliseconds; a negative value or NaN cancels at once, as 0 does. Infinity throws, even for an unknown job id. A grace period longer than about 24.8 days is honoured, not fired at once.

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.

In TypeScript, threshold is a number (a value below 1 opens on the first failure, as 1 does, and a fraction rounds up) and resetTimeout a finite number of milliseconds (a negative value or NaN means 0). resetTimeout: Infinity keeps the circuit open until resetCircuit(), and threshold: Infinity (or NaN) never opens it; a non-number throws when the app is created. A resetTimeout longer than about 24.8 days is honoured, not fired at once.

Event triggers

Section titled “Event triggers”

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

const app = new Bunqueue<{ id?: string; amount?: number }>('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<{ code: string }>('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(). Aging is best-effort: a tick whose job queries fail is skipped, and the next tick tries again. At most one tick runs at a time: if job queries take longer than interval, the firings that land during the tick are dropped rather than overlapping it, so a job is boosted at most once per tick and skipped ticks are not replayed in a burst.

In TypeScript, interval must be a finite number of milliseconds ≥ 1, minAge a number of milliseconds (a negative value means 0), boost any number but NaN (0 never ages), maxPriority any number (Infinity is no cap) and maxScan a finite number; anything else throws when the app is created.

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;

Some calls return before the broker answers: pause(), resume(), setGlobalRateLimit(), removeGlobalRateLimit(), setDlqConfig(), retryDlq(), purgeDlq(), the dlq constructor option, and the jobs that event triggers enqueue. If one of them fails (for example over TCP, when the broker is unreachable, commandTimeout elapses, or the app is already closed), the error event receives an Error whose command (for example Pause, RateLimit, SetDlqConfig, or add for a trigger) and queue properties say what failed. Without an error listener the failure is logged once with console.error. Either way it never crashes the process. Await the Async variants when your code needs the outcome.

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 shared TypeScript surface of bunqueue/client and bunqueue-client. Node.js and Deno use embedded: false; await the Async variants for authoritative remote queries and mutations. DLQ filtering and statistics come from the broker. The Python SDK mirrors the baseline in snake_case (get_job_counts, set_default_ttl, priority_aging, …): embedded mode is unavailable, 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 makes it the default; an explicit embedded: false still wins)
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 }

An invalid duration or count in retry, circuitBreaker, batch or priorityAging throws a TypeError or RangeError naming the option, for example Bunqueue: priorityAging.interval must be a finite number of milliseconds >= 1 (got NaN), before the queue and worker are created. undefined keeps the default.

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