The Oban-backed base for a Statifier.Invoke.Handler implementation:
use it and the host's <invoke type="..."> runs its work in an Oban
job on the host-supplied instance, with completion delivered back into
the run as done.invoke.<invoke_id> through
StatifierOban.Invoke.Delivery.
defmodule MyApp.AuthorizationHandler do
use StatifierOban.Invoke.Handler
@impl StatifierOban.Invoke.Handler
def config, do: MyApp.statifier_oban_config()
@impl StatifierOban.Invoke.Handler
def run(invoke) do
# `invoke.invoke_id` is the idempotency key upstream hands you:
# stable by construction across replays. Keying the write on it
# makes a re-run land on the prior attempt's row instead of
# authorizing the card a second time.
with {:ok, authorization} <-
MyApp.Payments.authorize_by_invoke_id(invoke.invoke_id, invoke.params) do
{:ok, %{"authorization_id" => authorization.id}}
end
end
endWork that keys on the run rather than on the invocation defines
run/2 instead, which is handed the job's run_ctx/0 alongside the
effect - the scope lives on the job row, not on the effect, so run/1
cannot see which run it is working for:
defmodule MyApp.ProvisioningHandler do
use StatifierOban.Invoke.Handler
@impl StatifierOban.Invoke.Handler
def config, do: MyApp.statifier_oban_config()
@impl StatifierOban.Invoke.Handler
def run(invoke, %{scope: scope}) do
with {:ok, record} <- MyApp.Provisioning.provision(scope, invoke.invoke_id) do
{:ok, %{"provision_id" => record.id}}
end
end
endDefine one arity or the other: a module defining both runs through
run/2, and one defining neither does not compile.
The division of labor follows st-ADR-0051 decision 4 exactly:
- The injected
start/2,cancel/2, andforward/3are pure planning callbacks. Each returns{:handler, __MODULE__, payload}instructions (or none) and touches nothing - not the config, not Oban.forward/3plans nothing by default and is overridable: this base gives a job-shaped invocation no autoforward path, because a job that has not run yet has nowhere to receive an event. - The injected
perform/2is the impure half: for a start it inserts oneStatifierOban.Invoke.Workerjob, unique on{scope, invoke_id, macrostep}(ADR-0003), into the config's:invoke_queueon the config's Oban instance; for a cancel it cancels every stored job under{scope, invoke_id}, whatever its macrostep. Both are idempotent by construction: the unique key makes a replayed insert conflict with the stored job (the same scheduling decision, not a new one), and a replayed cancel matches only jobs that have not run yet. That uniqueness is what this module enforces; what yourrun/1does with the outside world is yours to make idempotent the same way. run/1(orrun/2) is your work, executed inside the Oban job - at least once, so the MUST-be-idempotent-on-invoke_idcontract from statifier-ex'sdocs/extending.mdlands on it.{:ok, donedata}becomesdone.invoke.<invoke_id>with that donedata, delivered behind the run-liveness check;{:error, reason}makes the job retry, and a raise does the same. Exhausting those retries discards the job and deliverserror.communication.invoke.<invoke_id>into the run (see "Permanent failure surfaces into the chart" below).
config/0 is read at perform/2 time, never at planning time - which
is what keeps the planning callbacks pure even though the config is
host state. A config without :invoke_queue fails the first
perform/2 with {:error, {:missing_option, :invoke_queue}} rather
than falling back into any host queue.
The uniqueness key is {scope, invoke_id, macrostep} (ADR-0003) for
the same reason the timer jobs key on {scope, ordinal}: every
component is deterministic as of scheduling, byte-identical when a
crashed host re-runs the same drive. invoke_id is either the
author's literal id, used verbatim, or a deterministic
%MachineState{} counter (st-ADR-0008 as amended); it restarts per
chart run, so the scope (ctx.session_id, or the host's own durable
run id) keeps unrelated runs apart. macrostep is what tells a
replay from a re-entry: an authored id is byte-identical on every
re-entry of its state, and only the macrostep separates that fresh
scheduling decision from a redelivery of the old one. The unique
window is every state over an infinite period, and the unique fields
exclude :queue and the meta the delivery module rides on - a host
moving its invoke queue or reconfiguring its delivery must not turn a
replay into a second job.
At-least-once: the contract is upstream's
Jobs here are at-least-once, so a handler's work MUST be idempotent
on invoke_id. The contract's wording is statifier-ex's, not this
package's: read "At-least-once: handlers must be idempotent" in
statifier-ex's docs/extending.md - this module cites it rather than
restating it, because the seam that defines perform/2 owns the
rulebook. What this base adds on top is only the enqueue-side dedup
described above ({scope, invoke_id, macrostep} uniqueness).
Everything run/1
does to the outside world is the implementor's to key, and
invoke.invoke_id is the key upstream hands you for exactly that,
as the example above demonstrates.
Permanent failure surfaces into the chart
{:error, reason} from run/1 (and a raise or exit out of it) maps
to an Oban retry, never a cancel: the work is idempotent on
invoke_id by contract, so retrying is what at-least-once means.
When the retries are exhausted, Oban discards the job - and the run
hears about it. The discarding attempt delivers
error.communication.invoke.<invoke_id>into the chart, carrying %{"reason" => class, "attempts" => n, "detail" => text}, behind the same run-liveness check a completion
goes through. A chart that parks failed work for operator recovery
transitions on that event, or on the bare error.communication it
extends, and no longer hangs in the invoking state when your run/1
gives up for good.
The event and its payload are statifier-ex's call (st-ADR-0068); the
failure classes in "reason" are this package's, and
StatifierOban.Invoke.Worker lists them. Nothing is asked of your
run/1: keep returning {:error, reason} for anything worth
retrying, and the retry policy decides when that becomes permanent.
This was an open question in earlier versions of this module; ADR-0005
records how it was settled.
What this module deliberately does not do: interpret run/1's
donedata (<finalize> and namelist auto-assign are the session's, per
st-ADR-0051 decision 6), dedup your side effects (the at-least-once
contract is documented upstream and pinned by
Statifier.Testing.HandlerCase), or feed anything to a dead run (the
delivery seam discards a completed invoke against a dead or halted run
the same way a fired timer is discarded).
Summary
Types
The fan-out return: this invocation is N children, not one result.
The payload inside this base's {:handler, module, payload}
instructions: a start carrying the full effect, or a cancel carrying
the invoke_id.
Why a start or cancel could not be performed.
What the worker knows about the job it is running run/2 inside, and
the effect does not carry.
Callbacks
The host's StatifierOban.Config - the Oban instance, the
:invoke_queue, and the :invoke_delivery seam. Read at perform/2
time only, so implementations may read it from wherever host
configuration lives.
The work itself, executed inside the Oban job - at least once per
invoke_id, so it MUST be idempotent on it (statifier-ex
docs/extending.md, "At-least-once"). {:ok, donedata} is delivered
back to the run as done.invoke.<invoke_id>; {:error, reason} and
raises both make the job retry, and exhausting the retries delivers
error.communication.invoke.<invoke_id> instead.
The same work, handed the job's run_ctx/0 as well as the effect -
the arity to define when the work keys on the run rather than on
the invocation alone (sob-7b1).
Functions
The one implementation behind every use-ing module's perform/2:
routes a {:start, invoke} payload to perform_start/3 and a
{:cancel, invoke_id} payload to perform_cancel/3.
Cancels every stored invoke job under {scope, invoke_id} on the
handler's configured instance - every generation, whatever macrostep
each job was keyed under (ADR-0003).
Inserts the one Oban job that runs handler.run/1 for invoke.
Types
The fan-out return: this invocation is N children, not one result.
items is the evaluated list - sb-ADR-0009 decision 3 compiles
the items datamodel path into the <param> list and makes the
handler what evaluates it, and a job has the effect but no datamodel,
so what a handler that fans out reads off invoke.params is that
path: it evaluates the path itself and returns the resulting list
here (or builds the list however it likes). Only its length is
read by this package; what the items are is the handler's business,
and each one is bound to its child by the
StatifierOban.Invoke.ChildStarter seam.
opts is the handler's own, not the chart's. core.map declares no
concurrency field - sb-ADR-0009 decision 4 leaves that block field
deferred - so nothing arrives in params to pass along. A handler
with a width hint of its own passes max_concurrency: n, which is
shape-validated and then clamped to the queue's own limit in both
directions; a hint below that limit is not honoured - see
StatifierOban.Invoke.FanOut.
A handler returning this delivers no done.invoke: the invocation
stays open until the settlement side answers it once, on behalf of
all N. An empty items list is the one exception - a fan-out over
nothing succeeds over nothing (sb-ADR-0009 decision 8), so no child
starts and the invocation is answered immediately with []. A fan-out
refused before any child starts - over :max_fan_out, not a list, or
carrying an on that is neither "all" nor "first_error" - fails
the invocation on error.communication.invoke.<invoke_id> instead
(ADR-0007 decision 8). Those three refusals are the whole list;
StatifierOban.Invoke.FanOut.refusal/0 names them.
@type payload() :: {:start, Statifier.Effect.Invoke.t()} | {:cancel, String.t()}
The payload inside this base's {:handler, module, payload}
instructions: a start carrying the full effect, or a cancel carrying
the invoke_id.
@type perform_error() :: {:missing_option, :invoke_queue} | {:invalid_scope, term()} | StatifierOban.Invoke.JobArgs.encode_error() | Ecto.Changeset.t()
Why a start or cancel could not be performed.
What the worker knows about the job it is running run/2 inside, and
the effect does not carry.
:scope is the run the invocation belongs to - the same string the
base validated out of the planning ctx (ctx.session_id, or the
host's own durable run id) and stored on the job row, so work keyed
to the workflow instance can key on it. :invoke_id is the
idempotency key, repeated here so a handler destructuring the context
has the whole identity pair in one place; it is also on the effect,
and the two are always the same value.
The map is open by construction: this base may add keys, so match the
ones you need (def run(invoke, %{scope: scope})) rather than the
whole map.
Callbacks
@callback config() :: StatifierOban.Config.t()
The host's StatifierOban.Config - the Oban instance, the
:invoke_queue, and the :invoke_delivery seam. Read at perform/2
time only, so implementations may read it from wherever host
configuration lives.
@callback run(invoke :: Statifier.Effect.Invoke.t()) :: {:ok, donedata :: term()} | fan_out() | {:error, term()}
The work itself, executed inside the Oban job - at least once per
invoke_id, so it MUST be idempotent on it (statifier-ex
docs/extending.md, "At-least-once"). {:ok, donedata} is delivered
back to the run as done.invoke.<invoke_id>; {:error, reason} and
raises both make the job retry, and exhausting the retries delivers
error.communication.invoke.<invoke_id> instead.
{:fan_out, items} (or {:fan_out, items, opts}) is the third
answer, and it is not an answer at all: it says this invocation is
N children rather than one result. See fan_out/0.
@callback run(invoke :: Statifier.Effect.Invoke.t(), ctx :: run_ctx()) :: {:ok, donedata :: term()} | fan_out() | {:error, term()}
The same work, handed the job's run_ctx/0 as well as the effect -
the arity to define when the work keys on the run rather than on
the invocation alone (sob-7b1).
The effect is the invocation, so run/1 sees everything about what
was invoked; what it cannot see is which run invoked it, because the
scope lives on the job row rather than on the effect. Provisioning
keyed to the workflow instance, a write into a per-run table, a lookup
of the host's own record for the run: all of that needs the scope, and
this is the arity that has it.
Define run/1 or run/2, not both - a module defining both runs
through run/2, and the run/1 clause is dead code. A module
defining neither does not compile. The contract is otherwise
identical to run/1's, at-least-once and idempotent-on-invoke_id
included.
@impl StatifierOban.Invoke.Handler
def run(invoke, %{scope: scope}) do
with {:ok, record} <- MyApp.Provisioning.provision(scope, invoke.invoke_id) do
{:ok, %{"provision_id" => record.id}}
end
end
Functions
@spec perform(module(), payload(), Statifier.Invoke.Handler.ctx()) :: :ok | {:error, perform_error()}
The one implementation behind every use-ing module's perform/2:
routes a {:start, invoke} payload to perform_start/3 and a
{:cancel, invoke_id} payload to perform_cancel/3.
@spec perform_cancel(module(), String.t(), Statifier.Invoke.Handler.ctx()) :: :ok | {:error, perform_error()}
Cancels every stored invoke job under {scope, invoke_id} on the
handler's configured instance - every generation, whatever macrostep
each job was keyed under (ADR-0003).
6.4.3's cancellation for a job-shaped invocation: a job that has not
run yet is cancelled and never runs; a job that already reached a
terminal state keeps its outcome - a real-time <cancel> can lose to
work that already completed, and the delivery seam's liveness check is
the guard on that side, not this one. A cancel matching nothing
(including an invoke_id this handler never saw - a crash-recovering
host may replay a cancel whose start it never durably recorded) is a
no-op, never an error. The match ignores the queue, exactly as the
unique key does.
A cancel only ever reaches a job that has not run. The match is
restricted to the states a pending invocation can be in - scheduled,
available, retryable, and on Oban 2.21 or later suspended - so
a job that is executing right now is never swept, and neither is one
that already reached a terminal state.
Leaving executing out is what makes the common self-cancel safe, for
the same reason it does on the timer half (StatifierOban.Timer.cancel/3,
sob-uon). Spec 6.4.3 cancels an invocation when its invoking state is
exited, and the commonest way that state is exited is the invocation's
own completion: the job delivers done.invoke.<invoke_id>, the run
transitions out, and the exit runs <cancel> for the very invoke_id
being delivered - so the cancel and the invocation are the same Oban
job. Sweeping executing rows here would have
Oban.cancel_all_jobs/2 signal a :pkill at that job's own process
and kill it mid-delivery, leaving the row cancelled with
{:cancel, :shutdown}. A cancel raised from inside a job's own
execution must not kill that execution.
Cancelling a genuinely in-flight invocation from outside is therefore
not offered, exactly as it is not for timers: from the query's side
the self-cancel and the outside cancel are indistinguishable, and a
host that needs it holds the job id and can call Oban.cancel_job/2
itself.
@spec perform_start( module(), Statifier.Effect.Invoke.t(), Statifier.Invoke.Handler.ctx() ) :: :ok | {:error, perform_error()}
Inserts the one Oban job that runs handler.run/1 for invoke.
Unique on {scope, invoke_id, macrostep} over every state and an
infinite period, so performing the same instruction again - the
at-least-once replay st-ADR-0051 decision 4 says MAY happen -
conflicts with the stored job and inserts nothing, while a re-entered
state's fresh invocation (same authored id, later macrostep) inserts
a fresh job (ADR-0003). The job lands in the config's
:invoke_queue and carries the config's :invoke_delivery module in
its meta; meta is not part of the unique fields, so a replay under a
reconfigured delivery still conflicts with the stored job.
The config's :opaque_codec is fixed at enqueue time: read once from
handler.config(), and run over the effect's host-opaque params and
content before the job is stored. The module name travels on the row
alongside the encoded bytes, so the worker that later decodes the job
and runs handler.run/1 needs no configuration of its own.