Node worker

@dudousxd/durable-worker is the standalone Node worker SDK — the JavaScript twin of the Python client. It runs steps (and whole workflows) in a store-less process that owns no engine and no database: it consumes tasks off the broker, replays or executes, and publishes results back. The engine stays the sole owner of durable state, recovery and timers. This is the building block behind both the store-less tenant worker and the co-located worker — reach for it directly when you want a worker without the full NestJS module (a script, a purpose-built pod, a sidecar), or when you want to understand what those roles start under the hood.

pnpm add @dudousxd/durable-worker bullmq ioredis

The runtime + a runner

A worker is two pieces: a DurableWorkerRuntime (the pure routing core — it holds your registered workflows and steps) and a runner (runRedisWorker) that wires it to Redis. Register bodies, then start the runner:

import { DurableWorkerRuntime, runRedisWorker } from '@dudousxd/durable-worker';

const runtime = new DurableWorkerRuntime()
  // A step, routed by name — the same name a workflow's ctx.step('payments.charge-card', …) calls.
  .registerStep('payments.charge-card', async (input: { orderId: string; amountCents: number }) => {
    const res = await stripe.charge(input.orderId, input.amountCents);
    return { chargeId: res.id };
  })
  // A whole workflow, replayed turn-by-turn against the history the engine sends.
  .registerWorkflow('pipeline', async (ctx, input: { id: string }) => {
    const rows = await ctx.step('ingestion', { key: `/${input.id}/data.csv` });
    await ctx.sleep('1m');
    return { rows };
  });

const worker = await runRedisWorker({
  runtime,
  connection: { host: '127.0.0.1', port: 6379 }, // same Redis the engine's BullMQTransport uses
});

// on SIGTERM: await worker.close();  — drains + stops every queue and pub/sub

runRedisWorker starts one BullMQ Worker per registered name (workflows ∪ steps, deduped) — there's no single hand-declared "group" queue. Each name gets its own queue, so the engine's ctx.step('payments.charge-card', …) or a dispatched pipeline turn is routed to exactly the worker that registered it. The queue name is the routing; nothing else to declare.

Step-only vs workflow-authoring

DurableWorkerRuntime composes a StepWorker (runs remote steps → StepResult) and a WorkflowWorker (replays workflow turns → WorkflowDecision). Both are pure and transport-free — processTask(task) is a function of the task, so you can unit-test either without a broker:

import { StepWorker, WorkflowWorker } from '@dudousxd/durable-worker';

const steps = new StepWorker().register('reserve', (input) => reserve(input));
const result = await steps.processTask(task); // { status: 'completed', output, startedAt, … }

const workflows = new WorkflowWorker().register('checkout', async (ctx, input) => {
  await ctx.step('reserve', input);
  return { ok: true };
});
const decision = await workflows.processTask(workflowTask); // { status: 'continue' | 'completed' | …, commands }

A step handler receives (input, log) — the log is a step-event sink that rides back on the result (log.info(...), sub-process outcomes). An unknown name never throws the consumer down: it returns a failed result/decision with a clear message, so a misconfigured worker is a recorded failure, not a crash.

Authoring a workflow: the ctx

A registered workflow receives a WorkflowContext that implements the engine's WorkflowCtx, so a body written against WorkflowCtx runs unchanged here. Each op is keyed by a deterministic seq: a completed op replays its recorded result instead of re-running, and changing the op sequence under an in-flight run throws NondeterminismError rather than diverging silently.

The wire-expressible subset is supported:

Everything else on WorkflowCtx — transaction, callEntity, continueAsNew, sleepUntil, waitForEvent, task, fire-and-forget startChild, breakpoint, webhook, setEvent, onUpdate, patched — needs engine/store/transport features the remote wire can't express, so each throws UnsupportedOnThinWorker. Run such a workflow in-process on the engine instead.

Concurrency: fixed or adaptive

By default a runner processes one task per queue at a time (concurrency: 1). Raise it so a fanned-out batch (the N steps of a gather, say) runs in parallel instead of serially. Total parallelism is concurrency × distinct names × replicas.

await runRedisWorker({
  runtime,
  connection,
  concurrency: 8, // up to 8 in-flight tasks per subscribed queue, per process
});

await runRedisWorker({
  runtime,
  connection,
  // Self-tune the limit from a latency gradient, a RAM hard-brake, and an error/stall signal.
  concurrency: { mode: 'adaptive', min: 2, max: 32, ramCeilingPct: 85 },
});

You can normalise a concurrency option yourself with the exported resolveConcurrency(opt) (a bare number or undefined → { mode: 'fixed', fixed }; 'adaptive'/{ mode: 'adaptive', … } → { mode: 'adaptive', adaptive } with every default applied), and drive an AdaptiveController directly if you're building your own runner loop.

Runner options

await runRedisWorker({
  runtime,
  connection: process.env.REDIS_URL,
  prefix: 'durable',        // must equal BullMQTransport({ prefix })
  partition: 'acme-corp',   // this worker serves the acme-corp tenant's queues (<name>@acme-corp)
  instanceId: 'orders-worker-1',
  concurrency: 'adaptive',
});

The runner also, best-effort, subscribes to the <prefix>-control channel for cooperative cancellation, streams each local step's lifecycle live, and stamps a TTL'd liveness heartbeat per subscribed queue — none of which can block the worker from starting or processing. All of it mirrors the Python SDK byte-for-byte.

Starting runs without a database (startRun)

A store-less worker has no engine to call engine.start on. To request a new run from the control plane over the broker, use startRun — it publishes a StartRunMessage onto <prefix>-start-run, which the control plane turns into a durable run:

import { startRun } from '@dudousxd/durable-worker';

await startRun(connection, {
  tenant: 'acme-corp',
  workflow: 'pipeline',
  input: { id: 'order-42' },
  runId: 'run-order-42', // supply your own for idempotent, at-least-once-safe redelivery
});

See also

• Python worker — the same SDK in Python, with the full wire-protocol and cancellation treatment that applies equally here.
• Roles & config — the NestJS-module tenant/co-located worker this SDK powers.
• Cross-ecosystem interop — mixing Node, Python and Adonis workers on one control plane.
• BullMQ / Redis transport — the engine side of the same broker.