Observability

Runlane has three visibility surfaces:

• durable run history for audit, recovery, and operator workflows
• live observers for cheap in-process logs, counters, tests, and local handoff
• durable observation export for scalable out-of-band telemetry delivery

Use run history when correctness matters. Use live observers when the runtime process can do the work quickly after a durable write. Use durable export when telemetry provider latency, retries, or backpressure must stay outside task execution.

Live Observers

Configure observers on the runtime:

import { createRunlane, RunlaneObservationType } from '@runlane/core'

const runlane = createRunlane({
  lane,
  queues,
  tasks: { sendEmail },
  observability: {
    observers: [
      (observation) => {
        if (observation.type !== RunlaneObservationType.RunEvent) {
          return
        }

        metrics.increment('runlane.run_event', {
          environment: observation.environment.name,
          eventType: observation.eventType,
        })
      },
    ],
    onObserverError({ cause, observation }) {
      logger.warn({ cause, observation }, 'Runlane observer failed')
    },
  },
})

Observers run after storage confirms the event or step completion was persisted. A failed storage write does not emit an observation. A duplicate or stale race that storage rejects also does not emit an observation.

Observer failures do not roll back run state. Core catches observer rejections and reports them through observability.onObserverError when provided. If no error handler is configured, observer failures are ignored after the durable write.

Observers are live hooks, so slow observers add latency after the storage write that triggered them. Keep raw observers fast: increment a counter, write a local log line, enqueue to an application-owned bounded buffer, or hand off to an already-running telemetry loop. Do not perform one network export per observation in the observer callback.

Durable Observation Export

Durable export is a storage-owned stream over committed Runlane facts. Storage writes an observation record in the same durable operation as the run event or durable step completion. Runtime execution does not call telemetry backends, exporter loops, or provider SDKs.

Run an exporter in a separate process:

import { createRunlaneObservationExporter } from '@runlane/observability'
import { otlpObservationSink } from '@runlane/observability/opentelemetry'

const exporter = createRunlaneObservationExporter({
  consumer: 'production_otel',
  environment: { name: 'production' },
  storage,
  sink: otlpObservationSink({
    endpoint: 'http://otel-collector:4318/v1/logs',
    timeout: '10s',
  }),
})

await exporter.start({ signal })

The exporter reads the last checkpoint for one consumer, scans the next batch of observation records, passes that batch to the sink, and advances the checkpoint only after the sink accepts the batch. If the process crashes or the sink rejects, the next run resumes from the previous checkpoint.

Delivery is at least once. Sinks must tolerate duplicate delivery by record.id when duplicate external telemetry matters. The same durable run event or step completion maps to the same observation record id across exporter restarts.

Console Export

Use the console sink for local inspection, tests, or piping into a process supervisor:

import { consoleObservationSink, createRunlaneObservationExporter } from '@runlane/observability'

const exporter = createRunlaneObservationExporter({
  batchSize: 100,
  consumer: 'local_console',
  environment: { name: 'development' },
  storage,
  sink: consoleObservationSink(),
})

const result = await exporter.exportOnce()

exportOnce() performs one scan/export/checkpoint cycle and returns a structured tick result with current stats. Failed ticks include the original error so callers can inspect RunlaneError codes such as ObservationExportFailed or CapabilityUnsupported. start() keeps polling until its AbortSignal is aborted and rejects non-retryable RunlaneError failures instead of retrying bad configuration forever.

OTLP To Collector

Runlane's OTLP sink uses the official OpenTelemetry JavaScript OTLP/HTTP log exporter to send sanitized observation records to an OpenTelemetry Collector endpoint. Treat the Collector as the vendor boundary: it should receive, process, batch, retry, and export telemetry to Datadog, Honeycomb, Grafana, or another backend.

import { createRunlaneObservationExporter } from '@runlane/observability'
import { otlpObservationSink } from '@runlane/observability/opentelemetry'

const exporter = createRunlaneObservationExporter({
  batchSize: 500,
  consumer: 'production_otel',
  environment: { name: 'production' },
  idlePollInterval: '1s',
  retryDelay: '5s',
  storage,
  sink: otlpObservationSink({
    endpoint: process.env.OTLP_ENDPOINT ?? 'http://localhost:4318/v1/logs',
    headers: { 'x-runlane-source': 'production' },
    resourceAttributes: { 'service.namespace': 'jobs' },
  }),
})

