Queue Plugin

Background job queue plugin for Setu-TS with Memory, Redis, RabbitMQ, and SQS adapters.

Installation

deno add npm:@setu-ts/queue-plugin

Usage

Basic Setup (Memory Adapter)

import { QueuePlugin } from '@setu-ts/queue-plugin';

app.register(QueuePlugin({ adapter: 'memory' }));

Redis Adapter

import { QueuePlugin } from '@setu-ts/queue-plugin';

app.register(
  QueuePlugin({
    adapter: 'redis',
    url: 'redis://localhost:6379',
  }),
);

Connection errors of the ioredis client the adapter builds go to the application logger — the first error of an outage at warn, identical repeats at debug, the recovery at info — instead of ioredis printing every reconnect failure to console.error. An injected client gets no listener: it belongs to the caller.

node-redis. To use node-redis instead of ioredis, pass QueuePlugin({ adapter: 'redis', client: fromNodeRedis(createClient({ url })) }) (fromNodeRedis from @setu-ts/common, createClient from redis, installed in your application). The adapter provides eval, so every transition still runs as one script. See Using node-redis.

RabbitMQ Adapter

import { QueuePlugin } from '@setu-ts/queue-plugin';

app.register(
  QueuePlugin({
    adapter: 'rabbitmq',
    url: 'amqp://localhost:5672',
    // Queue names are derived per job name from this prefix
    // (`<prefix>.<name>.ready` / `.delay` / `.dead`); there is no `queues` map.
    prefix: 'he.queue',
  }),
);

Every job is published persistent through a confirm channel. The ready, delay and dead queues are durable, and a durable queue keeps only persistent messages across a broker restart — before 0.9.0 jobs were published transient, so a RabbitMQ restart silently discarded every job not yet processed. add() resolves only once RabbitMQ has accepted the job, and a retry or dead-letter acknowledges the reserved job only after its replacement is accepted, so a broker failure between the two can run a job twice but cannot lose it. publishTimeoutMs (default 15000) bounds each publish, queue declarations included, so a paused broker rejects add() rather than leaving it pending; a rejection is not proof the job was dropped. persistentMessages: false restores the transient publishes. An injected client without createConfirmChannel() keeps a plain channel and the adapter logs one warning saying its publishes are unconfirmed.

A queue name is an AMQP short string of at most 255 UTF-8 bytes, so <prefix>.<name>.delay must fit: with the default prefix a job name may be at most 240 bytes. A longer name is refused with a RangeError before the channel is touched — by add(), and at registration by process() and addRecurring(), so a declared processor with such a name stops start() rather than failing every poll — and a prefix over 247 bytes is refused at construction. Before 0.9.0 an oversized name reached amqplib, which had already claimed the channel’s reply slot, so every later queue declaration on the channel waited forever.

Independently of the backend, one job name whose reserve or recurring enqueue fails is reported and skipped for that tick; the processors and recurring jobs after it still run, and a recurring job whose enqueue failed stays due and is retried on the next tick.

SQS Adapter

import { QueuePlugin } from '@setu-ts/queue-plugin';

app.register(
  QueuePlugin({
    adapter: 'sqs',
    sqs: {
      region: 'us-east-1',
      queues: { default: 'https://sqs.us-east-1.amazonaws.com/123456789012/tasks' },
    },
  }),
);

Named Instances

import { QueuePlugin } from '@setu-ts/queue-plugin';

// Foreground queue
app.register(QueuePlugin({ adapter: 'memory', name: 'foreground' }));

// Background queue
app.register(QueuePlugin({ adapter: 'memory', name: 'background' }));

API

Adding Jobs

import { CAPABILITIES } from '@setu-ts/common';
import type { IQueue } from '@setu-ts/common';

const queue = ctx.services.get<IQueue>(CAPABILITIES.QUEUE);

// Basic job
await queue.add('send-email', { to: 'user@example.com' });

// Delayed job
await queue.add('send-email', { to: 'user@example.com' }, { delayMs: 5000 });

// Job with retry limit
await queue.add('process-data', { rows: 100 }, { maxAttempts: 5 });

Processing Jobs

import { CAPABILITIES } from '@setu-ts/common';
import type { ILogger, IMailer, IQueue } from '@setu-ts/common';

const queue = ctx.services.get<IQueue>(CAPABILITIES.QUEUE);
const mailer = ctx.services.get<IMailer>(CAPABILITIES.MAIL);
const logger = ctx.services.get<ILogger>(CAPABILITIES.LOGGER);

interface SendEmail {
  to: string;
}

queue.process<SendEmail>(
  'send-email',
  async (job) => {
    await mailer.send({ to: job.data.to, subject: 'Welcome', text: 'Hello' });
  },
  {
    concurrency: 3,
    // Invoked once when a job has exhausted its attempts, immediately before it
    // is dead-lettered — the only programmatic notice that work was abandoned.
    onFailed: (job, error) => {
      logger.error('job dead-lettered', { id: job.id, name: job.name, error: String(error) });
    },
  },
);

Recurring Jobs

import { CAPABILITIES } from '@setu-ts/common';
import type { IQueue } from '@setu-ts/common';

