Pass structured results between workflow steps.
Small results are stored inline on the step's workflow_nodes row — never in
oban_jobs.meta. Because the engine owns that table, writing a result never
contends with Oban's own writes (the data race the old meta-based approach
had).
Large results (above Baton.Config.inline_threshold_bytes/0) are transparently
spilled to Baton.ResultStore — gzipped and handed to a pluggable backend
(Postgres by default) — leaving only a small reference on the node row. This
keeps workflow_nodes small for the hot dependency-gating queries. Reads
resolve the reference automatically, so the public API below is unchanged and
callers never see a reference.
Storing a result (upstream worker)
Return {:ok, result} from perform_workflow/1; Baton.Worker
persists it automatically.
def perform_workflow(%Oban.Job{args: %{"api_key" => key}}) do
{:ok, %{token: fetch_token(key)}}
endReading a result (downstream worker)
case Results.get_result(job, :fetch_token) do
{:ok, %{"token" => token}} -> ...
{:error, reason} -> {:error, reason}
endReading all upstream results
results = Results.get_all_results(job)
# => %{"fetch_token" => %{"token" => "..."}, ...}Summary
Functions
Drop this job's checkpoint — the in-flight work it referred to is finished or abandoned. Idempotent.
Results from all of this job's dependencies that have stored one, keyed by step name. Steps still running, or with no result, are omitted.
The checkpoint this job stored on an earlier attempt, or
{:error, :no_checkpoint} when there is none (including for jobs with no
workflow node).
Fan-in for dynamically fanned-out steps: %{logical_step => [result, ...]},
one entry per named step, ordered by expansion.
The result this job stored on a previous attempt — the idempotency guard.
Fetch the stored result of a specific upstream step.
Record engine scratch for a step whose work spans several attempts.
Persist a result for the current job. Called automatically by the worker on
{:ok, result}. Returns :ok or {:error, reason}.
Functions
@spec clear_checkpoint(Oban.Job.t()) :: :ok
Drop this job's checkpoint — the in-flight work it referred to is finished or abandoned. Idempotent.
@spec get_all_results(Oban.Job.t()) :: %{required(String.t()) => term()}
Results from all of this job's dependencies that have stored one, keyed by step name. Steps still running, or with no result, are omitted.
@spec get_checkpoint(Oban.Job.t()) :: {:ok, map()} | {:error, :no_checkpoint}
The checkpoint this job stored on an earlier attempt, or
{:error, :no_checkpoint} when there is none (including for jobs with no
workflow node).
Fan-in for dynamically fanned-out steps: %{logical_step => [result, ...]},
one entry per named step, ordered by expansion.
The list-shaped counterpart to get_all_results/1. A static fan-out's
grouping is stamped on the reading job at compile time
(Baton.Flow.Runtime.fan_in/2); a dynamic one is only knowable from the
rows the expansion created, which is what this reads. A step whose
expansions produced nothing yields [] rather than being omitted, so a
reader always finds its dependency.
@spec get_own_result(Oban.Job.t()) :: {:ok, term()} | {:error, :no_result}
The result this job stored on a previous attempt — the idempotency guard.
Returns {:ok, value} if a result was stored, {:error, :no_result}
otherwise. Jobs without a workflow node return {:error, :no_result}.
@spec get_result(Oban.Job.t(), atom() | String.t()) :: {:ok, term()} | {:error, :no_workflow | :not_a_dep | :not_found | :no_result}
Fetch the stored result of a specific upstream step.
Errors:
:no_workflow— this job has no workflow node:not_a_dep— the named step isn't a declared dependency of this job:not_found— no such step in this workflow:no_result— the step exists but hasn't stored a result yet
@spec store_checkpoint(Oban.Job.t(), map()) :: :ok | {:error, term()}
Record engine scratch for a step whose work spans several attempts.
A checkpoint is not a result. A stored result means the step finished — the
idempotency guard in Baton.Worker completes any job that has one, and
dependents read it. A checkpoint means the opposite: work is still in flight,
and here is the handle needed to pick it back up. Nothing downstream can see
it.
Baton.LLMStep's batch mode uses it to carry a provider batch id across
snoozes:
%{"batch" => %{"id" => "msgbatch_abc", "custom_id" => "job-4171",
"submitted_at" => "2026-08-07T02:00:11Z"}}The map must survive a JSON round-trip; it comes back with string keys. Writing replaces any previous checkpoint for the job.
@spec store_result(Oban.Job.t(), term()) :: :ok | {:error, term()}
Persist a result for the current job. Called automatically by the worker on
{:ok, result}. Returns :ok or {:error, reason}.
Large results are spilled to Baton.ResultStore first; only a small reference
is written to the node row. {:error, :result_too_large} is returned if the
encoded result exceeds Baton.Config.max_result_bytes/0.