Overview

A state store persists workflow runs and step checkpoints — the source of truth for both durability and the dashboard. It's a pluggable StateStore driver, independent of the transport, so you can mix any store with any transport.

Stores are config-driven drivers: list the ones you use under stores in config/durable.ts, built with the stores factory, and pick the active one by name with store. The lucid driver lazily imports @adonisjs/lucid only when selected.

DriversCopy Anchor Link

The default — when config/durable.ts sets no store — is an in-memory store: zero setup, but state is lost on restart, so it's only for development and tests. For anything durable, select the Lucid store.

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

export default defineConfig({
  store: 'lucid',
  stores: {
    lucid: stores.lucid(),
  },
  transport: 'memory',
  transports: { memory: transports.memory() },
})

The interfaceCopy Anchor Link

A StateStore persists runs and checkpoints and exposes the primitives the engine needs for durability:

• Runs & checkpoints — createRun, updateRun, updateRunIf, getRun, deleteRun, saveCheckpoint, getCheckpoint, listCheckpoints. createRun must reject a duplicate id (throw — a SQL primary-key violation is exactly right) rather than overwrite: the engine treats the throw as "someone else already started this id", re-reads the existing run and returns its state, which is what makes start / signalWithStart / racing scheduler ticks converge on one run instead of forking or clobbering the winner.
• Recovery & leasing — listIncompleteRuns, listPendingRuns(limit) (so workers poll for pending runs), listDueTimers(nowMs, namespace?, limit?) (the due runs most overdue first, ascending wakeAt, with or without a limit; a suspended run carrying wakeRequestedAt is due whatever its wakeAt, and updateRun/getRun must round-trip that field; the engine passes its timerBatchSize as the limit to cap one timer pass, and a store may ignore the cap), and the atomic tryLockRun / renewRunLock / releaseRunLock that make one-instance-per-run leasing work. releaseRunLock(runId, owner?) is owner-fenced: when owner is given, release only if that owner still holds the lease (a conditional write, atomic like renewRunLock) — so a stale executor whose lease was taken over can't wipe the new owner's lease and open the door to a third concurrent executor. Omitting owner is the operator-style unconditional clear (requeue's stale-lease reset); a store implemented against the older single-argument signature simply ignores owner and degrades to the previous unfenced behavior.
• Signals — putSignalWaiter / takeSignalWaiter / listSignalWaiters / removeSignalWaiter, plus bufferSignal / takeBufferedSignal for the signal machinery.
• Events — bufferEvent / listBufferedEvents / removeBufferedEvent, so a published event that matched no live waiter is still consumed by a later ctx.waitForEvent.
• Querying — listRuns(query) for the dashboard and CLI. RunQuery carries the status/workflow/tag/namespace/attribute filters plus exact-match origin and the epoch-ms range pushdowns createdBefore / createdAfter / updatedBefore / wakeBefore — the last two are what let the retention/stalled sweeps and the blocked-run poll ask the store for candidate rows instead of fetching everything and filtering in process. Paging rides page (1-based, default 1) and size — the same offset shape @adonis-agora/filter takes, so every Agora listing is paged alike. Resolve the pair with the exported runPageWindow(query) helper rather than by hand: it returns the { limit, offset } your adapter spends, and it is what the conformance kit asserts against (no size means no bound — a store must not invent a default window).

The members below are optional, and the engine works without every one of them:

Everything else is required, including updateRunIf.

The Lucid driver implements every optional member in the table. updateRunIf is one conditional UPDATE … WHERE id = ? AND status IN (…), saveCheckpointIfPending is one conditional UPDATE … WHERE status = 'pending', never a read followed by a write, hasCheckpoints is one SELECT seq … ORDER BY seq DESC LIMIT 1 over the run's seq range, and countCheckpoints and countRuns are single COUNT(*) queries.

Two things a store must not try to do in SQL: decide whether a child failed, or count the children a folded checkpoint stands for. Both live in the payload ({ ok: false }, output.values), which a CodecStateStore may have encrypted. The engine loads those rows through queryCheckpoints and reads them after decoding.

updateRunIf — the compare-and-set writeCopy Anchor Link

updateRunIf(
  runId: string,
  expectedStatuses: RunStatus[],
  patch: Partial<WorkflowRun>,
): Promise<boolean>

Apply patch only if the run's currently persisted status is one of expectedStatuses at write time. It returns true when the write landed and false when the predicate did not match. A non-match is a normal outcome, never a throw.

Why it existsCopy Anchor Link

A workflow turn computes its outcome from a run snapshot loaded when the turn began — and a turn can run for minutes across remote steps. Meanwhile the run's persisted status can change through a completely different path: an operator cancels it, the execution-timeout sweep marks it cancelled, a poison run is dead-lettered, or ctx.all in failFast mode cancels a sibling.

With a plain updateRun, the turn's now-stale terminal write lands last and resurrects a terminal run: a cancelled run silently reports completed, its output is written, its parent is notified, and a run.completed event fires. updateRunIf moves the "is this run still eligible?" test into the same atomic write, so the loser of the race gets false, emits nothing, and reports the run's true current status instead.

The engine leans on it in five places — the execution-timeout sweep, all three terminal/suspend transitions of a settling turn, and parking a run as blocked — and the conformance kit covers it in three cases, including the two-sequential-writes race directly.

Here is the Lucid driver's implementation, as the shape to copy:

const affected = await trx
  .from(DURABLE_TABLES.runs)
  .where('id', runId)
  .whereIn('status', expectedStatuses)
  .update(row)

return rowsAffected(affected) === 1

Writing a custom storeCopy Anchor Link

Implement every required member above, then run the shared conformance contract against it — that is the definition of "correct", not the prose here:

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

runStateStoreContract('MyStateStore', async () => {
  const store = new MyStateStore(/* connection */)
  await store.ensureSchema?.()

  return {
    store,
    cleanup: async () => {
      await store.close()
    },
  }
})

The contract requires a few of the optional members above, because the engine's fallback for them is slow or unsafe: getLatestCheckpointByName, listCheckpointsByNamePrefix, countCheckpoints, hasCheckpoints, countRuns, saveCheckpointIfPending, saveCheckpointsIfPending, deleteRuns, runFacets, pruneBufferedEvents, the five schedule methods, and the checkpointStats + queryCheckpoints pair — whose answers are compared against in-memory references, and whose run detail and fan pages must equal the ones built from a full read. A store with withScope also has its scoping checked. A store that leaves one out fails its case. It also runs cases for settleCheckpointsIfPending (a refused entry, a duplicate in one call, every column a result does not carry left as stored, a batch wider than one statement) getCheckpointsForSettle (every row loses its input, everything else reads as getCheckpoints reads it) and listCheckpointsForReplay (only a completed row loses its input, everything else reads as listCheckpoints reads it) when the store has them. For the last two of the required members it also checks that both seq bounds are inclusive, that a count covers only its own run, that a refused write leaves a completed or failed row unchanged, and that the settling write replaces the pending row whole, clearing its wakeAt.

The factory runs once per test case, so each case gets a fresh store. Members that are optional for the engine but not required by the kit (listOrphanedRuns, listSignalWaitersByRunIds, hasSignalWaiter, registerChildWaits, compactCheckpoints, runValueFacets, tryAdmitSingleton, transaction) are checked only when the store has them. Two escape hatches exist for drivers that genuinely cannot satisfy a slice of the contract, returned from the factory next to store and cleanup: supportsAsyncTransaction: false (a synchronous driver) and supportsTagFilter: false. If the backing infrastructure is not available — no Docker for a testcontainers store — throw StateStoreUnavailableError from the factory and the suite skips instead of failing.

Auto-schemaCopy Anchor Link

By default the provider calls the selected store's ensureSchema() at boot — for the Lucid driver, the same idempotent createDurableTables(db, connection) the published migration runs. Set autoSchema: false in config/durable.ts to own the schema through migrations only; the in-memory store has no schema, so the flag is a no-op for it. See the Lucid page for the tables it creates and the upgrade caveat. It only ever adds the durable tables — it never touches yours.

Exactly-once DB writes with ctx.transactionCopy Anchor Link

A plain ctx.step is at-least-once: if the process dies after the step's business write commits but before its checkpoint does, the step re-runs on replay. ctx.transaction(name, fn) closes that gap by running fn and saving the step's checkpoint inside one store transaction, so they commit or fail together. It relies on the optional transaction(work) member: the store hands work its native transaction handle (raw) plus a saveCheckpoint that writes inside that same transaction.

For the Lucid driver, raw is the Lucid TransactionClientContract opened on the store's connection — run your writes on it, not on the app's regular db service, or they land outside the transaction. On a store without transaction(), ctx.transaction throws. See Transactional steps for a full example.

Retention & pruningCopy Anchor Link

Terminal runs (completed, failed, cancelled, dead) otherwise accumulate forever. The retention config — per-status ages plus scoped policies — is swept by the durable:work tick, throttled to one pass a minute per instance. The sweep needs nothing beyond the required members: it asks listRuns for up to 100 candidates per pass using the statuses and updatedBefore pushdowns, then deletes them (and their child subtrees) through deleteRuns in one call, or deleteRun per run on a store without it. Any store that implements the required interface supports retention. See Retention & archival.

Encrypting payloads at restCopy Anchor Link

CodecStateStore decorates any store and runs run/step payloads through a PayloadCodec — encoded on write, decoded on read — so inputs and outputs are never stored in the clear. Searchable metadata (id, status, workflow, tags, namespace, timestamps) is left untouched, so the dashboard, queries and recovery keep working.

import { createCipheriv, createDecipheriv, randomBytes } from 'node:crypto'
import { CodecStateStore, defineConfig, stores, type PayloadCodec } from '@adonis-agora/durable'
import env from '#start/env'

const key = Buffer.from(env.get('DURABLE_PAYLOAD_KEY'), 'base64') // 32 bytes

const aesCodec: PayloadCodec = {
  encode(value) {
    const iv = randomBytes(12)
    const cipher = createCipheriv('aes-256-gcm', key, iv)
    const body = Buffer.concat([cipher.update(JSON.stringify(value), 'utf8'), cipher.final()])
    return {
      iv: iv.toString('base64'),
      tag: cipher.getAuthTag().toString('base64'),
      body: body.toString('base64'),
    }
  },
  decode(value) {
    const { iv, tag, body } = value as { iv: string; tag: string; body: string }
    const decipher = createDecipheriv('aes-256-gcm', key, Buffer.from(iv, 'base64'))
    decipher.setAuthTag(Buffer.from(tag, 'base64'))
    const plain = Buffer.concat([decipher.update(Buffer.from(body, 'base64')), decipher.final()])
    return JSON.parse(plain.toString('utf8'))
  },
}

export default defineConfig({
  store: 'encrypted',
  stores: {
    // Build the Lucid store with its own factory, then wrap it. The factory resolves Lucid's `db`
    // from the container at boot; `@adonisjs/lucid/services/db` is still undefined at that point.
    encrypted: async (ctx) => new CodecStateStore(await stores.lucid()(ctx), aesCodec),
  },
})

encode and decode are synchronous and decode must be the exact inverse of encode — wrap an async KMS behind a local cache yourself. Beyond run and checkpoint payloads the decorator also encodes buffered signal and event payloads.

CodecStateStore never hides a capability of the store it wraps. saveCheckpointIfPending, saveCheckpointsIfPending, settleCheckpointsIfPending, getCheckpointsForSettle, listCheckpointsForReplay, countCheckpoints, hasCheckpoints, registerChildWaits, compactCheckpoints, hasSignalWaiter, deleteRuns, runFacets, pruneBufferedEvents, tryAdmitSingleton, withScope, the checkpointStats + queryCheckpoints pair and the five schedule methods are bound only when the inner store has them, so the engine's fallbacks still apply through the wrapper. A few members are always present on the wrapper and fall back inside it when the inner store lacks them: listOrphanedRuns (filters listIncompleteRuns), listSignalWaitersByRunIds (filters a full waiter scan), countRuns (counts a listing), runValueFacets (counts a bounded scan), getLatestCheckpointByName and listCheckpointsByNamePrefix (filter listCheckpoints), recordStepHeartbeat (a no-op) and transaction() (throws). Payload-carrying writes (updateRun and updateRunIf patches, saveCheckpointIfPending, the placeholder checkpoints of registerChildWaits, the folded checkpoint of compactCheckpoints, a schedule's spec) are encoded; countCheckpoints, hasCheckpoints and hasSignalWaiter read no payload and are forwarded verbatim. recordStepHeartbeat encodes the progress payload under extended coverage, the run reads (getRun, listRuns, listPendingRuns, listDueTimers, listOrphanedRuns) decode what they return, and transaction() encodes the checkpoint its saveCheckpoint writes. The batch and aggregate members follow the same rule: saveCheckpointsIfPending encodes every entry, settleCheckpointsIfPending encodes each settlement's output (and events and error under extended coverage), getCheckpointsForSettle and listCheckpointsForReplay decode the rows they return, queryCheckpoints decodes the rows it returns, checkpointStats, runFacets, deleteRuns and pruneBufferedEvents read no payload and are forwarded verbatim, and withScope wraps the inner store's scoped view in the same codec.