Point the sink at the Collector logs endpoint, normally /v1/logs. Do not import OpenTelemetry SDKs into @runlane/core or runtime task execution to make export work.

If you need a different OpenTelemetry transport or a preconfigured exporter, adapt it with openTelemetryObservationSink() instead of rebuilding Runlane's observation-to-log-record mapping:

import { OTLPLogExporter } from '@opentelemetry/exporter-logs-otlp-http'
import { openTelemetryObservationSink } from '@runlane/observability/opentelemetry'

const sink = openTelemetryObservationSink({
  exporter: new OTLPLogExporter({ url: 'http://localhost:4318/v1/logs' }),
  resourceAttributes: { 'service.namespace': 'jobs' },
})

Stream Semantics

The storage stream is scoped by environment and ordered by a storage-assigned stream position. Observation records include:

scanObservationRecords({ environment, cursor, limit }) returns records after the supplied cursor in stable forward order. The returned cursor is the checkpoint value after the final returned record; store it only after the sink accepts the batch.

getObservationCheckpoint() reads the current checkpoint for one named consumer. advanceObservationCheckpoint() requires the expected previous cursor, only accepts cursors that reference a real stream record in the same environment, and rejects stale, missing-record, or non-monotonic advances with StorageConflictKind.ObservationCheckpoint.

Observation Shape

The current observation types are:

Observations intentionally do not include task payload, successful output, durable step output, failure detail, release metadata, or operator-expanded storage rows. Those values can contain customer data and can be large. Fetch durable run state through operator APIs only when an operator workflow actually needs detail.

High-Scale Topology

The default exporter polls storage. That is enough for many deployments because task execution writes to primary storage while exporter reads happen out of band.

For higher volume:

• Run exporters from a read replica when replica lag is acceptable. Checkpoints still write to primary storage.
• Use one active exporter loop per (environment, consumer) checkpoint. Multiple active loops with the same consumer race by design and one loses the checkpoint advance.
• Give independent sinks independent consumer ids, such as production_otel and production_audit_index.
• When polling storage is no longer enough, stream the observation table through CDC or a message queue and preserve the same stable record ids and checkpoint semantics at the downstream boundary.

Primary storage writes should not wait on telemetry provider reads, Collector outages, vendor retries, or exporter backpressure. If storage itself is overloaded, that is a storage/runtime capacity issue, not a telemetry provider dependency.

Footguns

Do not tag metrics with run ids:

if (observation.type !== RunlaneObservationType.RunEvent) {
  return
}

metrics.increment('runlane.run_event', {
  eventType: observation.eventType,
  runId: observation.runId,
})

runId is high-cardinality. It belongs in logs, traces, and operator links, not metric labels. Prefer low-cardinality dimensions such as environment and eventType.

Do not put provider I/O in the observer:

const runlane = createRunlane({
  lane,
  observability: {
    observers: [
      async (observation) => {
        await fetch('https://telemetry.example.com/events', {
          body: JSON.stringify(observation),
          method: 'POST',
        })
      },
    ],
  },
})

That makes every observed durable write wait on the telemetry provider. During a provider outage, task execution, maintenance, signal resume, and delivery paths all inherit that latency.

Do not use unbounded buffers:

import type { RunlaneObservation } from '@runlane/core'

const pending: RunlaneObservation[] = []

const runlane = createRunlane({
  lane,
  observability: {
    observers: [
      (observation) => {
        pending.push(observation)
      },
    ],
  },
})

An unbounded array only moves the outage from latency to memory growth. If the process emits observations faster than the exporter drains them, the process can run out of memory while run history remains healthy.

Do not fetch run details from inside the observer by default:

const runlane = createRunlane({
  lane,
  observability: {
    observers: [
      async (observation) => {
        const run = await runlane.runs.get(observation.runId)
        if (run !== undefined) {
          auditLog.write({ observation, output: run.output })
        }
      },
    ],
  },
})

That adds a read for every event, can leak payload/output/failure/step detail into telemetry, and turns a narrow signal into a broad data export path.

Do not drive workflow behavior from observers or exporters. Observability is telemetry over facts that are already durable. Use task handlers, context.waitForRun(...), context.waitForSignal(...), schedules, operator APIs, and storage-backed run state for behavior that must be correct.

Do not assume export is exactly once. Runlane provides stable records and at-least-once delivery; the sink owns dedupe if the external system needs it.