Testing

The @adonis-agora/durable/testing subpath runs a whole workflow in a unit test — in-memory store and transport, a clock you control, and assertions that read back the recorded state. It ships in the main @adonis-agora/durable package and imports no test runner, so it works under Japa, vitest or anything that fails a test on a thrown error. vitest is an optional peer, needed only for the conformance suites on @adonis-agora/durable/testing/conformance:

npm i -D vitest # only if you run the conformance suites

A test engineCopy Anchor Link

createTestEngine() returns { engine, store, transport, clock, tick, run } — a self-contained engine wired to an InMemoryStateStore, an InMemoryTransport, and a controllable clock:

import { test } from '@japa/runner'
import { createTestEngine, assertRunStatus, assertOutput } from '@adonis-agora/durable/testing'

test('checkout completes', async () => {
  const t = createTestEngine()
  t.engine.register('checkout', '1', async (ctx) => {
    await ctx.localStep('reserve', () => reserve())
    return ctx.localStep('ship', () => ship())
  })

  await t.run('checkout', order, 'run1') // enqueue + wait for the run to settle
  await assertRunStatus(t.store, 'run1', 'completed')
})

t.run(workflow, input, runId, opts?) enqueues and waits for the run to settle (terminal or suspended) — use it when a test needs the outcome synchronously, since engine.start only enqueues. It is exactly await t.engine.start(workflow, input, runId, opts) followed by t.engine.waitForRun(runId), and resolves to the same RunResult ({ runId, status, output, error }).

Testing a BaseWorkflow classCopy Anchor Link

An app/workflows class registers on the test engine with registerWorkflowClass(engine, Class) from @adonis-agora/durable — the same call the provider makes at boot, so the class's static workflow identity, tags, singleton, concurrency, inputSchema, onError routes and colocated static schedule / static on triggers all come along. Then start it by class or by name:

import { registerWorkflowClass } from '@adonis-agora/durable'
import { createTestEngine, assertRunStatus } from '@adonis-agora/durable/testing'
import OrderWorkflow from '#workflows/order_workflow'

test('order workflow completes', async () => {
  const t = createTestEngine()
  await registerWorkflowClass(t.engine, OrderWorkflow) // async: it may resolve the instance through a factory
  await t.run(OrderWorkflow, { orderId: 'o1' }, 'run1')
  await assertRunStatus(t.store, 'run1', 'completed')
})

Pass a third factory argument ((Ctor) => instance, may be async) when the class needs constructor-injected dependencies — hand it fakes, or (Ctor) => app.container.make(Ctor) in an app-level test.

Dispatched steps — t.transport.handleCopy Anchor Link

A ctx.step(name, input) is always dispatched, so the test has to serve it: register a fake worker for each step name on the in-memory transport with t.transport.handle(name, fn) (the handler receives the step input). Because a dispatched step parks the run until its result comes back, t.run resolves on that first suspended state — wait for a terminal state instead:

t.engine.register('checkout', '1', async (ctx, order) => {
  await ctx.step('reserve', order)
  return ctx.step('ship', order)
})
t.transport.handle('reserve', async () => ({ reserved: true })) // stands in for the remote worker
t.transport.handle('ship', async () => ({ shipped: true }))

await t.engine.start('checkout', order, 'run1')                      // enqueues → pending
const result = await t.engine.waitForRun('run1', { terminal: true }) // skip the suspended hand-offs
await assertRunStatus(t.store, 'run1', 'completed')
await assertOutput(t.store, 'run1', { shipped: true })

waitForRun(runId, { terminal: true }) only resolves on completed / failed / cancelled / dead; add timeoutMs to bound it so a wedged run fails the test instead of hanging it.

Control time (durable sleep)Copy Anchor Link

t.tick(ms) advances the clock and resumes any durable sleep that is now due — no real waiting:

t.engine.register('digest', '1', async (ctx) => {
  await ctx.localStep('draft', () => draft())
  await ctx.sleep('7 days')
  await ctx.localStep('send', () => send())
})

await t.run('digest', {}, 'run1')   // settles on the durable sleep → suspended
await t.tick(7 * 24 * 60 * 60 * 1000) // the sleep is due → completes
await assertRunStatus(t.store, 'run1', 'completed')

t.tick(ms) is t.clock.advance(ms) followed by engine.resumeDueTimers(clock.now()), and returns the RunResult[] of the runs it resumed. A tick that leaves a sleep short of its deadline resumes nothing — the run stays suspended.

