Baton.Nodes (Baton v0.27.4)

Copy Markdown View Source

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

available_dependent_queues(workflow_id, completed_step)

@spec available_dependent_queues(String.t(), String.t()) :: [String.t()]

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.

checkpoint(oban_job_id)

@spec checkpoint(integer()) :: {:ok, map()} | {:error, :no_checkpoint}

This job's engine checkpoint, or {:error, :no_checkpoint}.

A missing node and a null checkpoint both mean "nothing in flight".

clear_checkpoint(oban_job_id)

@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.

completed_step_names(workflow_id)

@spec completed_step_names(String.t()) :: [String.t()]

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).

dep_states(workflow_id, dep_names)

@spec dep_states(String.t(), [String.t()]) :: [map()]

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: nilseeded_at is what lets the caller tell the two apart.

Returns a list of %{name, state, attempted_at, seeded_at}.

expansion_results(workflow_id, logical_ids)

@spec expansion_results(String.t(), [String.t()]) :: %{
  required(String.t()) => [term()]
}

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.

expansion_states(workflow_id, logical_id)

@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.

expansions(workflow_id)

@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.

for_job(oban_job_id)

@spec for_job(integer()) :: Baton.Node.t() | nil

Load the node for a given Oban job id, or nil.

insert_all(repo, rows)

@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.

own_result(oban_job_id)

@spec own_result(integer()) :: {:ok, term()} | {:error, :no_result}

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".

result_for(workflow_id, step_name)

@spec result_for(String.t(), String.t()) ::
  {:ok, term()} | {:error, :not_found | :no_result}

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(workflow_id, step_names)

@spec results_for(String.t(), [String.t()]) :: %{required(String.t()) => term()}

Results for all of the given steps that have stored one, keyed by step name. Steps with no result yet are omitted.

step_states(workflow_id)

@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.

store_checkpoint(oban_job_id, checkpoint)

@spec store_checkpoint(integer(), map()) :: :ok | {:error, term()}

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_result(oban_job_id, result)

@spec store_result(integer(), term()) :: :ok | {:error, term()}

Store (wrap) a step result on its node. Returns :ok or {:error, reason}.

waiting_dependents(workflow_id, completed_step)

@spec waiting_dependents(String.t(), String.t()) :: [
  %{id: integer(), deps: [String.t()]}
]

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.