ergon_db (ergon v0.5.0)
View SourceLow-level database access for Ergon.
Everything that touches ergon.jobs and ergon.job_edges lives here, so the
PostgreSQL 19 temporal features it relies on are in one place. The SQL itself is
in priv/queries/ and loaded through ergon_sql. Higher-level modules
(ergon_worker, the ergon facade) call these functions and never assemble SQL
of their own.
Two paths, deliberately separate
checkout/2 and apply_outcome/2 look symmetric and are not.
checkout/2 is a plain UPDATE ... FOR UPDATE SKIP LOCKED. It marks jobs
executing and increments attempt, both inside the statement that takes
the lock, because that is what makes checkout atomic across concurrent workers.
It does not use FOR PORTION OF.
apply_outcome/2 is UPDATE ... FOR PORTION OF valid_period, which closes the
job's current validity window at now() and writes a fresh row for the new
state, preserving history instead of overwriting it. It does not increment
attempt, since it persists the count the state machine already decided, and it is
where the retry backoff is applied, in SQL.
Collapsing the two would either double-count attempts or lose the history split.
Retry backoff
The delay itself is ergon.retry_backoff, a database function, so a caller that
writes its own SQL still gets it. What lives here is only the tuning, read from
the application environment and passed as bind parameters:
{ergon, [{ergon_db, [{retry_backoff, [
{strategy, full_jitter},
{base_ms, 1000},
{cap_ms, 100000}
]}]}]}full_jitter draws uniformly from [0, ceiling], equal_jitter from
[ceiling/2, ceiling], and none is the unjittered ceiling. The formulas, and
the article they come from, are documented at the function itself.
The defaults put the first retry's ceiling at 1 s and the hard ceiling at 100 s,
which are the endpoints Ergon has always had. What changed with jitter is the
expected wait: half the ceiling rather than all of it, so a job's full retry
budget is roughly halved. A host that wants the old total back doubles base_ms.
Summary
Functions
Persist the outcome of a state transition for a single job.
Cancel JobId and cascade to every descendant still in a cancellable state.
Atomically check out up to Limit available jobs from Queue, marking them
executing and consuming an attempt.
Insert a new job and return the materialised row. Get-or-create.
Application-time time travel: the job versions whose valid_period contained
Instant, i.e. the truth about the world as of that moment.
System-time time travel: what the database believed about each job as of
Instant, spanning live rows and the archived ergon.jobs_history twin.
Record a triggers edge from ParentId to ChildId.
Record a dependency edge (parent triggers child) in the workflow graph.
Types
-type db_error() :: empty_result | would_create_cycle | {job_not_found, ergon_job:job_id()} | {pgo_error, map()} | term().
-type fsm_outcome() :: #{state := job_state(), attempt := ergon_job:attempt(), last_error := binary() | pg_null()}.
-type job() :: #{id := ergon_job:job_id(), queue := binary(), worker := binary(), payload := json:decode_value(), state := job_state(), fingerprint := binary(), attempt := ergon_job:attempt(), max_attempts := pos_integer(), last_error := binary() | pg_null(), scheduled_at := pg_timestamp(), inserted_at := pg_timestamp()}.
-type job_state() :: available | executing | completed | failed | discarded.
-type new_job() :: #{queue := binary(), worker := binary(), payload := json:encode_value(), max_attempts := pos_integer(), uniqueness := uniqueness()}.
-type pg_null() :: null.
-type pg_timestamp() :: {calendar:date(), {0..23, 0..59, number()}} | infinity | '-infinity'.
-type uniqueness() :: not_unique | {unique_for, pos_integer()}.
Functions
-spec apply_outcome(ergon_job:job_id(), fsm_outcome()) -> {ok, job()} | {error, db_error()}.
Persist the outcome of a state transition for a single job.
The update uses UPDATE ... FOR PORTION OF, so the job's current validity window
is closed at now() and a fresh row is written for the new state. The legality of
the transition is checked twice over: by ergon_fsm:transition/2 before the call,
and by the jobs_transition_guard trigger regardless of caller.
On a retry the new scheduled_at comes from ergon.retry_backoff, tuned by the
retry_backoff configuration described in the module docs.
-spec cancel_cascade(ergon_job:job_id()) -> {ok, [job()]} | {error, db_error()}.
Cancel JobId and cascade to every descendant still in a cancellable state.
Each discard is a proper valid-time transition, so history is preserved and the transition guard is satisfied. Terminal descendants are left untouched; the return value is the jobs actually discarded.
-spec checkout(binary(), pos_integer()) -> {ok, [job()]} | {error, db_error()}.
Atomically check out up to Limit available jobs from Queue, marking them
executing and consuming an attempt.
FOR UPDATE SKIP LOCKED means concurrent workers never contend for the same job.
Only live rows are considered, so a superseded version is never handed out twice.
Insert a new job and return the materialised row. Get-or-create.
Uniqueness is enforced entirely by the database. The fingerprint is a
deterministic hash of (queue, worker, payload) generated by the ergon.jobs
table itself, so it can never disagree with the columns. For a unique job the
ergon.enqueue function sets a bounded dedup_period, and a second enqueue
whose window overlaps a live copy trips the temporal uniqueness EXCLUDE, which
the function catches and answers with the existing job rather than an error. A
non-unique job gets an empty dedup_period, which never overlaps, so duplicates
always insert.
-spec jobs_asof(pg_timestamp()) -> {ok, [job()]} | {error, db_error()}.
Application-time time travel: the job versions whose valid_period contained
Instant, i.e. the truth about the world as of that moment.
-spec jobs_asof_system(pg_timestamp()) -> {ok, [job()]} | {error, db_error()}.
System-time time travel: what the database believed about each job as of
Instant, spanning live rows and the archived ergon.jobs_history twin.
-spec link(ergon_job:job_id(), ergon_job:job_id()) -> ok | {error, db_error()}.
Record a triggers edge from ParentId to ChildId.
-spec link(ergon_job:job_id(), ergon_job:job_id(), binary()) -> ok | {error, db_error()}.
Record a dependency edge (parent triggers child) in the workflow graph.
Re-adding an existing edge is a no-op. An edge that would introduce a cycle,
including a self-loop, is rejected with {error, would_create_cycle}, decided in
the database by the would_create_cycle recursive reachability query.
The check and the insert run in one transaction and behind a
transaction-scoped advisory lock. The transaction alone is not enough: under READ
COMMITTED two concurrent calls adding opposite edges each see no cycle against
their own snapshot and both commit. The lock is what makes the pair mutually
exclusive. See priv/queries/graph/lock_edges.sql.
Returning {error, _} from inside the transaction commits rather than rolls
back, which is correct here: the rejection path has written nothing.