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

View Source

Postgrex-backed Raxol.Workflow.Checkpoint.Saver adapter.

Persists workflow checkpoints in a PostgreSQL table so resumable runs survive BEAM restarts and can be shared across nodes. Both state and metadata are stored as bytea via :erlang.term_to_binary/1, preserving arbitrary Erlang terms.

Optional dependency

Postgrex is declared optional: true in the umbrella's mix.exs. Consumers using this saver must add :postgrex to their own deps and start a Postgrex connection (typically under their own supervision tree). The saver does not start or own the connection; it expects :conn in the configuration to point at a live process.

Config

%{
  conn: MyApp.Postgrex,       # registered name or pid
  table: "raxol_workflow_checkpoints"  # optional, default below
}

:table defaults to "raxol_workflow_checkpoints". The table name is validated against a safe identifier pattern so config-level injection is not possible.

Schema

Run the SQL returned by create_table_sql/1 once (e.g. via Ecto migration or psql) before the first run. The table has two nullable columns — interrupt_reason and paused_at — populated only when the runtime writes a pause checkpoint. A partial index over them keeps list_paused/2 queries fast even when the active-runs table is large.

iex> Raxol.Workflow.Checkpoint.Saver.Postgrex.create_table_sql()
"""
CREATE TABLE IF NOT EXISTS raxol_workflow_checkpoints (
  thread_id        text NOT NULL,
  step             integer NOT NULL,
  parent_step      integer,
  state            bytea NOT NULL,
  metadata         bytea NOT NULL,
  interrupt_reason text,
  paused_at        timestamptz,
  created_at       timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (thread_id, step)
);
CREATE INDEX IF NOT EXISTS raxol_workflow_checkpoints_paused_idx
  ON raxol_workflow_checkpoints (paused_at DESC)
  WHERE interrupt_reason IS NOT NULL;
"""

Earlier deployments can migrate with:

ALTER TABLE raxol_workflow_checkpoints
  ADD COLUMN interrupt_reason text,
  ADD COLUMN paused_at        timestamptz;
CREATE INDEX raxol_workflow_checkpoints_paused_idx
  ON raxol_workflow_checkpoints (paused_at DESC)
  WHERE interrupt_reason IS NOT NULL;

Append-only contract

put/3 is idempotent: a second write to the same (thread_id, step) pair is a no-op via ON CONFLICT DO NOTHING, matching the contract documented on Raxol.Workflow.Checkpoint.Saver.

Cross-node safety

Multiple BEAM nodes pointing at the same database can read each other's checkpoints. Combined with Compiled.resume/4, this means a run interrupted on one node can be resumed on another. The ON CONFLICT DO NOTHING clause prevents racing writers from duplicating a step.

Summary

Functions

Create the checkpoint table and its index on conn.

Returns the CREATE TABLE IF NOT EXISTS SQL for the checkpoint table. Run once as part of the consumer's migration step.

Functions

create_table!(conn, table \\ "raxol_workflow_checkpoints")

@spec create_table!(term(), String.t()) :: :ok

Create the checkpoint table and its index on conn.

Runs each statement from create_table_statements/1 separately, since Postgrex cannot prepare a multi-statement query. Use this instead of passing create_table_sql/1 to Postgrex.query/4 directly.

create_table_sql(table \\ "raxol_workflow_checkpoints")

@spec create_table_sql(String.t()) :: String.t()

Returns the CREATE TABLE IF NOT EXISTS SQL for the checkpoint table. Run once as part of the consumer's migration step.

Pass a custom table name to match the value used in the saver config; defaults to "raxol_workflow_checkpoints".

create_table_statements(table \\ "raxol_workflow_checkpoints")

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

Returns create_table_sql/1 split into individual statements.

create_table_sql/1 is a two-statement DDL script (table + index) suitable for psql or a migration file. Postgrex.query/4 uses the extended protocol, which runs one command per query, so callers driving the DDL through Postgrex must run each statement separately. This is the list to iterate over.