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.

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 works

• 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; cross-instance control on <prefix>:control.

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

Options — transports.queue(config)

Stalled-job reclaim

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.

The default threshold of 30 minutes looks absurdly generous until you see why: a claim is never renewed while the worker processes it, so its age is the step's elapsed time, not a heartbeat gap. A threshold below your longest step re-delivers a job whose worker is merely slow — which double-runs the step.

So the rule is: raise stalledThresholdMs above your longest legitimate step, and make those steps idempotent. If you want tighter detection than a claim's age can give you, use a per-step timeoutMs instead — that one is heartbeat-aware.

transports.queue({
  connection: 'redis',
  stalledThresholdMs: 2 * 60 * 60 * 1000, // our longest import takes ~90 minutes
  maxStalledCount: 3,
  onError: (err) => logger.error({ err }, 'durable queue transport'),
})

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.

consumers — why a node ace script does not drain the queue

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 always consume eagerly.

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, event-emitter) are unaffected — there is no broker to compete for.

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.

Queue naming

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 pools

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 namespace

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' })