Queue API

Overview

OneBun provides a unified queue system for message-based communication. It supports multiple backends (in-memory, Redis, NATS, JetStream) with a consistent API. The queue system includes:

Message publishing and subscribing with pattern matching
Scheduled jobs (cron, interval, timeout)
Message guards for authorization
Auto and manual acknowledgment modes

Setup

The queue system is enabled when any one of three conditions holds: a controller in your application uses queue decorators (@Subscribe, @Cron, @Interval, @Timeout); queue.enabled: true is present in application options; or a backend is explicitly configured via queue.adapter, queue.options or queue.redis. Setting queue.enabled to false overrides all three and keeps the queue disabled. No explicit configuration is required for basic usage with the in-memory adapter, which remains the default.

queue.redis selects the Redis adapter

A queue.redis block both enables the queue and chooses the Redis adapter, so queue: { redis: { url } } connects to Redis. Writing queue: { adapter: 'redis', redis: { url } } is equivalent.

An explicit adapter always wins: queue: { adapter: 'memory', redis: { url } } runs in memory with the Redis settings staged but unused.

Because the block connects, an application whose queue.redis points at an unreachable Redis fails to boot; it never falls back to running in memory.

Application Configuration

Configure the queue backend via the queue option in OneBunApplication:

typescript

import { OneBunApplication } from '@onebun/core';
import { AppModule } from './app.module';

// Default: in-memory adapter, auto-detected
const app = new OneBunApplication(AppModule, {
  port: 3000,
});

// Explicit: Redis adapter
const app = new OneBunApplication(AppModule, {
  port: 3000,
  queue: {
    adapter: 'redis',
    redis: {
      useSharedProvider: true,   // Use shared Redis connection (recommended)
      prefix: 'myapp:queue:',   // Key prefix for Redis keys
    },
  },
});

// Explicit: Redis with dedicated connection
const app = new OneBunApplication(AppModule, {
  port: 3000,
  queue: {
    adapter: 'redis',
    redis: {
      useSharedProvider: false,
      url: 'redis://localhost:6379',
      prefix: 'myapp:queue:',
    },
  },
});

For NATS/JetStream or other backends, use a custom adapter constructor (see Custom adapter: NATS JetStream below).

QueueApplicationOptions

Option	Type	Default	Description
`enabled`	`boolean`	auto	Enable queue system. Auto-enabled when a controller has queue decorators, when set to `true`, or when a backend is configured via `adapter`, `options` or `redis`. An explicit `enabled: false` is the opt-out and wins over a configured backend
`adapter`	`'memory' \| 'redis'` or adapter class	`'memory'`	Built-in type or custom adapter constructor (e.g. for NATS JetStream)
`options`	inferred from adapter	-	Options passed to the custom adapter constructor — type-safe when `adapter` is a class
`redis.useSharedProvider`	`boolean`	`true`	Use shared Redis connection pool
`redis.url`	`string`	-	Redis URL (required if `useSharedProvider: false`)
`redis.prefix`	`string`	`'onebun:queue:'`	Key prefix for Redis keys

Registering Controllers with Queue Decorators

Classes that use @Subscribe, @Cron, @Interval, or @Timeout must be registered in a module's controllers array. The framework automatically discovers and registers queue handlers from all controllers during startup via initializeQueue().

typescript

import { Module, Controller, BaseController } from '@onebun/core';
import { Subscribe, Cron, CronExpression, Message } from '@onebun/core';

// Controller with queue decorators
@Controller('/orders')
class OrderProcessor extends BaseController {
  @Subscribe('orders.created')
  async handleOrderCreated(message: Message<{ orderId: string }>) {
    this.logger.info('Processing order', { orderId: message.data.orderId });
    // Process the order...
  }

  @Cron(CronExpression.EVERY_HOUR, { pattern: 'cleanup.expired' })
  getCleanupData() {
    return { timestamp: Date.now() };
  }
}

// Register in module's controllers array
@Module({
  controllers: [OrderProcessor],
})
class OrderModule {}

WARNING

Queue handlers are only discovered in classes registered in the controllers array of a @Module. Classes in providers will not be scanned for queue decorators.

The framework says so at startup rather than leaving you to find out. A provider carrying queue decorators produces one warning per class, naming the class and every decorated method:

Queue decorators on "ReindexService" are ignored: handleReindex(). Handler discovery walks
controllers only, and "ReindexService" is registered in a module's providers — move it into that
module's controllers array to run them.

If those were the application's only handlers the queue does not start, and the debug line says which case it is — "the queue decorators in this application are on providers, which handler discovery does not walk" — rather than reporting that no handlers exist.

Scheduled-only Controllers

If your controllers use only scheduling decorators (@Cron, @Interval, @Timeout) without @Subscribe, the queue system still auto-initializes with the in-memory adapter. No explicit queue configuration is needed. This works regardless of where in the module tree the controller is located — root, child, or deeply nested modules.

Scheduled Job Error Handling

Errors thrown inside @Cron, @Interval, and @Timeout handlers are caught and logged as warnings. The scheduler continues running — one failed job does not affect other scheduled jobs.

Interval overlap and the first run

A tick that arrives while the previous run is still going is dropped. That is the default and usually what you want: a collector on a 5-second interval that occasionally takes 7 seconds skips a beat rather than piling up on itself. Pass overlapStrategy: 'queue' to run it anyway, concurrently.

An interval job also runs once immediately when it starts, before the first period elapses. Pass runOnStart: false for "every hour, starting next hour".

typescript

import { BaseController, Controller, Interval } from '@onebun/core';

@Controller('/reports')
export class ReportsController extends BaseController {
  // Skips a beat rather than overlapping, and does not run at boot.
  @Interval(3_600_000, {
    pattern: 'reports.hourly',
    overlapStrategy: 'skip',
    runOnStart: false,
  })
  buildHourly(): { at: number } {
    return { at: Date.now() };
  }
}

runOnStart applies to @Interval only. @Cron takes the same overlapStrategy, also defaults to skipping an overlapping tick, and has no leading run; @Timeout fires once by definition.

Under the default 'skip', at most one run of an interval job is in flight at any moment, however much slower than its period the body is — a dropped tick publishes nothing.

Only starting a job runs it immediately. pause()/resume() and updateJob() continue the schedule rather than beginning one, so neither injects an execution the schedule did not ask for.

Producer-only apps

An application with zero queue decorators still gets a live queue as soon as queue.adapter, queue.options or queue.redis is set. Such a producer-only service can inject QueueService and call publish() without declaring a single @Subscribe handler, and the message reaches the broker. The trade-off: the adapter is constructed and connected during app.start(), so the application fails to boot when the broker is unreachable. Set queue.enabled: false if you want the queue to stay off despite a configured backend.

Error Handling in Handlers

typescript

@Controller('/orders')
class OrderProcessor extends BaseController {
  @Subscribe('orders.created', {
    ackMode: 'manual',
    retry: { attempts: 3, backoff: 'exponential', delay: 1000 },
  })
  async handleOrder(message: Message<{ orderId: string }>) {
    try {
      await this.processOrder(message.data);
      await message.ack();  // Acknowledge success
    } catch (error) {
      this.logger.error('Order processing failed', error);

      if (message.attempt && message.attempt >= (message.maxAttempts || 3)) {
        this.logger.error('Max retries reached, moving to DLQ', {
          orderId: message.data.orderId,
        });
        await message.nack(false);  // Terminate: never redeliver this message
      } else {
        await message.nack(true);  // Requeue for retry
      }
    }
  }
}

The memory, Redis and JetStream adapters populate attempt and maxAttempts. Core NATS tracks no delivery state at all, so both stay undefined there — and so does redelivered.

Under ackMode: 'auto' you do not have to write this at all: a handler that throws is retried up to retry.attempts times and then dropped, with onMessageFailed on every attempt. The recipe above is for 'manual', where nack(true) is uncapped — it is your instruction, not the framework's policy, and message.attempt is how a handler stops itself.

A handler written this way never throws, and it is still reported as a failure. That matters because it is the shape this page recommends: the catch swallows the exception and calls nack(), so the handler resolves normally and nothing about its control flow distinguishes it from a success. Every adapter reads the nack the message recorded and emits onMessageFailed with an error naming the message, rather than onMessageProcessed. A handler that nacks and then throws emits exactly one event, carrying the thrown error — it has a stack the synthesised one does not.

ack() and nack() are first-call-wins on every adapter: the first disposition the handler chooses is the one that counts, and a later call is a no-op. A handler's return value is never a signal — returning false or an error object still counts as success. The rule does not vary with ackMode, 'none' included: 'none' removes redelivery, not observability.

Quick Start

Queue handlers (@Subscribe, @Cron, @Interval, @Timeout) are only discovered in classes registered in a module's controllers array. Use a controller (not a provider) for queue handlers.

typescript

import {
  Module,
  Controller,
  BaseController,
  OneBunApplication,
  Subscribe,
  Cron,
  Interval,
  Message,
  CronExpression,
  OnQueueReady,
  QueueService,
} from '@onebun/core';

// Controller with queue handlers (must be in controllers array)
@Controller('/events')
class EventProcessor extends BaseController {
  constructor(private queueService: QueueService) {
    super();
  }

  @OnQueueReady()
  onReady() {
    this.logger.info('Queue connected and ready');
  }

  // Subscribe to messages
  @Subscribe('orders.created')
  async handleOrderCreated(message: Message<{ orderId: number }>) {
    this.logger.info('New order:', { orderId: message.data.orderId });
  }

  // Scheduled job: every hour
  @Cron(CronExpression.EVERY_HOUR, { pattern: 'cleanup.expired' })
  getCleanupData() {
    return { timestamp: Date.now() };
  }

  // Interval job: every 30 seconds
  @Interval(30000, { pattern: 'metrics.collect' })
  getMetricsData() {
    return { cpu: process.cpuUsage() };
  }

  // Publish messages programmatically
  async createOrder(data: { userId: string }) {
    await this.queueService.publish('orders.created', {
      orderId: Date.now(),
      userId: data.userId,
    });
  }
}

// Module: register in controllers so queue handlers are discovered
@Module({
  controllers: [EventProcessor],
})
class AppModule {}

// Application
const app = new OneBunApplication(AppModule, { port: 3000 });
await app.start();

The @Subscribe decorator marks a method as a message handler.

typescript

@Subscribe('orders.*')
async handleOrder(message: Message<OrderData>) {
  this.logger.info('Order received', { pattern: message.pattern, data: message.data });
}