The clock itself (t.clock, a MutableClock) is a standalone export, with set(ms) (jump to an absolute instant) alongside advance(ms) (move forward by a delta); now() reads it. createTestEngine starts it at 1_000. Reach for it directly when you need an absolute time, or when you're driving code that only takes a bare now: () => number:

import { MutableClock } from '@adonis-agora/durable/testing'

const clock = new MutableClock(1_720_000_000_000) // starts at this instant instead of the default 1_000
clock.set(1_720_100_000_000)                       // jump to an absolute instant
clock.advance(60_000)                              // ...then move forward a minute
clock.now()                                        // 1_720_100_060_000

Inject crashes & retriesCopy Anchor Link

failOnce / failTimes make a step throw before succeeding — to drive retries and resume:

import { failOnce, assertStepAttempts } from '@adonis-agora/durable/testing'

t.engine.register('wf', '1', async (ctx) =>
  ctx.localStep('charge', failOnce({ ok: true }), { retries: 3 }),
)
await t.run('wf', {}, 'run1')
await assertStepAttempts(t.store, 'run1', 'charge', 2) // failed once, then succeeded

Each factory returns a fresh step body that counts its own calls: failOnce(value) throws once then returns value; failTimes(n, value) throws the first n calls then returns it (failOnce is exactly failTimes(1)). Both take an optional custom Error as the last argument (default new Error('injected failure')).

Use failTimes to exhaust a retry budget — the classic "does a poison step actually fail the run?" test. On a ctx.localStep, retries is the total number of attempts (retries: 3 means three tries, not one plus three), so failing all three must fail the run:

import { failTimes, assertRunStatus, assertStepAttempts } from '@adonis-agora/durable/testing'

t.engine.register('wf', '1', async (ctx) =>
  ctx.localStep('flaky', failTimes(3, { ok: true }, new Error('upstream 503')), { retries: 3 }),
)
const result = await t.run('wf', {}, 'run1')
await assertRunStatus(t.store, 'run1', 'failed')       // result.error.message === 'upstream 503'
await assertStepAttempts(t.store, 'run1', 'flaky', 3)  // all attempts thrown → run failed

Inject failures with ctx.localStep, as above: a dispatched ctx.step retries on a different policy (a worker that reported an error is surfaced to the workflow rather than re-dispatched; re-dispatch is reserved for liveness timeouts), so it is not the path to drive with failOnce.

AssertionsCopy Anchor Link

All assertions read the store, so they work against any run the engine produced:

assertOutput compares by JSON.stringify, so key order matters. Each assertion throws a plain Error with the expected and actual values, so it works under Japa, vitest, or any runner that fails a test on a thrown error; an unknown runId throws run <id> not found.

Conformance kitsCopy Anchor Link

Write a custom store, transport, or admission backend and you can run the exact battery of behavioural tests the bundled adapters pass. That is the definition of "correct" for an adapter — the prose in these docs is a summary of it.

They live on two different subpaths, and which one a kit is on is not arbitrary:

The ones on /conformance generate their suite with vitest's describe / it, so they genuinely require vitest at import time. Keeping them on their own subpath is what lets createTestEngine and the assertions above stay importable from a Japa suite — a barrel re-exports everything, so a single subpath would have forced vitest on everyone. assertTransportConformance stays on /testing because it is a plain async function with no test-runner bindings, so it never needed vitest to begin with.

import { assertTransportConformance } from '@adonis-agora/durable/testing'
import { MyTransport } from '../src/my_transport.js'

// Throws on the first contract violation — call it from whichever runner you use.
await assertTransportConformance(new MyTransport({ group: 'conformance' }))

import { runStateStoreContract } from '@adonis-agora/durable/testing/conformance'
import { MyStateStore } from '../src/my_state_store.js'

// Registers its own describe/it — this call IS the suite.
runStateStoreContract('MyStateStore', async () => {
  const store = new MyStateStore(/* connection */)
  await store.ensureSchema?.()
  return { store, cleanup: () => store.close() }
})

The factory runs before every case and must return a fresh, empty store plus a cleanup. If it can't reach its backend at all (no Docker, no local Postgres), throw StateStoreUnavailableError (from @adonis-agora/durable/testing/conformance) instead of letting the connection error fail the suite — every case catches it, logs one SKIPPING warning, and passes as a no-op rather than reporting a false failure. The returned context can also opt out of two capabilities a driver genuinely lacks: supportsAsyncTransaction: false (a synchronous driver whose transaction() rejects a promise-returning callback) and supportsTagFilter: false (no listRuns({ tag }) filter); both default to true.