const queue = ctx.services.get<IQueue>(CAPABILITIES.QUEUE);

await queue.addRecurring('cleanup', {}, { cron: '0 0 * * *' }); // Daily at midnight

Options

Two options this table used to list do not exist and never did: region (SQS configuration lives under sqs) and queues (RabbitMQ derives its queue names from prefix).

QueueProcessorDefinition<T = unknown> uses method syntax: processor(job: IJob<T>): void | Promise<void>. Typed payload processors fit the declarative list, and methods retain their definition as this. The processor member is writable.

Declarative processors coexist with imperative queue.process() calls. Behaviours run in declared order; a short circuit acknowledges the job, while a throw follows the existing retry, final onFailed, and dead-letter path. With no behaviours, the processor receives the original job directly and no chain is allocated.

A behaviour also sees IngressContext.consumer, which the queue service sets to job.name — the dispatch identity a behaviour keys per-consumer state on (an idempotency record, for example). It is always present for a delivered job.

Retry span vs an ingress lease. A behaviour that claims a job for the duration of its work — the idempotentIngress() behaviour is the one shipped — refuses a redelivery inside the claim’s lease as in-progress, and every refusal consumes an attempt. With the default 3 attempts and the 2 s / 4 s backoff the retry span is 6 s, shorter than the default 30 s ingress lease, so a crashed holder’s job is dead-lettered rather than retried. For a job name you make idempotent, keep the queue’s retry span ABOVE the lease: with the default backoff that means defaultMaxAttempts ≥ 6.

Queue observations (M98f)

diagnostics exposes actual attempts and supported depths as minimized observations through the local diagnostics connector’s GET /v1/queues (@setu-ts/diagnostics-plugin). It never changes a processor, a retry, onFailed, or a settlement.

import { QueuePlugin } from '@setu-ts/queue-plugin';

app.register(
  QueuePlugin({
    adapter: 'memory',
    diagnostics: {
      enabled: true,
      instanceAlias: 'mailer',
      // Exact job-name -> display-alias allowlist; unlisted names are neither observed nor counted.
      queues: { 'send-welcome-email': 'welcome-emails' },
      // Optional, separately bounded depth counting; absent by default.
      depths: { intervalMs: 30000, timeoutMs: 2000, concurrency: 2 },
    },
  }),
);

Every instance — named or not, observed or not — registers one source under CAPABILITIES.QUEUE_DIAGNOSTICS as a multi provider, so named instances stay independently observable. Without diagnostics that source answers disabled and dispatch is untouched. enabled must be the literal true; aliases are 1–64 UTF-8 bytes with no control character (queue aliases unique, at most 64 queues), and approving an alias is authorizing its disclosure.

Each observed attempt carries the approved queue alias, a session-local j<N> job alias, the attempt number, monotonic timing, the processor outcome and the settlement — recorded only after the adapter’s settlement call returned, never inferred from the outcome. No payload, header, raw job id, claim token, credential or error text is captured.

Depth counting runs one non-overlapping cycle at bootstrap and one per intervalMs, over approved job names this instance has a processor for; a count still pending after timeoutMs is reported as timed out but keeps its slot until it settles. shared-backend counts from several instances or replicas are the same inventory — never sum them.

A depth row describes the latest cycle only. A name that cycle did not read (timed out, failed, or not reached because every slot was still held) has no row at all; the source’s failure and depthCoverage: 'partial' say why. A depth that cannot be read is never reported as a retained zero.

Seeing a job that failed for the last time

A job that exhausts its attempts is dead-lettered by every adapter, and used to be reachable only by opening a Redis client: IQueue has no getJob and no dead-letter accessor, the health payload was {"adapter":"RedisQueue"}, and /metrics carried no queue series at all. Three surfaces answer different questions about it:

The counter needs @setu-ts/metrics-plugin registered; without it no collector is built and behaviour is unchanged. Depths are reported by the memory and Redis adapters and OMITTED (never reported as zeros) on RabbitMQ and SQS, whose counts need a management API.

deadLetterTtlMs bounds how long the retained payload lives. Without it the jobs hash grows for the lifetime of the deployment — the retention exists for debugging, so dropping it by default would remove the value it is there for.

Retention is enforced per payload: each dead-letter sweeps the dead set — which is scored by dead-letter time — and deletes every entry older than the TTL along with its payload. A key-level EXPIRE alone would not do this, because each new dead-letter renews it, so a queue that keeps failing would retain its oldest payloads forever. The key-level expiry is kept beside the sweep as the backstop for the case the sweep cannot reach: a queue that dead-letters once and then goes quiet never runs another sweep.

The bound this gives is at least the TTL, and at most the TTL past the last dead-letter — so a payload that dead-letters just before a short burst that then stops can live for just under twice the TTL. Two mechanics produce that: the sweep only runs when a dead-letter arrives, so a payload is dropped at the first dead-letter after its deadline rather than at the deadline itself; and the backstop is armed from the dead-letter that armed it, because one shared key carries one deadline and arming it from the oldest survivor would take the newer payloads with it before their own TTL elapsed. It errs late by construction: dropping a payload early would discard exactly the debugging data the option exists to keep. Exact per-payload expiry is not expressible here — Redis has no per-member TTL on a sorted set, and the payloads share one hash.

