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, 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 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, 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.
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 - every generation, whatever macrostep
each job was keyed under (ADR-0003).
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 - 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.
@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.