Raxol.Workflow.Checkpoint.Saver behaviour (Raxol v2.6.0)

Behaviour for Raxol.Workflow.Checkpoint persistence.

Implementations decide the storage medium (ETS, DETS, Postgrex, etc.). Three adapters ship with raxol:

Raxol.Workflow.Checkpoint.Saver.Ets -- in-process, named ETS table. Default for tests and short-lived runs.
Raxol.Workflow.Checkpoint.Saver.Dets -- file-backed GenServer. Default for runs that should survive BEAM restarts.
Raxol.Workflow.Checkpoint.Saver.Postgrex -- Postgres-backed via the optional :postgrex dependency. Default for runs that need to be visible across BEAM nodes.

Append-only contract

put/3 must be a no-op (return :ok) when called with a (thread_id, step) pair that already exists. Saver implementations do not return errors for duplicate writes; that lets the runtime retry a checkpoint write idempotently without special-casing.

Configuration

Implementations receive a per-call config map provided by the caller. The caller is responsible for ensuring the config is consistent across calls for the same thread; the behaviour does not enforce this.

When a saver is wired into a compiled graph, the configuration is carried alongside the module:

Graph.compile(graph, saver: {Saver.Ets, %{table: :my_table}})
Graph.compile(graph, saver: {Saver.Dets, %{name: MyDets}})

The runtime destructures {module, config} and forwards config to every behaviour call. A bare module (no tuple) gets %{}.

Summary

Types

config()

Per-call configuration map passed to every callback.

paused()

Structured row returned by list_paused/2. The state is the workflow state at the moment of the pause; metadata is the full pause-checkpoint metadata including node_id, interrupt_reason, and paused_at.

thread_id()

Workflow run identifier; same shape as Checkpoint.thread_id.

Callbacks

delete_thread(config, thread_id)

Remove all checkpoints for the thread. Returns :ok even if the thread was unknown.

get_latest(config, thread_id)

Return the most-recently-written checkpoint for the thread, by step number. Returns {:error, :not_found} if the thread has no checkpoints.

list(config, thread_id, limit)

List checkpoints for the thread in newest-first order, up to limit. Returns {:ok, []} when the thread is unknown.

list_paused(config, limit)

Return up to limit paused threads, newest-paused-first.

put(config, thread_id, t)

Append a checkpoint to the thread. Must be idempotent: writing a checkpoint with a (thread_id, step) pair that already exists is a no-op and returns :ok.

Functions

accumulate_latest_per_thread(arg, acc)

Fold accumulator for ETS/DETS list_paused/2 implementations. Receives a single {{thread_id, step}, checkpoint} tuple plus the accumulator map and retains the highest-step checkpoint per thread_id. Shared so both adapter folds use the same comparison semantics.

list_paused(arg, limit)

Dispatch list_paused to the adapter, falling back to a generic per-thread scan for adapters that haven't implemented the optional callback.

normalize(module)

Normalize the saver opt into {module, config} form.

paused_rows_from_latest(latest_per_thread, limit)

Common post-fold pipeline for ETS and DETS adapters: a map of thread_id -> {step, checkpoint} (the highest-step checkpoint per thread) is filtered to pause checkpoints, sorted newest-paused-first, truncated to limit, and converted to paused() rows.

to_paused_row(checkpoint)

Convert a pause checkpoint (one whose metadata carries :interrupt_reason) into the structured paused() row returned by list_paused/2. Useful for adapters whose underlying scan produces Checkpoint.t() values.

Types

config()

@type config() :: map()

Per-call configuration map passed to every callback.

paused()

@type paused() :: %{
  thread_id: thread_id(),
  interrupt_reason: any(),
  paused_at: DateTime.t() | nil,
  state: any(),
  metadata: map()
}

thread_id()

@type thread_id() :: Raxol.Workflow.Checkpoint.thread_id()

Workflow run identifier; same shape as Checkpoint.thread_id.

Callbacks

delete_thread(config, thread_id)

@callback delete_thread(config(), thread_id()) :: :ok | {:error, term()}

Remove all checkpoints for the thread. Returns :ok even if the thread was unknown.

get_latest(config, thread_id)

@callback get_latest(config(), thread_id()) ::
  {:ok, Raxol.Workflow.Checkpoint.t()} | {:error, :not_found}

Return the most-recently-written checkpoint for the thread, by step number. Returns {:error, :not_found} if the thread has no checkpoints.

list(config, thread_id, limit)

@callback list(config(), thread_id(), limit :: pos_integer()) ::
  {:ok, [Raxol.Workflow.Checkpoint.t()]}

List checkpoints for the thread in newest-first order, up to limit. Returns {:ok, []} when the thread is unknown.

list_paused(config, limit)

(optional)

@callback list_paused(config(), limit :: pos_integer()) :: {:ok, [paused()]}

Return up to limit paused threads, newest-paused-first.

A thread is "paused" when its latest checkpoint carries :interrupt_reason in metadata. Resuming a paused thread writes a follow-up checkpoint with no :interrupt_reason, which implicitly removes the thread from this query's result set.

Optional callback: adapters that do not implement it inherit a default that delegates to list/3 per known thread (a slower path).

put(config, thread_id, t)

@callback put(config(), thread_id(), Raxol.Workflow.Checkpoint.t()) ::
  :ok | {:error, term()}

Append a checkpoint to the thread. Must be idempotent: writing a checkpoint with a (thread_id, step) pair that already exists is a no-op and returns :ok.

Functions

accumulate_latest_per_thread(arg, acc)

@spec accumulate_latest_per_thread(
  {{Raxol.Workflow.Checkpoint.thread_id(), non_neg_integer()},
   Raxol.Workflow.Checkpoint.t()},
  %{
    required(Raxol.Workflow.Checkpoint.thread_id()) =>
      {non_neg_integer(), Raxol.Workflow.Checkpoint.t()}
  }
) :: %{
  required(Raxol.Workflow.Checkpoint.thread_id()) =>
    {non_neg_integer(), Raxol.Workflow.Checkpoint.t()}
}

list_paused(arg, limit)

@spec list_paused(
  {module(), config()},
  pos_integer()
) :: {:ok, [paused()]}

Dispatch list_paused to the adapter, falling back to a generic per-thread scan for adapters that haven't implemented the optional callback.

Always returns {:ok, [paused()]}. The fallback can't enumerate threads, so for adapters without list_paused/2 it returns {:ok, []}. Bespoke adapters with their own thread index should implement the callback directly.

normalize(module)

@spec normalize(module() | {module(), config()} | nil) :: {module(), config()} | nil

Normalize the saver opt into {module, config} form.

Accepts a bare module, a {module, config} tuple, or nil. Returns nil for nil input, propagating the "no saver configured" case through the runtime without special branches at every call site.

paused_rows_from_latest(latest_per_thread, limit)

@spec paused_rows_from_latest(
  %{
    required(Raxol.Workflow.Checkpoint.thread_id()) =>
      {non_neg_integer(), Raxol.Workflow.Checkpoint.t()}
  },
  pos_integer()
) :: [paused()]

Adapter list_paused/2 implementations build the map via their storage-specific fold (:ets.foldl/3 or :dets.foldl/3) and hand the result to this helper so the rest of the pipeline stays in one place.

to_paused_row(checkpoint)

@spec to_paused_row(Raxol.Workflow.Checkpoint.t()) :: paused()