With a client exposing eval — always true of the client the adapter builds — the payload move and the dead-set insert run as one Lua script and are atomic (M101a). The rest of this paragraph describes an injected client without eval, which keeps separate commands. The payload move is issued before the dead-set insert, and that order is load-bearing: no two of these commands are atomic, and a sweep starts from the dead set, so a member that became visible before its payload was written can be swept by a concurrent deadLetter — deleting the member and stranding the payload where no later sweep can reach it. Writing first means a sweep either misses the id, and a later one collects it, or finds both together. Ordering closes that window; it does not make the sequence atomic, so a process that dies mid-sequence can still leave a payload behind.

Setting it MOVES a dead job’s payload from queue:<name>:jobs into queue:<name>:dead:jobs, and the expiry is applied to that key and the dead set. It is never applied to the live jobs hash: that key holds the payload of every job for the name, Redis keeps a key’s TTL across later writes, and a job whose payload has vanished cannot be returned by reserve — so expiring it would silently discard work that was only waiting to run.

Adapters

MemoryQueue

In-memory queue for testing and local development. Jobs are lost on restart.

Redis key namespaces

Set keyPrefix on QueuePlugin({ adapter: 'redis', keyPrefix: 'orders:' }) or new RedisQueue({ keyPrefix: 'orders:' }) to isolate applications sharing one Redis. The default remains queue: so existing jobs are found after upgrading. Every key, including recurring and dead-letter keys, uses the prefix. It must be non-empty, at most 128 characters and contain no control characters; invalid values are refused at construction. The RabbitMQ prefix option is separate and does not move Redis keys. Changing a deployed prefix requires draining or migrating its existing jobs.

Keys are the prefix followed by the job name, and job names may contain :. Isolation therefore holds only when no prefix on one Redis is a string prefix of another: app: with a job named b:x and app:b: with a job named x both use app:b:x:*. Choose sibling prefixes such as orders: and billing:, never nested ones. The adapter cannot check this, because one application never sees another’s prefix.

RabbitMQ recovery

Connection and channel faults start automatic reconnect with full-jitter backoff. Readiness is down during the fault window. An injected connection receives a fresh channel only; the caller owns reconnecting that connection. Queues are re-declared lazily after recovery. RabbitMQ returns unacked jobs for at-least-once redelivery: a job can run twice across an outage, and its redelivery can carry the original attempt count. Make processors idempotent. Disconnect stops scheduled reconnects and removes listeners. Each delivery has a fresh claim token; a pre-outage delivery cannot settle the redelivered message. A late ack, retry or dead-letter rejects with the internal StaleClaimError, reported by the service as a failed settlement.

RedisQueue

Redis-backed queue using sorted sets for delayed job storage. Supports persistence and distributed processing.

Since M101a each transition (enqueue, reserve, ack, requeue, dead-lettering) runs as one Lua script through the client’s eval, so a command that times out locally cannot leave a job half moved. An injected client without eval keeps the separate commands. The scripts touch several <keyPrefix><name>:* keys with no hash tag, so a Redis Cluster client is not supported: it would refuse them with CROSSSLOT.

Claims and dead workers

Redis 5 or later is required: the lease scripts use TIME before writes under script-effects replication. RedisQueueOptions.visibilityTimeoutMs and the flat QueuePluginOptions.visibilityTimeoutMs default to 30000. A duration from 1000 to 2147483647 is required, so renewal never runs more often than every 333 ms; invalid values throw RangeError at construction. Each reservation sets its lease deadline by Redis server time and mints a token with INCR <keyPrefix><name>:claimseq. Which jobs are due is decided by the reserving replica’s own clock, the same clock that wrote availableAtMs, so clock skew between an application host and Redis shifts neither delays nor retries. Reclaimed jobs become due at once. The service renews the claim every Math.floor(visibilityTimeoutMs / 3) ms while the processor runs, without overlapping renewal calls. A failed or false renewal is reported once per delivery; the processor continues.

On reserve, up to the requested limit of expired claims return to ready. Each reclaim counts as an attempt: a delivery has max(stored attempts, 1) + stalls. Once that exceeds maxAttempts, dispatch calls onFailed (if configured) and then dead-letters it without running the processor, so an abandoned job sends the same signal as a final failed attempt. The claim is renewed while onFailed runs, as it is during a processor, so a slow callback is not reclaimed; one that never settles holds its concurrency slot, like a processor that never settles. Orphan ids without payloads are removed. A timed-out reserve that Redis eventually applies is recovered by this same lease. Settlement clears claims and stalls; a stale ack, retry or dead-letter rejects with internal StaleClaimError and the service reports a failed settlement, leaving the current delivery intact. Payload bytes are unchanged by reclaim.

Delivery is at least once: a process death, GC pause or event-loop stall longer than the lease can run a job twice. Make processors idempotent and raise the lease for expected pauses. An alive but stuck processor keeps renewing and is not reclaimed. Drain old replicas before upgrading, since pre-upgrade processing entries are immediately reclaimable. Claim tokens fence deliveries; they are not authorization. A Redis writer can read them and already has access to delete jobs; Redis ACLs are the security boundary.

