PhoenixKit.Jobs.StateMachine (phoenix_kit v2.55.1)

Copy Markdown View Source

The transitions of a job run, as pure functions (no database, no clock, no Oban): transition/3 takes a run, an event and the time, and answers the fields to change and the effects to carry out, or why the event does not apply.

PhoenixKit.Jobs.Engine does the I/O — it locks the run row, calls this, writes the changes, inserts the Oban job a {:dispatch, seconds} effect asks for and the Activity entry a {:log, …} effect asks for, all in one transaction. Keeping the rules here makes them testable without a database (dev_docs/plans/2026-10-03-job-runs.md, §3.2, §14 R1, R2, R5, R7).

Events

  • {:claim, token} — a batch is about to work. Needs queued or running and no claim held. A pending restart is taken in here.
  • {:checkpoint, token, outcome} — the batch holding token is done. outcome is {:more, progress, opts}, {:done, progress, result}, {:snooze, seconds}, {:fail, message} (final) or {:release, message} (an error Oban will retry).
  • {:pause, actor}, {:resume, actor}, {:cancel, actor}
  • :request_restart — a trigger asked for a fresh pass
  • {:heartbeat, token}
  • {:rescue, limit} — the sweeper found the dispatch lost; refused while a batch holds the run
  • {:fail, message} — the run cannot go on (a job the sweeper found discarded); refused while a batch holds the run — a batch ends its own run through {:checkpoint, token, {:fail, message}}

Precedence when a batch ends

StateBatch returns more / snoozesBatch is done
running, no restart pendingnext batch dispatchedcompleted
running, restart pendingpass restarts (queued, counters reset)pass restarts
pausingpaused, nothing dispatchedcompleted (nothing is left to pause)
cancellingcancelledcancelled

A pending restart never overrides a pause or a cancel; it waits.

Summary

Types

actor()
effect()
error()
event()
outcome()
progress()

Functions

transition(run, arg2, now)

Applies event to run at now: {:ok, changes, effects} — changes is a map of fields to write, effects what to do besides — or {:error, reason}.

Types

actor()

@type actor() :: String.t() | nil

effect()

@type effect() :: {:dispatch, non_neg_integer()} | {:log, String.t(), map()}

error()

@type error() ::
  :inactive
  | :claimed
  | :claim_lost
  | :already_pausing
  | :already_paused
  | :already_cancelling
  | :draining
  | :not_paused
  | :cancelling
  | :finished
  | :not_active

event()

@type event() ::
  {:claim, String.t()}
  | {:checkpoint, String.t(), outcome()}
  | {:pause, actor()}
  | {:resume, actor()}
  | {:cancel, actor()}
  | :request_restart
  | {:heartbeat, String.t()}
  | {:rescue, pos_integer()}
  | {:fail, String.t()}

outcome()

@type outcome() ::
  {:more, progress(), keyword()}
  | {:done, progress(), map()}
  | {:snooze, non_neg_integer()}
  | {:fail, String.t()}
  | {:release, String.t()}

progress()

@type progress() :: %{
  optional(:cursor) => map(),
  optional(:done) => non_neg_integer(),
  optional(:failed) => non_neg_integer(),
  optional(:total) => non_neg_integer() | nil
}

Functions

transition(run, arg2, now)

@spec transition(PhoenixKit.Jobs.Run.t(), event(), DateTime.t()) ::
  {:ok, map(), [effect()]} | {:error, error()}

Applies event to run at now: {:ok, changes, effects} — changes is a map of fields to write, effects what to do besides — or {:error, reason}.