Queue (@adonisjs/queue)

The queue driver carries steps over @adonisjs/queue — the path for steps that run in another process. It rides any @adonisjs/queue adapter (Redis, a database adapter, or the fake adapter in tests), so it reuses the queue infrastructure you already run. To weigh it against the other drivers, see the comparison.

It ships inside @adonis-agora/durable — there's nothing extra to install. @adonisjs/queue is an optional peer, imported lazily only when you select the queue transport, so install and configure it where you use the driver:

npm i @adonisjs/queue
node ace configure @adonisjs/queue

You define the adapter and its host/credentials once, in config/queue.ts (the standard @adonisjs/queue place), as a named connection. The durable queue transport then just references that connection by name — exactly like stores.lucid({ connection }) or stores.redis({ connection }):

import { defineConfig, drivers } from '@adonisjs/queue'

export default defineConfig({
  default: 'redis',
  adapters: {
    redis: drivers.redis({ connectionName: 'main' }), // @adonisjs/redis connection
  },
})

How it worksCopy Anchor Link

• A dispatched ctx.step goes to a per-name task queue: <prefix>:tasks:<token>, where the routing token is tenantGroup(sanitizeQueueToken(name), partition) — a dedicated queue per handler name (with an optional partition suffix), never a single shared group queue.
• A worker that serves that name consumes it, runs the handler, and adds the result to the shared <prefix>:results queue.
• The engine-side instance consumes results and checkpoints them.
• Liveness flows on <prefix>:heartbeats. <prefix>:control exists for the ControlPlane interface, but it is point-to-point — one instance takes each message — so it does not fan out across pods; use controlPlanes.redis for that.
• A remote workflow turn rides the same per-name task queue (the worker tells a turn from a step by its shape); its decision returns on <prefix>:decisions, and the local steps it runs mid-turn on <prefix>:step-events.
• A store-less tenant pod reaches its control plane over the same backend.

The default prefix is durable and the default pollIntervalMs is 200.

Run one instance engine-side (consumes results) and one or more worker processes that serve the step handlers (each subscribes to a task queue per handler name it serves):

import { defineConfig, transports } from '@adonis-agora/durable'

export default defineConfig({
  transport: 'queue',
  transports: {
    // References the connection named in config/queue.ts. Omit `connection` to
    // use that file's `default`.
    queue: transports.queue({ connection: 'redis' }),
  },
})

The idiomatic worker is the bundled ace command, which boots the app and reads the same config/durable.ts — no adapter wiring to repeat. It serves whatever step handlers the app registers (the app/steps @Step/defineStep convention is auto-discovered), subscribing to a task queue per served name:

node ace durable:work

Supported @adonisjs/queue versionsCopy Anchor Link

The transport works with @adonisjs/queue 0.6 (@boringnode/queue 0.6) and 0.7 (@boringnode/queue 0.9). It drives the adapter directly — not the queue manager's Worker — and picks the right adapter contract from each claim: on 0.7 every claim carries a lease token, which is what completes, retries or fails it.

Upgrading an app from 0.6 to 0.7:

• Follow the @adonisjs/queue 0.7 upgrade notes: stop every process on the old version (do not run both side by side), then run node ace queue:migrate before starting the new ones. Durable keeps no schedules or dedup ids in the queue backend, so its own queues need nothing more — but your app's jobs and schedules do.
• defaultJobOptions (maxRetries, retry, timeout, failOnTimeout) and worker.unknownJobRetries only apply to jobs run by queue:work. Durable's jobs are never run by it: retries, timeouts and backoff stay governed by the step options (retries, timeoutMs…) and this transport's settings.
• Don't point queue:work at durable's queues (<prefix>:…). On 0.7 a worker that meets a job whose class it doesn't know puts it back in the queue up to unknownJobRetries times, then fails it.

Options — transports.queue(config)Copy Anchor Link

Concurrency and idle pollingCopy Anchor Link

Every step name a worker serves gets its own task queue and poll loop. By default that loop runs one job at a time: a backlog of 100 sendEmail steps is worked through serially per worker process. Raise concurrency to run several at once:

queue: transports.queue({
  connection: 'redis',
  concurrency: 8,        // up to 8 jobs of each step name in flight per worker
  resultConcurrency: 1024, // the engine holds up to 1024 results at once (at least 512)
}),

