ergon (ergon v0.5.0)
View SourcePostgreSQL-native background job and workflow processing.
This is the module a host application uses. It covers the whole lifecycle: enqueue jobs, wire up workflow dependencies, ask what the graph is doing, and start workers to execute them.
{ok, Job} = ergon:enqueue(
ergon_new_job:on_queue(
ergon_new_job:new(~"send_email", #{~"to" => ~"a@b.com"}),
~"mailers")),
{ok, _Worker} = ergon:start_worker(
ergon_queue:with_concurrency(ergon_queue:new(~"mailers"), 4),
fun handle_email/1).The schema is installed by ergon_migrate, and the connection pool starts with
the application.
Where the work actually happens
Almost none of it is here. Uniqueness, retry backoff, state-transition legality,
history, and dependency blocking are all enforced by PostgreSQL. See
ergon_db, and the migrations under priv/migrations. This module and the
worker processes are the thin part.
One consequence worth stating plainly: a job with incomplete parents is not
checked out. depends_on/2 is a scheduling constraint, not an annotation. A
parent that ends failed or discarded therefore leaves its children blocked
until an operator intervenes, which is the honest behaviour for a dependency.
cancel/1 exists to tear a stuck subtree down deliberately.
Summary
Functions
Cancel Job and cascade to every descendant still running or waiting, returning
the jobs actually discarded.
Declare that Parent completing should release Child, adding a triggers
edge.
Enqueue a job and return the inserted row.
Add a labelled dependency edge to the workflow graph.
The ids of every job whose workflow parents have all completed.
Start a supervised consumer draining a pgmq queue.
Start a supervised worker that drains Queue, running Handler on each job.
Stop a consumer started by start_consumer/2, along with its pools.
Stop a worker started by start_worker/2, along with its executor pool.
The ids of the available jobs a completed Parent directly unblocks.
Types
-type db_error() :: empty_result | would_create_cycle | {job_not_found, ergon_job:job_id()} | {pgo_error, map()} | term().
-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 pgmq_handler() :: fun((pgmq_message()) -> ok | {error, binary()}).
-type pgmq_message() :: #{id := non_neg_integer(), read_ct := non_neg_integer(), message := json:decode_value(), headers := json:decode_value() | pg_null()}.
-type pgmq_queue() :: #{name := binary(), poll_interval := pos_integer(), batch_size := pos_integer(), visibility_timeout := pos_integer(), concurrency := pos_integer(), handler_timeout := timeout() | undefined, notify_channel := binary() | undefined, read_strategy := pgmq_read_strategy()}.
-type pgmq_read_strategy() :: plain | grouped | grouped_head | grouped_rr | {long_poll, MaxSeconds :: pos_integer(), IntervalMs :: pos_integer()} | {long_poll, plain | grouped | grouped_head | grouped_rr, pos_integer(), pos_integer()}.
-type queue() :: #{name := binary(), poll_interval := pos_integer(), batch_size := pos_integer(), concurrency := pos_integer(), handler_timeout := timeout()}.
-type uniqueness() :: not_unique | {unique_for, pos_integer()}.
Functions
-spec cancel(ergon_job:job_id()) -> {ok, [job()]} | {error, db_error()}.
Cancel Job and cascade to every descendant still running or waiting, returning
the jobs actually discarded.
Terminal descendants are left alone. This is the way to clear a subtree blocked behind a parent that will never complete.
-spec depends_on(ergon_job:job_id(), ergon_job:job_id()) -> ok | {error, db_error()}.
Declare that Parent completing should release Child, adding a triggers
edge.
Until Parent reaches completed, Child is withheld from checkout. Rejected
with {error, would_create_cycle} if the edge would close a loop.
Enqueue a job and return the inserted row.
Enqueuing inside your own transaction
This is an ordinary query on Ergon's pool, so calling it inside
ergon_repo:transaction/1 enlists it in that transaction: pgo binds the
transaction's connection in the process dictionary and every nested query rides
it. A job and the rows that justify it therefore commit or roll back together.
ergon_repo:transaction(fun() ->
{ok, _} = ergon_repo:query("UPDATE orders SET status = 'paid' WHERE id = $1", [OrderId]),
{ok, Job} = ergon:enqueue(ergon_new_job:new(~"send_receipt", #{~"order" => OrderId}))
end).There is no window in which the order was marked paid but the receipt job was lost, and none in which the job runs for an order whose update rolled back. This is the transactional outbox pattern without the outbox: the table, the poller and the drift reconciler that pattern needs all exist to close a gap that does not open when the queue lives in the same database as the data.
The same holds for depends_on/2 and link/3, so an entire workflow can be
declared atomically with the business change that motivates it.
-spec link(ergon_job:job_id(), ergon_job:job_id(), binary()) -> ok | {error, db_error()}.
Add a labelled dependency edge to the workflow graph.
-spec ready_children() -> {ok, [ergon_job:job_id()]} | {error, db_error()}.
The ids of every job whose workflow parents have all completed.
Observability, not scheduling: workers pick these up on their own, because checkout already excludes anything still blocked.
-spec start_consumer(pgmq_queue(), pgmq_handler()) -> {ok, pid()} | {error, term()}.
Start a supervised consumer draining a pgmq queue.
The other queue. ergon.jobs is for work with retries, dependencies and history;
pgmq is a durable message transport for streaming at volume, with at-least-once
delivery from visibility timeouts rather than state transitions. Configure it
with ergon_pgmq_queue, create the queue itself with
ergon_pgmq:create_queue/1.
{ok, _} = ergon:start_consumer(
ergon_pgmq_queue:with_concurrency(ergon_pgmq_queue:new(~"events"), 8),
fun handle_event/1).Start a supervised worker that drains Queue, running Handler on each job.
Returns the pid of the queue's supervisor, for stop_worker/1.
-spec stop_consumer(pid()) -> ok | {error, not_found}.
Stop a consumer started by start_consumer/2, along with its pools.
-spec stop_worker(pid()) -> ok | {error, not_found}.
Stop a worker started by start_worker/2, along with its executor pool.
-spec unblocked_by(ergon_job:job_id()) -> {ok, [ergon_job:job_id()]} | {error, db_error()}.
The ids of the available jobs a completed Parent directly unblocks.