An injected client without eval keeps the old multi-command behavior, with no claims or reclaim. connect() reports once through the connection-error reporter that visibilityTimeoutMs is inactive. The client the adapter builds always has eval.

RabbitMqQueue

RabbitMQ-backed queue polling per-name durable queues through the default exchange. Delayed jobs and retries wait in a per-name delay queue whose message TTL dead-letters them into the ready queue. Jobs are published persistent through a confirm channel (see RabbitMQ Adapter). Requires the amqplib package at runtime (or an injected IAmqpQueueConnection). Each connection attempt the adapter makes is bounded at 10 seconds in total, TCP and AMQP handshake together, so a broker that never answers, or answers too slowly, is retried rather than waited on; a connection that opens after its attempt expired is closed at once. A message in a ready queue whose body is not a stored job (not JSON, not an object, or missing a string id or name, data, or a finite attempts, maxAttempts or availableAtMs) is moved with its bytes unchanged into the dead queue, acknowledged, and reported once as a warning that does not quote the body; the rest of the batch is reserved normally. If the dead queue refuses the move (a reject-publish length policy, or a blocked broker past publishTimeoutMs), the warning says so, the jobs already claimed in that call are returned, fetching stops, and the message stays unacknowledged until the channel closes, when RabbitMQ redelivers it and the move is tried again.

SqsQueue

AWS SQS-backed queue using the AWS SDK @aws-sdk/client-sqs. Supports visibility timeouts, delayed jobs (up to 15 minutes via DelaySeconds), and manual DLQ forwarding. The maxAttempts field in the job envelope drives the DLQ promotion logic. Requires the @aws-sdk/client-sqs package at runtime (or an injected transport).

SQS Retry and DLQ

When a job reaches maxAttempts, SqsQueue.deadLetter() forwards the original message body to the configured DLQ queue URL (via sqs.deadLetterQueues), then deletes the source message. If DLQ send fails, the source message is not deleted (preventing silent data loss). If the source delete fails after a successful DLQ send, a duplicate-risk warning is logged. Note: AWS SQS does not support managed DLQ redrive for this adapter — DLQ forwarding is manual via the deadLetterQueues option.

ElasticMQ E2E

The SQS adapter includes guarded E2E tests against ElasticMQ (SQS_ENDPOINT_URL environment variable). These tests verify enqueue→reserve→ack round-trips, queue isolation, visibility retry progression, and DLQ promotion.

Trace propagation

A queue job is deferred work whose whole diagnostic value is “which request caused this”. With a CAPABILITIES.TELEMETRY provider registered, a job joins the trace of the request that enqueued it: queue.add() runs inside an enqueue <name> producer span and writes its W3C traceparent into AddJobOptions.headers, and every processor runs inside a process <name> consumer span parented from IJob.headers.

Every registration route is covered — the processors arm, a processors factory, and an imperative queue.process() through the resolved capability — and all four adapters carry the map end to end. An M86 ingress behaviour reads it as IngressContext.headers, the same member the 'messaging' arm uses.

import { QueuePlugin } from '@setu-ts/queue-plugin';
import { TelemetryPlugin } from '@setu-ts/telemetry-plugin';
import { CAPABILITIES, type IQueue } from '@setu-ts/common';

app.register(
  TelemetryPlugin({
    serviceName: 'orders',
    exporter: 'otlp',
    endpoint: 'http://localhost:4318/v1/traces',
  }),
);
app.register(QueuePlugin({ adapter: 'redis', url: 'redis://localhost:6379' }));

app.router.post('/orders', async (ctx) => {
  const queue = ctx.services.get<IQueue>(CAPABILITIES.QUEUE);
  // Both the enqueue and the eventual processing join THIS request's trace.
  await queue.add('order.export', { id: 'o-1' }, { headers: { 'x-tenant': 't-1' } });
  return ctx.response.json({ accepted: true });
});

headers is a public option and its delivery never depends on which capabilities are registered: a map you pass reaches the processor with or without telemetry. What telemetry adds is the traceparent alongside it.

Semantics. IJob.headers mirrors MessageMetadata.headers exactly — {} means the channel was read and carried nothing; absent means there was no channel. It is never substituted, so “this job had no cause recorded” stays distinguishable from “the cause was lost”.

Without telemetry nothing is injected and behaviour is byte-identical to before. A job carrying no traceparent starts a root span rather than failing. addRecurring is deliberately untraced: a recurring job fires on a schedule, so the call that registered it is not the cause of any run.

Known limitation. With behaviors configured, a behaviour runs OUTSIDE the consumer span, because the behaviour chain is applied by a service subclass that is necessarily outermost at dispatch. The processor’s own work is correctly parented regardless, and with no behaviours configured the question does not arise.

License

MIT

Health indicator

Registered under the queue’s capability token. Since M70c it reports two signals: the adapter’s lifecycle (isReady()) and its reachability (isHealthy()).

data reports { adapter, reachable }, where reachable is true, false, or 'unknown' when the adapter has no liveness check.

