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
| event | measurements | metadata |
|---|---|---|
[: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, anddurationis 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:
reasonis the exit reason, anddurationthe 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
Functions
Runs the step specs and returns results keyed by step id.
Types
Functions
@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 toSystem.schedulers_online(), never more than the phase size.:timeout— per-step timeout in milliseconds (default120_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:stopor:exception.:after_phaselets every step of the phase finish, then returns the first failure in completion order; later phases never start. Any other value raisesArgumentError.