The slots share the step's single poll loop — it keeps popping while a slot is free and waits for one to free when all are busy — so concurrency costs no extra polling. Every in-flight claim is renewed, and close() waits (within its 5s budget) for all of them, like a single one. Fleet-wide parallelism for a step is concurrency × worker processes; use a flow-control queue when you need a cap across the fleet.

Adaptive concurrencyCopy Anchor Link

Pass 'adaptive' instead of a number to let the worker tune its own limit. It is the same controller the BullMQ transport uses: while steps saturate the limit at a steady latency it doubles it until the first move down (re-deciding after each limit's worth of settled steps, not only on the tick), then grows by ⌊√limit⌋ a tick; every tickMs it shrinks it by the latency gradient when steps start to queue (the gradient (2 × baseline + 5 ms) / p50 below 0.7, so the jitter of millisecond steps is not read as queuing), sheds one on a >20% error rate, a stall or the CPU ceiling, and cuts it by 20% when RSS reaches ramCeilingPct of the memory limit (the cgroup limit, else the host's memory).

queue: transports.queue({
  connection: 'redis',
  concurrency: { mode: 'adaptive', min: 1, max: 16, start: 2, tickMs: 2000, ramCeilingPct: 85, cpuCeilingPct: 90 },
}),

A non-finite or out-of-range field falls back to its default. One controller serves every step name of the worker, and each step name's task loop reads the live limit on every poll, so a move applies at once — a lower limit lets jobs already running finish and stops taking new ones until in-flight work is under it. Only step durations feed the latency window. The results, heartbeat and control loops are not affected (resultConcurrency stays fixed).

Whatever the mode, each worker's liveness heartbeat carries the controller's snapshot — mode and live limit (with min/max when adaptive), in-flight count, RSS, CPU, throughput, p95 and the last adjustment with its reason — which the dashboard's worker cards render.

Idle pollingCopy Anchor Link

Polling is the cost of a backend-agnostic queue: each loop asks the adapter for a job every pollIntervalMs, and a worker serving 30 step names runs 30 task loops. So idle task, heartbeat and control loops back off — the sleep doubles after each empty poll up to idleBackoffMax × pollIntervalMs (800ms by default), and the first job found snaps the loop back to the base interval and drains the backlog without sleeping. The price is up to that much pickup latency for the first job after a quiet spell; set idleBackoffMax: 1 for the old fixed cadence. The results loop never backs off, since every step's result is on its workflow's critical path.

Worker advertisementCopy Anchor Link

A queue hands each job to one consumer, so a worker's descriptor and heartbeat cannot travel over it: every engine has to read every worker's. They go to a key store instead — by default the app database's durable_transport_worker_keys table (created on first use, and by the migration configure publishes), the same one the db transport uses. With no database and no workerKeys, nothing is advertised and the engine routes without descriptors, as before.

Worker healthCopy Anchor Link

Every worker process beats a liveness key for each step name it serves — <prefix>-worker-heartbeat:<token>:<instanceId> in the worker-key store, refreshed every workerHeartbeatIntervalMs (10s) and live for 35s, carrying the WorkerStatus snapshot described above. A process that only produces (a console/repl process with deferred consumers) beats nothing until it starts consuming.

From those keys and the queue, the transport implements groupHealth, so engine.workerHealth(), the dashboard's /api/workers and /api/health and the Telescope worker panel report queue-transport groups:

• liveWorkers: the unexpired liveness keys of the group, with each worker's instanceId, lastBeatAt and status. A worker that stops cleanly removes its keys at once; one that dies drops out within 35s. Reads of the keys are reused for workerKeysCacheMs (5s).
• depth: the task queue's pending jobs, from the adapter's sizeOf. This is narrower than BullMQ's depth (waiting + active + delayed + prioritized): the @adonisjs/queue adapter contract has no count for jobs in flight or delayed retries, so those are not included. depth > 0 with no live worker is still the alert state, since pending jobs are the ones no worker has picked up.

groupHealth needs the worker-key store: with workerKeys: false (or no database and no store), no worker can advertise liveness, so the transport does not implement it and the panels stay empty rather than flagging every group as starved.

Stalled-job reclaimCopy Anchor Link

A worker that throws is easy: the job is redelivered a poll interval later. A worker that vanishes — killed mid-step, its pod evicted — throws nothing and leaves a claim that nobody will ever complete. The reclaim sweep is the net for that case: every stalledCheckIntervalMs, claims older than stalledThresholdMs are re-delivered.