Probes and readiness (M119b, V10-22). The reachability check stays cached for 5 s and bounded at 2 s. The RabbitMQ adapter no longer reads a supervisor flag — its check opens and closes a throwaway AMQP channel on the connection, so a paused broker can no longer report reachable: true; a refused round trip answers false. For every queue adapter, a check that does not answer within the 2 s bound now reports up with reachable: 'unknown' and no depth counts, instead of down: a hung probe is not evidence the backend is down. A check that answers false — a stopped broker — still reads down at once. The indicator counts toward readiness by default; the health plugin’s readiness option overrides it by name.

Exports

Generated from the package barrel by pnpm run docs:exports; pnpm run check:docs fails when it drifts.

Full API

Every export and option is documented in PUBLIC_API.md.

API reference

Queue (@setu-ts/queue-plugin)

Provides background job queue with Memory and Redis adapters.

Exports

• QueuePlugin — Plugin factory for registering the queue service
• QueueAdapterType — 'memory' | 'redis' | 'rabbitmq' | 'sqs'
• QueuePluginOptions — Plugin configuration options (includes client, url, prefix?, sqs?)
• MemoryQueue — In-memory queue adapter for development/testing
• RedisQueue — Redis-backed queue adapter for production
• RedisQueueOptions — Redis adapter configuration, including keyPrefix? (default queue:) and visibilityTimeoutMs? (default 30000)
• RabbitMqQueue — RabbitMQ queue adapter via amqplib (polling via basicGet, TTL+DLX for delays)
• RabbitMqQueueOptions — RabbitMQ adapter configuration (includes url, client, prefix?, persistentMessages?, publishTimeoutMs?, reportWarning?)
• SqsQueue — AWS SQS queue adapter via @aws-sdk/client-sqs (receipt-handle bookkeeping, ApproximateReceiveCount attempt ladder, visibility-timeout backoff, dead-letter ordering)
• SqsQueueOptions — SQS adapter configuration (queues, deadLetterQueues?, region?, credentials?, endpoint?, client?)
• ISqsTransport — Structural SQS transport port (injected via SqsQueueOptions.client)
• SqsReceivedMessage — SQS received message shape (body, receiptHandle, approximateReceiveCount)
• SnsPublisher — SNS publisher for fan-out (SNS→SQS pairing)
• SnsPublisherOptions — SNS publisher configuration
• ISnsTransport — Structural SNS transport port
• adaptSqsModule / loadSqsModule — SQS SDK adapter and lazy loader
• SqsSdkModule — Shape of the SQS SDK module adaptSqsModule consumes; exported so a consumer can type a substitute module
• adaptSnsModule / loadSnsModule — SNS SDK adapter and lazy loader
• SnsSdkModule — Shape of the SNS SDK module adaptSnsModule consumes
• QueueBackendUnavailableError — Thrown when a cloud queue backend is unavailable (e.g., SQS on Cloudflare Workers)
• SqsDelayTooLongError — Thrown when SQS delay exceeds 900 s
• SqsQueueNotConfiguredError — Thrown when a job name has no queue URL mapping
• IQueue — Queue service interface (re-exported from @setu-ts/common)
• IJob<T> — Job interface (re-exported)
• JobProcessor<T> — Job processor type (re-exported)
• AddJobOptions — Options for queue.add() (re-exported)
• ProcessOptions — Options for queue.process() (re-exported)
• QueueDepths — { ready, processing, dead } per job name, as the health indicator publishes them
• RecurringOptions — Options for queue.addRecurring() (re-exported)
• QueueLogger — Minimal error/warn logger surface the service reports background failures through (structurally compatible with ILogger)
• QueueDiagnosticsOptions — Opt-in minimized attempt/depth observations for the diagnostics connector (M98f): { enabled: true, instanceAlias, queues, depths? }
• QueueDepthDiagnosticsOptions — Separately bounded depth collection: { intervalMs, timeoutMs, concurrency }

Registration

import { QueuePlugin } from '@setu-ts/queue-plugin';

// Memory adapter (development/testing)
app.register(
  QueuePlugin({
    adapter: 'memory',
    pollIntervalMs: 1000,
    defaultMaxAttempts: 3,
  }),
);

// Redis adapter (production)
app.register(
  QueuePlugin({
    adapter: 'redis',
    url: config.get('REDIS_URL'),
    pollIntervalMs: 1000,
    defaultMaxAttempts: 3,
  }),
);

// Named instance for multi-queue support
app.register(
  QueuePlugin({
    adapter: 'memory',
    name: 'background',
    pollIntervalMs: 2000,
  }),
);

// RabbitMQ adapter (production, requires amqplib)
app.register(
  QueuePlugin({
    adapter: 'rabbitmq',
    url: config.get('RABBITMQ_URL'),
    prefix: 'myapp.queue',
    pollIntervalMs: 1000,
    defaultMaxAttempts: 3,
  }),
);

// SQS adapter (production, requires @aws-sdk/client-sqs)
app.register(
  QueuePlugin({
    adapter: 'sqs',
    sqs: {
      region: 'us-east-1',
      queues: {
        'send-welcome-email': 'https://sqs.us-east-1.amazonaws.com/123456789012/welcome-emails',
        'process-payment': 'https://sqs.us-east-1.amazonaws.com/123456789012/payments',
      },
      deadLetterQueues: {
        'process-payment': 'https://sqs.us-east-1.amazonaws.com/123456789012/payments-dlq',
      },
    },
    pollIntervalMs: 1000,
    defaultMaxAttempts: 3,
  }),
);

