ergon_job_runner (ergon v0.5.0)
View SourceExecutes one job: runs the host's handler, then persists the outcome.
worker_pool starts concurrency of these per queue and ergon_worker casts
each checked-out job to the pool. The pool is used purely as a supervised,
bounded set of executors. None of its worker-selection strategies apply here,
because FOR UPDATE SKIP LOCKED in checkout already decided which job goes to
which node and process. The database is the dispatcher.
The handler runs in a child process
A handler is host code and may do anything: return garbage, raise, exit, or hang forever. All four have to cost one job rather than an executor, so the handler runs in a spawned child and this process waits on it with a deadline.
That the child is spawned with spawn_request/2 rather than spawn_monitor/1
is deliberate. For a local spawn the two are otherwise equivalent, but this
process is a gen_server that worker_pool also manages, so its mailbox carries
traffic that is not ours. {reply_tag, ergon_spawn} and {monitor, [{tag, ergon_down}]} keep Ergon's two possible replies unmistakable instead of
competing for the generic 'DOWN' tag, and the ReqId doubles as the monitor
reference, so one term covers both "the spawn failed" and "the process died".
{reply, error_only} drops the success reply, leaving exactly one message on the
happy path.
A timeout is an outcome, not an abandonment
On deadline the child is killed and the job is recorded as {errored, <<"handler timeout">>}, which consumes an attempt and lets the database's
jittered backoff reschedule it. This is why max_overrun_warnings is left at
infinity in the pool configuration: wpool's own overrun enforcement kills the
executor, which would abandon the job mid-flight with no outcome written,
leaving the row executing until the reconciler reclaims it. The deadline here
kills the handler and still writes the outcome.
The in-flight counter
The counter is decremented in an after clause, so a slot is returned even if
persisting the outcome fails. Leaking slots would starve the queue permanently:
ergon_worker sizes each checkout against free capacity, so a counter that only
ever rises eventually stops all fetching.
Summary
Types
-type args() :: #{queue := binary(), handler := handler(), handler_timeout := timeout(), in_flight := counters:counters_ref()}.
-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 pg_null() :: null.
-type pg_timestamp() :: {calendar:date(), {0..23, 0..59, number()}} | infinity | '-infinity'.