ergon_fsm (ergon v0.5.0)

View Source

The job lifecycle state machine, as a pure function.

transition/2 computes the next state from a job and an event with no database access, which makes the transition rules exhaustively testable and keeps the authoritative definition of "what may follow what" in one place. ergon_db:apply_outcome/2 persists the result; the jobs_transition_guard trigger enforces the same rules in the database as defense in depth, for callers that bypass this module entirely.

What this module does not do

The attempt increment on checkout is not here. transition(Job, fetched) exists and is correct, but the worker path never calls it: checkout.sql marks the job executing and increments attempt in the same statement that locks it, because doing those together under FOR UPDATE SKIP LOCKED is what makes checkout atomic across concurrent workers. Routing checkout through this module would count every attempt twice.

Retry backoff is not here either. The delay is computed by ergon.retry_backoff, a database function apply_outcome.sql calls: a capped exponential in the attempt count with jitter drawn under it. This module decides only whether to retry, which is what keeps it pure. Deciding when could not be, since the answer is deliberately random.

Summary

Functions

Compute the outcome of applying Event to Job.

Types

fsm_event()

-type fsm_event() :: fetched | succeeded | {errored, binary()} | cancelled.

fsm_outcome()

-type fsm_outcome() ::
          #{state := job_state(), attempt := ergon_job:attempt(), last_error := binary() | pg_null()}.

invalid_transition()

-type invalid_transition() :: #{from := job_state(), event := fsm_event()}.

job()

-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()}.

job_state()

-type job_state() :: available | executing | completed | failed | discarded.

pg_null()

-type pg_null() :: null.

pg_timestamp()

-type pg_timestamp() :: {calendar:date(), {0..23, 0..59, number()}} | infinity | '-infinity'.

Functions

transition/2

-spec transition(job(), fsm_event()) -> {ok, fsm_outcome()} | {error, invalid_transition()}.

Compute the outcome of applying Event to Job.

The retry decision lives here: an errored job goes back to available while attempts remain, and to failed once they are exhausted.