Testing

Unit-test workflows with an in-memory engine harness, crash/flaky-step injection, and replay assertions — no Postgres, no Redis, no real time.

@dudousxd/nestjs-durable-testing 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.

pnpm add -D @dudousxd/nestjs-durable-testing

A test engine

import { createTestEngine, assertRunStatus, assertOutput } from '@dudousxd/nestjs-durable-testing';

const t = createTestEngine(); // { engine, store, transport, clock, tick }
t.engine.register('checkout', '1', async (ctx) => {
  await ctx.step('reserve', {});
  return ctx.step('ship', {});
});
t.transport.handle('reserve', () => reserve()); // register each step's handler on the transport
t.transport.handle('ship', () => ship());

const { runId } = await t.engine.start('checkout', order, 'run1'); // enqueues → { status: 'pending' }
await t.engine.waitForRun(runId);                                  // resolves when the run settles
await assertRunStatus(t.store, 'run1', 'completed');

One call instead of start + waitForRun

t.run(workflow, input, runId, opts) is the same start then waitForRun in one shot, for tests that only care about the settled result:

const result = await t.run('checkout', order, 'run1');
expect(result.status).toBe('completed');

Control time (durable sleep)

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

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

const { runId } = await t.engine.start('digest', {}, 'run1'); // enqueues → { status: 'pending' }
await t.engine.waitForRun(runId);                             // 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 is t.clock.advance(ms) followed by engine.resumeDueTimers(clock.now()) — the clock itself (MutableClock) is a standalone piece of createTestEngine, with set(ms) (jump to an absolute instant) alongside advance(ms) (move forward by a delta). Reach for t.clock directly, without tick, when you need an absolute time before a run starts, or you're driving code that only takes a bare now: () => number and isn't wired into resumeDueTimers:

import { MutableClock } from '@dudousxd/nestjs-durable-testing';

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

Inject crashes & retries

failOnce and failTimes are step-body factories that throw before succeeding, so a test can drive the engine's retry/resume path deterministically instead of hoping a real dependency flakes. Each 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.

import { failOnce, failTimes, assertStepAttempts } from '@dudousxd/nestjs-durable-testing';

// Throws once, then succeeds — proves the step is retried and the run resumes on attempt 2.
t.engine.register('wf', '1', async (ctx) => ctx.step('charge', {}, { retries: 3 }));
t.transport.handle('charge', failOnce({ ok: true }));
const { runId } = await t.engine.start('wf', {}, 'run1');
await t.engine.waitForRun(runId);
await assertStepAttempts(t.store, 'run1', 'charge', 2); // failed once, then succeeded

Use failTimes(n, …) to exhaust a retry budget — the classic "does a poison step actually fail the run after retries?" test — optionally with a specific error so the assertion is precise:

// retries: 2 means 3 total attempts; failing all 3 must fail the run.
t.engine.register('wf', '1', async (ctx) => ctx.step('flaky', {}, { retries: 2 }));
t.transport.handle('flaky', failTimes(3, { ok: true }, new Error('upstream 503')));
const { runId } = await t.engine.start('wf', {}, 'run1');
await t.engine.waitForRun(runId);
await assertRunStatus(t.store, 'run1', 'failed');
await assertStepAttempts(t.store, 'run1', 'flaky', 3); // all attempts thrown → run failed

Assertions

assertRunStatus, assertOutput, assertStepsRan, assertStepAttempts, and recordedSteps — all read the store, so they work against any run the engine produced.

Replay a captured run in CI

A code change can rename, reorder, or remove a step at a position an in-flight run already recorded a checkpoint for. The engine catches that as a NonDeterminismError on replay — but you want to catch it in CI, before it reaches a real in-flight run on deploy. assertReplayable(register, history) replays a captured run's history against the CURRENT workflow code and rethrows if they diverged:

import { InMemoryStateStore, WorkflowEngine } from '@dudousxd/nestjs-durable-core';
import { assertReplayable, type RunHistory } from '@dudousxd/nestjs-durable-testing';

// Capture a real run's history — do this once, then commit the result as a fixture.
async function recordHistory(register: (e: WorkflowEngine) => void): Promise<RunHistory> {
  const store = new InMemoryStateStore();
  const engine = new WorkflowEngine({ store });
  register(engine);
  await engine.start('pipeline', {}, 'run1');
  await engine.waitForRun('run1');
  const run = await store.getRun('run1');
  if (!run) throw new Error('no run');
  return { run, checkpoints: await store.listCheckpoints('run1') };
}

const registerPipeline = (engine: WorkflowEngine) =>
  engine.register('pipeline', '1', pipeline.run); // register exactly as the app does

it('replays the recorded fixture against the current workflow code', async () => {
  const fixture = await recordHistory(registerPipeline); // or: JSON committed from a real run
  await assertReplayable(registerPipeline, fixture);      // throws NonDeterminismError on divergence
});

Testing a custom store/transport/admission backend

If you're writing a new StateStore, Transport, or AdmissionBackend adapter, don't write your own behavioral spec from scratch — run the shared conformance suite the shipped adapters already run against. Each one throws (or fails its it block) on any divergence from the canonical (in-memory) semantics.

State store — runStateStoreContract(name, makeStore) takes a StateStoreFactory that returns a fresh, empty store plus a cleanup:

import { runStateStoreContract } from '@dudousxd/nestjs-durable-testing';
import { MyStateStore } from './my-state-store';

runStateStoreContract('MyStateStore', async () => ({
  store: new MyStateStore(/* fresh, empty */),
  cleanup: async () => {
    /* close the connection, drop the schema, etc. */
  },
}));

If the factory can't reach its backend at all (no Docker, no local Postgres), throw StateStoreUnavailableError from it instead of letting the connection error fail the suite — the contract skips that case gracefully rather than reporting a false failure.

Transport — assertTransportConformance(transport, idPrefix?) proves a durable remote step round-trips to a worker and back, for both success and a throwing handler:

import { assertTransportConformance } from '@dudousxd/nestjs-durable-testing';
import { MyTransport } from './my-transport';

it('conforms to the Transport contract', async () => {
  await assertTransportConformance(new MyTransport());
});

Admission backend — runAdmissionBackendContract(name, makeBackend) takes an AdmissionBackendFactory ((clock) => AdmissionBackend) and asserts concurrency caps, rate limits, priority, FIFO/LIFO ordering, and fairness:

import { runAdmissionBackendContract } from '@dudousxd/nestjs-durable-testing';
import { MyAdmissionBackend } from './my-admission-backend';

runAdmissionBackendContract('MyAdmissionBackend', (clock) => new MyAdmissionBackend(clock));

If your store is a new ORM adapter, also assert its physical column names against the canonical snake_case contract, so a mismatch fails a unit test instead of "Unknown column" at runtime: assertDurableColumns(resolve) walks every table/property in DURABLE_CANONICAL_COLUMNS and returns the mismatches (empty when conformant) — expect(assertDurableColumns(resolve)).toEqual([]). If your adapter intentionally preserves camelCase columns instead of following the snake_case contract, build the matching expectation map with preserveColumnExpectation() and pass it in place of the default.