Durable operations available to an Absurd.Task callback.
A context carries the current queue, task and run identities, attempt,
worker ID, and immutable headers. new/4 preloads the checkpoints visible to
the claimed run. A short-lived linked state process then tracks the local
checkpoint cache and deterministic occurrence names for this execution only;
PostgreSQL remains the durable authority.
Contexts are normally created and closed by Absurd.WorkerPool. A custom
executor must call close/1 after the task callback finishes.
Step callbacks use the same tagged result contract as tasks:
Absurd.Context.step(context, "fetch-profile:v1", fn ->
{:ok, %{"name" => "Ada"}}
end)Use a decomposed step when an external system needs an idempotency key before the effect happens:
with {:ok, step} <- Absurd.Context.begin_step(context, "charge:v1") do
if step.done do
{:ok, step.value}
else
key = Absurd.Context.idempotency_key(context, step)
with {:ok, charge} <- Payments.charge(amount, idempotency_key: key) do
Absurd.Context.complete_step(context, step, %{"charge_id" => charge.id})
end
end
endLong work can extend its lease, while sleeps and waits release the worker slot durably:
:ok = Absurd.Context.heartbeat(context, 60_000)
:ok = Absurd.Context.sleep_for(context, "backoff:v1", 5_000)
{:ok, payload} = Absurd.Context.await_event(context, "approved:order_123")
{:ok, child} = Absurd.Context.await_task_result(context, spawned_child)Calls that durably suspend a run unwind the task callback through a private control signal. The worker consumes that signal without completing or failing the run.
Summary
Types
A notification invoked after the database extends the active lease.
The execution context passed to a task attempt.
Functions
Returns an event payload or durably suspends until the event or timeout.
Polls a child task in another queue and checkpoints its terminal snapshot.
Allocates one deterministic checkpoint occurrence and returns its handle.
Stops the context's short-lived local state process.
Persists a value for a handle returned by begin_step/2.
Emits a first-write-wins event payload on the current queue.
Extends the current run's claim.
Derives a stable external idempotency key for a concrete step handle.
Builds an execution context from a claimed task and preloads its checkpoints.
Sleeps durably for a non-negative number of milliseconds.
Sleeps durably until an absolute UTC DateTime.
Runs a checkpointed step or replays its previously committed value.
Types
@type lease_notifier() :: (pos_integer() -> any())
A notification invoked after the database extends the active lease.
@type t() :: %Absurd.Context{ attempt: pos_integer(), claim_timeout: pos_integer(), control_ref: reference(), db: Absurd.Client.queryable(), headers: %{optional(String.t()) => Absurd.JSON.value()}, lease_notifier: lease_notifier(), query_options: keyword(), queue: String.t(), run_id: binary(), state: pid(), task_id: binary(), task_name: String.t(), worker_id: String.t() }
The execution context passed to a task attempt.
Functions
@spec await_event(t(), String.t(), keyword()) :: {:ok, Absurd.JSON.value()} | {:error, Absurd.Error.t()}
Returns an event payload or durably suspends until the event or timeout.
Options are :step_name and a millisecond :timeout (default :infinity).
The default step name is $awaitEvent:<event_name>. An elapsed timeout
returns an Absurd.Error with kind :timeout exactly once for an execution.
@spec await_task_result(t(), Absurd.SpawnResult.t() | binary(), keyword()) :: {:ok, Absurd.TaskResult.t()} | {:error, Absurd.Error.t()}
Polls a child task in another queue and checkpoints its terminal snapshot.
task may be a 16-byte task ID or an Absurd.SpawnResult. Options are
:queue, :step_name, and a millisecond :timeout. Same-queue waits are
rejected before polling because they can deadlock worker capacity. Unknown
children fail immediately.
@spec begin_step(t(), String.t()) :: {:ok, Absurd.Step.t()} | {:error, Absurd.Error.t()}
Allocates one deterministic checkpoint occurrence and returns its handle.
Repeated uses of a logical name become name, name#2, and so on. A handle
with done: true contains the committed value in value. Names that collide
with an occurrence already allocated in this execution are rejected.
@spec close(t()) :: :ok
Stops the context's short-lived local state process.
Closing a context never changes durable task state and is idempotent.
@spec complete_step(t(), Absurd.Step.t(), Absurd.JSON.value()) :: {:ok, Absurd.JSON.value()} | {:error, Absurd.Error.t()}
Persists a value for a handle returned by begin_step/2.
An already-completed handle returns its cached value without overwriting the checkpoint. New values must be JSON-compatible.
@spec emit_event(t(), String.t(), Absurd.JSON.value()) :: :ok | {:error, Absurd.Error.t()}
Emits a first-write-wins event payload on the current queue.
@spec heartbeat(t(), pos_integer() | nil) :: :ok | {:error, Absurd.Error.t()}
Extends the current run's claim.
duration is expressed in milliseconds and defaults to the original claim
timeout. Successful extension also resets the worker's local lease observer.
@spec idempotency_key(t(), Absurd.Step.t()) :: String.t()
Derives a stable external idempotency key for a concrete step handle.
The key combines the lowercase hexadecimal task UUID with the allocated checkpoint name. Calling this function does not allocate another occurrence.
@spec new(Absurd.Client.queryable(), String.t(), Absurd.ClaimedTask.t(), keyword()) :: {:ok, t()} | {:error, Absurd.Error.t()}
Builds an execution context from a claimed task and preloads its checkpoints.
Options are:
:worker_id- non-empty worker identity, defaulting to"worker";:claim_timeout- active lease in milliseconds, defaulting to120_000;:query_options- options passed to Postgrex queries;:lease_notifier- callback invoked with the effective millisecond lease after a checkpoint or heartbeat extends it.
Sub-second claim durations round up to a whole database second so the local lease observer never fires before the persisted lease.
@spec sleep_for(t(), String.t(), non_neg_integer()) :: :ok | {:error, Absurd.Error.t()}
Sleeps durably for a non-negative number of milliseconds.
The chosen absolute wake time is checkpointed under step_name. If the wake
time is still in the future, the run is scheduled using the database clock
and the worker slot is released.
@spec sleep_until(t(), String.t(), DateTime.t()) :: :ok | {:error, Absurd.Error.t()}
Sleeps durably until an absolute UTC DateTime.
Replay always uses the checkpointed wake time rather than a newly supplied value. A past wake time returns immediately.
@spec step(t(), String.t(), (-> Absurd.Task.result())) :: Absurd.Task.result()
Runs a checkpointed step or replays its previously committed value.
The zero-arity callback must return {:ok, json_value} or
{:error, reason}. Error results and raised exceptions are not checkpointed.