Connection errors. The 'redis' adapter routes the 'error' events of the ioredis client it builds to the application logger through createConnectionErrorReporter (common): the first error of an outage at warn, identical repeats at debug, the recovery at info. Without a listener ioredis prints every reconnect failure to console.error, bypassing the logger and redaction. An injected client gets no listener — it belongs to the caller. Health semantics are unchanged; the indicator still reports the outage.

Command bound. Since M101a the 'redis' adapter’s commandTimeoutMs option (default 15000, 0 disables) is the ioredis commandTimeout on the client it builds, so a paused server makes add() reject and a poll record the failure instead of waiting forever. Not applied to an injected client. A value outside 0–2147483647 throws RangeError when QueuePlugin(...) is called. A depth row in the observations describes the latest cycle only: a name that cycle did not read has no row, and the source’s failure plus depthCoverage: 'partial' say why — an unreadable depth is never a retained zero.

RabbitMQ durability. Since 0.9.0 the 'rabbitmq' adapter publishes every job persistent (persistentMessages, default true) through a publisher-confirm channel. Its ready/delay/dead queues are durable, and a durable queue keeps only persistent messages across a broker restart — before 0.9.0 a restart discarded every job not yet processed. add() resolves once RabbitMQ has accepted the job; a retry or dead-letter acknowledges the reserved job only after its replacement is accepted, so a broker failure in between can run a job twice but never loses it. publishTimeoutMs (default 15000, 0 disables) bounds each publish, queue declarations included; a value outside 0–2147483647 throws RangeError when QueuePlugin(...) is called. A rejection is not proof the job was dropped. An injected connection without createConfirmChannel() keeps a plain channel and the adapter reports one warning through the logger.

Redis namespaces

RedisQueueOptions.keyPrefix and the flat QueuePluginOptions.keyPrefix default to queue:, preserving existing keys. Every Redis key, including recurring and dead-letter keys, uses this prefix. Use a distinct prefix per application on a shared Redis. It must be a non-empty string of at most 128 characters with no control characters; invalid values throw RangeError at adapter or Redis plugin construction. The RabbitMQ-only prefix option does not configure Redis.

RabbitMQ recovery (0.10.0). Connection or channel faults drive automatic reconnect with full-jitter backoff through the common supervisor. Readiness reports down during the fault window. An injected connection receives only a new channel. Successful recovery clears local delivery bookkeeping and lazily re-declares queues. Unacked jobs return to the broker for at-least-once redelivery, possibly with the original attempt count. Disconnect stops scheduled retries and disposes listeners.

Delivery claims (0.10.0). RabbitMQ uses a fresh runtime UUID for each delivery. Redis 5+ uses atomic Lua scripts, server TIME for lease deadlines, the reserving replica’s nowMs (the clock that wrote availableAtMs) to decide which jobs are due, and INCR <keyPrefix><name>:claimseq tokens shared across replicas. Redis’s visibilityTimeoutMs (also a flat QueuePluginOptions field) defaults to 30000, must be from 1000 to 2147483647, and is validated at adapter and Redis plugin construction. The service renews every Math.floor(visibilityTimeoutMs / 3) ms with no overlapping calls and reports a failed or false renewal once per delivery, while the processor continues. Each reserve reclaims up to its limit of expired claims and removes payload-less orphans. Attempts become max(stored attempts, 1) + stalls; dispatch dead-letters an over-budget job without running it. Successful settlement clears the claim and stalls. Redis and RabbitMQ reject stale ack, requeue and dead-letter calls with internal StaleClaimError (not barrel-exported); the service reports failed settlement and the new delivery remains intact. These tokens fence deliveries, not Redis writers; ACLs remain the boundary. Delivery is at least once, including on pauses longer than the Redis lease. A live stuck processor keeps renewing. Drain older replicas before upgrading: their processing entries are immediately reclaimable. An injected Redis facade without eval retains the prior multi-command behavior, no claims or reclaim, and connect() reports once through the connection-error reporter, when one is configured, that visibilityTimeoutMs is inactive.

Declarative processors and behaviours

QueueProcessorDefinition<T = unknown> declares processor(job: IJob<T>): void | Promise<void> as a method. Typed definitions and typed inline processors fit the processors list. The method is called with its definition as this; its member is no longer readonly. QueueProcessorEntry is unchanged.

Both arms coexist with imperative queue.process() calls. With no behaviours, a processor receives the original job directly and no chain is allocated.

Queue observations (M98f)

The plugin accepts an opt-in diagnostics option that exposes actual attempts and supported depths as minimized observations through the diagnostics connector’s GET /v1/queues. It never changes a processor, a retry, a failure callback, or a settlement.

app.register(
  QueuePlugin({
    adapter: 'redis',
    url: config.get('REDIS_URL'),
    diagnostics: {
      enabled: true,
      instanceAlias: 'mailer',
      // Exact job-name -> display-alias allowlist; unlisted names are neither observed nor counted.
      queues: { 'send-welcome-email': 'welcome-emails' },
      // Optional, separately bounded depth counting; absent by default.
      depths: { intervalMs: 30000, timeoutMs: 2000, concurrency: 2 },
    },
  }),
);

