StatifierOban.Invoke.JobArgs (StatifierOban v0.8.0)

Copy Markdown View Source

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 uniqueness keys read 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_id is 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, and macrostep is what tells a state re-entry's fresh invocation apart from a crash replay of the old one.
  • The three host-opaque fields, params, content and caller_context, are arbitrary terms with no JSON shape, so they ride as tagged :erlang.term_to_binary/1 payloads (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/1 reads whatever tag the stored row carries, regardless of what the reading caller passed. caller_context is st-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 rules StatifierOban.Timer.JobArgs states 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 to nil, which is st-ADR-0063's own "no context attached" - so the field is additive over stored rows, not a migration.
  • handler is the module name of the StatifierOban.Invoke.Handler implementation, 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

args()

@type args() :: %{optional(String.t()) => term()}

String-keyed args map as Oban stores and redelivers it.

decode_error()

@type decode_error() ::
  {:missing_field, String.t()}
  | {:invalid_field, String.t(), term()}
  | StatifierOban.OpaqueTerm.decode_error()

encode_error()

@type encode_error() ::
  {:codec_failed, String.t(), StatifierOban.OpaqueTerm.encode_error()}

Why from_invoke/4 could not build an args map.

Functions

child_opts(args)

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

child_position(args)

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

for_child_start(args, index, count, policy)

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.ChildStartWorker is unique on the four-component {scope, invoke_id, macrostep, index} (ADR-0007 decision 4), and Oban's uniqueness keys read 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 reach StatifierOban.Invoke.ChildStarter.start_child/5 at 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.

from_invoke(scope, handler, invoke, codec \\ nil)

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

identity(args)

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

to_invoke(args)

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