RequestSeal.Replay.Postgres (RequestSeal v0.5.2)

Optional Postgrex atomic replay adapter. The caller owns connections and DDL.

Prefer RequestSeal.Replay.AshOnetime for durable replay with ash_onetime's retention guards and externally managed cleanup. This Postgrex-only adapter remains available for callers selecting its explicit table contract.

The optional dependency has runtime: false; callers explicitly start the Postgrex application and their connection/pool, then pass it to store/2 with an explicit table name. The caller executes the SQL from ddl/1 and ddl_index/1. Queries propagate the remaining operation deadline. Errors expose no SQL, connection, namespace, key, or backend exception. Conflicts never update retention and claims never evict. Required claims fail closed on outages.

For production, the caller obtains the sweep/3 time from its trusted store clock and reconciles it with verification clocks; injected historical verification clocks require matching sweep times. Multi-node operators sweep at now - max_internode_skew to preserve claims while any verifier still accepts the authenticated input. One namespace should use one freshness bound because duplicates never extend retention. PostgreSQL persistence follows the caller's database/storage configuration; this adapter does not start a database, pool, timer, or supervisor.

Summary

Functions

claim(arg1, claim, context)

Atomically insert claim unless the table already holds its namespace and key.

ddl(name)

Return the SQL the caller executes to create the replay table.

ddl_index(name)

Return the SQL the caller executes to index the replay table for sweeps.

store(conn, arg2)

Reference an existing caller connection or pool and an explicit replay table.

sweep(arg1, now, context)

Delete rows whose exclusive retention end is at or before now.

Functions

claim(arg1, claim, context)

Atomically insert claim unless the table already holds its namespace and key.

Implements RequestSeal.Replay.claim/3; most callers use RequestSeal.Replay.claim/3 with the store from store/2. The insert uses ON CONFLICT (namespace, key) DO NOTHING, so a duplicate never updates retention.

Parameters

ref - {conn, table}, the reference built by store/2.
claim - a RequestSeal.Replay.Claim.
context - a RequestSeal.Custody.Context; its remaining time is the query timeout.

Returns

:claimed - one row was inserted.
:already_claimed - the namespace and key were already stored.
{:error, :timeout} - the deadline had expired before the query; or, from 0.5.0, the query ran past the deadline (the pool closed the connection at the client timeout), the pool dropped the queued request (DBConnection :queue_timeout), or the server canceled the statement or could not take a lock (query_canceled, lock_not_available). The claim may have committed.
{:error, :unavailable} - the connection process is not running or exited; or, from 0.5.0, the connection was lost before the deadline or the server is shutting down or refusing connections (admin_shutdown, crash_shutdown, cannot_connect_now, too_many_connections).
{:error, :failure} - an invalid table name or claim, any other database error, or any other failure.

Examples

iex> context = %RequestSeal.Custody.Context{
...>   owner: self(),
...>   deadline: System.monotonic_time(:millisecond) + 1_000
...> }
iex> claim = %RequestSeal.Replay.Claim{namespace: "orders", key: "", retain_until: 100}
iex> RequestSeal.Replay.Postgres.claim({:replay_db, "replay_claims"}, claim, context)
{:error, :failure}

Fragment (requires a running PostgreSQL connection and the ddl/1 table; not executed here):

{:ok, conn} = Postgrex.start_link(database: "replay", hostname: "db.example")
store = RequestSeal.Replay.Postgres.store(conn, table: "replay_claims")
claim = %RequestSeal.Replay.Claim{namespace: "orders", key: "nonce-7f3a", retain_until: 1_900_000_000}
:claimed = RequestSeal.Replay.claim(store, claim, timeout: 2_000)
:already_claimed = RequestSeal.Replay.claim(store, claim, timeout: 2_000)

ddl(name)

@spec ddl(binary()) :: binary() | {:error, RequestSeal.Replay.Error.t()}

Return the SQL the caller executes to create the replay table.

The table has a bytea namespace/key primary key and a bigint retention end. This function only builds the statement; it does not run it.

Parameters

name - the table name, under the same rules as the :table option of store/2.

Returns

binary() - one CREATE TABLE statement with the name double-quoted.
{:error, %RequestSeal.Replay.Error{}} - reason :invalid_store for a malformed name. Meanings: RequestSeal.Replay.Error.

Examples

iex> RequestSeal.Replay.Postgres.ddl("replay_claims")
~S|CREATE TABLE "replay_claims" (namespace bytea NOT NULL, key bytea NOT NULL, retain_until bigint NOT NULL, PRIMARY KEY (namespace, key))|
iex> RequestSeal.Replay.Postgres.ddl("1_claims")
{:error, %RequestSeal.Replay.Error{reason: :invalid_store, retryable: false}}

