MimirWorkflows.Runner (mimir_workflows v0.7.0)

Copy Markdown View Source

Executes a list of step specs in topological phase order.

Steps whose dependencies are all satisfied by earlier phases share a phase and run concurrently via Task.async_stream; phases execute sequentially. A failure in any step halts the run and returns an error tuple carrying the step id; the :halt option decides whether the rest of the failing phase is stopped or allowed to finish first.

Step spec shape

%{
  id: term(),
  module: module(),   # implements MimirWorkflows.Step
  params: map(),
  depends_on: [term()]
}

This shape is a host-seam contract, not a runtime-validated one: unlike MimirWorkflows.Spec.parse/1's schema-validated IR, run/2 does not check that a step map carries all four keys, that module implements MimirWorkflows.Step, or that params/depends_on have the right container type. Dag.waves/1 validates values it can reason about (unknown depends_on ids, cycles) but assumes the shape is already correct. Callers that build specs from a schema-validated %Spec{} (e.g. via MimirWorkflows.Compiler.compile/2) get this for free; callers who hand-build specs are responsible for the shape themselves — a malformed entry raises (e.g. KeyError, UndefinedFunctionError) rather than returning a diagnostic.

Return value

{:ok, %{step_id => result}}
| {:error, {:invalid, :cyclic | {:unknown_dependency, term()}}}
| {:error, {:step_failed, step_id, reason}}
| {:error, {:step_crashed, step_id, reason}}

Telemetry

eventmeasurementsmetadata
[:mimir_workflows, :run, :start]%{system_time}%{run_ref} ∪ telemetry_meta
[:mimir_workflows, :run, :stop]%{duration} (native)%{run_ref, status: :ok | :error} ∪ telemetry_meta
[:mimir_workflows, :step, :start]%{system_time}%{run_ref, step_id, phase} ∪ telemetry_meta
[:mimir_workflows, :step, :stop]%{duration}%{run_ref, step_id, phase} ∪ telemetry_meta
[:mimir_workflows, :step, :exception]%{duration}%{run_ref, step_id, phase, reason} ∪ telemetry_meta

A step whose task the runner has to reap, or that exits without reporting, gets its :exception event from the runner process instead, and its duration is a bound, not a measurement:

  • killed on timeout: reason: :timeout, and duration is the configured deadline, a lower bound on the time since the task was spawned (just before :start); the timer fires late, never early.
  • any other exit, which reaches the runner only when the caller traps exits: reason is the exit reason, and duration the time since the step's phase began, an upper bound.

A step the runner stops in an :immediate halt emits no :stop or :exception; see :halt.

Handlers for a step's own events run inside its task, so a slow :stop handler can push the task past its deadline; that step then gets :stop followed by a timeout :exception, and the run fails.

The :telemetry_meta option is how hosts thread correlation ids (workflow ids, request ids) into every event without this library knowing what they mean. Emission is hand-rolled rather than :telemetry.span/3 because a step's {:error, _} tuple is data, not a raise.

Summary

Types

Opaque step identifier; compared by value only, never atomized.

One entry in the list passed to run/2. See "Step spec shape" above — this type documents the expected shape but run/2 does not validate it at runtime.

Functions

Runs the step specs and returns results keyed by step id.

Types

error()

@type error() ::
  {:invalid, :cyclic | {:unknown_dependency, step_id()}}
  | {:step_failed, step_id(), term()}
  | {:step_crashed, step_id(), term()}

step_id()

@type step_id() :: term()

Opaque step identifier; compared by value only, never atomized.

step_spec()

@type step_spec() :: %{
  id: step_id(),
  module: module(),
  params: map(),
  depends_on: [step_id()]
}

One entry in the list passed to run/2. See "Step spec shape" above — this type documents the expected shape but run/2 does not validate it at runtime.

Functions

run(steps, opts \\ [])

@spec run(
  [step_spec()],
  keyword()
) :: {:ok, %{required(step_id()) => term()}} | {:error, error()}

Runs the step specs and returns results keyed by step id.

Options:

  • :max_concurrency — cap on in-phase parallelism; defaults to System.schedulers_online(), never more than the phase size.
  • :timeout — per-step timeout in milliseconds (default 120_000); a timed-out step's task is killed and surfaces as {:step_crashed, step_id, :timeout}.
  • :telemetry_meta — map merged into every telemetry event's metadata (default %{}).
  • :halt — what a failure does to the rest of its phase. :immediate (the default) stops the phase's other steps at once; a stopped step emits no :stop or :exception. :after_phase lets every step of the phase finish, then returns the first failure in completion order; later phases never start. Any other value raises ArgumentError.