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.EnrichHandler 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
# creating a second enrichment.
with {:ok, enrichment} <-
MyApp.Enrichments.upsert_by_invoke_id(invoke.invoke_id, invoke.params) do
{:ok, %{"enrichment_id" => enrichment.id}}
end
end
endThe 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}, into the config's:invoke_queueon the config's Oban instance; for a cancel it cancels every stored job under that pair. Both are idempotent oninvoke_idby 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 still in a non-terminal state. That uniqueness is what this module enforces; what yourrun/1does with the outside world is yours to make idempotent the same way. run/1is 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; a raise does the same.
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} for the same reason the
timer jobs key on {scope, ordinal}: invoke_id is a deterministic
%MachineState{} counter (st-ADR-0008 as amended), byte-identical when
a crashed host re-runs the same drive, and it restarts per chart run,
so the scope (ctx.session_id, or the host's own durable run id) keeps
unrelated runs apart. 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} 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.
Open question: surfacing permanent failure 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 retries are exhausted, Oban discards the job, and today nothing
is fed back into the run: the failure is observable on the job row
(discarded state, {:run_failed, reason} in the errors) and
nowhere else. Whether - and as what - a permanently failed invocation
should surface into the chart (an error.* event on the run, for
example) is deliberately open: the event vocabulary is
statifier-ex's call, so this package documents the gap rather than
deciding it.
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 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.
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.
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.
Inserts the one Oban job that runs handler.run/1 for invoke.
Types
@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()} | Ecto.Changeset.t()
Why a start or cancel could not be performed.
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()} | {: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.
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.
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.
@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} 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. 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.