All database access for workflow_nodes.
Centralizing node queries here keeps the engine modules (Check, Results,
Reschedule) free of Ecto and gives the library a single, testable data
layer. Every function resolves its repo at runtime via
Baton.Config.repo/0.
Results are stored wrapped as %{"value" => term} so that non-map results
(lists, scalars, strings) round-trip through the jsonb column. Callers always
receive the unwrapped value.
Summary
Functions
Distinct Oban queues of completed_step's dependents that are now available.
This job's engine checkpoint, or {:error, :no_checkpoint}.
Drop this job's checkpoint. Idempotent: clearing an absent checkpoint (or a
pruned node) is :ok, since the post-condition already holds.
Step names in a workflow that count as complete for dependency promotion:
those whose Oban job is completed, plus seeded nodes (complete by
construction — they never had a job).
Dependency states for the named steps in a workflow.
Stored results of every expansion of the named logical steps, grouped by
logical step and ordered by item_index — the fan-in list for a dynamic
fan-out.
Every expansion of a dynamically fanned-out step, in item order, with the current Oban state of each.
Every dynamically created expansion in a workflow, grouped by the logical
step it expands and ordered by item_index.
Load the node for a given Oban job id, or nil.
Bulk-insert node rows within an existing transaction.
The result this job stored on a previous attempt, for the idempotency guard.
The stored result of a specific step in a workflow.
Results for all of the given steps that have stored one, keyed by step name. Steps with no result yet are omitted.
The current Oban state of every step in a workflow.
Write this job's engine checkpoint, replacing any previous one.
Store (wrap) a step result on its node. Returns :ok or {:error, reason}.
Nodes that list completed_step as a dependency and whose job is currently
scheduled (i.e. snoozing). Returns %{id, deps} for the completion-triggered
reschedule to evaluate. Uses a native array membership test on deps.
Functions
Distinct Oban queues of completed_step's dependents that are now available.
These are exactly the jobs Baton.Reschedule just promoted when the step
completed; Baton.RescheduleReporter nudges these queues so their producers
dispatch immediately instead of waiting for the Stager. A queue that no longer
has a ready job is harmless to nudge — the woken job re-checks its deps.
This job's engine checkpoint, or {:error, :no_checkpoint}.
A missing node and a null checkpoint both mean "nothing in flight".
@spec clear_checkpoint(integer()) :: :ok
Drop this job's checkpoint. Idempotent: clearing an absent checkpoint (or a
pruned node) is :ok, since the post-condition already holds.
Step names in a workflow that count as complete for dependency promotion:
those whose Oban job is completed, plus seeded nodes (complete by
construction — they never had a job).
Dependency states for the named steps in a workflow.
LEFT JOINs to oban_jobs, so a dep whose job has been pruned comes back with
state: nil (detected as pruned by the caller) rather than vanishing. A
seeded dep also has state: nil — seeded_at is what lets the caller tell
the two apart.
Returns a list of %{name, state, attempted_at, seeded_at}.
Stored results of every expansion of the named logical steps, grouped by
logical step and ordered by item_index — the fan-in list for a dynamic
fan-out.
Expansions with no result are omitted rather than held as nil, matching
Baton.Flow.Runtime.fan_in/2: a list of results should not need
nil-guarding, and the alternative hides a missing result inside a value a
prompt renders. A logical step whose expansions have all produced nothing is
absent from the map; callers supply the empty list.
@spec expansion_states(String.t(), String.t()) :: [ %{name: String.t(), state: String.t() | nil, item_index: integer() | nil} ]
Every expansion of a dynamically fanned-out step, in item order, with the current Oban state of each.
This is what Baton.Expansion.finish/1 reads to decide the expander's own
outcome. A child whose job was pruned comes back with state: nil through
the LEFT JOIN, which counts as "did not complete" — the same reading
Baton.Check gives a pruned dep.
@spec expansions(String.t()) :: %{ required(String.t()) => [%{id: String.t(), item_index: integer() | nil}] }
Every dynamically created expansion in a workflow, grouped by the logical
step it expands and ordered by item_index.
A run's compiled_graph snapshot is immutable and was written before these
existed, so this is the other half a run view needs to render what actually
ran. Empty for a workflow with no dynamic fan-out, which is every workflow
that predates one.
@spec for_job(integer()) :: Baton.Node.t() | nil
Load the node for a given Oban job id, or nil.
@spec insert_all(module(), [map()]) :: {non_neg_integer(), nil}
Bulk-insert node rows within an existing transaction.
Takes the repo explicitly because this runs inside the same transaction as
the Oban job insert (see Baton.insert/2), where the repo is already
resolved.
The result this job stored on a previous attempt, for the idempotency guard.
Returns {:ok, value} or {:error, :no_result}. A missing node or a null
result both yield :no_result — both correctly mean "nothing cached".
The stored result of a specific step in a workflow.
Distinguishes a missing step (:not_found) from a step that exists but
hasn't stored a result yet (:no_result).
Results for all of the given steps that have stored one, keyed by step name. Steps with no result yet are omitted.
@spec step_states(String.t()) :: [ %{name: String.t(), state: String.t() | nil, seeded_at: DateTime.t() | nil} ]
The current Oban state of every step in a workflow.
LEFT JOINs to oban_jobs, so a step whose job has been pruned comes back with
state: nil. Returns a list of %{name, state, seeded_at}. Used by
Baton.Completion to decide whether a workflow has fully settled.
seeded_at is what separates the two ways a step can have no job: a seeded
step never had one and is complete by construction, while a pruned one lost
its job and can make no further progress. Both read state: nil, and
treating them alike would call every seeded workflow a failure.
Write this job's engine checkpoint, replacing any previous one.
Unlike store_result/2 the map is stored unwrapped — a checkpoint is a map
by contract, so there is no non-map case to protect against.
Store (wrap) a step result on its node. Returns :ok or {:error, reason}.
Nodes that list completed_step as a dependency and whose job is currently
scheduled (i.e. snoozing). Returns %{id, deps} for the completion-triggered
reschedule to evaluate. Uses a native array membership test on deps.