External tasks & control primitives

This page collects three smaller authoring primitives that share a theme: each one hands a boundary — a foreign system, the database, a human operator — a clean, durable seam.

• ctx.task — call something that will report back later, suspend until it does. The async-completion primitive.
• ctx.transaction — run a step whose business write and "done" checkpoint commit in the same database transaction.
• ctx.breakpoint — pause a run at a point and let a human resume it from the dashboard.

External async-completion tasks — ctx.taskCopy Anchor Link

A durable webhook waits for an inbound HTTP callback. But plenty of "fire at a foreign system and wait for it to finish" flows don't come back over HTTP at all — you drop a message on SQS and a non-durable consumer eventually publishes a result, you kick off a long-running Python job, you enqueue work for a human review tool. The shape is always the same: dispatch once, suspend with zero compute, resume when the far side reports a result back by an id. That is ctx.task.

ctx.task<TResult>(
  name: string,               // unique per run — the correlation id the far side reports back on
  dispatch: () => Promise<void>, // fire the request (runs once, checkpointed)
  options?: StepOptions,      // retries/backoff for the dispatch call, not for the wait
): Promise<TResult>

Under the hood ctx.task runs dispatch exactly once (as a checkpointed local step named task:dispatch:<name>, so it never re-fires on replay/recovery), then suspends the run on the signal token task:<runId>:<name>. Nothing runs, nothing polls, no thread is held. When the far side is done, your code calls engine.completeTask(runId, name, result) (or engine.failTask) and the run resumes with the result.

name must be unique per run: it is the id both the dispatch checkpoint and the completion key on. The far side therefore needs two things to report back, the run id (ctx.runId) and the task name, so put both in the message you dispatch. options is a StepOptions (retries, backoff, backoffMs, …) and governs the dispatch call only; the wait itself has no timeout.

import { BaseWorkflow } from '@adonis-agora/durable'
import type { WorkflowCtx } from '@adonis-agora/durable'
import transcodeQueue from '#services/transcode_queue' // your SQS / broker client

interface TranscodeResult {
  outputUrl: string
  durationMs: number
}

export default class TranscodeWorkflow extends BaseWorkflow {
  static workflow = { name: 'transcode', version: '1' }

  async run(ctx: WorkflowCtx, job: { id: string; src: string }) {
    // Dispatch to a foreign, NON-durable worker (an SQS queue, a Python service, a GPU box),
    // then suspend with zero compute until it reports back with this run id + task name.
    const result = await ctx.task<TranscodeResult>('transcode', async () => {
      await transcodeQueue.send({ runId: ctx.runId, task: 'transcode', src: job.src })
    })

    await ctx.step('publish-asset', { jobId: job.id, url: result.outputUrl })
    return result
  }
}

Reporting the result back — engine.completeTask / engine.failTaskCopy Anchor Link

The far side eventually finishes and calls back into your app — typically a webhook, a queue consumer, or a callback controller — which resolves the task:

// Report success: the run resumes with `result` as the return value of ctx.task.
engine.completeTask(runId: string, name: string, result: unknown): Promise<RunResult | null>

// Report failure: the run resumes and the ctx.task call throws a FatalError.
engine.failTask(runId: string, name: string, error: string): Promise<RunResult | null>

Both deliver a signal on task:<runId>:<name>, so they inherit signal buffering:

• When the run is waiting, they resume it and return its RunResult.
• When no run is waiting yet (the far side answered before the run reached the wait), the completion is buffered and they return null. The run consumes it when it reaches ctx.task, so a fast worker never loses its result.
• A duplicate or late delivery, after the run has moved past the task, also returns null. It can't corrupt the run: nothing waits on that token again.

failTask makes ctx.task throw a FatalError with the message task "<name>" failed: <error>. A fatal error is not retried, so catch it in the workflow if you want a fallback branch.

The dashboard exposes the same two calls as POST /api/runs/:id/tasks/:name/complete and .../fail, so an operator can settle a stuck task by hand.

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

export default class TranscodeCallbacksController {
  // The transcode worker POSTs here when a job finishes, echoing the runId + task it was sent.
  async handle({ request, response }: HttpContext) {
    const { runId, task, status, outputUrl, durationMs, error } = request.body()

    if (status === 'ok') {
      await engine.completeTask(runId, task, { outputUrl, durationMs })
    } else {
      await engine.failTask(runId, task, error ?? 'transcode failed')
    }

    return response.noContent()
  }
}

