Baton.ResultStore behaviour (Baton v0.27.4)

Tiered storage for step results, with a pluggable large-payload backend.

Baton.Results calls into this module so the size-tiering is transparent to workers: small results stay inline on the workflow_nodes row exactly as before; large results are gzipped and handed to a backend, leaving only a small reference envelope inline. Downstream reads resolve the envelope back to the original value automatically.

Tiering

Encode (encode/2) — JSON-encode the result. If it is at or below Baton.Config.inline_threshold_bytes/0, return it unchanged for inline storage. Otherwise gzip it, put/3 it to the backend, and return a reference envelope. Results whose encoded size exceeds Baton.Config.max_result_bytes/0 are rejected with {:error, :result_too_large}.
Decode (decode/1, decode_all/1) — pass plain values through; for a reference envelope, fetch the bytes from the backend, verify the SHA-256, gunzip, and JSON-decode.

Backend behaviour

A backend stores opaque key => bytes pairs. The default is Baton.ResultStore.Postgres; configure another with config :baton, :result_store, MyApp.S3Store.

@callback put(key, bytes, opts) :: :ok | {:error, term}
@callback get(key) :: {:ok, binary} | {:error, term}
@callback delete(keys) :: :ok | {:error, term}
@callback get_many(keys) :: {:ok, %{key => binary}} | {:error, term}  # optional

opts passed to put/3 carry :workflow_id, :step_name, :oban_job_id, :codec, :byte_size, :sha256, and :content_type so a backend can index and later reclaim its own storage.

Switching backends

A reference resolves against the currently configured backend. Change the backend only after every in-flight workflow that wrote large results has settled, or old references will not resolve.

Summary

Types

key()

opts()

Callbacks

delete(list)

get(key)

get_many(list)

put(key, binary, opts)

Functions

backend()

The configured backend module.

decode(value)

Resolve a stored inline value back to the original term.

decode_all(map)

Resolve a name => stored_value map, batching backend fetches for any references. Returns {:ok, %{name => term}} or {:error, reason}.

encode(job, result)

Prepare a result for storage on the node row.

key_of(arg1)

Extract the backend key from a reference envelope, or nil.

reference?(arg1)

True if a stored inline value is a backend reference envelope.

Types

key()

@type key() :: String.t()

opts()

@type opts() :: keyword()

Callbacks

delete(list)

@callback delete([key()]) :: :ok | {:error, term()}

get(key)

@callback get(key()) :: {:ok, binary()} | {:error, term()}

get_many(list)

(optional)

@callback get_many([key()]) :: {:ok, %{required(key()) => binary()}} | {:error, term()}

put(key, binary, opts)

@callback put(key(), binary(), opts()) :: :ok | {:error, term()}

Functions

backend()

@spec backend() :: module()

The configured backend module.

decode(value)

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

Resolve a stored inline value back to the original term.

Plain values pass through; a reference envelope is fetched from the backend, verified, gunzipped, and decoded. Returns {:ok, term} or {:error, reason}.

decode_all(map)

@spec decode_all(%{optional(String.t()) => term()}) ::
  {:ok, %{optional(String.t()) => term()}} | {:error, term()}

Resolve a name => stored_value map, batching backend fetches for any references. Returns {:ok, %{name => term}} or {:error, reason}.

encode(job, result)

@spec encode(Oban.Job.t(), term()) :: {:ok, term()} | {:error, term()}

Prepare a result for storage on the node row.

Returns {:ok, inline_term} — either the original result (small) or a reference envelope (large, after writing the payload to the backend) — or {:error, :result_too_large} / a backend error.

key_of(arg1)

@spec key_of(term()) :: key() | nil

Extract the backend key from a reference envelope, or nil.

reference?(arg1)

@spec reference?(term()) :: boolean()

True if a stored inline value is a backend reference envelope.