While a step runs, the transport renews its claim every half stalledThresholdMs through the adapter's renewJobs — the same heartbeat @adonisjs/queue's own worker runs, on both 0.6 and 0.7. A claim's age therefore measures how long its worker has been silent, not how long the step has run:

• a dead worker's job is re-delivered within about stalledThresholdMs (default 2 minutes) plus one sweep interval;
• a healthy step is never re-delivered, however long it runs — a 3-hour import keeps its claim.

What can still look like a dead worker is a live one whose event loop is blocked longer than the threshold — a synchronous, CPU-heavy step body starves the renewal timer. Keep heavy work async or off the main thread, or raise the threshold above the longest block:

transports.queue({
  connection: 'redis',
  stalledThresholdMs: 5 * 60 * 1000, // a report renderer blocks the loop for up to ~3 minutes
  maxStalledCount: 3,
  onError: (err) => logger.error({ err }, 'durable queue transport'),
})

When a renewal finds a claim it no longer holds — the sweep re-delivered it while this worker was silent — the transport reports it to onError (lost N of M claim(s) …): the step may be running twice, so make steps idempotent on the stepId they receive.

Set stalledCheckIntervalMs: 0 to switch the sweep off — reasonable when an adapter already redelivers stalled jobs itself. Adapters that cannot report stalled jobs are detected and skipped silently, in which case engine.redispatchPending(runId) is the recovery path.

Poison jobsCopy Anchor Link

A job reclaimed more than maxStalledCount times is one whose workers keep vanishing while they hold it, so the sweep fails it instead of re-delivering it again. For a step task, the transport then publishes a failed result for that step (remote step worker failed: stalled more than N times, marked retryable) on the results queue, the same bridge the BullMQ transport runs for a task job that fails terminally. The engine treats it like any failed attempt: it re-dispatches while the step has retries left, then fails the step into the workflow (a try/catch around the step sees a RemoteStepError; uncaught, the run fails). Without this, the run would wait on that step forever.

This needs @adonisjs/queue 0.7: there the adapter hands a job that stalled too often back to the transport to fail. On 0.6 the adapter deletes such a job inside its own sweep and returns only a count, so the transport never sees it and no result is published. On 0.6, set the step's timeoutMs or the engine's remoteRedispatchMs (see lost remote dispatch) so such a step does not wait forever. Stalled result, heartbeat and control jobs are failed without publishing anything.

consumers — why a node ace script does not drain the queueCopy Anchor Link

Broker queues are point-to-point: a job goes to exactly one consumer. That makes a short-lived process subscribing to them actively harmful — it claims step jobs it then dies holding, steals results addressed to the long-lived engine, and, with jobs queued, never exits because the drain loop keeps feeding it.

So by default a console or repl process does not start the broker consumer loops. It can still dispatch runs and read the store — it is a pure producer. durable:work re-enables consumption for itself, so the worker command behaves identically either way, and web and test processes consume eagerly. The setting has three values:

A deferred queue transport still consumes the inboxes that belong to it alone (its run-reply and tenant-event queues), since no other process could take those jobs.

What this changes in practice: a node ace script that starts a run and waits for a remote step to come back will wait forever, because nothing in that process is listening for the result. Either let the worker fleet do the work (start the run and exit), or opt the process back in:

export default defineConfig({
  consumers: 'always', // every booted process consumes eagerly
})

Reach for 'always' only for a script that genuinely must round-trip a remote step inline, and be aware it competes with the real worker fleet for jobs while it runs. In-process transports (memory, eventEmitter) are unaffected — there is no broker to compete for. The same setting works on db and bullmq.

transports.queue(config) returns the lazy driver thunk used in config/durable.ts; the underlying class is QueueTransport, also exported from @adonis-agora/durable for constructing a worker-side instance directly.

Tenant pods over the queue transportCopy Anchor Link

A tenant pod owns no store: every read, control action and run start goes to the control plane as a request, and the answer comes back. The queue transport carries that protocol, so a control-plane and its tenant pods can share one @adonisjs/queue backend instead of Redis plus bullmq. Point both at the same connection and prefix:

