PubSub broadcasting for workflow step lifecycle events.
Workers broadcast on two topic levels so LiveViews can subscribe at whatever granularity they need:
"workflow:all"— every event from every workflow (for the index view)"workflow:{workflow_id}"— events for one specific workflow (for the detail view)
Event shapes
# Step changed state
{:workflow_step_updated, %{
workflow_id: "uuid",
workflow_label: "my-etl",
step_name: "fetch_token",
worker: "MyApp.Workers.FetchToken",
state: "completed", # see the state list below
job_id: 123,
attempt: 1,
has_result: true,
error: nil, # or error string on failure
detail: nil, # or a state-specific map (see "awaiting")
timestamp: ~U[...]
}}state is one of "executing", "snoozed", "awaiting", "expanded",
"completed", "retryable", "discarded", or "cancelled". Consumers that
match on it should keep a catch-all clause; new states are added as the
engine grows ways to wait.
"expanded" is not a resting state: it reports that a dynamic fan-out step
just created detail.count new steps (Baton.Expansion) and is about to
park until they finish. It is the one event that means the graph itself
changed shape.
"awaiting" means the step is parked on work happening somewhere else — a
provider batch that will take hours — as opposed to "snoozed", which means
it is waiting for a resource of its own: detail carries
%{reason: r, seconds: n} where reason is "deps" (upstream dependency
pending), "rate_budget" (host rate limiter starved), "provider_limit"
(provider 429/529), "batch_slot" (host batch-submission slot unavailable),
or "step" (the step snoozed itself), and seconds is the announced wait —
what the step asked Oban for, an upper bound on the actual wait.
"awaiting"'s detail map is %{batch_id: "msgbatch_...", seconds: n} and
repeats on every poll, so a UI can show what is being waited on rather
than a step that appears to flicker.
# Whole workflow reached terminal state
{:workflow_finished, %{
workflow_id: "uuid",
workflow_label: "my-etl",
outcome: :completed, # or :failed
failed_steps: ["step_b"],
timestamp: ~U[...]
}}
Summary
Functions
Broadcast that a step is waiting on out-of-band work, e.g. a provider batch.
Broadcast that a step was cancelled (cascaded from a failed dep).
Broadcast that a step completed successfully.
Broadcast that a step was permanently discarded.
Broadcast that a step expanded itself into count new steps.
Broadcast that a step failed (will retry).
Broadcast that a step snoozed, and why.
Broadcast that a step has started executing.
Broadcast that a whole workflow has reached a terminal state.
Subscribe to events for all workflows. Use in the index LiveView.
Subscribe to events for one specific workflow. Use in the detail LiveView.
Unsubscribe from a specific workflow's topic.
Functions
Broadcast that a step is waiting on out-of-band work, e.g. a provider batch.
Emitted on submit and again on every poll. detail describes what is being
waited on and for how long the step parked itself this time
(%{batch_id: id, seconds: poll_interval}) and rides along in the payload.
Broadcast that a step was cancelled (cascaded from a failed dep).
Broadcast that a step completed successfully.
Broadcast that a step was permanently discarded.
Broadcast that a step expanded itself into count new steps.
Emitted once by Baton.Expansion, immediately before the expander parks to
wait for the children it just created. detail is %{count: n}. A UI
holding a compiled graph should treat this as "re-read the graph": the
workflow has more steps than it did a moment ago.
Broadcast that a step failed (will retry).
Broadcast that a step snoozed, and why.
reason names what the step is waiting for — :deps (upstream dependency
not yet terminal), :rate_budget (host rate limiter starved),
:provider_limit (provider 429/529), :batch_slot (host batch-submission
slot unavailable), or :step (the step snoozed itself without saying why).
seconds is the announced wait — the duration the step asked Oban for,
an upper bound on the actual wait since completion-triggered rescheduling
can wake a dep-snoozed job early. Both ride the payload's detail map as
%{reason: "deps", seconds: 15}.
Broadcast that a step has started executing.
@spec broadcast_workflow_finished( String.t(), String.t() | nil, :completed | :failed, [String.t()], keyword() ) :: :ok
Broadcast that a whole workflow has reached a terminal state.
Sent once by Baton.Completion when the last outstanding step settles.
outcome is :completed (every step succeeded) or :failed (at least one
step was cancelled or discarded); failed_steps lists the offending step
names. Also emits [:baton, :workflow, :finished] telemetry.
Subscribe to events for all workflows. Use in the index LiveView.
Subscribe to events for one specific workflow. Use in the detail LiveView.
Unsubscribe from a specific workflow's topic.