Transactional steps — ctx.transactionCopy Anchor Link

A plain ctx.step/ctx.localStep writes its checkpoint after the body runs. That's exactly-once for pure or idempotent work, but for a non-idempotent database write there's a window: if the process crashes after the write commits but before the checkpoint is saved, recovery re-runs the body and the write happens twice.

ctx.transaction(name, fn) closes that window. It runs your write and persists the step's checkpoint inside one store transaction, so the business row and the "done" marker commit atomically — the write can never be done-but-not-checkpointed. On replay it returns the recorded output without re-running fn.

ctx.transaction<TOutput>(
  name: string,
  fn: (tx: unknown) => Promise<TOutput>, // receives the store-native transaction handle
): Promise<TOutput>

fn receives the store-native transaction handle — with the Lucid store, a Lucid TransactionClientContract opened on the store's own connection; a custom store hands over whatever its own transaction object is. Do your writes on that handle (not on the app's db service) so they land in the same transaction as the checkpoint. That also means the tables you write must live in the same database as the durable tables.

Like any step, the call is position-checked on replay: renaming name on a workflow with runs in flight throws a NonDeterminismError, so treat it as a versioned change.

import { BaseWorkflow } from '@adonis-agora/durable'
import type { WorkflowCtx } from '@adonis-agora/durable'
import type { TransactionClientContract } from '@adonisjs/lucid/types/database'

export default class PayoutWorkflow extends BaseWorkflow {
  static workflow = { name: 'payout', version: '1' }

  async run(ctx: WorkflowCtx, payout: { id: number; accountId: number; cents: number }) {
    // The ledger insert commits ATOMICALLY with the step's "done" checkpoint. A crash mid-flight
    // either commits both or neither — the payout is never double-recorded on recovery.
    const ledgerId = await ctx.transaction('record-payout', async (raw) => {
      const trx = raw as TransactionClientContract
      await trx.from('accounts')
        .where('id', payout.accountId)
        .decrement('balance_cents', payout.cents)
      const [row] = await trx.table('ledger_entries')
        .insert({ payout_id: payout.id, cents: payout.cents })
        .returning('id')
      return row.id as number
    })

    return { payoutId: payout.id, ledgerId }
  }
}

Human-in-the-loop pauses — ctx.breakpointCopy Anchor Link

ctx.breakpoint(label?) pauses a run at a point and waits for a human to resume it — the durable equivalent of a debugger breakpoint. It records a visible pending checkpoint (so the pause shows up in the run's timeline in the dashboard), then suspends with zero compute. An operator resumes it from the dashboard (or that page's POST <path>/api/runs/:id/continue), or you call engine.continue(runId) yourself — the run replays past the breakpoint and carries on. engine.continue returns the resumed RunResult, or null when the run is not paused at a breakpoint, so calling it blindly is safe.

ctx.breakpoint(label?: string): Promise<void>

The optional label names the pause point in the timeline (breakpoint:<label>), which matters when a workflow has more than one. Gate a breakpoint on your own config so it's opt-in per run — invaluable for staging a risky pipeline where you want a human to eyeball the extracted data before it's committed:

import { BaseWorkflow } from '@adonis-agora/durable'
import type { WorkflowCtx } from '@adonis-agora/durable'

export default class ImportWorkflow extends BaseWorkflow {
  static workflow = { name: 'import', version: '1' }

  async run(ctx: WorkflowCtx, job: { id: string; dryRun: boolean }) {
    // The string form returns `unknown`, so name the payload with a generic (or pass a
    // @Step reference, which infers it).
    const extracted = await ctx.step<{ rows: Record<string, unknown>[] }>('extract', { jobId: job.id })
    await ctx.setEvent('preview', extracted) // publish so a reviewer can inspect it

    // On a supervised import, park here until a human approves from the dashboard.
    if (job.dryRun) {
      await ctx.breakpoint('after-extraction')
    }

    await ctx.step('load', { jobId: job.id, rows: extracted.rows })
    return { jobId: job.id, loaded: extracted.rows.length }
  }
}

Resuming from a controller is a one-liner:

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

export default class ImportsController {
  // POST /imports/:runId/continue
  async continue({ params, response }: HttpContext) {
    const result = await engine.continue(params.runId)
    return result ? response.ok({ resumed: true }) : response.notFound({ resumed: false })
  }
}