Every QueuePlugin instance — named or not, observed or not — registers one IQueueDiagnosticsSource under CAPABILITIES.QUEUE_DIAGNOSTICS with { multi: true } and WITHOUT claiming the token in provides, so named instances never collide in the resolver. An absent diagnostics option registers an inert source answering disabled and adds nothing to dispatch: no ring, no alias map, no timer, no hook. enabled is the LITERAL true, refused otherwise when QueuePlugin(...) is called; every option is validated there with fixed messages that never echo a value. instanceAlias and each queues alias are a SHAPE — 1–64 UTF-8 bytes with no control character, queue aliases unique, at most 64 queues — and approving an exact alias IS authorizing its disclosure. The instance alias never derives from the plugin name.

An observed attempt is recorded from dispatch until the adapter’s settlement call RETURNED: the approved queue alias, a session-local j<N> alias for the raw job id (a bounded 4,096-entry LRU map; an eviction is counted and gives a later retry a new alias), the attempt number, monotonic durationMs/ageMs, the processor outcome (completed / retryable-error / terminal-error), and the settlement: acknowledged / requeued / dead-lettered when the call completed on an adapter that confirms it, failed when it rejected (the rejection is rethrown exactly as before), and unknown when it completed on an adapter that cannot confirm it. The queue metrics keep their timing — they are recorded before the settlement call — so an outcome is never presented as settlement proof. At most 1,024 attempts are retained and 2,048 observed in flight; droppedAttempts counts an attempt dropped because that bound was reached, because its persisted attempt number was not a positive safe integer, or because its runner failed before reporting a settlement (which releases the slot). No payload, header, raw id, claim token, credential, attempt limit or thrown value is captured: the observer signatures cannot accept them.

depths is its own opt-in because counting costs backend work: one cycle at bootstrap and one per intervalMs (1,000–300,000), never overlapping, at most concurrency (1–4) count calls in flight, each reported timed-out after timeoutMs (1–30,000) while it keeps its slot until it actually settles. Only approved job names this instance has a processor for are counted. A diagnostic read never counts, reserves, acknowledges, retries or dead-letters anything. unavailable is never reported as zero, and shared-backend counts from several sources or replicas describe the same inventory and must never be summed. The collector closes first on shutdown, clearing its timers and everything it retained.

Trace propagation across the queue hop

A queue job is deferred work whose whole diagnostic value is “which request caused this”. When a CAPABILITIES.TELEMETRY provider is registered, the plugin carries a W3C traceparent from the enqueue to the processor, so a job joins the trace of the request that enqueued it:

• queue.add() runs inside a enqueue <name> producer span and writes the span’s traceparent into AddJobOptions.headers, merged on top of anything the caller supplied.
• Every processor — registered through the processors arm, through a processors factory, or imperatively through the resolved capability — runs inside a process <name> consumer span parented from IJob.headers.
• The map is carried end to end by all four adapters (memory, redis, rabbitmq, sqs) and is exposed to an M86 ingress behaviour as IngressContext.headers, the same member the 'messaging' arm uses.

Without a telemetry capability nothing is injected and behaviour is byte-identical to before: a job carries no framework-written header. A headers map the CALLER passed is still delivered — it is a public option on a public contract, so its delivery never depends on which capabilities happen to be registered.

A job carrying no traceparent — enqueued before this channel existed, or by a non-framework producer — starts a root span rather than failing. addRecurring is deliberately untraced: a recurring job fires on a schedule, so the call that registered it is not the cause of any particular run.

Known limitation. When behaviors are configured, a behaviour runs OUTSIDE the consumer span, because the behaviour chain is applied by a service subclass that is necessarily outermost at dispatch. The processor’s own work is correctly parented either way, and with no behaviours configured the question does not arise. messaging-plugin composes the other way round and runs behaviours inside the span.

Adding Jobs

import type { AddJobOptions, IQueue } from '@setu-ts/queue-plugin';

app.router.post('/users', async (ctx) => {
  const queue = ctx.services.get<IQueue>('queue');
  const user = await createUser(ctx.request.body);

  // Add a background job
  await queue.add('send-welcome-email', {
    userId: user.id,
    email: user.email,
  });

  // Add a delayed job (delayMs in milliseconds)
  await queue.add(
    'send-reminder',
    {
      userId: user.id,
    },
    { delayMs: 86400000 },
  ); // 24 hours

  // Add with custom max attempts
  const options: AddJobOptions = { maxAttempts: 5 };
  await queue.add('process-payment', paymentData, options);

  return ctx.response.status(201).json(user);
});

Processing Jobs

import type { IJob, IQueue } from '@setu-ts/queue-plugin';

app.register({
  name: 'job-processors',
  version: '1.0.0',
  dependencies: ['queue', 'mail'],
  register(ctx) {
    const queue = ctx.services.get<IQueue>('queue');

    queue.process(
      'send-welcome-email',
      async (job: IJob<{ userId: string; email: string }>) => {
        await sendEmail({
          to: job.data.email,
          subject: 'Welcome!',
          body: 'Thank you for joining.',
        });
      },
      { concurrency: 3 },
    );

    queue.process('send-reminder', async (job) => {
      await sendReminder(job.data.userId);
    });
  },
});

