PhoenixKit.Jobs.Engine (phoenix_kit v2.55.1)

Copy Markdown View Source

The transactional core of job runs. Not the public API — that is PhoenixKit.Jobs, which checks permissions and takes a Scope; this module trusts its caller and does the I/O the pure PhoenixKit.Jobs.StateMachine decides.

Every transition is one transaction: the row is locked (FOR UPDATE), the state machine says what changes and what to do besides, the changes are written, the Oban job a dispatch asks for is inserted, and the Activity entry is inserted — all or nothing. The PubSub broadcast, the feed's broadcast and the kind's on_finish/2 happen after the commit.

Because the row is locked, a pause that races a checkpoint is serialised: it either commits first (the checkpoint then sees pausing and dispatches nothing) or after (it then sees the batch is done). That is what a re-read of the state before an enqueue could not give.

Not inside another transaction

The promise above — nothing is published or finished before the commit — can only be kept when this module's transaction is the outermost one. Called from inside a host's Repo.transaction/1 or an Ecto.Multi, the commit is the host's, and an event or an on_finish/2 would escape a transaction that may still roll back. Every entry point therefore refuses to run there and answers {:error, :in_transaction} before writing anything: call PhoenixKit.Jobs after your transaction commits.

Dispatch

A dispatch bumps generation (the state machine puts it in the changes) and inserts a PhoenixKit.Jobs.RunWorker job carrying that generation. A job of an older generation finds a mismatch when it tries to claim and does nothing. When Oban is not running (a web-only node, a script) the insert is skipped, the run keeps its new generation and no job, and the sweeper — or the next node that runs Oban — recovers it.

Summary

Types

ctx()
scope()

Functions

active_run(kind, scope)

The active run of kind for scope, or nil.

checkpoint(run_uuid, token, outcome, opts \\ [])

The batch holding token is done: records its outcome (see PhoenixKit.Jobs.StateMachine) and, unless a pause or cancel was waiting, dispatches the next batch — in one transaction. :dispatch false (inline execution) leaves the dispatch to the caller.

claim(run_uuid, generation)

Takes the claim on a run for the Oban job of generation. {:ok, run, token} (the run as it is now, a pending restart already taken in), or {:skip, reason}: :obsolete (an older generation, or the run is gone), :inactive (paused, cancelled, finished…) or :busy (a batch still holds it).

claim_inline(run_uuid, owner \\ Ecto.UUID.generate())

Claims a run for inline execution (PhoenixKit.Jobs.run_inline/3): any generation, and the run's generation is bumped so a pending Oban job of the old one is inert. Refuses a run a batch already holds — unless that is a script's claim whose lease ran out (Run.lease_expired?/2), which a new script may take over. The claim is marked inline: the sweeper has no Oban job to ask about it, and judges it only by its lease.

heartbeat(run)

Records a sign of life from the batch holding token. Cheap: no transaction.

recover(run_uuid, cutoff, decide, opts \\ [])

Recovers one run the sweeper suspects stopped without saying so — decided under the row lock.

start(kind, scope, opts \\ [])

Starts a run of kind for scope, or says which one is already active.

transition(run_uuid, event, opts \\ [])

