Interface QueuePluginOptions

The adapter type to use (default 'memory').

Ingress behaviours wrapped around every processor — the queue arm of the transport-neutral behaviour chain shared with the websocket, scheduler, and messaging plugins (IIngressBehavior in @setu-ts/common).

Each behaviour observes an IngressContext carrying kind: 'queue', the job name as name, the delivered IJob as payload, and attempt equal to IJob.attempts, and runs in declared order ahead of the processor. A behaviour that returns without calling next() short-circuits: the processor never sees the job and the job is acknowledged. A behaviour that throws follows the processor's own failure path — requeue with backoff, and ProcessOptions.onFailed plus the dead-letter on the final attempt. Every processor registration is wrapped, imperative process() calls included, so a mixed application cannot leave a handler unchained.

With no behaviours configured, dispatch is byte-identical to the pre-chain behaviour: the processor is handed the job directly, with no chain allocated.

When an entry is a FACTORY, dispatch is HELD until onInit has resolved the whole chain, so nothing reaches a handler through a partial one. That gate covers every registration — this plugin's declared entries and any a later plugin makes imperatively through the resolved capability — which is why no registration's timing has to change. It is released once and costs nothing thereafter.

Instance entries are handed to the service at register(); factory entries are resolved in the onInit phase and a throwing factory rejects start() naming QueuePlugin({ behaviors }) and the entry's index in THIS declared array.

Injected client (bypasses lazy import).

Bound on one Redis command, in milliseconds (adapter 'redis' only; ignored by every other adapter). Default 15000; 0 disables it. A paused or partitioned server keeps its socket open, so without a bound a command waits forever; with it add() rejects and a poll records the failure. Applied as ioredis commandTimeout to the client the adapter BUILDS — never to an injected client, which keeps its own configuration. A value outside 0–2147483647 (including NaN) throws RangeError.

A timeout rejects the command LOCALLY; the server may still apply it once it answers. So a rejected add() is not proof the job was not enqueued — a caller that retries on rejection can run the job twice, and should give the job an idempotent handler. Each queue transition runs as one Lua script (see IRedisQueueClient.eval), so a timeout leaves it whole or absent; a reserve applied after its timeout leaves the job and its payload in the processing set. With eval, an expired lease is reclaimed on reserve. A client without eval keeps the previous behaviour: no claims or reclaim.

How long a dead-lettered job's payload is retained, in milliseconds.

Omitted (the default) keeps today's behaviour: the payload is retained indefinitely "for debugging", so the jobs hash grows without bound for the lifetime of the deployment (X8-4). Applied only by the Redis adapter, and only when the injected client exposes expire.

Enforced PER PAYLOAD: each dead-letter sweeps the dead set — scored by dead-letter time — and deletes every entry older than this, so a queue that keeps failing still drops its oldest payloads. Setting it also relocates a dead job's payload from queue:<name>:jobs, which holds every queued job's payload for that name, into queue:<name>:dead:jobs.

The retention it delivers is a bound rather than a deadline: AT LEAST this long, and AT MOST this long past the LAST dead-letter on the queue — so a payload that dies just before a short burst can live for just under twice this value. The sweep runs only when a dead-letter arrives, and the key-level backstop beside it carries one deadline for a shared key, which must be the newest or it would take newer payloads with it. It errs late deliberately: dropping a payload early discards the debugging data this option exists to keep.

Default max attempts for jobs (default 3).

Opt-in, minimized attempt and depth observations for the local diagnostics connector (M98f). Absent (the default) registers an inert disabled source under CAPABILITIES.QUEUE_DIAGNOSTICS and observes nothing: no ring, no alias map, no timer and no added hook on dispatch. Validated when QueuePlugin(...) is called, with fixed messages that never echo a supplied value.

Redis key namespace, default queue:. Non-empty, at most 128 characters, without control characters. Does not read the RabbitMQ prefix option.

Instance name for multi-instance support.

Publish every job persistent (delivery_mode 2) — adapter 'rabbitmq' only. Default true.

The adapter's ready/delay/dead queues are durable, and a durable queue keeps only PERSISTENT messages across a broker restart: before 0.9.0 every job was published transient, so a RabbitMQ restart silently discarded every job not yet processed. false restores that behaviour.

Poll interval for worker loop (default 1000ms).

Queue name prefix for RabbitMQ adapter (default 'he.queue'). Queue names are <prefix>.<name>.ready|.delay|.dead, each at most 255 UTF-8 bytes; a prefix over 247 bytes is refused at construction, and a job name that does not fit is refused before the channel is touched.

Processors registered declaratively, as an alternative to calling queue.process(name, processor, options) imperatively after start(). Each entry — instance or RegistryFactory — produces one process() call, so a processor can be declared where the plugin is composed instead of after the application has started.

Instance entries register during the plugin's register() phase, identical to the imperative timing. The one exception is an array that also carries a FACTORY: the whole array then registers in onInit, in DECLARED order, because process() is last-wins on a job name and registering instances first would let a leading factory beat a trailing instance. Factory entries are resolved in the onInit phase — the first at which the registry holds every capability — so a factory can build its processor from a resolved capability. A factory that throws rejects start() with an error naming QueuePlugin({ processors }) and the entry's index in THIS declared array, not its position among the factories.

Registering two entries under one job name keeps the service's existing last-wins behaviour: exactly what two imperative process() calls with the same name do — decided by DECLARED order, so a trailing entry beats a leading one whichever arm each uses.

Bound on one job publish, in milliseconds — adapter 'rabbitmq' only. Default 15000; 0 waits without a bound.

It covers every broker round trip the publish makes — the job's queue declarations and, on a confirm channel (a real amqplib connection always opens one), the wait for RabbitMQ to accept the job. add() resolves once RabbitMQ has accepted the job and rejects when it refuses it or the bound expires; a retry or dead-letter acknowledges the reserved job only after its replacement is accepted, so a broker failure between the two cannot lose it. A rejection is not proof the job was dropped: the broker may still accept it after the bound. A value outside 0–2147483647 (including NaN) throws RangeError when QueuePlugin(...) is called.

SQS-specific options (required when adapter is 'sqs').

Connection URL (used when adapter is 'redis' or 'rabbitmq').

Redis lease length in ms, default 30000; from 1000 to 2147483647. Requires eval; otherwise inactive with one connection-error report. Renewed every third of this duration. Redis 5+ required.