import { runRunDispatchContract } from '@adonis-agora/durable/testing/conformance'
import { MyTransport } from '../src/my_transport.js'

// Every transport `make()` returns within one case must share that case's queue (two engines on two
// of them are two instances); `setup` runs before each case to give it a queue of its own.
let prefix = ''
runRunDispatchContract('MyTransport', {
  setup: () => {
    prefix = `runs-${crypto.randomUUID()}`
  },
  make: () => new MyTransport({ prefix }),
})

The run-dispatch contract runs real engines over an in-memory store: a start for a new pending state is delivered while the previous one is still being handled, a ctx.all and ctx.allSettled fan's child starts go through the queue in batches (with and without concurrency), two instances split a fan and run every child (and every step of a step fan) once, each namespace keeps its own queue, a run carries its priority, a failed enqueue leaves the run for the pending-run poll, the consumer skips a run that is no longer pending, a draining instance takes no more runs, and starts, child starts, continue-as-new, retries, recovery, signal and child-completion wakes, due timers and freed singleton slots all go through the queue. It also runs assertMidTurnChildWake (exported next to it) for ctx.all, ctx.allSettled and ctx.child: a child finishes while its parent's turn, on another instance, still holds the lease, and the parent must finish without waiting for reconcileMs.

import { runAdmissionBackendContract } from '@adonis-agora/durable/testing/conformance'
import { MyAdmissionBackend } from '../src/my_admission_backend.js'

// The factory receives a TEST clock, so rate-limit windows are deterministic.
runAdmissionBackendContract('MyAdmissionBackend', (clock) => new MyAdmissionBackend(clock))

Replay CI — assertReplayableCopy Anchor Link

The failure the determinism rules guard against lands on deploy: a code change renames, reorders or removes a step at a position an in-flight run's history already recorded, and the replay silently feeds the wrong checkpoint into the wrong step. assertReplayable catches that in CI instead — replay a recorded run's history against the current workflow code, and fail the build on divergence.

The loop is capture → commit → assert:

1. Capture a fixture from a real (ideally representative, or in-flight) run. From the terminal, against your running app:

node ace durable:export ord-42 --out tests/fixtures/checkout-run.json

Or in code, captureHistory(engine, runId) returns the same RunHistory ({ run, checkpoints }) from any live engine (or null for an unknown run) — including a createTestEngine() engine, so a test can record its own fixture. The onEvict archival hook receives the same pair as two arguments, (run, checkpoints), so writing { run, checkpoints } from it makes an archived run a fixture.

2. Commit the JSON next to your tests. It is the frozen contract: "code that cannot replay this history cannot be deployed over runs shaped like it."

3. Assert it in CI. parseRunHistory revives the JSON's date fields; assertReplayable seeds a throwaway in-memory engine with the history, registers the workflow exactly as your app does, and replays. Nothing is dispatched — there is no transport — so completed steps replay from their checkpoints and any positional mismatch surfaces as a thrown NonDeterminismError:

import { readFileSync } from 'node:fs'
import { test } from '@japa/runner'
import { assertReplayable, parseRunHistory } from '@adonis-agora/durable/testing'
import { checkout } from '../app/workflows/checkout.js'

test('checkout replays its recorded production history', async () => {
  const fixture = parseRunHistory(readFileSync('tests/fixtures/checkout-run.json', 'utf8'))
  await assertReplayable((engine) => engine.register('checkout', '1', checkout), fixture)
})

register is called synchronously, so for a BaseWorkflow class bind its run with engine.register rather than the async registerWorkflowClass:

import { workflowMeta } from '@adonis-agora/durable'
import CheckoutWorkflow from '#workflows/checkout_workflow'

const meta = workflowMeta(CheckoutWorkflow)! // { name, version } from `static workflow` (version defaults to '1')
await assertReplayable(
  (engine) => engine.register(meta.name, meta.version, (ctx, input) => new CheckoutWorkflow().run(ctx, input)),
  fixture,
)

A passing test means the current code replays the recorded history checkpoint-for-checkpoint. A failing one is the exact break that would have corrupted an in-flight run on deploy — fix it with a version bump or a ctx.patched gate rather than by re-recording the fixture, unless no runs of the old shape can still be in flight.

assertReplayable only throws on a non-determinism divergence (as an error named NonDeterminismError); a replay that fails for another reason does not throw, so it is not a substitute for asserting the run's outcome.