Events & interceptors

The dashboard, OpenTelemetry and Telescope integrations are all built on the same two hooks, and both are public. Reach for them when you need something none of those give you: a metric with your own labels, an audit trail in your own tables, per-step tracing your tracer understands.

engine.subscribe — the lifecycle event streamCopy Anchor Link

subscribe(listener) receives every lifecycle event the engine emits, and returns the function that detaches it:

import engine from '@adonis-agora/durable/services/main'
import logger from '@adonisjs/core/services/logger'

const unsubscribe = engine.subscribe((event) => {
  if (event.type !== 'run.failed') return

  logger.error(
    { runId: event.runId, workflow: event.workflow, err: event.error },
    'durable run failed',
  )
})

A listener that throws cannot break workflow execution — the engine isolates each delivery — but it runs synchronously on the engine's emit, so it can still slow the run down. Keep the body cheap and hand anything expensive to a queue. The engine does not await a listener: if you pass an async function, catch its errors yourself, or a rejection surfaces as an unhandledRejection.

Events from other instancesCopy Anchor Link

With a control plane configured, every instance also receives the lifecycle events every other instance produced. That is what lets a dashboard pod live-tail a run executing on a worker pod. A listener therefore sees two kinds of event: the ones its own engine emitted, and copies re-delivered from the fleet. Anything that records or counts events (a metric, an audit row, a trace) should keep only the local ones, or a three-pod fleet records every event three times:

import engine from '@adonis-agora/durable/services/main'
import { isRemoteEvent } from '@adonis-agora/durable'

engine.subscribe((event) => {
  if (isRemoteEvent(event)) return // counted on the instance that produced it
  if (event.type === 'run.completed') runsCompleted.inc({ workflow: event.workflow })
})

isRemoteEvent(event) remembers re-delivered events by identity (a WeakSet), so it adds no field to the event and nothing to the wire. Since 0.51 the built-in consumers (collectMetrics, attachDurableMetrics, attachDurableOtel and the diagnostics bridge) all skip remote events. A listener that only reacts to a run (an in-memory UI push, a cache invalidation on every pod) can keep both kinds.

The event typesCopy Anchor Link

EngineEvent.type is one of twelve values (eleven are emitted today). Every event carries runId and at; the rest of the payload depends on the type:

namespace is stamped on the run.* events only, never on step events. kind is local, remote, sleep or signal. queueMs is how long a remote step waited before a worker picked it up, and durationMs how long it then ran — the pair that separates "the step is slow" from "the fleet is saturated".

The two block events carry a diagnostics payload with everything needed to explain the block without guessing: the routing token, what the step requires, which capabilities were missing, how many live workers were considered, and the descriptors of both the control plane and the workers.

Prometheus counters, for freeCopy Anchor Link

collectMetrics(engine) subscribes for you and keeps counters you can scrape:

import router from '@adonisjs/core/services/router'
import engine from '@adonis-agora/durable/services/main'
import { collectMetrics } from '@adonis-agora/durable'

const metrics = collectMetrics(engine)

router.get('/metrics/durable', ({ response }) => {
  response.header('content-type', 'text/plain; version=0.0.4')
  return metrics.prometheus()
})

The exposition carries durable_runs_total{event} (started, completed, failed, suspended, routed, unrouted), durable_steps_total{event}, durable_runs_by_workflow_total{workflow,event}, durable_runs_routed_total{code}, and durable_step_duration_ms_sum / durable_step_duration_count for an average step duration. metrics.snapshot() returns the same numbers as a plain object, and metrics.stop() detaches. With the dashboard mounted you may not need this route at all: GET <path>/api/metrics already serves these counters plus backlog and worker gauges. See Metrics.

Counters are per process, count only the events this process produced, and reset when it restarts, so scrape every instance and sum in Prometheus rather than expecting one pod to know the fleet's totals.

Bridging to the diagnostics busCopy Anchor Link

If your app uses @adonis-agora/diagnostics, every engine event is already on that bus — the provider calls attachDurableDiagnostics(engine) at boot whenever diagnostics is installed, with no configuration. Call it yourself only when you are running the engine outside the AdonisJS provider:

import { attachDurableDiagnostics } from '@adonis-agora/durable'

const detach = attachDurableDiagnostics(engine)

engine.use — middleware around a local stepCopy Anchor Link

use(interceptor) wraps the real execution of every local step in an onion, and returns the function that removes it:

import engine from '@adonis-agora/durable/services/main'
import logger from '@adonisjs/core/services/logger'

engine.use(async (invocation, next) => {
  const startedAt = performance.now()

  try {
    return await next()
  } catch (error) {
    logger.warn(
      { workflow: invocation.workflow, step: invocation.stepName, attempt: invocation.attempt },
      'durable step threw',
    )
    throw error
  } finally {
    histogram.observe(
      { workflow: invocation.workflow, step: invocation.stepName },
      performance.now() - startedAt,
    )
  }
})

The interceptor receives the step's runId, workflow, stepName, seq and 1-based attempt, plus next(). Whatever it returns becomes the step's result — so an interceptor can transform an output — and whatever it throws fails the step, which then follows its normal retry policy.

Three properties matter more than the signature:

• First registered is outermost. Interceptors compose as an onion in registration order, so a logging interceptor registered first wraps a caching one registered second.
• They never fire on replay. A replayed step returns its recorded output without executing, so the interceptor is not called. That is what makes timings measured here real execution timings rather than replay artefacts — and it is also why an interceptor is the wrong place to enforce an invariant that must hold on every turn.
• They wrap local steps only. A dispatched ctx.step executes on a worker, in a different process, so there is no local body to wrap. The worker side does not run engine interceptors either: a transport runs a step handler directly. To time or log a remote step, wrap the handler body itself, or read durationMs / queueMs off its step.completed event.

currentStep() — step logs from deep inside a stepCopy Anchor Link

A step body receives a step logger, and the lines it records are checkpointed with the step and shown on the step in the dashboard. Code a few calls below the step (a batch inserter, an HTTP client) usually cannot take that logger as a parameter. currentStep() returns the logger of the step executing on the current async path, or undefined outside one:

import { currentStep } from '@adonis-agora/durable'

export async function insertBatch(rows: Row[]) {
  await db.table('orders').multiInsert(rows)
  currentStep()?.info('inserted batch', { rows: rows.length })
}

It works on local steps and on remote step handlers running on a worker, and concurrent steps each see their own logger. A line recorded through currentStep() lands in the same list as one recorded through the argument. Keep the optional chain: outside a step there is nothing to record to, and the same helper keeps working in a unit test with no run around it. currentStep() is new in 0.51 and is the TypeScript twin of the Python SDK's current_step().