Pattern Syntax

Pattern	Example Match	Description	NATS subject	Redis key glob
`orders.created`	`orders.created`	Exact match	`orders.created`	`orders.created`
`orders.*`	`orders.created`, `orders.updated`	Single-level wildcard	`orders.*`	`orders.*`
`events.#`	`events.user.created`, `events.order.paid`	Multi-level wildcard	`events.>`	`events.*`
`orders.{id}`	`orders.123` → `{ id: '123' }`	Named parameter	`orders.*`	`orders.*`

A captured value is read from message.params, on every adapter:

typescript

import {
  BaseController,
  Controller,
  Subscribe,
  type Message,
} from '@onebun/core';

// A controller, not a provider: queue handlers are only discovered in a module's
// `controllers` array — see "Registering Controllers with Queue Decorators".
@Controller('/orders')
export class OrderConsumer extends BaseController {
  @Subscribe('orders.{id}.{event}')
  async handle(message: Message<{ total: number }>): Promise<void> {
    // Delivered `orders.123.created`:
    message.params.id;      // '123'
    message.params.event;   // 'created'
    message.pattern;        // 'orders.123.created' — the topic as delivered
  }
}

{name} matches exactly one token, like *; the difference is that it also names what it matched. A pattern with no {name} gives params === {} rather than undefined.

The NATS subject column applies to the NatsQueueAdapter and JetStreamQueueAdapter only. NATS has no equivalent of a named parameter, so {name} widens to * on the wire and the pattern is re-checked in process — a subscription never receives a message its pattern does not match.

The Redis key glob column applies to the RedisQueueAdapter, which uses it to SCAN for topics that already hold a backlog. The mapping is {name} → *, * → *, and a trailing # → * — a Redis glob * spans separators, so events.* also matches the key for events.user.created. Every translation only widens: the in-process matcher, built from the original pattern, decides what a handler actually receives.

On all three adapters # must be the final token. On NATS it translates to the > wildcard, which NATS accepts only at the end of a subject; Redis could serve *.created perfectly well, and rejects it anyway so that one pattern means one thing everywhere. A # anywhere else (#.created, events.#.created) throws — from the JetStreamQueueAdapter constructor for a stream declaration, and from the awaited publish() / subscribe() call otherwise.

typescript

@Subscribe('orders.*', {
  ackMode: 'manual',        // 'auto' (default), 'manual' or 'none'
  group: 'order-processors', // One durable consumer per (group, pattern) — a PERMANENT server resource
  prefetch: 10,             // Messages to process in parallel
  ackTimeout: 30_000,       // ms — max time a handler may hold a message before redelivery
  retry: {
    attempts: 3,
    backoff: 'exponential',
    delay: 1000,
  },
})
async handleOrder(message: Message<OrderData>) {
  try {
    await this.processOrder(message.data);
    await message.ack();
  } catch (error) {
    await message.nack(true); // requeue
  }
}

This recipe assumes an adapter that acknowledges. On NatsQueueAdapter it does not fail — it does nothing: core NATS has no acknowledgement protocol, so ack() and nack() are permanent no-ops there and ackMode has no effect on the wire. A nack is still reported as onMessageFailed so the drop is visible, but the message is not requeued, because nothing in the protocol can redeliver it. See NatsQueueAdapter and the Feature Support Matrix.

Whatever the adapter, a handler that nacks and returns normally is reported as a failure, not a success: onMessageFailed fires with an error naming the message, and onMessageProcessed does not fire at all.

retry

RetryOptions decides how many times a failing handler is delivered to, and how long the wait between attempts is:

Field	Meaning	Default
`attempts`	Total deliveries, not extra ones — `attempts: 3` runs the handler at most three times	`1`
`backoff`	`'fixed'` → `delay`, `'linear'` → `delay * n`, `'exponential'` → `delay * 2^(n-1)`, where `n` is the attempt that just failed	`'fixed'`
`delay`	Base delay in milliseconds	`100`

The default of 1 means an unconfigured subscription delivers once — retries are opt-in, so a handler with a non-idempotent side effect is never quietly upgraded to three of them.

attempts counts total deliveries because that is what the error-handling recipe compares against: message.attempt is 1-based, and attempt >= maxAttempts is the terminal delivery.

Honoured by the memory, Redis and JetStream adapters — all three fields, computed by one shared function, so a ladder means the same thing wherever it is declared. Core NATS tracks no delivery state and ignores it. Under ackMode: 'none' it goes inert on every adapter, along with deadLetter and the attempt / maxAttempts / redelivered fields.

Where the wait actually happens differs, and it is worth knowing which one you are running on: memory sleeps in process, Redis parks the retry in its delayed sorted set, and JetStream sends the delay with the nak, so the server holds the message. On JetStream a delivery lost some other way — a handler that never returns, a process that dies — still comes back at ack_wait, because nothing nak'd it. The consumer's own backoff field is deliberately left alone: JetStream's backoff list replaces ack_wait per delivery, so writing a one-second base delay there would also cut the handler's acknowledgement window to one second.

Retries apply to ackMode: 'auto', where the framework owns the decision. Under 'manual', nack(true) is uncapped — it is your instruction, not a policy — and message.attempt is how a handler stops itself.

ackMode: 'none'

Fire-and-forget. The message is delivered exactly once and nothing is acknowledged, so there is no redelivery and no dead-letter routing on any adapter. Choose it when losing a message costs less than processing it twice — a telemetry firehose, say, where a replayed sample is worse than a missing one.

typescript

@Subscribe('telemetry.samples', { ackMode: 'none' })
async ingest(message: Message<Sample>) {
  await this.write(message.data);   // if this throws, the sample is gone
}

Everything that depends on the broker tracking delivery state goes inert with it:

Option or field	Under `'none'`
`retry.attempts`	ignored — nothing is redelivered
`deadLetter`	ignored — there is no failure event to route
`ackTimeout`	ignored — no effect under `ackMode: 'none'`
`ack_wait`	not sent to the server
`max_deliver`	not sent to the server
`Message.attempt`	reported but inert — stays 1, nothing is redelivered
`Message.maxAttempts`	reported but inert — no attempt can be the last
`Message.redelivered`	always `false`

A handler that throws still emits onMessageFailed, so failures remain observable — the message itself is simply not retried. message.ack() and message.nack() become no-ops.

Per adapter: JetStream creates the consumer with ack_policy: none — and refuses the mode at subscribe() on a work-queue stream, where a task leaves the stream only by being acknowledged; Redis skips both the requeue and the dead-letter branch; the in-memory adapter suppresses nack(true), which would otherwise resurrect a message in the one mode that promises a single delivery; and NatsQueueAdapter gains nothing, because core NATS never had an acknowledgement protocol — 'none' is simply the only mode that describes it truthfully.

Message Interface

typescript

interface Message<T> {
  id: string;              // Unique message ID
  pattern: string;         // Message pattern/topic — the topic as DELIVERED
  params: Record<string, string>; // Values captured by the pattern's {name} parameters; {} when none
  data: T;                 // Message payload
  timestamp: number;       // Unix timestamp in ms
  metadata: MessageMetadata;
  redelivered?: boolean;   // True if redelivered — always false under ackMode: 'none'
  attempt?: number;        // Current attempt number — inert under ackMode: 'none'
  maxAttempts?: number;    // Max attempts allowed — inert under ackMode: 'none'
  
  ack(): Promise<void>;    // Acknowledge (manual mode)
  nack(requeue?: boolean): Promise<void>; // Negative ack; requeue=false means never redeliver
}

interface MessageMetadata {
  headers?: Record<string, string>;
  authorization?: string;  // Bearer token
  serviceId?: string;      // Calling service ID
  traceId?: string;        // Distributed tracing
  spanId?: string;
  parentSpanId?: string;
  traceFlags?: number;     // W3C sampling decision; bit 0 is "sampled"
}

The four trace fields are filled by publish() and publishBatch() from the trace the publisher was running in — an HTTP request, a @Traced method, a queue handler, a scheduler tick or a WebSocket handler. The delivery then starts its own span as a child of that one, so a message and the work that sent it are one trace.

Two cases where they are absent, both deliberate:

You set traceId yourself. Then nothing is overwritten. A service relaying on behalf of someone else is stating the causal trace, and the ambient one is not it.
There is no trace to join — a publish from onModuleInit, from a lifecycle hook, or with tracing disabled. The fields stay empty rather than being invented: an id that names no span is worse than no id, because it looks joinable and resolves to nothing in the backend.

params carries what the subscriber's pattern captured from the delivered topic: @Subscribe('orders.{id}') receiving orders.123 reads message.params.id === '123'.

It is always an object — {} for an exact pattern and for a * or # wildcard — so a handler never guards the access. It is captured per subscription, not per publish: one topic reaching two subscriptions with different patterns gives each handler its own values, in its own object, and it is therefore not part of the published envelope and does not travel over the wire. A metadata key of the same name would be the publisher's, and would be wrong for exactly that reason.

Treat it as read-only: one object is captured per delivery and reused across every retry of that delivery, so mutating it changes what the next attempt of the same delivery sees.

A @Cron, @Interval or @Timeout method receives no message at all — it is the data provider, and what it returns is published to the job's pattern. A @Subscribe handler matching that pattern captures from it like any other.

nack(true) asks for the message to be delivered again. nack(false) — and the bare nack(), since requeue defaults to false — says the opposite: do not deliver this message again.

What "do not deliver again" costs is adapter-specific, and only the first line below holds everywhere:

Adapter	`nack(false)`
JetStream	terminated on the server; never redelivered, and the delivery does not count against `maxDeliver`
Redis	republished to `deadLetter.queue` if configured, otherwise dropped
In-memory	dropped; this adapter has no dead-letter queue
NATS (pub/sub)	no effect — plain NATS has no acknowledgement, so neither form does anything

So nack(false) never redelivers, but it does not universally mean "send to the dead-letter queue". Publish the payload somewhere yourself if you need to keep it.

The memory, Redis and JetStream adapters populate attempt and maxAttempts; core NATS leaves both undefined.

Scheduling Decorators

@Cron

Executes on a cron schedule. The decorated method returns data to publish.

typescript

import { Cron, CronExpression } from '@onebun/core';

// Daily at 9 AM
@Cron('0 0 9 * * *', { pattern: 'reports.daily' })
getDailyReportData() {
  return { type: 'daily', date: new Date() };
}

