Postgres persistence for the durable run aggregate and its operational delivery state.
A claim is the current authority to commit a run, not evidence that only one process exists. An expired holder can overlap a new holder after a steal, but only the freshly stored token can refresh, release, abandon, or satisfy the commit fence.
Claim acquisition dispatches through the ClaimPolicy resolved in the backend context. The selected engine builds and decodes one data-only admission plan; this module alone executes that plan as one PostgreSQL statement. Dispatchers call the entrypoint outside any transaction that spans vehicle execution, so no connection is held while node code runs.
Committed Docket.Run fields and backend-owned operational fields share a
row but have separate clocks. In particular, updated_at belongs to the
last committed run document; claims, refreshes, releases, and poison
transitions never rewrite it.
Claim tokens are redacted from schema inspection, and token-bearing refresh/release queries disable Ecto's ordinary SQL query log. Repo telemetry remains a trusted instrumentation boundary and may observe bind parameters, like any other database telemetry subscriber.
Run insertion, moment commit, signal mutation, and poison recovery
announce a wake due at or before the database clock with pg_notify on
the docket_wake channel, carrying the context prefix (empty string when
unprefixed) as payload. The notification runs on the write's connection
and joins any transaction the write executes in, so PostgreSQL exposes it
only after that transaction commits and drops it on rollback. Claim
release and abandonment record their wakes without a notification, so the
dispatcher's poll interval alone bounds their redispatch latency.
Summary
Functions
Hands back a pre-execution claim under a token-and-sequence fence.
Executes one admission plan from the ClaimPolicy selected by ctx.
Commits an exact-next run document under the current claim fence.
Fetches the last committed run under an explicit SQL-enforced scope.
Inserts one initialized running run under its explicit owner scope.
Fetches the committed run plus token-free backend operational state.
Lists lightweight run summaries under an explicit SQL-enforced scope.
Serializes and applies one pure run mutation without requiring a claim fence.
Refreshes the claim timestamp when claim_token is still current.
Idempotently releases the exact current token and records an immediate wake.
Clears poison from a non-terminal run and records an immediate wake.
Types
Functions
@spec abandon_claim( ctx(), :system, String.t(), Docket.Backend.RunStore.claim_token(), Docket.Backend.RunStore.abandon_policy() ) :: Docket.Backend.RunStore.abandon_result()
Hands back a pre-execution claim under a token-and-sequence fence.
One conditional UPDATE reverses the acquisition's attempt increment,
counts the abandon, and either records the policy's retry wake or — once
the abandon count reached the policy maximum — poisons the run with
reason "max_claim_abandons_exceeded". A stale token or an advanced
checkpoint sequence matches nothing and reports {:ok, :stale}.
A :non_poisoning policy never poisons; with :backoff the recorded
wake grows exponentially with the durable abandon count up to the cap.
@spec claim_due(ctx(), :system, Docket.Backend.RunStore.claim_policy()) :: {:ok, Docket.Backend.RunStore.claim_batch()} | {:error, term()}
Executes one admission plan from the ClaimPolicy selected by ctx.
Admission requires a configured root or transaction context carrying one already-resolved policy value. Bare Repo contexts remain valid for non-admission storage operations but are rejected here.
@spec commit(ctx(), Docket.Backend.scope(), Docket.Backend.RunStore.commit_proposal()) :: {:ok, Docket.Run.t()} | {:error, :stale_fence | :invalid_commit | :not_found}
Commits an exact-next run document under the current claim fence.
@spec fetch_run(ctx(), Docket.Backend.scope(), String.t()) :: {:ok, Docket.Run.t()} | {:error, :not_found}
Fetches the last committed run under an explicit SQL-enforced scope.
@spec insert_run( ctx(), Docket.Backend.owner_scope(), Docket.Run.t(), Docket.Checkpoint.type(), DateTime.t() ) :: {:ok, Docket.Run.t()} | {:error, term()}
Inserts one initialized running run under its explicit owner scope.
The graph version must already exist. Initialization is the only insert
shape: the committed run has a positive checkpoint sequence, start and
update timestamps, :run_initialized checkpoint metadata, and an explicit
first wake.
Before inserting the run, the same transaction ensures its canonical claim partition exists. A concurrently created row wins unchanged. Failed inserts and caller rollbacks remove both lifecycle writes; committed partition rows remain dormant indefinitely when their last run is later pruned.
@spec inspect_run(ctx(), Docket.Backend.scope(), String.t()) :: {:ok, Docket.RunInfo.t()} | {:error, :not_found}
Fetches the committed run plus token-free backend operational state.
@spec list_runs(ctx(), Docket.Backend.scope(), Docket.Backend.RunStore.list_query()) :: {:ok, Docket.RunPage.t()}
Lists lightweight run summaries under an explicit SQL-enforced scope.
The query selects only summary columns, reads one row beyond the requested
limit, and uses the immutable (started_at, run_id) key for stable
newest-first pagination.
@spec mutate_run( ctx(), Docket.Backend.scope(), String.t(), Docket.Backend.RunStore.mutation() ) :: Docket.Backend.RunStore.mutation_result()
Serializes and applies one pure run mutation without requiring a claim fence.
@spec refresh_claim( ctx(), :system, String.t(), Docket.Backend.RunStore.claim_token(), DateTime.t() ) :: :ok | {:error, :claim_lost}
Refreshes the claim timestamp when claim_token is still current.
Expiry is deliberately absent from the predicate. A token remains valid after its TTL until another claimant actually replaces it.
The write uses the database clock and never regresses:
GREATEST(claimed_at, CURRENT_TIMESTAMP). A refresh that was delayed in
the pool behind a :retain_claim commit therefore cannot move
claimed_at backward past the commit's fresher stamp and re-expose the
run to steal. The caller's now is validated but not written.
@spec release_claim( ctx(), :system, String.t(), Docket.Backend.RunStore.claim_token(), DateTime.t() ) :: :ok
Idempotently releases the exact current token and records an immediate wake.
A missing or stale token changes nothing. A matching release never changes
checkpoint_seq, claim_attempts, or the committed run's updated_at;
it clears claim authority and restores wake_at while retaining any
durable tenant admission.
@spec retry_poisoned_run( ctx(), Docket.Backend.scope(), String.t(), DateTime.t() ) :: {:ok, Docket.Run.t()} | {:error, :not_found | :inactive_run}
Clears poison from a non-terminal run and records an immediate wake.