The Ecto StatifierPersistence.Storage.Adapter: the storage contract
over the schemas a host generates with use StatifierPersistence.Ecto
(ADR-0002), against the tables the versioned migrations helper
creates. Requires the optional ecto_sql dependency (ADR-0005).
defmodule MyApp.Persistence do
use StatifierPersistence.Ecto, repo: MyApp.Repo
end
{:ok, store} =
StatifierPersistence.Storage.new(
StatifierPersistence.Storage.Ecto,
persistence: MyApp.Persistence
)Options init/1 accepts:
:persistence- required, a module that calleduse StatifierPersistence.Ecto. The repo, the schema modules, and the table names all come from its resolved configuration, so this adapter adds no knobs of its own (ADR-0002 decision 3).:input_log_cap- the per-execution cap on ADR-0010's input log::infinity(the default, no cap) or a positive integer. It counts entries, and past itappend_input/3refuses and closes the execution's log with a marker (decision 6). It has no bounded default, because one would be this package silently truncating a host's log.:sandbox- whentrue,isolate/1checks out anEcto.Adapters.SQL.Sandboxconnection: the hook a test suite (this package's conformance suite included) uses to wrap each test in its own transaction. Defaultfalse, andisolate/1is then a no-op.
Engine identities (content_hash, session_id, execution_id) are
stored verbatim in text columns and blobs in bytea columns, so
both round-trip byte-identically (ADR-0002 decision 1, ADR-0003
decision 1). The identity guard lives in
StatifierPersistence.Storage, above this adapter like above every
other one (ADR-0003 decision 2); nothing here decodes a blob.
insert_execution/2's :execution_exists refusal rides the V01 unique index on
execution_id - one atomic insert, never a check-then-insert. A backend
failure a callback cannot observe as a value (the database down, a
timeout) raises the driver's own exception rather than being
flattened into a default (this package's errors-are-events rule).
Summary
Functions
Appends one input at the execution's next ordinal (the optional
StatifierPersistence.Storage.Adapter.append_input/3).
Fetches the chart stored under content_hash, or :chart_not_found.
Fetches the execution stored under execution_id, or :execution_not_found.
Fetches the position stored for session_id, or
:position_not_found.
Resolves the :persistence host module into the handle every other
callback takes: the host's repo, its three generated schema modules,
and its executions table name (for the unique-constraint mapping).
Inserts execution_record, refusing a duplicate execution_id with
{:error, :execution_exists}.
Per-test isolation (the optional
StatifierPersistence.Storage.Adapter.isolate/1): checks out an
Ecto.Adapters.SQL.Sandbox connection when this handle was built
with sandbox: true, and is a no-op otherwise.
The indexed status projection over a metadata match (the optional
StatifierPersistence.Storage.Adapter.list_execution_states_by_metadata/2).
Lists the executions whose stored metadata contains every key/value
pair in metadata (ADR-0006 decision 3's equality-match list helper).
Lists an execution's whole log in ascending seq (the optional
StatifierPersistence.Storage.Adapter.list_inputs/2), or
:execution_not_found for an execution this adapter does not hold.
Executions fun under per-execution mutual exclusion for execution_id (the optional
StatifierPersistence.Storage.Adapter.lock_execution/3, ADR-0004
decision 5 as amended 2026-08-22).
Stores chart_record, idempotent on its content_hash: an insert
with on_conflict: :nothing against the unique index, so a repeated
save of the same hash neither duplicates the row nor rewrites it.
Stores position_record under its session_id, overwriting any
position already stored for that session: an upsert replacing the
record columns (and updated_at) on the unique index.
Declares outcome support (the optional
StatifierPersistence.Storage.Adapter.supports_execution_outcome?/1):
this adapter stores an execution's answer in the V03 outcome_blob column,
under the configured :blob_type like every other blob column.
Declares input log support (the optional
StatifierPersistence.Storage.Adapter.supports_input_log?/1): this
adapter keeps ADR-0010's log in the V05 inputs table, on every Ecto
backend.
Declares metadata support (the optional
StatifierPersistence.Storage.Adapter.supports_metadata?/1): this
adapter stores an execution's metadata in the V02 jsonb column (ADR-0006
decision 3), on a Postgres repo.
Overwrites the execution stored under execution_record's execution_id with the
full record, or refuses with :execution_not_found.
Functions
@spec append_input( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.execution_id(), StatifierPersistence.Storage.Adapter.input_record() ) :: {:ok, StatifierPersistence.Storage.Adapter.seq()} | {:error, StatifierPersistence.Storage.Adapter.error()}
Appends one input at the execution's next ordinal (the optional
StatifierPersistence.Storage.Adapter.append_input/3).
The ordinal is the execution's current maximum plus one, read and written
inside the exclusion the caller already holds
(StatifierPersistence.Executions' serialized unit). The V05 unique index
on (execution_id, seq), not the read, is what makes denseness true: a
lost race fails the write with {:adapter, :seq_conflict} rather
than duplicating an ordinal.
Past the configured input_log_cap: the log closes itself. The cap's
last slot is written as a row with a nil input_blob - decision
6's marker - and this call and every later one for that execution return
{:error, :input_log_full}. The step that produced the input is not
failed by it: the log records its own truncation instead
(ADR-0010 decision 5).
@spec fetch_chart( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.content_hash() ) :: {:ok, StatifierPersistence.Storage.Adapter.chart_record()} | {:error, StatifierPersistence.Storage.Adapter.error()}
Fetches the chart stored under content_hash, or :chart_not_found.
@spec fetch_execution( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.execution_id() ) :: {:ok, StatifierPersistence.Storage.Adapter.execution_record()} | {:error, StatifierPersistence.Storage.Adapter.error()}
Fetches the execution stored under execution_id, or :execution_not_found.
@spec fetch_position( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.session_id() ) :: {:ok, StatifierPersistence.Storage.Adapter.position_record()} | {:error, StatifierPersistence.Storage.Adapter.error()}
Fetches the position stored for session_id, or
:position_not_found.
@spec init(StatifierPersistence.Storage.Adapter.opts()) :: {:ok, StatifierPersistence.Storage.Adapter.opts()} | {:error, StatifierPersistence.Storage.Adapter.error()}
Resolves the :persistence host module into the handle every other
callback takes: the host's repo, its three generated schema modules,
and its executions table name (for the unique-constraint mapping).
Refuses a module that never called use StatifierPersistence.Ecto
with {:error, {:adapter, {:not_a_persistence_host, module}}}.
Makes no database call: reachability surfaces on first use, per
call site.
@spec insert_execution( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.execution_record() ) :: :ok | {:error, StatifierPersistence.Storage.Adapter.error()}
Inserts execution_record, refusing a duplicate execution_id with
{:error, :execution_exists}.
The refusal is the V01 unique index on execution_id speaking: the insert
carries a unique_constraint/3 on that index's name, so two
concurrent inserts of one execution_id cannot both return :ok and no
separate existence check ever executions.
metadata is stored in the V02 jsonb column, NULL for the empty
map. jsonb holds only JSON-representable values, which makes term
narrower here than in Elixir (ADR-0006 decision 3): a tuple, a pid, a
reference, an atom, a struct, or a binary that is not valid UTF-8 has
no jsonb form. This adapter refuses such a map at open with
{:error, :metadata_unsupported} - the failure shape ADR-0006
decision 3 leaves to the implementation - rather than letting the
encoder raise from inside a transaction or, worse, storing something
that is not what the caller handed over. Refusing at open is the
principle the decision already sets for an adapter that cannot store a
map; a value it cannot store is the same answer at a finer grain.
@spec isolate(StatifierPersistence.Storage.Adapter.opts()) :: :ok | {:error, StatifierPersistence.Storage.Adapter.error()}
Per-test isolation (the optional
StatifierPersistence.Storage.Adapter.isolate/1): checks out an
Ecto.Adapters.SQL.Sandbox connection when this handle was built
with sandbox: true, and is a no-op otherwise.
@spec list_execution_states_by_metadata( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.metadata() ) :: {:ok, [StatifierPersistence.Storage.Adapter.execution_state()]} | {:error, StatifierPersistence.Storage.Adapter.error()}
The indexed status projection over a metadata match (the optional
StatifierPersistence.Storage.Adapter.list_execution_states_by_metadata/2).
The same jsonb containment predicate list_executions_by_metadata/2
issues - which V03's GIN jsonb_path_ops index serves - with a
three-column select: in place of the whole row. No blob column is
read, which is the point: a fan-out of N children asks this question N
times, and the listing would move N identity and position blobs each
time.
child_index is extracted from this package's own reserved linkage
namespace inside metadata, and is nil for a matched execution carrying
no linkage.
Takes the same non-empty string-keyed map, with the same
ArgumentError for anything else.
Off Postgres this refuses with {:error, :metadata_unsupported}
rather than issuing SQL the backend cannot parse - the same answer
supports_metadata?/1 already gives the facade, given directly to a
caller who reached the callback itself.
@spec list_executions_by_metadata( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.metadata() ) :: {:ok, [StatifierPersistence.Storage.Adapter.execution_record()]} | {:error, StatifierPersistence.Storage.Adapter.error()}
Lists the executions whose stored metadata contains every key/value
pair in metadata (ADR-0006 decision 3's equality-match list helper).
Equality match on all pairs is the whole query surface: no ranges, no partial matches, no containment operators exposed to the caller, and no ordering guarantee. A host needing more than that queries its own column directly - ADR-0002's configurable table names already make that a supported thing to do.
The query is one jsonb containment predicate, which a GIN index on
the column serves directly; V02 ships no index, because which pairs a
host queries by is the host's call (ADR-0006 decision 4).
metadata must be a non-empty map of string keys: a zero-pair
"contains every given pair" matches every execution with any metadata at
all, which is a caller bug far more often than a request, so it
raises ArgumentError rather than answering it.
StatifierPersistence.Storage.Ecto.list_executions_by_metadata(
store.opts,
%{"tenant_id" => "acct_01H8X"}
)Returns records in fetch_execution/2's shape.
Off Postgres this refuses with {:error, :metadata_unsupported}
rather than issuing SQL the backend cannot parse - the same answer
supports_metadata?/1 already gives the facade, given directly to a
caller who reached the callback itself.
@spec list_inputs( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.execution_id() ) :: {:ok, [StatifierPersistence.Storage.Adapter.input_record()]} | {:error, StatifierPersistence.Storage.Adapter.error()}
Lists an execution's whole log in ascending seq (the optional
StatifierPersistence.Storage.Adapter.list_inputs/2), or
:execution_not_found for an execution this adapter does not hold.
One index-ordered read of the V05 unique index. No filter, no range, no limit: the whole log is what a replay consumes and the cap is what bounds it (ADR-0010 decision 2).
@spec lock_execution( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.execution_id(), (-> result) ) :: {:ok, result} | {:error, StatifierPersistence.Storage.Adapter.error()} when result: term()
Executions fun under per-execution mutual exclusion for execution_id (the optional
StatifierPersistence.Storage.Adapter.lock_execution/3, ADR-0004
decision 5 as amended 2026-08-22).
Everything happens inside one transaction that spans fun. It takes
pg_advisory_xact_lock(hashtextextended(execution_id, 0)) first -
unconditional per-execution exclusion whether or not the execution row exists
yet - and then SELECT ... FOR UPDATE on the execution row when it does,
keeping the row itself locked against every other writer for the
rest of the transaction. Both locks are transaction-scoped, so any
exit from fun releases them: a normal return commits, and a raise
rolls back and propagates to the caller with nothing leaked.
@spec save_chart( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.chart_record() ) :: :ok | {:error, StatifierPersistence.Storage.Adapter.error()}
Stores chart_record, idempotent on its content_hash: an insert
with on_conflict: :nothing against the unique index, so a repeated
save of the same hash neither duplicates the row nor rewrites it.
@spec save_position( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.position_record() ) :: :ok | {:error, StatifierPersistence.Storage.Adapter.error()}
Stores position_record under its session_id, overwriting any
position already stored for that session: an upsert replacing the
record columns (and updated_at) on the unique index.
@spec supports_execution_outcome?(StatifierPersistence.Storage.Adapter.opts()) :: boolean()
Declares outcome support (the optional
StatifierPersistence.Storage.Adapter.supports_execution_outcome?/1):
this adapter stores an execution's answer in the V03 outcome_blob column,
under the configured :blob_type like every other blob column.
@spec supports_input_log?(StatifierPersistence.Storage.Adapter.opts()) :: boolean()
Declares input log support (the optional
StatifierPersistence.Storage.Adapter.supports_input_log?/1): this
adapter keeps ADR-0010's log in the V05 inputs table, on every Ecto
backend.
Unconditional, unlike supports_metadata?/1, which answers false
off Postgres because the containment SQL its listings issue does not
parse there. Nothing in the input log needs a Postgres-only feature -
a table, four columns and a unique index (ADR-0010 decision 9) - so
there is nothing here for a backend to decline.
@spec supports_metadata?(StatifierPersistence.Storage.Adapter.opts()) :: boolean()
Declares metadata support (the optional
StatifierPersistence.Storage.Adapter.supports_metadata?/1): this
adapter stores an execution's metadata in the V02 jsonb column (ADR-0006
decision 3), on a Postgres repo.
On any other Ecto adapter this answers false, which is ADR-0006
decision 3's refusal-at-open arm rather than a new one. The capability
that record defines is the column and the equality-match list
helper, and the helper is Postgres-only SQL: both
list_executions_by_metadata/2 and list_execution_states_by_metadata/2 are
jsonb containment (@>) with a -> ... ->> extraction, which a
non-Postgres backend does not parse. Declaring the capability true
there would strand a durable subchart or a fan-out at the far end of
a listing that cannot answer, with children already started that
nothing could then settle (sp-11w).
The two listings consult this answer themselves, so a caller holding
the raw callback gets the same {:error, :metadata_unsupported} the
facade gives rather than a raise from the driver (sp-4eo). V03's
metadata index is skipped on the same adapters, for the same
reason; sp-5lm tracks this surface.
@spec update_execution( StatifierPersistence.Storage.Adapter.opts(), StatifierPersistence.Storage.Adapter.execution_record() ) :: :ok | {:error, StatifierPersistence.Storage.Adapter.error()}
Overwrites the execution stored under execution_record's execution_id with the
full record, or refuses with :execution_not_found.
One update_all/3 keyed on execution_id: the match count is the
existence check, so refusal and overwrite are a single statement.
metadata is not in the set: list, and that is the documented
exception to the full overwrite: the map is write-once (ADR-0006
decision 1 grants it at create and grants no way to change it), so the
stored column is left exactly as insert_execution/2 wrote it and the given
record's metadata is ignored.
outcome_blob is the second exception, and it joins the set: list
only when the given record carries one: a nil leaves the stored
column alone, so an ordinary step of an execution that has already answered
does not erase its answer.