ddl_index(name)

@spec ddl_index(binary()) :: binary() | {:error, RequestSeal.Replay.Error.t()}

Return the SQL the caller executes to index the replay table for sweeps.

The primary key from ddl/1 serves claims; sweep/3 deletes by retain_until, which without this index is a full-table scan that grows with the table. Execute this once beside ddl/1; it builds statements and runs nothing.

Parameters

name - the table name, under the same rules as the :table option of store/2, with one further bound: at most 46 bytes, so the derived <table>_retain_until_idx identifier (17 more bytes) fits PostgreSQL's 63-byte name limit without server-side truncation.

Returns

binary() - one CREATE INDEX statement with the name double-quoted.
{:error, %RequestSeal.Replay.Error{}} - reason :invalid_store for a malformed or over-long name. Meanings: RequestSeal.Replay.Error.

Examples

iex> RequestSeal.Replay.Postgres.ddl_index("replay_claims")
~S|CREATE INDEX "replay_claims_retain_until_idx" ON "replay_claims" (retain_until)|
iex> RequestSeal.Replay.Postgres.ddl_index("1_claims")
{:error, %RequestSeal.Replay.Error{reason: :invalid_store, retryable: false}}
iex> RequestSeal.Replay.Postgres.ddl_index(String.duplicate("x", 47))
{:error, %RequestSeal.Replay.Error{reason: :invalid_store, retryable: false}}

store(conn, arg2)

@spec store(term(), keyword()) ::
  RequestSeal.Replay.Store.t() | {:error, RequestSeal.Replay.Error.t()}

Reference an existing caller connection or pool and an explicit replay table.

Building the reference does not contact the database or check that the table exists.

Parameters

conn - a running Postgrex connection or pool: a pid or registered name.
opts - keyword list; see Options.

Options

:table - required, and the only accepted option: the table name. Names contain only lowercase ASCII letters, digits, and underscores, begin with a letter, and occupy 1..63 bytes.

Returns

RequestSeal.Replay.Store - adapter RequestSeal.Replay.Postgres with {conn, table} as its reference.
{:error, %RequestSeal.Replay.Error{}} - reason :invalid_store for a missing, extra or malformed option. Meanings: RequestSeal.Replay.Error.

Examples

iex> store = RequestSeal.Replay.Postgres.store(:replay_db, table: "replay_claims")
iex> store.adapter
RequestSeal.Replay.Postgres
iex> RequestSeal.Replay.Postgres.store(:replay_db, table: "Replay-Claims")
{:error, %RequestSeal.Replay.Error{reason: :invalid_store, retryable: false}}

sweep(arg1, now, context)

Delete rows whose exclusive retention end is at or before now.

Implements RequestSeal.Replay.sweep/3. The caller schedules sweeps; this adapter never sweeps on its own.

Parameters

ref - {conn, table}, the reference built by store/2.
now - Unix seconds from 0 through 253,402,300,799.
context - a RequestSeal.Custody.Context; its remaining time is the query timeout.

Returns

{:ok, removed} - the number of rows deleted.
{:error, :timeout} - the deadline had expired before the query; or, from 0.5.0, the query ran past the deadline (the pool closed the connection at the client timeout), the pool dropped the queued request (DBConnection :queue_timeout), or the server canceled the statement or could not take a lock (query_canceled, lock_not_available). The sweep's deletions were not committed when the statement was canceled or never sent, and may or may not have committed when the connection closed at the deadline; a later sweep with the same now removes whatever remains.
{:error, :unavailable} - the connection process is not running or exited; or, from 0.5.0, the connection was lost before the deadline or the server is shutting down or refusing connections (admin_shutdown, crash_shutdown, cannot_connect_now, too_many_connections).
{:error, :failure} - now is outside the range, the table name is invalid, any other database error, or any other failure.

Examples

iex> context = %RequestSeal.Custody.Context{
...>   owner: self(),
...>   deadline: System.monotonic_time(:millisecond) + 1_000
...> }
iex> RequestSeal.Replay.Postgres.sweep({:replay_db, "replay_claims"}, -1, context)
{:error, :failure}

Fragment (requires a running PostgreSQL connection and the ddl/1 table; not executed here):

context = %RequestSeal.Custody.Context{
  owner: self(),
  deadline: System.monotonic_time(:millisecond) + 5_000
}

{:ok, removed} =
  RequestSeal.Replay.Postgres.sweep({conn, "replay_claims"}, trusted_now, context)