Applies a state machine event to the run, in one transaction. Returns the updated run, or {:error, reason} (:not_found or the machine's).

Types

ctx()

@type ctx() :: %{actor: String.t() | nil, mode: String.t(), dispatch: boolean()}

scope()

@type scope() :: :site | {String.t(), String.t()}

Functions

active_run(kind, scope)

@spec active_run(String.t(), scope()) :: PhoenixKit.Jobs.Run.t() | nil

The active run of kind for scope, or nil.

checkpoint(run_uuid, token, outcome, opts \\ [])

@spec checkpoint(
  String.t(),
  String.t(),
  PhoenixKit.Jobs.StateMachine.outcome(),
  keyword()
) ::
  {:ok, PhoenixKit.Jobs.Run.t()} | {:error, term()}

The batch holding token is done: records its outcome (see PhoenixKit.Jobs.StateMachine) and, unless a pause or cancel was waiting, dispatches the next batch — in one transaction. :dispatch false (inline execution) leaves the dispatch to the caller.

claim(run_uuid, generation)

@spec claim(String.t(), integer()) ::
  {:ok, PhoenixKit.Jobs.Run.t(), String.t()}
  | {:skip, :obsolete | :inactive | :busy}
  | {:error, :in_transaction}

Takes the claim on a run for the Oban job of generation. {:ok, run, token} (the run as it is now, a pending restart already taken in), or {:skip, reason}: :obsolete (an older generation, or the run is gone), :inactive (paused, cancelled, finished…) or :busy (a batch still holds it).

claim_inline(run_uuid, owner \\ Ecto.UUID.generate())

@spec claim_inline(String.t(), String.t()) ::
  {:ok, PhoenixKit.Jobs.Run.t(), String.t()}
  | {:error, :claimed | :inactive | :not_found | :in_transaction}

Claims a run for inline execution (PhoenixKit.Jobs.run_inline/3): any generation, and the run's generation is bumped so a pending Oban job of the old one is inert. Refuses a run a batch already holds — unless that is a script's claim whose lease ran out (Run.lease_expired?/2), which a new script may take over. The claim is marked inline: the sweeper has no Oban job to ask about it, and judges it only by its lease.

owner identifies the invocation (run_inline/3 makes one and passes it to every batch of its loop). It is kept on the run between the batches, while the script waits out a delay, and a different owner is refused (:claimed) while that owner's lease stands — the wait belongs to the invocation that asked for it. A new invocation may take the run over once the lease has run out.

heartbeat(run)

@spec heartbeat(PhoenixKit.Jobs.Run.t()) :: :ok | {:error, :claim_lost}

Records a sign of life from the batch holding token. Cheap: no transaction.

recover(run_uuid, cutoff, decide, opts \\ [])

@spec recover(
  String.t(),
  DateTime.t(),
  (PhoenixKit.Jobs.Run.t() ->
     :leave | {atom(), PhoenixKit.Jobs.StateMachine.event()}),
  keyword()
) :: {:ok, atom()} | {:error, term()}

Recovers one run the sweeper suspects stopped without saying so — decided under the row lock.

The caller lists candidates cheaply; whether a candidate really needs recovery is only knowable once nothing else can move it, so the decision is made here: the run is locked and reloaded, must still be unchanged since cutoff and waiting on a batch (queued, running, pausing, cancelling), and decide is then called with the run as it is now and answers :leave or {tag, event} — the state machine event that puts it right, and a tag (:released, :failed, :rescued) for the caller's tally. A candidate that a live worker advanced, rescued or claimed in the meantime is therefore left alone, whatever the sweeper saw before. A release is followed by one more decision on the released run (a dead job is judged like any waiting one), in the same transaction.

{:ok, tag} (:left when nothing was done) or {:error, reason}.

start(kind, scope, opts \\ [])

@spec start(module(), scope(), keyword()) ::
  {:ok, PhoenixKit.Jobs.Run.t(), :started | :existing} | {:error, term()}

Starts a run of kind for scope, or says which one is already active.

Options: :args (a map), :actor_uuid, :mode (default "manual"), :retry_of (a run uuid, kept in the args). Returns {:ok, run, :started}, {:ok, run, :existing} (an active run of this kind and scope exists; the kind's restart/0 decided whether it was asked for a fresh pass) or {:error, reason}. {:error, :raced} — the active run finished as this start arrived, three attempts in a row — means the trigger was not recorded.

transition(run_uuid, event, opts \\ [])

@spec transition(String.t(), PhoenixKit.Jobs.StateMachine.event(), keyword()) ::
  {:ok, PhoenixKit.Jobs.Run.t()} | {:error, term()}

Applies a state machine event to the run, in one transaction. Returns the updated run, or {:error, reason} (:not_found or the machine's).