// Using CronExpression enum
@Cron(CronExpression.EVERY_HOUR, { pattern: 'health.check' })
getHealthData() {
  return { status: 'ok' };
}

An expression that can never match is rejected, not registered. '0 0 30 2 *' — February 30th — parses cleanly and has no reachable run, so the job would sit in the schedule and never fire. Registration throws instead, naming the job and the expression, and the application does not start. The same applies to updateJob(), which leaves the job on its previous schedule rather than stripping it. A job that runs out of reachable runs while the application is live is reported through the scheduler's error handler, which the application logs as a warning.

Day-of-month and day-of-week are OR'd when both are restricted, as crontab(5) specifies. @Cron('0 0 0 1 * 1') — the 5-field equivalent is '0 0 1 * 1' — fires on the 1st of every month and on every Monday, not only on a Monday that happens to be the 1st. When either field is * only the other one applies, which is the common case and behaves as you would expect.

CronExpression Constants

Constant	Expression	Description
`EVERY_SECOND`	`* * * * * *`	Every second
`EVERY_5_SECONDS`	`/5 * * * *`	Every 5 seconds
`EVERY_MINUTE`	`0 * * * * *`	Every minute
`EVERY_5_MINUTES`	`0 /5 * * *`	Every 5 minutes
`EVERY_HOUR`	`0 0 * * * *`	Every hour
`EVERY_DAY_AT_MIDNIGHT`	`0 0 0 * * *`	Daily at midnight
`EVERY_DAY_AT_NOON`	`0 0 12 * * *`	Daily at noon
`EVERY_WEEKDAY`	`0 0 0 * * 1-5`	Mon-Fri at midnight
`EVERY_WEEK`	`0 0 0 * * 0`	Sunday at midnight
`EVERY_MONTH`	`0 0 0 1 * *`	1st of month

@Interval

Executes at fixed intervals.

typescript

// Every 60 seconds
@Interval(60000, { pattern: 'metrics.collect' })
getMetrics() {
  return { cpu: process.cpuUsage() };
}

@Timeout

Executes once after a delay.

typescript

// After 5 seconds
@Timeout(5000, { pattern: 'init.complete' })
getInitData() {
  return { startedAt: this.startTime };
}

Message Guards

Guards control access to message handlers, similar to WebSocket guards. A refused message never reaches the handler: it is nacked without requeue — a guard decision is deterministic, so redelivery would only be denied again — and the adapter reports it through onMessageFailed, not onMessageProcessed. A guard that throws also denies, because on the queue there is no filter chain to carry the exception; the framework logs the guard's name and the error instead.

Built-in Guards

typescript

import { 
  UseMessageGuards,
  MessageAuthGuard,
  MessageServiceGuard,
  MessageHeaderGuard,
  MessageTraceGuard,
} from '@onebun/core';

// Require authorization token
@UseMessageGuards(MessageAuthGuard)
@Subscribe('secure.events')
async handleSecure(message: Message) {}

// Require specific service
@UseMessageGuards(new MessageServiceGuard(['payment-service']))
@Subscribe('internal.events')
async handleInternal(message: Message) {}

// Require header
@UseMessageGuards(new MessageHeaderGuard('x-api-key'))
@Subscribe('api.events')
async handleApi(message: Message) {}

// Require trace context
@UseMessageGuards(MessageTraceGuard)
@Subscribe('traced.events')
async handleTraced(message: Message) {}

These guards trust what the publisher sent

MessageAuthGuard passes any message whose metadata.authorization is non-empty, and MessageServiceGuard compares metadata.serviceId against its allow-list. Both fields are written by the publisher and travel with the message. Neither the broker nor the framework authenticates them, so anything able to publish to the subject can set them to whatever it likes — the queue-side twin of the x-user-roles hazard described under RolesGuard.

They earn their keep as a filter — "this consumer only handles messages that claim to come from payment-service" — and as a defence-in-depth check behind a broker that already authenticates publishers per subject. They are not an authorization decision on their own. If the sender's identity has to be trusted, sign the payload and verify the signature in a guard of your own.

MessageTraceGuard passes a message that carries metadata.traceId, which publish() fills in whenever the publisher was inside a trace. It therefore refuses exactly the messages sent from outside one — a lifecycle hook, or an application with tracing off — and a refusal is nacked and logged, not silently consumed.

A producer does not need to set traceId by hand to pass this guard. One that does is not overwritten: a hand-set traceId is kept as it is and satisfies the guard like any other.

Composite Guards

typescript

import { MessageAllGuards, MessageAnyGuard } from '@onebun/core';

// All guards must pass
@UseMessageGuards(new MessageAllGuards([
  MessageAuthGuard,
  new MessageServiceGuard(['allowed-service']),
]))
@Subscribe('strict.events')
async handleStrict(message: Message) {}

// Any guard can pass
@UseMessageGuards(new MessageAnyGuard([
  new MessageServiceGuard(['internal-service']),
  MessageAuthGuard,
]))
@Subscribe('flexible.events')
async handleFlexible(message: Message) {}

Both composites construct their children themselves, in their own constructor — a child passed as a class gets a bare new guard() at decoration time, before any module exists, so a guard with a constructor dependency receives nothing. Pass such a guard as an already constructed instance, or list the guards directly (@UseMessageGuards(A, B)): that form is resolved through the module owning the consumer, at startup, with full constructor DI per guard.

Neither composite checks the transport either. The built-in leaves each deny on a non-queue context, so a composite built only from them denies too — but that is their children's doing, not the composite's. A hand-written child that reads context.getMetadata() will throw if the same composite is also listed under @UseGuards on an HTTP route or WebSocket handler, where that accessor does not exist. Narrow with isQueueContext(context) in your own guards; createMessageGuard() already does it for you.

Custom Guards

typescript

import { createMessageGuard } from '@onebun/core';

const customGuard = createMessageGuard((context) => {
  const metadata = context.getMetadata();
  return metadata.headers?.['x-custom'] === 'expected';
});

@UseMessageGuards(customGuard)
@Subscribe('custom.events')
async handleCustom(message: Message) {}

Interceptors

@UseInterceptors() reaches queue handlers — same decorator as HTTP and WebSocket — but only in its class-level form. Interceptors wrap handler execution for logging, timing, or other cross-cutting concerns.

typescript

@UseInterceptors(LoggingInterceptor)
@Controller('/processor')
class EventProcessor extends BaseController {
  @Subscribe('events.created')
  async handleEvent(message: Message<{ id: string }>) {
    // handler code
  }
}

Method-level @UseInterceptors never runs on a @Subscribe handler

Applied to a method, the decorator records itself on the class prototype, while QueueService.registerService() reads interceptors off the class — the metadata store is keyed on the exact object and does not walk prototypes, so the interceptor is silently dropped and the handler runs unwrapped. Put it on the consumer class, or split handlers that need different interceptors into separate consumers. Guards are unaffected: @UseMessageGuards writes to the class, and the method form of @UseGuards is read back by walking the prototype chain.

@Cron, @Interval and @Timeout handlers get no interceptors and no guards at all, at either level — the scheduler is handed the bound method directly.

See Interceptors for full documentation.

Using QueueService from module lifecycle hooks

Announcing an instance on boot and on shutdown is the ordinary reason to publish from a lifecycle hook, and both ends work:

From	`publish()` / `publishBatch()`	Everything else on `QueueService`
`onModuleInit`	accepted, sent once the queue is up	throws — the adapter and the handlers do not exist yet
`onApplicationInit`	sent immediately	available
request / message handlers	sent immediately	available
`onModuleDestroy`	sent immediately — the transport is still open	scheduler and consumers already stopped
`beforeApplicationDestroy`	sent immediately	available

A publish() issued from onModuleInit returns as soon as the message is accepted, not when it reaches the broker — the flush cannot run until your hook returns. The value it resolves to is the real message id, and it is delivered after the application's own @Subscribe handlers are registered, so a subscriber declared in the same application receives it. If the queue is not enabled at all, the call still throws the "Queue is not enabled" error naming the three ways to enable one.

Lifecycle Decorators

