Low-level, process-free wrapper around the upstream Absurd PostgreSQL functions.
Every operation accepts an explicit Postgrex pool, registered name, or checked-out connection as its first argument. Passing a transaction connection therefore keeps Absurd operations in the caller's transaction; this module never checks out a second connection itself.
SQL statements are fixed and all application values use Postgrex parameters.
The module validates protocol values before issuing a query and maps database
failures to Absurd.Error.
SDK 0.2.x targets exactly schema 0.5.0:
iex> Absurd.SQL.supported_schema_version()
"0.5.0"Passing a checked-out connection keeps application data and task creation in one transaction:
Postgrex.transaction(MyApp.AbsurdDB, fn connection ->
Postgrex.query!(connection, "INSERT INTO orders (id) VALUES ($1)", [order_id])
case Absurd.SQL.spawn_task(
connection,
"workflows",
"fulfil-order",
%{"order_id" => order_id},
idempotency_key: "fulfil:#{order_id}"
) do
{:ok, spawned} -> spawned
{:error, error} -> Postgrex.rollback(connection, error)
end
end)Summary
Types
A cancellation policy whose durations are expressed in seconds.
Options forwarded to Postgrex.query/4.
A retry strategy persisted in spawn options.
Options accepted by spawn_task/5.
Functions
Atomically resolves or durably suspends an event wait.
Cancels a task without attempting to stop arbitrary external effects.
Claims up to :batch_size runs for a worker.
Runs one policy-driven cleanup batch for all queues or one selected queue.
Deletes one bounded batch of events older than :ttl milliseconds.
Deletes one bounded batch of terminal tasks older than :ttl milliseconds.
Completes a running run with a JSON result.
Creates a queue idempotently.
Drops a queue if it exists.
Emits a first-write-wins event payload on a queue.
Extends a running claim by a positive millisecond duration.
Fails a run and lets the upstream retry policy choose its next state.
Returns a queue policy, or nil when the queue does not exist.
Returns one checkpoint, or nil when it is not visible.
Returns committed checkpoints visible to the supplied run attempt.
Returns the current task snapshot, or nil when no task exists.
Lists queue names in database order.
Retries a failed task in place or creates a new logical task.
Schedules a running run at an absolute UTC time and releases its claim.
Schedules a run relative to the Absurd database clock.
Returns the schema version reported by absurd.get_schema_version/0.
Updates the official SDK queue-policy fields.
Persists a task checkpoint and optionally extends its run claim.
Spawns a task through absurd.spawn_task/4.
Returns the upstream Absurd schema version supported by this release.
Verifies that the database reports the supported Absurd schema version.
Types
@type cancellation() :: [ max_duration: non_neg_integer(), max_delay: non_neg_integer() ]
A cancellation policy whose durations are expressed in seconds.
@type query_options() :: keyword()
Options forwarded to Postgrex.query/4.
@type retry_strategy() :: [ kind: :none | :fixed | :exponential, base_seconds: number(), factor: number(), max_seconds: number() ]
A retry strategy persisted in spawn options.
@type spawn_option() :: {:max_attempts, pos_integer()} | {:retry_strategy, retry_strategy()} | {:headers, %{optional(String.t()) => Absurd.JSON.value()}} | {:cancellation, cancellation()} | {:idempotency_key, String.t()} | {:query_options, query_options()}
Options accepted by spawn_task/5.
Functions
@spec await_event( Absurd.Client.queryable(), String.t(), binary(), binary(), String.t(), String.t(), keyword() ) :: {:ok, Absurd.EventWait.t()} | {:error, Absurd.Error.t()}
Atomically resolves or durably suspends an event wait.
@spec cancel_task(Absurd.Client.queryable(), String.t(), binary(), query_options()) :: :ok | {:error, Absurd.Error.t()}
Cancels a task without attempting to stop arbitrary external effects.
@spec claim_tasks(Absurd.Client.queryable(), String.t(), String.t(), keyword()) :: {:ok, [Absurd.ClaimedTask.t()]} | {:error, Absurd.Error.t()}
Claims up to :batch_size runs for a worker.
@spec cleanup_all_queues(Absurd.Client.queryable(), String.t() | nil, query_options()) :: {:ok, [Absurd.CleanupResult.t()]} | {:error, Absurd.Error.t()}
Runs one policy-driven cleanup batch for all queues or one selected queue.
@spec cleanup_events(Absurd.Client.queryable(), String.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, Absurd.Error.t()}
Deletes one bounded batch of events older than :ttl milliseconds.
@spec cleanup_tasks(Absurd.Client.queryable(), String.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, Absurd.Error.t()}
Deletes one bounded batch of terminal tasks older than :ttl milliseconds.
@spec complete_run( Absurd.Client.queryable(), String.t(), binary(), Absurd.JSON.value(), query_options() ) :: :ok | {:error, Absurd.Error.t()}
Completes a running run with a JSON result.
@spec create_queue(Absurd.Client.queryable(), String.t(), keyword()) :: :ok | {:error, Absurd.Error.t()}
Creates a queue idempotently.
:storage_mode is :unpartitioned by default and may instead be
:partitioned. Queue policy is configured separately with
set_queue_policy/4.
@spec drop_queue(Absurd.Client.queryable(), String.t(), query_options()) :: :ok | {:error, Absurd.Error.t()}
Drops a queue if it exists.
@spec emit_event( Absurd.Client.queryable(), String.t(), String.t(), Absurd.JSON.value(), query_options() ) :: :ok | {:error, Absurd.Error.t()}
Emits a first-write-wins event payload on a queue.
@spec extend_claim( Absurd.Client.queryable(), String.t(), binary(), pos_integer(), query_options() ) :: :ok | {:error, Absurd.Error.t()}
Extends a running claim by a positive millisecond duration.
@spec fail_run( Absurd.Client.queryable(), String.t(), binary(), Absurd.JSON.value(), keyword() ) :: :ok | {:error, Absurd.Error.t()}
Fails a run and lets the upstream retry policy choose its next state.
@spec get_queue_policy(Absurd.Client.queryable(), String.t(), query_options()) :: {:ok, Absurd.QueuePolicy.t() | nil} | {:error, Absurd.Error.t()}
Returns a queue policy, or nil when the queue does not exist.
@spec get_task_checkpoint_state( Absurd.Client.queryable(), String.t(), binary(), String.t(), keyword() ) :: {:ok, Absurd.Checkpoint.t() | nil} | {:error, Absurd.Error.t()}
Returns one checkpoint, or nil when it is not visible.
@spec get_task_checkpoint_states( Absurd.Client.queryable(), String.t(), binary(), binary(), query_options() ) :: {:ok, [Absurd.Checkpoint.t()]} | {:error, Absurd.Error.t()}
Returns committed checkpoints visible to the supplied run attempt.
@spec get_task_result( Absurd.Client.queryable(), String.t(), binary(), query_options() ) :: {:ok, Absurd.TaskResult.t() | nil} | {:error, Absurd.Error.t()}
Returns the current task snapshot, or nil when no task exists.
@spec list_queues(Absurd.Client.queryable(), query_options()) :: {:ok, [String.t()]} | {:error, Absurd.Error.t()}
Lists queue names in database order.
@spec retry_task(Absurd.Client.queryable(), String.t(), binary(), keyword()) :: {:ok, Absurd.SpawnResult.t()} | {:error, Absurd.Error.t()}
Retries a failed task in place or creates a new logical task.
Supported options are :max_attempts, :spawn_new, and :query_options.
The returned Absurd.SpawnResult.created distinguishes the two modes.
@spec schedule_run( Absurd.Client.queryable(), String.t(), binary(), DateTime.t(), query_options() ) :: :ok | {:error, Absurd.Error.t()}
Schedules a running run at an absolute UTC time and releases its claim.
Prefer schedule_run_after/5 when a relative duration should use the
database clock.
@spec schedule_run_after( Absurd.Client.queryable(), String.t(), binary(), non_neg_integer(), query_options() ) :: :ok | {:error, Absurd.Error.t()}
Schedules a run relative to the Absurd database clock.
@spec schema_version(Absurd.Client.queryable(), query_options()) :: {:ok, String.t()} | {:error, Absurd.Error.t()}
Returns the schema version reported by absurd.get_schema_version/0.
@spec set_queue_policy( Absurd.Client.queryable(), String.t(), keyword(), query_options() ) :: :ok | {:error, Absurd.Error.t()}
Updates the official SDK queue-policy fields.
Interval values are PostgreSQL interval strings. Supported fields are
:partition_lookahead, :partition_lookback, :cleanup_ttl,
:cleanup_limit, :detach_mode, and :detach_min_age.
@spec set_task_checkpoint_state( Absurd.Client.queryable(), String.t(), binary(), String.t(), Absurd.JSON.value(), binary(), keyword() ) :: :ok | {:error, Absurd.Error.t()}
Persists a task checkpoint and optionally extends its run claim.
@spec spawn_task( Absurd.Client.queryable(), String.t(), String.t(), Absurd.JSON.value(), [spawn_option()] ) :: {:ok, Absurd.SpawnResult.t()} | {:error, Absurd.Error.t()}
Spawns a task through absurd.spawn_task/4.
Parameters may be any Absurd.JSON.value/0. Options are validated and
normalized to the upstream snake-case JSON object; Postgrex performs the
single JSON encoding step.
@spec supported_schema_version() :: String.t()
Returns the upstream Absurd schema version supported by this release.
@spec verify_schema_version(Absurd.Client.queryable(), query_options()) :: :ok | {:error, Absurd.Error.t()}
Verifies that the database reports the supported Absurd schema version.
A mismatch returns an Absurd.Error with kind :schema_incompatible.