Queries & updates

A run is durable but it is not opaque. While it is in flight — suspended on a sleep, waiting on a remote step, parked on a signal — you often need to read what it has done so far, and sometimes to steer it from the outside. The engine gives you two distinct primitives, mirroring Temporal's split between queries (read-only) and updates (mutating, validated):

• Queries publish a named value from inside the run (ctx.setEvent) and read the latest one from outside (engine.getEvent) with no side effect on the run.
• Updates expose a named update point inside the run (ctx.onUpdate) that suspends until an external engine.update delivers an argument — gated by an optional validator that can reject the update before it touches the run.

The examples import the engine singleton with import engine from '@adonis-agora/durable/services/main'. It is bound on the standalone and control-plane roles; a store-less tenant pod has no engine, so getEvent / update are not available there (the dashboard's routes for them answer 404 on that topology). See Roles & config.

QueriesCopy Anchor Link

A query is a side-effect-free read of a run's state. The run publishes values; readers observe them. Neither side resumes, suspends, or otherwise perturbs the run.

Publishing a value — ctx.setEvent(key, value)Copy Anchor Link

Inside the workflow body, call ctx.setEvent(key, value) to publish a named, queryable value. The latest value for a given key is what a reader sees; calling setEvent again with the same key overwrites it. It is checkpointed and replay-safe — on replay the publish is idempotent, so it never double-fires.

engine.register('video-encode', '1', async (ctx, job: EncodeJob) => {
  await ctx.setEvent('progress', { phase: 'probing', pct: 0 })

  const segments = await ctx.localStep('probe', () => encoder.probe(job.src))

  for (let i = 0; i < segments.length; i++) {
    await ctx.step(encodeSegment, { jobId: job.id, segment: segments[i] })
    // Publish progress after each segment — readers see the latest snapshot.
    await ctx.setEvent('progress', {
      phase: 'encoding',
      pct: Math.round(((i + 1) / segments.length) * 100),
      done: i + 1,
      total: segments.length,
    })
  }

  await ctx.setEvent('progress', { phase: 'done', pct: 100 })
  return { jobId: job.id, segments: segments.length }
})

Use setEvent for anything an outside observer might want before the run completes: progress, a partial result, a status string, an intermediate id. Because each call overwrites the previous value for that key, you can publish a key as often as you like without growing unbounded state — only the latest survives.

Reading a value — engine.getEvent(runId, key)Copy Anchor Link

From outside the run — a controller, a polling endpoint, another service — read the latest published value with engine.getEvent(runId, key). It returns the most recent value for that key, or undefined if the run never published it. Crucially, the read has no effect on the run: it does not resume a suspended run, does not consume a logical position, and does not appear in the run's history.

import type { HttpContext } from '@adonisjs/core/http'
import engine from '@adonis-agora/durable/services/main'

export default class JobsController {
  async progress({ params, response }: HttpContext) {
    const progress = await engine.getEvent<{ phase: string; pct: number }>(params.runId, 'progress')
    return response.ok(progress ?? { phase: 'pending', pct: 0 })
  }
}

getEvent works against a live run and a finished one — the published values live in the run's checkpoints, so they remain queryable after the run completes (useful for surfacing a final summary without unpacking the full output). The read is typed: pass the value type as the generic (getEvent<TValue>), and the engine returns TValue | undefined.

The dashboard exposes the same read over HTTP as GET <path>/api/runs/:id/events/:key (path defaults to /durable). It answers { value } — {} when the run never published that key, 404 for an unknown run — and passes the value through the checkpoint redaction hook, so any HTTP client can poll a run's progress without holding the engine.

1async run(ctx: WorkflowCtx, job: EncodeJob) {2  const segments = await ctx.step(this.encoder.probe, job.src);3 4  for (let i = 0; i < segments.length; i++) {5    await ctx.step(this.encoder.encodeSegment, segments[i]);6    // overwrite the 'progress' key each pass — only the latest survives7    await ctx.setEvent('progress', {8      pct: Math.round(((i + 1) / segments.length) * 100),9    });10  }11  return { jobId: job.id, segments: segments.length };12}

UpdatesCopy Anchor Link

A query reads; an update steers. An update point inside the run suspends until an external caller delivers an argument — and, unlike a raw signal, an update can be validated and rejected before it ever touches the run. That validator runs in the caller's request, so a rejected update returns a reason synchronously and leaves the run exactly as it was.

The update point — ctx.onUpdate(name, { timeoutMs? })Copy Anchor Link

Inside the workflow, await ctx.onUpdate(name) suspends the run with zero compute until an engine.update(runId, name, arg) delivers arg, then resumes with it. The name is run-scoped — unique within this run, not globally. Pass { timeoutMs } to bound the wait; if the deadline passes, the call throws SignalTimeoutError.

import { FatalError, SignalTimeoutError } from '@adonis-agora/durable'

engine.register('expense-approval', '1', async (ctx, expense: Expense) => {
  await ctx.setEvent('status', { state: 'awaiting-approval', amountCents: expense.amountCents })

  let decision: { approved: boolean; approver: string; note?: string }
  try {
    // Suspends here — no compute — until engine.update delivers a decision, or 7 days pass.
    decision = await ctx.onUpdate('decision', { timeoutMs: 7 * 24 * 60 * 60 * 1000 })
  } catch (err) {
    if (err instanceof SignalTimeoutError) {
      await ctx.setEvent('status', { state: 'expired' })
      throw new FatalError('approval timed out', 'expired')
    }
    throw err
  }

  if (!decision.approved) {
    await ctx.setEvent('status', { state: 'rejected', by: decision.approver })
    return { reimbursed: false }
  }

  await ctx.localStep('reimburse', () => ledger.reimburse(expense, decision.approver))
  await ctx.setEvent('status', { state: 'reimbursed', by: decision.approver })
  return { reimbursed: true }
})

1// the run parks on a decision, bounded by a deadline2async run(ctx: WorkflowCtx, expense: Expense) {3  await ctx.setEvent('status', { state: 'awaiting-approval' });4 5  let decision: Decision;6  try {7    // suspends with zero compute until engine.update delivers — or 7 days pass8    decision = await ctx.onUpdate('decision', { timeoutMs: 7 * DAY });9  } catch (err) {10    if (!(err instanceof SignalTimeoutError)) throw err;11    // nobody decided in time — take the default branch and fail cleanly12    await ctx.setEvent('status', { state: 'expired' });13    throw new FatalError('approval timed out', 'expired');14  }15 16  await ctx.step(this.ledger.reimburse, { expense, by: decision.approver });17  return { reimbursed: true };18}

The validator — engine.registerUpdateValidator(workflow, name, validate)Copy Anchor Link

Register a validator for a (workflow, update name) pair. It runs before the update is delivered, in the calling request — so it can enforce a business rule and reject the update without ever disturbing the run. To reject, either return a non-empty reason string or throw (the error's message becomes the reason); to accept, return nothing. The validator may be async, so it can check a database or an authorization service. Its type is UpdateValidator<TArg> = (arg: TArg) => void | string | Promise<void | string>.

Register validators once at boot, next to your workflow registrations — for example in a preloaded start/durable.ts:

import engine from '@adonis-agora/durable/services/main'
import Person from '#models/person'

type Decision = { approved: boolean; approver: string; note?: string }

engine.registerUpdateValidator<Decision>('expense-approval', 'decision', async (arg) => {
  if (!arg?.approver) return 'approver is required'
  const approver = await Person.findBy('handle', arg.approver)
  if (!approver?.canApproveExpenses) return `${arg.approver} is not allowed to approve expenses`
  if (!arg.approved && !arg.note) throw new Error('a rejection must include a note')
  // returning nothing accepts the update
})

Only one validator lives per (workflow, name); registering again replaces it. If no validator is registered, every update is accepted and delivered as-is.

Delivering an update — engine.update(runId, name, arg)Copy Anchor Link

From outside, call engine.update(runId, name, arg). It runs the validator first, then (if accepted) delivers arg to the suspended ctx.onUpdate and resumes the run. The return type makes the outcome explicit:

type UpdateResult =
  | { accepted: false; reason: string }
  | { accepted: true; run: RunResult | null }

• { accepted: false, reason } — the validator rejected it, or the run does not exist (reason is run <id> not found). The run is untouched; reason is the validator's returned string or thrown message.
• { accepted: true, run } — accepted and delivered. run is the resumed run's result, or null when nothing was waiting at that update point yet (a too-early or duplicate update).

import type { HttpContext } from '@adonisjs/core/http'
import engine from '@adonis-agora/durable/services/main'

export default class ExpensesController {
  async decide({ params, request, response }: HttpContext) {
    const result = await engine.update(params.runId, 'decision', request.body())
    if (!result.accepted) return response.badRequest({ error: result.reason })
    return response.ok({ status: result.run?.status ?? 'pending' })
  }
}

The dashboard exposes the same validated path as POST <path>/api/runs/:id/update/:name with a { arg } body: a rejection answers 422 with the reason, and nothing is delivered.

1// same run as the timeout example — but a decision arrives in time2async run(ctx: WorkflowCtx, expense: Expense) {3  await ctx.setEvent('status', { state: 'awaiting-approval' });4  // suspended here — zero compute — until engine.update delivers a decision5  const decision = await ctx.onUpdate('decision', { timeoutMs: 7 * DAY });6 7  await ctx.step(this.ledger.reimburse, { expense, by: decision.approver });8  return { reimbursed: true };9}

Why a validator beats a raw signalCopy Anchor Link

You can already steer a run with ctx.waitForSignal + engine.signal. The difference is where the rejection happens. A signal always lands; if the argument is bad, the workflow body has to detect it after resuming and somehow get the run back to a waiting state. An update with a validator rejects in the caller's request, synchronously, with a reason — the run never wakes for a bad update, and the caller gets immediate, actionable feedback.

Search attributesCopy Anchor Link

getEvent answers "what is this one run doing?" — you already know the run id. Search attributes answer the other question: "which runs match this business predicate?" — find every checkout run for tier pro with an amount over 200, regardless of id. They are typed, indexed labels you stamp on a run so the dashboard and engine.listRuns can filter on them — the durable counterpart of Temporal's search attributes.

An attribute set is a flat map of primitives:

type SearchAttributes = Record<string, string | number | boolean>

Unlike a free-form setEvent value, attributes are typed and queryable: the Lucid store explodes them into an indexed side-table so amount >= 200 is a real indexed predicate, not an in-memory filter over every row.

Stamping attributes on a runCopy Anchor Link

There are two moments to set them. At start time, pass searchAttributes in the start options (or the class dispatch options):

import CheckoutWorkflow from '#workflows/checkout_workflow'

await CheckoutWorkflow.dispatch(order, {
  runId: `checkout:${order.id}`,
  searchAttributes: { amount: order.totalCents, tier: order.tier, priority: order.rush },
})

// Or via the engine, for a string-named workflow:
await engine.start('checkout', order, `checkout:${order.id}`, {
  searchAttributes: { amount: order.totalCents, tier: order.tier },
})

From inside the run, refine them as the run learns more with ctx.upsertSearchAttributes(attrs). It shallow-merges into the run's existing attributes (keys you don't pass are kept), and is durable and exactly-once — recorded at this position on the first run and skipped on replay, so it does one write, not one per turn:

engine.register('checkout', '1', async (ctx, order: Order) => {
  await ctx.upsertSearchAttributes({ stage: 'charging' })

  const charge = await ctx.step(chargeCard, { orderId: order.id, amountCents: order.totalCents })

  // Merge in the outcome — `amount`/`tier` set at start time are preserved.
  await ctx.upsertSearchAttributes({ stage: 'shipped', chargeId: charge.chargeId })
  return { status: 'shipped', chargeId: charge.chargeId }
})

Typed attributes — a search-attributes schemaCopy Anchor Link

By default any flat string | number | boolean map is accepted. To pin the shape, declare a Standard Schema (zod, valibot, arktype, …) whose output is search-attribute-shaped — on a class workflow as static workflow = { searchAttributes }, or on a string-named one as engine.register(name, version, fn, { searchAttributesSchema }). Then:

• every ctx.upsertSearchAttributes validates the merged attributes (the schema describes the whole run's bag, so make keys that arrive later optional) and a non-conforming write throws a SchemaValidationError (a non-retryable FatalError, code schema_validation) in the body — nothing is written or recorded;
• typing run's ctx as WorkflowCtx<InferSearchAttributes<typeof schema>> narrows upsertSearchAttributes to Partial<A> at compile time;
• a schema whose output has a nested object or array value is a compile-time error, and its validate must be synchronous (an async refinement could make replay nondeterministic).

import { BaseWorkflow, type InferSearchAttributes, type WorkflowCtx } from '@adonis-agora/durable'
import { z } from 'zod'

const checkoutAttrs = z.object({
  tier: z.enum(['free', 'pro']),
  amount: z.number(),
  stage: z.enum(['charging', 'shipped']).optional(),
})

export default class CheckoutWorkflow extends BaseWorkflow {
  static workflow = { name: 'checkout', searchAttributes: checkoutAttrs }

  async run(ctx: WorkflowCtx<InferSearchAttributes<typeof checkoutAttrs>>, order: Order) {
    await ctx.upsertSearchAttributes({ tier: order.tier, amount: order.totalCents, stage: 'charging' })
    // ...
  }
}

Read them back typed with attributesOf(CheckoutWorkflow, run) (the schema is resolved from the class) or readSearchAttributes(schema, run). Both are deliberately lenient: an older run whose attributes predate the schema yields {} instead of throwing, so a read path never crashes on old history.

import { attributesOf } from '@adonis-agora/durable'
import engine from '@adonis-agora/durable/services/main'
import CheckoutWorkflow from '#workflows/checkout_workflow'

const run = await engine.getRun(runId)
const { tier, amount } = attributesOf(CheckoutWorkflow, run ?? {})

Querying by attribute — engine.listRunsCopy Anchor Link

Filter runs with an attributes array on a RunQuery. Each entry is an AttributeFilter, and the query ANDs them all together:

type AttributeOp = 'eq' | 'ne' | 'gt' | 'gte' | 'lt' | 'lte' | 'in'
type AttributeValue = string | number | boolean

type AttributeFilter =
  | { key: string; op: Exclude<AttributeOp, 'in'>; value: AttributeValue }
  | { key: string; op: 'in'; values: AttributeValue[] } // OR inside one predicate

Pair the attribute predicates with the coarse workflow / status / tag filters so the store can bound the scan before applying them:

import type { HttpContext } from '@adonisjs/core/http'
import engine from '@adonis-agora/durable/services/main'

export default class RunsController {
  // GET /admin/big-paid-checkouts
  async bigPaidCheckouts({ response }: HttpContext) {
    const runs = await engine.listRuns({
      workflow: 'checkout', // coarse filter — bounds the scan
      status: 'suspended',
      attributes: [
        { key: 'amount', op: 'gte', value: 20000 }, // amount >= 20000 cents
        { key: 'tier', op: 'in', values: ['pro', 'enterprise'] }, // AND tier is pro OR enterprise
      ],
      page: 1,
      size: 50,
    })
    return response.ok(runs)
  }
}

A few semantics worth knowing:

• A missing key never matches. If a run never set tier, it fails any filter on tier — including ne (there is simply no value to compare). Filter on keys you know you stamp.
• Booleans are compared by eq/ne only (stored as "true"/"false" under the hood); range ops (gt/gte/lt/lte) are meant for numbers (a string operand compares lexicographically).
• in matches when the value equals any member of values — the only way to OR on one key, since two eq predicates on the same key are ANDed. An empty values matches nothing.
• Empty / absent attributes matches everything — the query degrades to a plain listRuns.
• Paging is page (1-based) + size; omit size for every matching run.
• On the Lucid store the predicates push down into the indexed side-table (EXISTS subqueries); the in-memory test store applies the identical logic in-process, so a query behaves the same across adapters.