The wire shape between a %Statifier.Effect.Invoke{} and the args map
an invoke-handler Oban job stores.
Oban args live as JSON, so this module owns the (de)serialization of
the one effect an invoke job carries, plus the scope it was enqueued
under and the handler module whose run/1 the worker calls back into.
The rules are StatifierOban.Timer.JobArgs's, applied to the invoke
effect:
- Every deterministic field rides as an explicit JSON value: the dedup
triple (
scope,invoke_id,macrostep- ADR-0003) at the top level - Oban's uniquenesskeysread args at the top level - and the effect's remaining position row data (state_index,invoke_index,microstep,round) beside it, self-describing in the store during an incident.invoke_idis the authored id used verbatim, or the deterministic%MachineState{}counter, st-ADR-0008 (as amended) blesses as the idempotency key; scoping is mandatory because that counter restarts per chart run, andmacrostepis what tells a state re-entry's fresh invocation apart from a crash replay of the old one. - The three host-opaque fields,
params,contentandcaller_context, are arbitrary terms with no JSON shape, so they ride as tagged:erlang.term_to_binary/1payloads (StatifierOban.OpaqueTerm) and come back byte-identical.from_invoke/4's optional codec runs over both fields' bytes and tags the payload with its module name (StatifierOban.OpaqueTerm.Codec);to_invoke/1reads whatever tag the stored row carries, regardless of what the reading caller passed.caller_contextisst-ADR-0063's opaque host slot, stamped by the macrostep that executed the<invoke>; it is stored so the answer event can inherit it days later on another node, and the two durability rulesStatifierOban.Timer.JobArgsstates for the timer half bind a host's choice of term here identically. A row written before this field existed carries no"caller_context"key and decodes tonil, which isst-ADR-0063's own "no context attached" - so the field is additive over stored rows, not a migration. handleris the module name of theStatifierOban.Invoke.Handlerimplementation, written from a validated module at enqueue time and resolved back by the worker - a resolution failure there is deploy-shaped (the module was renamed or removed after the job was stored) and retries, exactly like the timer worker's delivery module.
A fan-out child start job stores the same map with three fields
added, for_child_start/4's "index", "child_count" and
"policy" (ADR-0007 decision 4). The base map is unchanged by that
widening, so a start job's row decodes through to_invoke/1 exactly
as an ordinary invoke job's does and the two kinds stay mutually
readable during an incident. All three ride as explicit JSON values
rather than through the opaque-term codec: they are this package's own
scheduling facts with a JSON shape, not host terms, and a value the
codec owns cannot be read off the row during an incident.
to_invoke/1 is the exact inverse of from_invoke/4 for every
%Statifier.Effect.Invoke{} the base handler enqueues: what the job
carries is enough to hand the handler's run/1 the same effect the
planning callback saw.
Summary
Types
String-keyed args map as Oban stores and redelivers it.
Why from_invoke/4 could not build an args map.
Functions
Reads a child start job's seam options back off its args.
Reads a child start job's {index, count} back off its args.
Widens an invoke job's args into a fan-out child start job's args.
Builds the args map for an invoke job from the scope, the handler module, and the effect.
Rebuilds just the identity pair - the scope and the invoke id - from a job's args.
Rebuilds the scope, the handler module name, and the
%Statifier.Effect.Invoke{} from a job's args.
Types
String-keyed args map as Oban stores and redelivers it.
@type decode_error() :: {:missing_field, String.t()} | {:invalid_field, String.t(), term()} | StatifierOban.OpaqueTerm.decode_error()
@type encode_error() :: {:codec_failed, String.t(), StatifierOban.OpaqueTerm.encode_error()}
Why from_invoke/4 could not build an args map.
Functions
@spec child_opts(args()) :: {:ok, StatifierOban.Invoke.ChildStarter.opts()} | {:error, decode_error()}
Reads a child start job's seam options back off its args.
Returns the keyword list
StatifierOban.Invoke.ChildStarter.start_child/5 is handed:
[policy: :all] or [policy: :first_error].
A row carrying no "policy" reads as [policy: :all] rather than
as an error. :all is the aggregation an absent on parameter means
on the way in, so it is what an absent field means on the way out too,
and a start job stored before this field existed then starts the child
it was always going to start. A "policy" carrying anything else is
{:invalid_field, _, _}, on to_invoke/1's rule: a value that is not
one of the two words is a fact about the row, and the worker decides
what to do with it.
@spec child_position(args()) :: {:ok, non_neg_integer(), pos_integer()} | {:error, decode_error()}
Reads a child start job's {index, count} back off its args.
The rules are to_invoke/1's: a missing or malformed field is a typed
error about the row rather than a raise, and the worker decides what
to do with it. A "child_count" of zero, or an "index" that is not
inside it, is {:invalid_field, _, _} for the same reason a negative
index would be - the row cannot be a child of any invocation.
@spec for_child_start( args(), non_neg_integer(), pos_integer(), StatifierOban.Invoke.ChildStarter.policy() ) :: args()
Widens an invoke job's args into a fan-out child start job's args.
A child start job carries the whole invocation - the same fields
from_invoke/4 laid out, opaque payloads and codec tag included, so
the starter seam is handed the effect the planning callback saw - plus
the three values that say which child of which fan-out it is:
"index"is the item's zero-based position in the fanned-out list. It rides at the top level because it is a key component:StatifierOban.Invoke.ChildStartWorkeris unique on the four-component{scope, invoke_id, macrostep, index}(ADR-0007 decision 4), and Oban's uniquenesskeysread args at the top level."child_count"is the list's length, and is row data rather than a key component - two starts of the same index under the same invocation are the same scheduling decision whatever the count says. It travels because the seam's callback takes it: a starter that records the child's slot needs to know how many slots there are, and re-deriving it later would mean re-reading the parent."policy"is the invocation's aggregation policy as its wire word,"all"or"first_error". It is row data for the same reason"child_count"is, and it travels for a sharper one: the settlement side records the policy on each child's own linkage at creation, so the value has to reachStatifierOban.Invoke.ChildStarter.start_child/5at every index rather than be looked up once.
The policy is one of two constants rather than an arbitrary term, so
it is written as itself and read back by child_opts/1; nothing here
goes through the opaque-term codec.
@spec from_invoke(String.t(), module(), Statifier.Effect.Invoke.t(), module() | nil) :: {:ok, args()} | {:error, encode_error()}
Builds the args map for an invoke job from the scope, the handler module, and the effect.
The caller (StatifierOban.Invoke.Handler.perform_start/3) has already
validated the scope; this function only lays fields out on the wire.
params and content are encoded through a with, so the first codec
failure short-circuits and no partially-encoded args map is ever
returned.
@spec identity(args()) :: {:ok, String.t(), String.t()} | {:error, decode_error()}
Rebuilds just the identity pair - the scope and the invoke id - from a job's args.
to_invoke/1 fails the whole row when any field is undecodable,
including the two host-opaque payloads, which is the common way a row
goes bad. This reads only the two plain-string fields that name the
invocation, so a caller holding an otherwise undecodable row can still
tell the run which invocation it is about
(StatifierOban.Invoke.Worker delivers error.communication that
way before cancelling). The rules are to_invoke/1's own, because
this is the same fetch_binary/2: a missing, empty, or non-string
field is a typed error, and then the row names nothing and there is
nobody to tell.
@spec to_invoke(args()) :: {:ok, String.t(), String.t(), Statifier.Effect.Invoke.t()} | {:error, decode_error()}
Rebuilds the scope, the handler module name, and the
%Statifier.Effect.Invoke{} from a job's args.
The handler comes back as the stored string, not a resolved module: resolution is the worker's call, because an unresolvable name is a retryable environment fact where every error here is a fact about the row. Returns a typed error rather than raising: an undecodable job is a fact about the row, and the worker decides what to do with it.