StatifierOban.Invoke.Handler behaviour (StatifierOban v0.10.0)

Copy Markdown View Source

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
end

Work 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
end

Define 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, and forward/3 are pure planning callbacks. Each returns {:handler, __MODULE__, payload} instructions (or none) and touches nothing - not the config, not Oban. forward/3 plans 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/2 is the impure half: for a start it inserts one StatifierOban.Invoke.Worker job, unique on {scope, invoke_id, macrostep} (ADR-0003), into the config's :invoke_queue on 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 your run/1 does with the outside world is yours to make idempotent the same way.
  • run/1 (or run/2) is your work, executed inside the Oban job - at least once, so the MUST-be-idempotent-on-invoke_id contract from statifier-ex's docs/extending.md lands on it. {:ok, donedata} becomes done.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 delivers error.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

fan_out()

@type fan_out() ::
  {:fan_out, items :: list()} | {:fan_out, items :: list(), opts :: keyword()}

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.

payload()

@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.

perform_error()

@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.

run_ctx()

@type run_ctx() :: %{
  :scope => String.t(),
  :invoke_id => String.t(),
  optional(atom()) => term()
}

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

config()

@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.

run(invoke)

(optional)
@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.

run(invoke, ctx)

(optional)
@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

perform(handler, arg, ctx)

@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.

perform_cancel(handler, invoke_id, ctx)

@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.

perform_start(handler, invoke, ctx)

@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.