Lifecycle handlers run only when the class is registered as a controller (in a module's controllers array).

typescript

import {
  Controller,
  BaseController,
  OnQueueReady,
  OnQueueError,
  OnMessageReceived,
  OnMessageProcessed,
  OnMessageFailed,
} from '@onebun/core';

@Controller('/events')
class EventProcessor extends BaseController {
  @OnQueueReady()
  handleReady() {
    this.logger.info('Queue connected');
  }

  @OnQueueError()
  handleError(error: Error) {
    this.logger.error('Queue error', error);
  }

  @OnMessageReceived()
  handleReceived(message: Message) {
    this.logger.info('Received', { id: message.id });
  }

  @OnMessageProcessed()
  handleProcessed(message: Message) {
    this.logger.info('Processed', { id: message.id });
  }

  @OnMessageFailed()
  handleFailed(message: Message, error: Error) {
    this.logger.error('Failed', { id: message.id, error });
  }
}

Queue Adapters

InMemoryQueueAdapter

In-process message bus. Good for development and testing. This is the default adapter when queue.adapter is not specified:

typescript

const app = new OneBunApplication(AppModule, {
  queue: { adapter: 'memory' },
});

This snippet by itself enables the queue: an explicit queue.adapter is a backend configuration, so no @Subscribe, @Cron, @Interval or @Timeout decorator is needed anywhere in the application.

Supported Features:

Pattern subscriptions
Delayed messages
Priority
Retry (in-process, non-persistent)
Scheduled jobs

Retries here live in the process: the attempt counter is a closure variable, so a restart loses it along with the message. That is the adapter, not a caveat on retry — nothing in this adapter survives a restart. There is no dead-letter queue, so a message that exhausts its attempts is dropped, having been reported through onMessageFailed on every attempt.

RedisQueueAdapter

Distributed queue using Redis. Uses SharedRedisProvider by default:

typescript

const app = new OneBunApplication(AppModule, {
  queue: {
    adapter: 'redis',
    redis: { useSharedProvider: true, prefix: 'myapp:queue:' },
  },
});

Or with a dedicated Redis connection:

typescript

const app = new OneBunApplication(AppModule, {
  queue: {
    adapter: 'redis',
    redis: { useSharedProvider: false, url: 'redis://localhost:6379' },
  },
});

Supported Features:

Publishing and delivery, pattern subscriptions, delayed messages, priority messages, consumer groups, retry, dead-letter queue, scheduled jobs

Dead-letter queue

DeadLetterOptions.queue is a queue pattern, not a Redis key. On the terminal delivery the message is republished through the normal publish path to that pattern, so an ordinary subscriber consumes it:

typescript

@Subscribe('orders.created', {
  deadLetter: { queue: 'orders.dead', maxRetries: 3 },
})
async handleOrder(message: Message<OrderData>) {
  await this.processOrder(message.data);
}

@Subscribe('orders.dead')
async handleDeadOrder(message: Message<OrderData>) {
  this.logger.error('order gave up', { origin: message.metadata['dlq.originalPattern'] });
}

The attempt cap is retry.attempts ?? deadLetter.maxRetries ?? 1 — the same precedence JetStream uses for max_deliver. With neither option set a failing handler still gets exactly one delivery, and is then dead-lettered if a queue is configured or dropped if not.

The republished envelope keeps the original id, data, timestamp and metadata, and gains dlq.originalPattern, dlq.deliveryCount and dlq.error. Its pattern is the dead-letter queue — the origin lives in the metadata. A message is dead-lettered at most once: one arriving with dlq.originalPattern already set is dropped rather than routed again, so a dead-letter queue whose own subscriber throws does not republish to itself forever.

Under ackMode: 'none' the whole path is skipped, along with retry — that mode tracks no delivery, so there is no terminal delivery to detect.

A republish that fails is reported through onError with the broker's rejection as cause, rather than swallowed: a dead letter that cannot be written is a message lost in the one path that exists to not lose it.

How delivery works

Delivery is list-based. A message is pushed onto a Redis list keyed by its topic, and a wake-up frame naming that topic is published on one shared channel. The list is the only delivery path: a consumer claims a message with an atomic LPOP, so two subscriptions that both match a topic compete rather than both receive it — whether they are two replicas on one pattern or two different patterns in one process. Per-subscription params do not change that: only the subscription that wins the claim gets the message, and it gets its own captures. The channel carries no payload — it only wakes the drain, which is why a message published while nothing was subscribed is still delivered when a subscriber starts.

A pattern subscription (orders.{id}, events.#) cannot know its topics in advance, so it also polls: the pattern is translated to a Redis glob and matching topic keys are found with SCAN, then drained through the same LPOP claim. The wake channel short-circuits the wait; the scan is what finds a backlog that predates the subscription.

Do not publish to the wake channel by hand expecting delivery: the frame is a signal, and the message has to be on the list to be taken.

A failing poll is reported, not swallowed. A command that cannot reach Redis emits onError and the poll backs off exponentially from pollInterval to a one-second ceiling, resetting on the first poll that gets through. Both halves matter: unreported, an unreachable broker would look exactly like an empty queue — messages sitting in Redis, no consumer, no log — and reported at the full 100 ms rate it would mean ten error events a second per subscription.

The application logs every adapter onError through its own logger, so an operator sees it without writing an @OnQueueError handler. Adding one is still worth it when you want to act on the failure rather than read about it.

For delivery that survives the broker itself — persistence, server-tracked redelivery, durable consumers — use the NATS/JetStream adapter (@onebun/nats).

Custom adapter: NATS JetStream

To use a custom backend (e.g. NATS JetStream), pass the adapter constructor and options in queue:

typescript

import { OneBunApplication, type QueueAdapter } from '@onebun/core';
import { Module, Controller, BaseController, Subscribe, Message } from '@onebun/core';
import { AppModule } from './app.module';

// If you have an adapter class (e.g. from @onebun/nats or your own):
class NatsJetStreamAdapter implements QueueAdapter {
  readonly name = 'nats-jetstream';
  readonly type = 'jetstream';
  constructor(private opts: { servers: string; streams?: Array<{ name: string; subjects: string[] }> }) {}
  async connect() { /* connect to NATS */ }
  async disconnect() { /* disconnect */ }
  isConnected() { return true; }
  async publish() { return ''; }
  async publishBatch() { return []; }
  async subscribe() { return { unsubscribe: async () => {}, pause: () => {}, resume: () => {}, pattern: '', isActive: true }; }
  supports() { return false; }
  on() {}
  off() {}
}

@Controller('/jobs')
class JobHandler extends BaseController {
  @Subscribe('jobs.created')
  async handle(message: Message<{ id: string }>) {
    this.logger.info('Job', { id: message.data.id });
  }
}

@Module({ controllers: [JobHandler] })
class AppModule {}

const app = new OneBunApplication(AppModule, {
  port: 3000,
  queue: {
    adapter: NatsJetStreamAdapter,
    options: {
      servers: 'nats://localhost:4222',
      // A stream must bind every subject the application subscribes to. `JOBS` binds
      // `jobs.created`; with a real JetStream adapter, declaring only `events.>` here would
      // refuse to boot rather than attach the handler to a stream that never receives it.
      streams: [{ name: 'JOBS', subjects: ['jobs.>'] }],
    },
  },
});
await app.start();

The framework instantiates the adapter with new Adapter(queue.options) and uses it as the queue backend. When you pass a class constructor as adapter, options is automatically typed to match the adapter's constructor argument — no type assertions needed. For a ready-made NATS/JetStream adapter, use the @onebun/nats package if available and pass its adapter class and options the same way.

If your connect() can fail after it has opened a connection, close that connection in connect()'s own catch before rethrowing. A failed app.start() calls disconnect() as part of its rollback, but a disconnect() that returns early for an adapter that never finished connecting leaves the socket open, and the process never exits. The built-in adapters release it in connect().

Connecting to NATS

Both @onebun/nats adapters take the same connection options. servers is the only required one:

Option	Meaning
`servers`	Server URL, or a list of them. Required.
`name`	Connection name, as it appears in `nats server report connections`.
`token` / `user` / `pass`	Credentials.
`maxReconnectAttempts`	Reconnection attempts before the client gives up.
`reconnectTimeWait`	Milliseconds between reconnection attempts.
`timeout`	Connection timeout in milliseconds.
`tls`	`true` negotiates TLS with the driver's defaults.
`inboxPrefix`	Subject prefix for the inboxes this client generates.
`driverOptions`	Anything else nats.js accepts, merged last.

inboxPrefix: connecting to a multi-tenant broker

A broker that separates tenants by subject grants each user SUBSCRIBE on its own inbox space rather than the global one — ["<app>.>", "_INBOX_<tenant>_<app>.>"] and nothing wider. The driver's default inbox is _INBOX, which such a grant does not cover, and every JetStream operation is request/reply over the inbox: the manager's API calls, a publish's PubAck, every pull fetch. The result is not a degraded connection but one that opens and can then do nothing at all.

inboxPrefix is what the grant names:

typescript

import { OneBunApplication } from '@onebun/core';
import { JetStreamQueueAdapter } from '@onebun/nats';

const app = new OneBunApplication(AppModule, {
  queue: {
    adapter: JetStreamQueueAdapter,
    options: {
      servers: 'nats://nats.internal:4222',
      user: 'acme-api',
      pass: process.env.NATS_PASSWORD,
      inboxPrefix: '_INBOX_acme_api',
      streams: [{ name: 'ACME_EVENTS', subjects: ['acme.events.>'] }],
    },
  },
});

driverOptions: everything OneBun does not name

The options above are the ones OneBun names itself, and it translates where the two spellings disagree — tls: true becomes the driver's empty options object, for instance. driverOptions carries everything else @nats-io/transport-node accepts, and is merged last, so it also overrides anything above it:

typescript

import { JetStreamQueueAdapter } from '@onebun/nats';

const adapter = new JetStreamQueueAdapter({
  servers: 'nats://localhost:4222',
  streams: [{ name: 'EVENTS', subjects: ['events.>'] }],
  driverOptions: {
    noEcho: true,
    pingInterval: 30_000,
    // Full TLS settings, where the boolean above is not enough.
    tls: { caFile: '/etc/ssl/nats-ca.pem' },
  },
});

This exists so the named list is not a ceiling: without it, a nats.js option OneBun has not named yet could not be reached until @onebun/nats shipped a release that names it.

NatsQueueAdapter

NATS pub/sub for lightweight messaging (no persistence). Pass the adapter class in application options — the framework handles instantiation and connection automatically:

typescript

import { OneBunApplication } from '@onebun/core';
import { NatsQueueAdapter } from '@onebun/nats';

const app = new OneBunApplication(AppModule, {
  queue: {
    adapter: NatsQueueAdapter,
    options: { servers: 'nats://localhost:4222' },
  },
});
await app.start();

QueueService is automatically available for injection in any controller or service across all modules — no additional imports or configuration required.

Supported Features:

Pattern subscriptions
Consumer groups
Scheduled jobs

Acknowledgements do not exist here. Core NATS is fire-and-forget: the protocol has no acknowledgement, no redelivery and no delivery state, so message.ack() and message.nack() are permanent no-ops on this adapter and ackMode changes nothing on the wire. 'auto', 'manual' and 'none' are all accepted — 'none' is simply the only one that describes what this adapter really does. A handler written against the ackMode: 'manual' recipe therefore runs unchanged and silently accomplishes nothing; if you need the acknowledgements that recipe implies, the adapter you want is JetStreamQueueAdapter.

One thing the nack does still do is report. A handler that catches its own exception and calls message.nack() returns normally, so nothing distinguishes it from a success by control flow — the subscription reads the nack and emits onMessageFailed rather than onMessageProcessed, so a dropped message is visible to metrics and tracing instead of being counted as handled. The message itself is still gone: requeue cannot be honoured by a protocol that never redelivers.

JetStreamQueueAdapter

NATS JetStream for persistent, reliable messaging. Pass the adapter class in application options:

typescript

import { OneBunApplication } from '@onebun/core';
import { JetStreamQueueAdapter } from '@onebun/nats';

const app = new OneBunApplication(AppModule, {
  queue: {
    adapter: JetStreamQueueAdapter,
    options: {
      servers: 'nats://localhost:4222',
      streamDefaults: {
        retention: 'limits',
        storage: 'file',
        replicas: 1,
      },
      streams: [
        {
          name: 'EVENTS',
          subjects: ['events.>'],
        },
        {
          name: 'agent_events',
          subjects: ['agent.events.>'],
          maxAge: 7 * 24 * 60 * 60 * 1e9,
        },
        {
          name: 'agent_dlq',
          subjects: ['agent.dlq.>'],
          maxAge: 7 * 24 * 60 * 60 * 1e9,
          storage: 'memory',
        },
      ],
    },
  },
});
await app.start();

Declared streams are reconciled during startup, not recreated: a stream the server does not have yet is created, and a stream that already exists is reconciled only when its configuration hash changed. The reconcile runs inside the adapter's connect(): a stream the server refuses (replicas: 3 on a single node, a server without JetStream) fails app.start(), and the adapter closes the connection it opened before the error reaches you. streamDefaults is merged into each stream definition (per-stream values take priority). QueueService is automatically available for injection in any controller or service. When using @Subscribe('agent.events.task.done'), the adapter resolves the stream from these declarations — agent_events, whose agent.events.> binds it — and, since that subscription declares no group, its consumer is ephemeral: every process running it gets its own, and each one delivers only what is published after it starts, never the backlog already stored in agent_events. Passing adapter: JetStreamQueueAdapter alone enables the queue, so a producer-only service with zero @Subscribe handlers still connects to NATS during app.start().

A declaration says two things, and they can be separated

Declaring a stream means both "resolve my subscriptions against this" and "reconcile this on the broker". The second is not always this application's to do, so each half can be turned off on its own:

streams may be empty or omitted. That is a publish-only unit: it declares nothing, touches no stream, and can still publish anywhere the broker routes. A subscribe() on such an adapter is refused with a message saying why the two are not symmetric.
manage: false on a stream declares it for resolution only. No STREAM.INFO probe, no create, no update. This is how a stream that belongs to a platform operator, an infrastructure repository or another team is declared — a tenant-scoped user is usually granted no write on its configuration, and the reconcile pass would fail the boot rather than converge. It governs the stream's own config and nothing else: subscribe() still creates and reconciles its consumer on that stream, which is a separate grant ($JS.API.CONSUMER.*.<stream>.>) and the one a consuming tenant is normally given.
manageStreams: false is the same switch, adapter-wide. A per-stream manage overrides it in either direction, which is what a unit that owns some of its streams and only reads from the rest needs. streamDefaults: { manage: false } says the same thing and is likewise overridable per stream.

typescript

import { JetStreamQueueAdapter } from '@onebun/nats';

// Owns its own stream; reads from one the platform provisions.
const worker = new JetStreamQueueAdapter({
  servers: 'nats://localhost:4222',
  streams: [
    { name: 'WORKER_JOBS', subjects: ['worker.jobs.>'] },
    { name: 'PLATFORM_EVENTS', subjects: ['platform.events.>'], manage: false },
  ],
});

// Publishes into streams other units own, and declares nothing.
const producer = new JetStreamQueueAdapter({
  servers: 'nats://localhost:4222',
});

What turning reconciliation off costs you

The subject-narrowing guard, the create-only divergence guard and the create-if-missing branch all ride on the reconcile pass. An unmanaged stream that is missing, or bound to different subjects, is therefore not caught at app.start() — it surfaces at the first publish() or subscribe() instead. That is the trade: the declaration stops being an assertion about the broker and becomes a local fact about routing.

Stream resolution: exactly one, or the application does not start

consumers.add takes a stream name, so subscribe() has to choose one. It chooses only from the streams this application declares:

Streams whose declared subjects cover the whole pattern (natsSubjectCovers). Exactly one wins.
Otherwise streams that merely overlap it (natsSubjectsOverlap) — events.* against a stream declared ['events.created', 'events.updated'] really does deliver, so that stream qualifies.
Nothing matches, or more than one matches on either pass: app.start() throws, naming the pattern, the NATS subject it translated to, and every stream you declared.

An adapter that declares no streams at all is the degenerate case of rule 3: every subscribe() throws, and the message says that this is how a publish-only unit is configured rather than listing an empty set of candidates. A manage: false stream is a candidate like any other — turning reconciliation off does not take the declaration out of resolution, which is the whole reason to keep declaring it.

There is no broker-side check to fall back on

Measured against nats-server 2.10: a filter_subject completely unrelated to what the stream holds is accepted, stored verbatim, and its consumer sits at zero pending forever — a subscription that is alive, healthy and permanently empty, with nothing logged on either side. Falling back to some other declared stream when no declaration matches would produce exactly that. The declared stream set is the only oracle there is, which is why a miss is fatal rather than a warning.

Two declarations that both qualify is also refused. Picking one is not a neutral choice: the durable consumer name is derived from the group and the pattern and does not include the stream, so resolving differently on a later boot creates the same durable on another stream and orphans the first along with its delivery position — and streams carry their own retention, limits and storage. A catch-all archive stream declared beside topic streams is the usual way to hit it, as is a ORDERS ['orders.>'] / ORDERS_DLQ ['orders.dlq.>'] pair with @Subscribe('orders.dlq.failed'). Narrow the declarations until exactly one binds each subscribed subject, or drop the stream this service does not consume from.

publish() is not symmetric and needs no declaration at all. It addresses a subject and lets the server route it, so a producer-only service declares nothing — streams omitted entirely, and the reconcile pass has nothing to do. subscribe() cannot do that, because the API it calls demands a stream name; on an adapter with no declarations it refuses and says so rather than guessing.

Supported Features:

Pattern subscriptions
Consumer groups
Dead letter queue
Retry
Scheduled jobs

Reconciliation on connect. Each declared stream this application manages is reconciled during startup rather than blindly rewritten; one declared with manage: false is skipped entirely, probe included. The adapter first asks the server for the stream; only a genuine "stream not found" rejection counts as absence and leads to a create. Any other rejection — permissions denied, JetStream disabled, a transport timeout — is reported as itself, instead of being read as a missing stream and turned into the opaque stream name already in use that the follow-up create would raise. When the stream already exists, the update carries only the keys this application declared. A limit the declaration says nothing about is never sent, so maxMsgs, maxBytes and maxAge set out of band survive every connect: a stream pre-provisioned by an operator keeps the limits that operator gave it. retention, storage and the replica default apply on creation only — update cannot change the first two, and applying a default on update would rewrite a value the application never asked about. A limit or replica count declared in streamDefaults counts as declared and is sent on both paths; retention and storage are creation-only whether you declare them or not, and reach the divergence guard rather than the wire.

An unchanged stream configuration performs no write at all. A changed one is updated in place, carrying every pre-existing metadata key forward.

Replicas that start together. Several processes declaring the same stream the server does not have yet all find it absent, and all of them create it. The server accepts the first create and refuses the others with stream name already in use, because each create carries its own reconciliation timestamp. A refused replica does not fail: a peer created the stream first, so it reads the stream back and reconciles against it like a stream it had found on the first probe. The same guards apply, so a peer stream this declaration would narrow, or one with a different retention, still fails startup exactly as it would have one second later. This happens once per stream: a second refusal is reported as the create failure it is.

Startup fails loudly, and onError fires, when an existing stream cannot be reconciled safely — the adapter fails startup rather than diverging silently:

A declaration that would narrow an existing stream. Subject coverage is checked against the server's own subject list before anything else, so a stale stamp left by an out-of-band nats stream edit cannot wave a narrowing declaration through. The error names every subject the stream stores today that the declaration would no longer cover, because applying it would stop those subjects being stored and silently drop their messages. Widening is always allowed: every service sharing a stream must declare identical subjects, or at least a superset of what the stream already binds.
A storage or retention that differs from the server's. Neither can be changed on an existing stream, so this is checked before the configuration hash too — a matching hash must never mask a field an update cannot carry. The error names the diverging field and the only way forward, nats stream rm <STREAM>, which discards every message the stream holds.
A reconcile cycle. Two processes writing different definitions for the same stream, where the desired configuration was applied moments ago and has since been replaced; the error names the fields they disagree on.

The reconciliation stamp lives in JetStream stream metadata, so streams, like consumers, require nats-server 2.10 or newer. A rejection that names a server-version requirement is re-reported with that requirement stated.

Consumer defaults. consumerConfig supplies the values every subscription starts from:

typescript

const app = new OneBunApplication(AppModule, {
  queue: {
    adapter: JetStreamQueueAdapter,
    options: {
      servers: 'nats://localhost:4222',
      streams: [{ name: 'EVENTS', subjects: ['events.>'] }],
      consumerConfig: {
        ackWait: 30 * 1e9,  // nanoseconds the server waits for an acknowledgement
        maxDeliver: 5,      // deliveries before the message goes to the dead-letter queue
        maxAckPending: 200, // messages delivered but not yet acknowledged
      },
    },
  },
});

Per-subscription options win over these defaults, field by field:

max_ack_pending — the rule is prefetch overrides consumerConfig.maxAckPending, and 100 applies when neither is set. The pull batch is the smaller of the resolved max_ack_pending and prefetch (10 when prefetch is absent), so a batch can never outrun the acknowledgement window.
max_deliver — retry.attempts on @Subscribe, then deadLetter.maxRetries, override consumerConfig.maxDeliver, and 3 applies when none is set. retry.attempts wins over deadLetter.maxRetries when both are set, so adding a dead-letter queue to a subscription with a retry ladder does not change its max_deliver.
ack_wait — ackTimeout on @Subscribe overrides consumerConfig.ackWait, and 30 seconds (30_000_000_000 nanoseconds) applies when neither is set.

retry.backoff and retry.delay reach no consumer field at all — the adapter sends the computed wait with the nak instead, so ack_wait keeps meaning "how long the handler may hold a message" and the backoff keeps meaning "how long before the next attempt". JetStream's native backoff conflates the two: it is a list of per-delivery ack_wait values, so putting the ladder there would shrink the acknowledgement window to the base delay.

ackTimeout: how long a handler may hold a message. It is the per-subscription form of ack_wait — the window the server waits for an acknowledgement before it assumes the consumer died and redelivers. Mind the units: ackTimeout is in milliseconds, like every other duration in @onebun/core, while consumerConfig.ackWait is in nanoseconds, because it is a NATS-native value passed through untouched. The adapter converts, so ackTimeout: 30_000 and ackWait: 30 * 1e9 describe the same window.

Set it per handler when one subscription's work is slower than the rest: a handler that routinely runs longer than ack_wait is redelivered while it is still working, and the same message is then processed twice concurrently. Raising ackTimeout for that one handler costs nothing elsewhere.

Changing ackTimeout does not require deleting anything. It belongs to the hashed consumer configuration rather than to the durable name, so an existing durable consumer is reconciled in place on the next start, keeping its position and its backlog — no nats consumer rm, and no second consumer splitting the group's load. Two processes that disagree on it fail startup with a reconcile-cycle error naming ack_wait and both values, rather than rewriting each other's consumer on every deploy.

ackTimeout has no effect under ackMode: 'none': the server tracks no acknowledgements there, so ack_wait is never sent and there is no redelivery for it to govern.

Consumers are created with an explicit acknowledgement policy under ackMode: 'auto' and ackMode: 'manual', and with ack_policy: none under ackMode: 'none'. Between the first two, ackMode selects who acknowledges — the adapter on the handler's behalf, or the handler itself. 'none' is the one that decides whether the server tracks acknowledgements at all, and server-side tracking is what makes ackWait, maxDeliver, maxAckPending, message.ack(), message.nack(), retries and the dead-letter queue work.

The policy is fixed when the consumer is created and cannot be changed in place, so a durable whose stored policy disagrees with the mode its subscription declares fails startup naming both — in either direction. Changing a subscription's ackMode between 'none' and the other two therefore requires nats consumer rm on its durable.

Negative acknowledgement. nack(true) asks the server to redeliver the message immediately; it counts against maxDeliver, and once that is exhausted the message stops. nack(false), and the bare nack(), terminate the message instead: the server drops it permanently, it is never redelivered, and the delivery does not consume a maxDeliver attempt. Terminating does not move the payload anywhere — publish it yourself first if you need it kept.

A consumer that vanishes. If the consumer is deleted server-side — by an operator, by an ephemeral timing out, or by a stream being recreated — the client does not fail. It retries internally forever, so the subscription simply stops receiving while isConnected() keeps reporting healthy. The adapter therefore watches the consumer's notification channel: a deleted or missing consumer emits @OnQueueError and is re-created automatically from the same configuration the subscription was built with, and delivery resumes without a restart.

One consequence is unavoidable and worth planning for: deleting a consumer destroys its position on the server, so messages published while it was gone are not delivered to the re-created one. They remain in the stream, subject to its retention, but this subscription will not see them. Nothing the client does can recover a position the server no longer has. A work-queue stream is the exception: its consumers are deliver-all, so the re-created one receives every task still stored — those published while it was gone, and any that had exhausted their deliveries.

heartbeats_missed and stream_not_found are reported to @OnQueueError as well but are deliberately not repaired: a missing stream is an operator problem, and re-creating a consumer against a stream that does not exist cannot help.

Messages the adapter cannot read. A payload that fails to parse, or one that parses as JSON but carries no data field and so is not a OneBun envelope, is term()ed — terminated on the server and never redelivered — and the failure is emitted to @OnQueueError with the subject and the underlying parse error. It is not acknowledged and it is not retried: a payload that does not parse will not parse on a redelivery either, so maxDeliver attempts would only spend the acknowledgement window on a message that can never succeed. The message is never handed to your handler with an undefined payload. This most often means another producer publishes to a subject this application subscribes to.

Errors from the consume loop itself — a vanished consumer, a broken connection — reach @OnQueueError as well, so a failing consumer is never indistinguishable from an idle one.

Consumer identity. A subscription that declares a group gets a durable consumer named ${group}--${filterSubject}--${digest}: the group and the subject with every character outside [-\w] — the ., * and > a subject is made of among them — replaced by an underscore, followed by a short digest of the raw pair. The digest is not decoration. Sanitisation is lossy: orders.* and orders.> both reduce to orders__, as do orders.new and orders_new, and -- is itself legal inside a group name — without the digest those pairs would share one consumer. A durable is therefore per (group, pattern), not per group. Two handlers sharing a group but filtering different subjects are two consumers, each fed by its own subject; naming both after the group alone would let the second silently take the first's consumer over, so one subject would receive everything and the other nothing. The same pattern in the same group resolves to a single durable, which is exactly what lets several instances of a service share the work.

A subscription without a group is ephemeral. It is named after a freshly generated identifier no concurrent subscription can produce, it is never shared with another handler or another process, and it receives only messages published after the subscription starts: whatever the stream already holds is not replayed to it. A durable behaves the same way the first time it is created, and afterwards resumes where it left off — a consumer's name, its durability and its delivery policy are fixed when it is created, and reconciliation never rewrites them. On a stream declared retention: 'workqueue' neither holds: a group is required there, and the durable receives everything the stream holds — see Work-queue streams.

Reconciliation on subscribe. Each @Subscribe reconciles its consumer during startup rather than blindly creating one. The adapter first asks the server for the consumer; only a genuine "consumer not found" rejection counts as absence and leads to a create. Any other rejection — permissions denied, JetStream disabled, a transport timeout — is reported as itself instead of being mistaken for a missing consumer. When the consumer already exists, the adapter compares its stored configuration against the desired one, and an unchanged consumer configuration performs no update at all. A changed configuration is rewritten in place, carrying every pre-existing metadata key forward. Replicas of one group and pattern that start at the same moment all find the consumer absent and all create it. The server accepts the first create and refuses the others with consumer already exists, since each create carries its own reconciliation timestamp. A refused replica does not fail: it reads back the consumer its peer created and reconciles against it like a consumer it had found on the first probe. That happens once per subscription, and a second refusal is reported as the create failure it is. Startup fails loudly, and onError fires, when the existing consumer cannot be reconciled safely: a subscription without a group whose generated name is already taken, a consumer whose acknowledgement policy is not explicit, or a consumer that two processes are rewriting with different configurations in a reconcile cycle. On a work-queue stream it also fails, before asking the server anything, for the three shapes a workqueue cannot hold.

The reconciliation stamp lives in JetStream consumer metadata, which requires nats-server 2.10 or newer.

Publishing requires a bound subject. JetStream stores a message only if some stream on the broker binds its subject, and the server's own rejection for an unbound subject is jetstream is not enabled — which names neither the subject nor the streams, and sends operators looking for a JetStream that is disabled when it is not. The adapter re-reports it with the context the server omits, keeping the original rejection as the error's cause:

Failed to publish OneBun pattern "orders.created" to JetStream subject "orders.created". The most common cause is that no stream on the broker binds that subject. This application declares "EVENTS" (events.>). ...

There is no pre-flight check against the streams this application declares. A subject may legitimately be bound by a stream some other service owns, so the publish is always attempted and only the broker decides.

Graceful shutdown waits for running handlers. unsubscribe() and disconnect() do not resolve until the handler each subscription is currently executing has finished. This matters because an acknowledgement published after the connection closes is buffered and then dropped without raising anything: the handler completes, its ack() goes nowhere, the server redelivers after ackWait, and the work runs a second time on the next boot. Since QueueService.stop() unsubscribes every subscription during graceful shutdown, without the wait that would happen on every deploy — to exactly the handler shape that charges a card or sends an email.

The wait is bounded by a fixed 30 seconds, matching the default ackWait: past that point the server has already redelivered the message, so waiting longer cannot prevent the duplicate. A handler that has not returned by then is left running and shutdown proceeds without it. The timeout is not configurable — keep handlers shorter than ackWait, and if a unit of work genuinely takes longer, acknowledge it early and track its completion yourself.

Shutdown stops pulling before it waits: disconnect() marks every subscription as no longer running first, so each consume loop hands its next message back to the server instead of starting one more handler for the drain to wait on. At most one handler per subscription is ever in flight.

A consumer with the wrong acknowledgement policy must be deleted once

A consumer whose acknowledgement policy is not explicit tracks no acknowledgements: on it, ackWait, maxDeliver, maxAckPending, retries and the dead-letter queue are all inert, and a handler that throws is never redelivered. A consumer's acknowledgement policy cannot be changed in place, so under ackMode: 'auto' or 'manual' subscribe() refuses to start against such a consumer and names it in the error. Delete it once and let OneBun recreate it:

bash

nats consumer rm <STREAM> <CONSUMER>

Work-queue streams

A stream declared retention: 'workqueue' — on the stream itself or through streamDefaults — is a task queue: each message goes to exactly one consumer, and acknowledging it deletes it. nats-server enforces that shape with consumer rules of its own. OneBun derives what it can from the declaration, so there is no extra option to set:

typescript

import {
  BaseController,
  Controller,
  Module,
  OneBunApplication,
  Subscribe,
  type Message,
} from '@onebun/core';
import { JetStreamQueueAdapter } from '@onebun/nats';

@Controller('/tasks')
class TaskWorker extends BaseController {
  @Subscribe('tasks.run', { group: 'workers', ackMode: 'manual' })
  async run(message: Message<{ id: string }>): Promise<void> {
    // ... do the work ...
    await message.ack(); // on a workqueue, this deletes the task
  }
}

@Module({ controllers: [TaskWorker] })
class WorkerModule {}

const worker = new OneBunApplication(WorkerModule, {
  queue: {
    adapter: JetStreamQueueAdapter,
    options: {
      servers: 'nats://localhost:4222',
      streams: [
        { name: 'TASKS', subjects: ['tasks.>'], retention: 'workqueue' },
      ],
    },
  },
});

// Receives every task TASKS already holds — including those published before any worker
// ever subscribed — and then the new ones.
await worker.start();

What the declaration changes, and what it requires:

deliver_policy: all. Every consumer on the stream is created deliver-all, the only policy nats-server accepts on a workqueue. A task published before the first worker subscribed is delivered, and so is everything published while the workers were down. There is nothing to "replay": an acknowledged task no longer exists. Limits and interest streams keep deliver_policy: new, exactly as described under consumer identity.
A group is required. A workqueue allows one consumer per subject, so the group-less contract — a consumer of its own in every process, never shared — cannot exist there: the second replica would be refused, and so would a restart within seconds of a crash, while the dead process's consumer still holds the subject. subscribe() refuses a group-less subscription instead. Replicas with the same group and pattern share one durable and split the tasks, and they can all start at the same moment — see Reconciliation on subscribe above.
ackMode: 'none' is refused. A workqueue removes a task only when it is acknowledged, so the server requires explicit acknowledgements there. 'auto' and 'manual' both work.
One consumer per subject, stream-wide. A second group on the same subject, or an overlapping pattern in any group, is rejected by the server. That includes a dead-letter subject kept in the same stream: a subscription reading tasks.failed overlaps a worker on tasks.*, and whichever starts second is refused. Give dead letters a stream of their own. The startup error names the overlapping subscription when it is in this application. When it is not, the other consumer is most often a durable an earlier deployment left behind after its group was renamed or its pattern changed. Remove it with deleteDurableConsumer(oldPattern, oldGroup) or nats consumer rm. On a workqueue that loses no tasks: they live in the stream, not in the consumer, and the next consumer, being deliver-all, receives them.
Partial-token parameter patterns are refused. jobs.v{version} filters jobs.* on the wire, and the adapter acknowledges a delivered message the pattern does not match, such as jobs.x, so it cannot wedge the consumer. On a workqueue that acknowledgement deletes the task, silently. subscribe() refuses such a pattern there. A whole-token parameter (jobs.{version}) translates exactly and is fine; check the value in the handler.
Exhausted tasks stay stored. A task whose handler fails max_deliver times is not deleted. It stays in the stream, keeps counting against maxMsgs and maxBytes, and is never delivered to that consumer again. Raising retry.attempts or consumerConfig.maxDeliver updates the consumer in place and does not bring it back. Only a recreated consumer receives it — one you deleted, or one OneBun re-creates after it vanished — and then together with every other stranded task at once. Configure deadLetter: on the last attempt the task is republished to the dead-letter subject and the original is terminated, which on a workqueue deletes it.
term() deletes. A payload that does not parse, or one that is not a OneBun envelope, is terminated (see Messages the adapter cannot read above). On a workqueue that removes it from the stream for good, leaving the @OnQueueError event as the only trace. The same goes for nack(false) without a dead-letter queue. A producer that does not publish OneBun envelopes therefore should not share a workqueue subject with a OneBun worker.
Declare the owner's exact retention. The declaration is trusted, never probed. For a managed stream a declared retention that differs from the server's fails app.start(). A manage: false stream is not checked: declaring workqueue for a stream that is really limits makes a new durable deliver-all, and it replays the stream once. Declaring nothing, or limits, for a stream that really is a workqueue fails the subscription. The error says to declare retention: 'workqueue'.

An interest stream is not a workqueue: it keeps deliver_policy: new and none of the rules above, but it discards a message published while no consumer exists at all, so a service whose consumers are all down loses what is published meanwhile.

Consumer Lifecycle (JetStream)

A subscription that declares no group gets a framework-generated ephemeral consumer. Nothing outside the process can address it, so it is deleted when the subscription ends — on unsubscribe() and on disconnect(). Without that, every restart would leave another consumer behind on the server.

A subscription that declares a group gets a durable, and a durable is never deleted implicitly. That is deliberate rather than an omission: QueueService.stop() unsubscribes every subscription on graceful shutdown, so deleting durables here would discard their position on every deploy and redeliver everything already acknowledged. A durable exists precisely to survive restarts. Delete one when you actually mean to, with nats consumer rm <stream> <consumer>.

typescript

// Ephemeral: no group. The consumer is gone once the subscription ends.
const temporary = await queueService.subscribe('events.created', handler);
await temporary.unsubscribe();   // the server-side consumer is deleted

// Durable: group set. Survives unsubscribe, restart and redeploy.
const worker = await queueService.subscribe('events.created', handler, { group: 'workers' });
await worker.unsubscribe();      // the consumer, and its position, remain

Pausing loses nothing. pause() stops the adapter pulling new batches, and a message already pulled when the pause lands is nak()ed and redelivered on resume(), not abandoned — an abandoned message would occupy the acknowledgement window until ackWait expired.

A group names a permanent server resource — do not template it per run or per deploy. The durable is named from the group and the pattern together (${group}--${filterSubject}--${digest}), so a group built from a build number, a pod name or a timestamp creates a brand-new consumer every time and leaves the previous one behind forever. Streams accumulate them, each holding its own position and its own max_ack_pending budget. Pick a group that names the role — order-processors — not the deployment.

group also means two different things depending on the adapter, and the difference is not cosmetic. On NatsQueueAdapter it is a stateless NATS queue group: the broker round-robins between whoever is connected right now, and nothing survives them. On JetStream it provisions a durable consumer that outlives every process. The same option, the same value, one transient and one permanent.

Decommissioning one. When a durable genuinely has to go — a retired handler, or a trail left by a templated group — deleteDurableConsumer() is the supported way. It is not on the QueueAdapter interface, because only JetStream has durables, so it needs a downcast:

typescript

import { JetStreamQueueAdapter } from '@onebun/nats';

const adapter = queueService.getAdapter() as JetStreamQueueAdapter;

await adapter.deleteDurableConsumer('orders.created', 'order-processors');
// true  — the consumer was removed
// false — there was no such consumer; safe to call twice

It resolves the stream strictly: if no declared stream binds the pattern — or if more than one does — it throws and names every stream the application declares, rather than falling back to the first one. On a destructive call a mistyped pattern must not delete a consumer on an unrelated stream. A permissions denial is rethrown as itself — only a genuine consumer-not-found produces false.

In teardown, use tryDeleteDurableConsumer(). The strictness above is right for a call you make on purpose and wrong for one an afterEach makes unconditionally: on an adapter that never connected, after a case failed before connect(), or on a pattern nothing binds, the strict form throws — and a throw in teardown replaces the assertion failure in the output. The real breakage disappears behind a JetStream error from the cleanup.

typescript

afterEach(async () => {
  await adapter.tryDeleteDurableConsumer('orders.created', 'order-processors');
});

It never throws. Not connected is a quiet false — nothing was attempted, so reporting it would make every clean teardown noisy. Anything else that goes wrong, including a permissions denial or an unbound pattern, is a false plus an onError event: swallowed for the caller, not for a listener. It calls the strict form, so the two cannot drift.

Subscribing again with the same (pattern, group) after a delete creates a fresh durable under the same name. Fresh means fresh — except on a work-queue stream: on a limits or interest stream a new consumer carries deliver_policy: new, so it starts from the moment it is created and does not replay what the stream still holds, while on a stream declared retention: 'workqueue' it carries deliver_policy: all and receives every task still stored, because there a task lives in the stream rather than in the consumer (see Work-queue streams). Deleting a durable on a limits or interest stream therefore discards its position permanently — that is the cost, and it is why unsubscribe() never does it for you.

Dead Letter Queue

A message whose handler keeps failing eventually runs out of attempts. Without a dead-letter queue the server simply stops redelivering it and the application never hears about it again. deadLetter gives that message somewhere to go:

typescript

@Subscribe('orders.created', {
  deadLetter: { queue: 'orders.dlq', maxRetries: 5 },
})
async handleOrder(message: Message<OrderData>) {
  await this.processOrder(message.data);   // throws often enough and it lands in orders.dlq
}

How many attempts. maxRetries sets the server's max_deliver, behind retry.attempts. The full order is retry.attempts → deadLetter.maxRetries → consumerConfig.maxDeliver → 3. retry.attempts comes first, so a subscription that declares a retry ladder keeps exactly that max_deliver whether or not it also declares a dead-letter queue.

What happens on the last attempt. The payload is republished to deadLetter.queue first, and only once that succeeded is the original terminated. If the republish fails, the original is deliberately left alone — the server redelivers it or exhausts max_deliver exactly as it does without a dead-letter queue — and the failure is reported to @OnQueueError. Terminating first and republishing second would turn a failed republish into a lost message, which is the one outcome a dead-letter queue exists to prevent.

The same routing happens when a handler under ackMode: 'manual' calls message.nack(false) — or the bare message.nack(), which means the same thing. That is deliberate: nack(false) already means never deliver this again, so with a dead-letter queue configured it parks the message rather than discarding it. nack(true) still asks for redelivery.

What arrives in the queue. The republished message keeps the original id, and its metadata gains three provenance keys alongside whatever the producer set:

typescript

@Subscribe('orders.dlq')
async inspectFailures(message: Message<OrderData>) {
  message.metadata['dlq.originalPattern'];  // 'orders.created'
  message.metadata['dlq.deliveryCount'];    // 5
  message.metadata['dlq.error'];            // the handler's error message
}

deadLetter.queue must be a literal subject, and a declared stream must bind it. A wildcard (*, #, >) is rejected, because a message is published to exactly one subject; so is a queue equal to the subscription's own pattern, which would hand every dead letter straight back to the handler that just rejected it. Those two throw from subscribe() at startup, not on the first failure.

The stream requirement is on you, and is not checked at startup: dead letters are republished through publish(), which addresses a subject and lets the server route it rather than resolving a stream. An unbound dead-letter subject therefore surfaces the first time a message is actually dead-lettered, as a republish failure — the original is deliberately left un-terminated at that point, so it is redelivered rather than lost. The agent_dlq stream in the JetStreamQueueAdapter example above is the shape to copy — the dead-letter subject needs a stream just like any other.

What happens to the original. term() removes it under retention: 'workqueue' and 'interest'. Under the default retention: 'limits' the payload stays in the source stream until retention evicts it — terminating only stops redelivery. Either way the dead-letter copy is a normal stored message with its own retention, so it survives independently of the original. On a work-queue stream that removal is what you want: without a dead-letter queue a task that exhausts its deliveries is not deleted, it stays stored and undelivered.

Redis honours the same deadLetter options

The Redis adapter republishes to deadLetter.queue through its normal publish path, keeps the original id, adds the same three dlq.* metadata keys and dead-letters a message at most once — see RedisQueueAdapter. Two things differ: its attempt cap is retry.attempts ?? deadLetter.maxRetries ?? 1, with no consumerConfig step and a default of 1 rather than 3; and a republish that fails is reported through onError, but there is no server to redeliver the original, so that message is lost.

Feature Support Matrix

Feature	Memory	Redis	NATS	JetStream
Pattern subscriptions	✅	✅	✅ ¹	✅ ¹
Delayed messages	✅	✅	❌	❌
Priority	✅	✅	❌	❌
Consumer groups	❌	✅	✅	✅
Dead letter queue	❌	✅	❌	✅
Retry	✅	✅	❌	✅
Scheduled jobs	✅	✅	✅	✅
Persistence	❌	✅	❌	✅
Publish deduplication	❌	❌	❌	✅
Acknowledgments	✅	✅	❌ ²	✅

¹ On NATS and JetStream # must be the final token of a pattern — see Pattern Syntax.

² NatsQueueAdapter accepts every ackMode and rejects nothing, but core NATS has no acknowledgement protocol at all, so message.ack() and message.nack() are permanent no-ops there — see NatsQueueAdapter.

Publishing Messages

QueueService: availability and injection

QueueService is available via DI. You can inject it in the constructor of controllers, providers, WebSocket gateways, and middleware (e.g. constructor(private queueService: QueueService)). The framework registers a proxy in the module before creating controllers; when the queue is enabled during app.start(), the proxy delegates to the real QueueService. No wrapper or getQueueService() is required for normal DI-based code.

When is the real QueueService created?
The queue system is initialized during app.start(), after the module is set up, inside initializeQueue(). It is only created when the queue is enabled, which happens when any one of three conditions holds: at least one controller has queue decorators (@Subscribe, @Cron, @Interval, @Timeout); queue.enabled: true is present in application options; or a backend is explicitly configured via queue.adapter, queue.options or queue.redis. An explicit queue.enabled: false overrides a configured backend and keeps the queue disabled — one warning is logged and the adapter is never constructed.

If the queue is not enabled but you injected QueueService:
The injected instance is a proxy. Any call to a method (e.g. publish(), subscribe()) will throw an error with a message explaining how to enable the queue (register a controller with queue decorators, set queue.enabled: true in application options, or configure a backend via queue.adapter, queue.options or queue.redis — the message also states that an explicit queue.enabled: false overrides a configured backend and keeps the queue disabled).

Getting QueueService without DI:
Use app.getQueueService() when you do not have DI (e.g. bootstrap scripts or code that only has the app reference). It returns a QueueService and throws when there is none — with the same explanation the injected proxy gives, and naming which of the three states applies: never enabled, still starting, already stopped. It never returns null, so no caller has to null-check the result. Wrap it in try/catch if a missing queue is a case your code handles rather than a misconfiguration.

QueueService

Inject QueueService in your controller, provider, middleware, or gateway constructor:

typescript

import { QueueService } from '@onebun/core';

@Service()
class OrderService extends BaseService {
  constructor(private queue: QueueService) {
    super();
  }

  async createOrder(data: OrderData) {
    // Publish with options
    await this.queue.publish('orders.created', data, {
      delay: 1000,           // Delay 1 second
      priority: 10,          // Higher = more important
      messageId: 'custom-id',
      metadata: {
        authorization: 'Bearer token',
        serviceId: 'order-service',
        traceId: 'trace-123',
      },
    });

    // Batch publish
    await this.queue.publishBatch([
      { pattern: 'orders.created', data: order1 },
      { pattern: 'orders.created', data: order2 },
    ]);
  }
}

delay and priority together. delay decides when a message becomes deliverable and priority decides who goes first among those that are — the two never trade places. On the in-memory adapter a message due in a minute is delivered then, whatever priority a message due in an hour carries; messages that come due in the same pass are delivered highest priority first. A far-future high-priority message never holds back an earlier-due one.

messageId and deduplication. On JetStream the id is sent as the Nats-Msg-Id header, and the server refuses a second message carrying an id it has already seen inside the stream's deduplication window. That makes an outbox safe to replay: publishing the same logical message twice stores it once. The duplicate is dropped server-side and publish() still resolves normally — there is no error and no local signal, so do not treat a resolved publish as proof that a new message was stored.

Only an id you supply enables this. When messageId is omitted — or empty — the adapter generates one, uses it as Message.id, and sends no header: a generated id is unique per call and would only grow the server's deduplication index. On the memory, Redis and core-NATS adapters messageId is echoed back as Message.id and nothing more.

The window is per stream and defaults to two minutes, which is usually too short for a crash-and-restart cycle. Widen it with duplicateWindow, in nanoseconds:

typescript

streams: [
  {
    name: 'ORDERS',
    subjects: ['orders.>'],
    duplicateWindow: 10 * 60 * 1e9,   // 10 minutes
  },
]

It is sent only when you declare it, on both the create and the update path, so a window an operator set out of band is never reset by a deploy.

Dynamic Job Management

QueueService provides programmatic control over scheduled jobs at runtime.

Adding Jobs

typescript

// Cron job
queueService.addJob({
  type: 'cron',
  name: 'cleanup',
  expression: '0 * * * *',
  pattern: 'jobs.cleanup',
});

// Interval job
queueService.addJob({
  type: 'interval',
  name: 'heartbeat',
  intervalMs: 5000,
  pattern: 'jobs.heartbeat',
});

// Timeout job (one-time)
queueService.addJob({
  type: 'timeout',
  name: 'warmup',
  timeoutMs: 3000,
  pattern: 'jobs.warmup',
});

Querying Jobs

typescript

const job = queueService.getJob('cleanup');
const allJobs = queueService.getJobs();
const exists = queueService.hasJob('cleanup');

// Filter by origin — decorator-created vs dynamic
const decoratorJobs = allJobs.filter(j => j.declarative);
const dynamicJobs = allJobs.filter(j => !j.declarative);

ScheduledJobInfo fields: name, type, pattern, paused, declarative, schedule, lastRun, nextRun, isRunning. The declarative field is true for jobs created via @Cron/@Interval/@Timeout decorators, false for jobs added via addJob().

Controlling Jobs

typescript

queueService.pauseJob('cleanup');   // Pause
queueService.resumeJob('cleanup');  // Resume
queueService.removeJob('cleanup');  // Delete

Updating Jobs

typescript

queueService.updateJob({ type: 'cron', name: 'cleanup', expression: '*/5 * * * *' });
queueService.updateJob({ type: 'interval', name: 'heartbeat', intervalMs: 10000 });

Jobs created via decorators (@Cron, @Interval, @Timeout) are also accessible through this API by their name (defaults to method name, overridable via name option in decorator). The name has to be unique across the application, not just within a controller: registering a second job under a name already taken throws, naming both jobs and their patterns, so a job is never silently replaced.

Cron Parser

OneBun includes a built-in cron parser (no external dependencies).

typescript

import { 
  parseCronExpression,
  getNextRun,
  isValidCronExpression,
} from '@onebun/core';

// Parse expression
const schedule = parseCronExpression('0 30 9 * * 1-5');
// { seconds: [0], minutes: [30], hours: [9], ...,
//   daysOfMonthRestricted: false, daysOfWeekRestricted: true }

// Get next run time
const nextRun = getNextRun(schedule);
console.log('Next run:', nextRun);

// Validate expression
if (isValidCronExpression('0 0 * * *')) {
  console.log('Valid!');
}

Supported Syntax

Field	Values	Special
Seconds	0-59	``, `/N`, `N-M`, `N,M`
Minutes	0-59	``, `/N`, `N-M`, `N,M`
Hours	0-23	``, `/N`, `N-M`, `N,M`
Day of month	1-31	``, `/N`, `N-M`, `N,M`
Month	1-12	``, `/N`, `N-M`, `N,M`
Day of week	0-6 (0=Sun)	``, `/N`, `N-M`, `N,M`

The two day fields combine the way crontab(5) says they do, and it is not an AND:

Day of month	Day of week	A day matches when
restricted	restricted	either field matches — `0 0 1 * 1` is the 1st or any Monday
restricted	`*`	day-of-month matches
`*`	restricted	day-of-week matches
`*`	`*`	always

"Restricted" means written as something other than a wildcard. A leading * is unrestricted even with a step, so */2 in the day-of-week field does not switch the pair into OR mode. The parsed CronSchedule carries this as daysOfMonthRestricted and daysOfWeekRestricted, because the value arrays cannot express it — * and 1-31 parse to the same 31 numbers.

Pattern Matcher

typescript

import {
  matchQueuePattern,
  isQueuePatternMatch,
  createQueuePatternMatcher,
} from '@onebun/core';

// Match with parameter extraction
const result = matchQueuePattern('orders.{id}.status', 'orders.123.status');
// { matched: true, params: { id: '123' } }

// Simple match check
if (isQueuePatternMatch('events.*', 'events.user')) {
  console.log('Matched!');
}

// Create reusable matcher (optimized)
const matcher = createQueuePatternMatcher('orders.*.status');
matcher('orders.123.status'); // { matched: true, params: {} }
matcher('orders.456.status'); // { matched: true, params: {} }

Queue API ​

Overview ​

Setup ​

Application Configuration ​

QueueApplicationOptions ​

Registering Controllers with Queue Decorators ​

Interval overlap and the first run ​

Error Handling in Handlers ​

Quick Start ​

Subscribe Decorator ​

Pattern Syntax ​

Subscribe Options ​

retry ​

ackMode: 'none' ​

Message Interface ​

Scheduling Decorators ​

@Cron ​

CronExpression Constants ​

@Interval ​

@Timeout ​

Message Guards ​

Built-in Guards ​

Composite Guards ​

Custom Guards ​

Interceptors ​

Using QueueService from module lifecycle hooks ​

Lifecycle Decorators ​

Queue Adapters ​

InMemoryQueueAdapter ​

RedisQueueAdapter ​

Custom adapter: NATS JetStream ​

Connecting to NATS ​

inboxPrefix: connecting to a multi-tenant broker ​

driverOptions: everything OneBun does not name ​

NatsQueueAdapter ​

JetStreamQueueAdapter ​

A declaration says two things, and they can be separated ​

Stream resolution: exactly one, or the application does not start ​

Work-queue streams ​

Consumer Lifecycle (JetStream) ​

Dead Letter Queue ​

Feature Support Matrix ​

Publishing Messages ​

QueueService: availability and injection ​

QueueService ​

Dynamic Job Management ​

Adding Jobs ​

Querying Jobs ​

Controlling Jobs ​

Updating Jobs ​

Cron Parser ​

Supported Syntax ​

Pattern Matcher ​

Queue API

Overview

Setup

Application Configuration

QueueApplicationOptions

Registering Controllers with Queue Decorators

Interval overlap and the first run

Error Handling in Handlers

Quick Start

Subscribe Decorator

Pattern Syntax

Subscribe Options

retry

ackMode: 'none'

Message Interface

Scheduling Decorators

@Cron

CronExpression Constants

@Interval

@Timeout

Message Guards

Built-in Guards

Composite Guards

Custom Guards

Interceptors

Using QueueService from module lifecycle hooks

Lifecycle Decorators

Queue Adapters

InMemoryQueueAdapter

RedisQueueAdapter

Custom adapter: NATS JetStream

Connecting to NATS

inboxPrefix: connecting to a multi-tenant broker

driverOptions: everything OneBun does not name

NatsQueueAdapter

JetStreamQueueAdapter

A declaration says two things, and they can be separated

Stream resolution: exactly one, or the application does not start

Work-queue streams

Consumer Lifecycle (JetStream)

Dead Letter Queue

Feature Support Matrix

Publishing Messages

QueueService: availability and injection

QueueService

Dynamic Job Management

Adding Jobs

Querying Jobs

Controlling Jobs

Updating Jobs

Cron Parser

Supported Syntax

Pattern Matcher