export default defineConfig({
  role: 'control-plane',
  transport: 'queue',
  transports: { queue: transports.queue({ connection: 'redis' }) },
  store: 'lucid',
  stores: { lucid: stores.lucid({ connection: 'pg' }) },
  verifyTenant: hmacTenantVerifier(process.env.DURABLE_TENANT_SECRET!),
})

export default defineConfig({
  role: 'tenant',
  transport: 'queue',
  // No database on the pod: its advertisement and subscriptions go to the control plane.
  // The transport's `partition` decides which `<name>@acme-corp` task queues the pod consumes.
  transports: { queue: transports.queue({ connection: 'redis', workerKeys: false, partition: 'acme-corp' }) },
  partition: 'acme-corp',
  tenant: { token: process.env.DURABLE_TENANT_TOKEN },
})

Set partition on the transport as well as at the top level: the provider passes the top-level one to the run gateway and the worker runtime, not to the transport that subscribes the task queues (see Roles and config).

Starting a run and sending a read or control request are plain jobs on <prefix>:start-run and <prefix>:run-request, which one control-plane instance consumes. The two parts BullMQ broadcasts on Redis pub/sub are different here, because a queue job has exactly one consumer:

• Replies. Every instance that sends requests consumes its own <prefix>:run-replies:<instanceId> queue, and stamps that instance id on each request. The control-plane instance that handles the request pushes the reply onto that queue only, so two pods never receive or take each other's replies.
• Live-tail events. A pod that subscribes to its tenant's events (runGateway.subscribe) announces the subscription on <prefix>:inbox-registrations, and again every third of inboxTtlMs. The control plane records it as a key in its worker-key store — the pod itself needs no store — and pushes one copy of each event of that tenant onto <prefix>:tenant-events:<instanceId> of every pod with a live subscription. Pods of other tenants receive nothing.

A pod with no key store of its own forwards its worker advertisement the same way, so the control plane still sees its workers for capability routing and convention dispatch.

How this compares with bullmq, which carries the same protocol over Redis pub/sub:

• Order. Events from one control-plane instance reach a pod in the order they were published, as on pub/sub. Events from several instances are not ordered against each other on either transport.
• Latency. Replies and events are polled, not pushed. The reply loop never backs off (one pollIntervalMs); the event loop backs off like the task loops (up to idleBackoffMax × pollIntervalMs when idle).
• Subscribing. A new subscription reaches the control plane after its next registration poll plus tenantSubscribersCacheMs, so events published in the first second or so after subscribing can be missed. Pub/sub misses only the events published before the subscription.
• Cost. Each event is one job per subscribed pod of its tenant (pub/sub sends it once). Each pod that sends requests or subscribes sends one small registration job every third of inboxTtlMs.
• Trust. As on bullmq, anything with access to the broker can subscribe to any tenant's events. Requests are checked against verifyTenant on the control plane; subscriptions are not.
• Requirements. Live-tail needs a worker-key store on the control plane (the default database table); without one, tenant events reach no pod and the control plane reports it once through onError, and the janitor is off. Replies, starts and reads need no key store.

Queue namingCopy Anchor Link

Because the payload is the documented RemoteTask / StepResult JSON, multiple worker processes can share the same queues — scale out by running more durable:work processes that serve the same step names.

Namespaces — sharing one broker across poolsCopy Anchor Link

A worker-pool namespace (set on the engine in config/durable.ts) partitions a deployment so several non-interchangeable pools can safely share one state store and one broker without stealing each other's work. The engine stamps the namespace on every run, scopes its poll/recovery/timer paths to it, and propagates it to the transport — so the queue transport folds the namespace into every queue name:

import { defineConfig, transports } from '@adonis-agora/durable'

export default defineConfig({
  namespace: 'tenant-a', // ← partitions the store AND the queue names
  transport: 'queue',
  transports: { queue: transports.queue({ connection: 'redis' }) },
})

A non-'default' namespace inserts a -<namespace> segment after the prefix; 'default' (and unset) keeps the exact un-namespaced names, so existing deployments are unaffected — zero migration. An engine and its durable:work workers must run the same namespace to share queues.

Filtering runs by namespaceCopy Anchor Link

Read surfaces (the dashboard, getRun, the CLI) stay namespace-agnostic by default. To scope a run search to one pool, pass namespace to a RunQuery — it is ANDed with the other predicates only when provided:

// Only runs in the 'tenant-a' pool, ANDed with the other filters.
await engine.store.listRuns({ namespace: 'tenant-a', status: 'running' })