Recurring Jobs

import type { IQueue, RecurringOptions } from '@setu-ts/queue-plugin';

const queue = ctx.services.get<IQueue>('queue');

// Every hour using cron expression
const hourlyOptions: RecurringOptions = { cron: '0 * * * *' };
await queue.addRecurring('cleanup-old-sessions', {}, hourlyOptions);

// Every day at 9 AM
await queue.addRecurring('daily-report', { type: 'summary' }, { cron: '0 9 * * *' });

Failure Reporting

The worker loop, the recurring-schedule loop, and the job runner all report failures through the logger capability when one is registered — a failing job logs at error with the job id, name, attempt count and the retry delay (or the dead-letter decision), and an adapter outage logs the poll failure. Nothing is required: with no LoggerPlugin registered the queue keeps running and reports nowhere, and a throwing logger can never take the worker loop down.

Dead-Lettered Jobs

A job that fails on its final attempt is dead-lettered and never delivered again. MemoryQueue exposes its dead set for assertions in tests; the Redis transport keeps its dead set in Redis, and the RabbitMQ transport keeps its dead set in a per-name dead queue (he.queue.<name>.dead).

import { MemoryQueue } from '@setu-ts/queue-plugin';

const adapter = new MemoryQueue();
// ... jobs fail through all their attempts ...

// getDeadLetters<T>(name: string): readonly StoredJob<T>[]
const dead = adapter.getDeadLetters('send-email');
console.log(dead.length, dead[0]?.attempts); // 1 3

Type Reference

// AddJobOptions
interface AddJobOptions {
  readonly delayMs?: number; // Delay before job becomes available (ms)
  readonly maxAttempts?: number; // Maximum retry attempts (default: 3)
  // Transport headers to carry with the job, delivered as `IJob.headers`.
  // Delivered whether or not telemetry is registered; with a telemetry
  // capability present the framework ADDS a `traceparent` alongside them.
  readonly headers?: Readonly<Record<string, string>>;
}

// ProcessOptions
interface ProcessOptions {
  readonly concurrency?: number; // Jobs processed concurrently (default: 1)
  // Invoked ONCE when a job has exhausted its attempts, immediately before it
  // is dead-lettered. Does NOT fire on an attempt that will be retried. A
  // callback that throws or rejects is reported through the logger and
  // swallowed — the dead-letter still happens.
  readonly onFailed?: (job: IJob, error: unknown) => void | Promise<void>;
}

// RecurringOptions
interface RecurringOptions {
  readonly cron: string; // Cron expression (e.g., '0 * * * *')
}

// IJob<T>
interface IJob<T = unknown> {
  readonly id: string;
  readonly name: string;
  readonly data: T;
  readonly attempts: number;
  // Transport headers, mirroring `MessageMetadata.headers` exactly: `{}` means
  // the channel was read and carried nothing; ABSENT means there was no
  // channel. Never substituted — an implementation whose transport cannot
  // carry a header omits the member.
  readonly headers?: Readonly<Record<string, string>>;
}

// IQueue interface
interface IQueue {
  add<T>(name: string, data: T, options?: AddJobOptions): Promise<string>;
  process<T>(name: string, processor: JobProcessor<T>, options?: ProcessOptions): void;
  addRecurring<T>(name: string, data: T, options: RecurringOptions): Promise<void>;
}

Health status

Since M70c the indicator reports two signals: the adapter’s lifecycle (isReady()) and its reachability (isHealthy()).

Since M70k the payload also carries queues, keyed by job name, when the adapter can count its states cheaply — the memory adapter (in process) and the Redis adapter (ZCARD). The key is OMITTED rather than reported as zeros on RabbitMQ and SQS, whose counts need a management API or GetQueueAttributes: “this adapter cannot tell you” and “there is nothing there” are different answers, and an operator acting on a dead-letter alert needs to tell them apart.

Since M90b the payload also carries backlog whenever at least one depth read succeeded: the sum of each successfully-read name’s ready + processing — unfinished work. dead is deliberately EXCLUDED: a dead-lettered job is terminal, already visible per name, and would otherwise make an attended dead-letter queue read as “backlog”. A partial read sums what it could read. backlog is a fact, not a threshold — no status changes because of it and no admission policy reads it.

The counts are read only once the adapter has reported itself reachable, so a down payload carries { adapter, reachable } and no queues. Counting against a backend already known to be unreachable would cost a failing round trip per name on every probe interval and tell an operator nothing that reachable: false does not. The reachability probe itself is cached for 5 seconds and bounded at 2 seconds through createCachedProbe, so concurrent health polls share one backend round trip per TTL and a hung backend cannot hold the response past the bound.

{
  "adapter": "RedisQueue",
  "reachable": true,
  "backlog": 3,
  "queues": { "thumbnail": { "ready": 0, "processing": 0, "dead": 1 } }
}

data reports { adapter, reachable }, where reachable is true, false, or 'unknown' when the adapter has no liveness check.