# `PhoenixKit.Jobs.Engine`
[🔗](https://github.com/BeamLabEU/phoenix_kit/blob/v2.55.1/lib/phoenix_kit/jobs/engine.ex#L1)

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.

# `ctx`

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

# `scope`

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

# `active_run`

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

The active run of `kind` for `scope`, or nil.

# `checkpoint`

```elixir
@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`

```elixir
@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`

```elixir
@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`

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

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

# `recover`

```elixir
@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`

```elixir
@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`

```elixir
@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).

---

*Consult [api-reference.md](api-reference.md) for complete listing*