Statifier.Send.Processor behaviour (Statifier v2.15.0)

Copy Markdown View Source

The behaviour a host implements for an Event I/O Processor type it registers with Statifier.Session.start_link/2's :send_types option (ADR-0069). A <send> whose type is registered is handed to the module registered for it, with the event already built (Statifier.Send.Event.build/3), and the library delivers nothing for it itself.

The split is Statifier.Invoke.Handler's (ADR-0051 decision 4):

  1. deliver/3 and cancel/2 are pure planning callbacks, called from Statifier.Session.Effects.plan/2's fold with no process, no clock and no I/O. They return instructions for an executor to perform and perform nothing themselves. A processor's usual instruction is {:handler, __MODULE__, payload}, which the executor routes back to perform/2.
  2. perform/2 is the impure half, and MAY be called more than once for the same send. After a crash and a retry a host may perform the same effect again, so a processor MUST be idempotent on the ADR-0054 decision 3 dedup key's components read off the effect: the send id, the step counters, c_index, owner, and ordinal (every registered-type send carries one - ADR-0059 decision 5 as amended), with the session scope the host supplies.

What deliver/3 is handed

The %Statifier.Effect.Send{} or %Statifier.Effect.SendDelayed{} the core produced, and the event built from it with the default origin and origintype. target is the processor's own opaque route string: the library never parses it. A processor that wants a reply to reach its own address rather than the sender's session builds the event again with :origin and :origintype.

A delayed send is the processor's timer. For a %Statifier.Effect.SendDelayed{} the session schedules nothing: the processor owns the delay, and spec 6.2's discard at termination is its fire-time check (ADR-0054 decision 4). A processor that holds the delay in a process it starts from perform/2 may hand that process to Statifier.Session.HaltNotice.watch/2: a Statifier.Session then sends it a message when the session halts, so the fire-time check reads the process's own mailbox and never has to call a session that may be busy.

What cancel/2 is handed

A <cancel> whose send id names a delayed send this processor was handed reaches cancel/2 with the %Statifier.Effect.Cancel{} itself, whose send_id and the host's session scope are ADR-0054 decision 3's cancellation key. The session keeps which processor holds each such send id, so a cancel reaches only the processors that were handed a delayed send under that id. A processor MUST tolerate a cancel for a send it has already fired.

Those holds are the live session's own state. They are not part of the persisted position (Statifier.Position), so a session resumed from a position holds nothing: a <cancel> it runs for a delayed send handed over before the position was saved reaches no processor. Routing that cancel is the host's job, as re-arming timers after a resume is (ADR-0060 decision 7): the resumed session still emits the %Statifier.Effect.Cancel{}, and the host matches it under ADR-0054 decision 3's cancellation key against the send id it recorded when its processor was handed the delayed send.

A hold is released only when a <cancel> naming its send id is routed; the library never learns that a processor's timer fired. A live session therefore keeps one entry per distinct send id handed to a processor and not cancelled since: an author-written id reused by later sends stays one entry, and a generated id adds one per uncancelled send. The entries go with the session process when it stops.

Where the fire time of such a send lives, which transaction a cancel commits in, and which side drops a fire that races a cancel are ADR-0074's: the host stores the absolute fire time when first handed the send, commits the cancel with the position of the step that ran it, and drops a stale fire itself; the session drops none.

When the host cannot deliver

A processor that cannot deliver a send while its sender still exists reports the miss through Statifier.Session.failed_send/3, called by the host, never by deliver/3 or cancel/2: the sender then sees C.1's error.communication, carrying the send id, on its internal queue. When the sender has reached a final state or no longer exists there is no queue to write, and the host records the miss as a dead letter keyed by the send's dedup key, with its reason. failed_send/3's own documentation states both rules.

The _ioprocessors entry

Spec 5.10 binds _ioprocessors to one entry for each Event I/O Processor a session supports, and a registered type is one. The session's _ioprocessors carries an entry keyed by each registered type string, whose value is the map the optional Statifier.Send.Processor.ioprocessors_entry/1 callback returns for that type, or an empty map when the processor does not implement it. Statifier.Send.Types.from_send_types/1 reads the value and Statifier.MachineState.new/2 writes it, once, when the session starts; Statifier.Send.Types.from_send_types/1 says how it reads after a resume.

An entry that must address one session, such as a location a receiver POSTs to, cannot be built from the type alone. A processor that implements the optional Statifier.Send.Processor.ioprocessors_entry/2 is asked for its entry with the type string and a context carrying the session id and the registration's options, when the session starts (ADR-0075 decision 3); it is asked instead of /1, and a processor that implements only /1 is asked as before.

Refusing a registration

A processor that needs something from its registration, such as an option its entry is built from, implements the optional Statifier.Send.Processor.check_registration/2. A fresh start asks it for every registration whose module exports it, before any session process is spawned, and a registration it rejects refuses the start with a named value instead of a process that exits. A resume does not ask it, because the entries the persisted position carries stand (ADR-0075 decision 3). A check that raises, throws, exits or answers anything other than :ok or {:error, reason} refuses nothing; the callback's own doc says what the start does then.

Registration options

A :send_types value is a bare module or {module, opts} (ADR-0075 decision 8, point b). The options reach Statifier.Send.Processor.ioprocessors_entry/2's context as :opts and, for a {module, opts} registration only, the plan context deliver/3 and cancel/2 receive, under :opts. A bare-module registration's plan context carries no :opts key, exactly as before.

Any other value is outside that shape: a string, a {module, opts} whose opts is not a list, a tuple whose first element is not an atom, or a tuple of another size. Statifier.Session.start_link/2 has no named refusal for it. It asks no processor's check_registration/2 about a map that holds one, spawns the session, and the session process stops while it starts, so the start answers {:error, {:function_clause, stacktrace}}. The stacktrace's first frame is the library's split of the registration, called with the value as given:

{Statifier.Send.Types, :split, [value], _location}

The start does not raise in the caller. Statifier.Session.replace_send_type/3 takes the same two shapes and raises FunctionClauseError in the caller for any other value, before the session is asked.

ctx

The plan context Statifier.Session.Effects.plan/2 threads through its fold, handed over unchanged: a plain map carrying session_id (spec 5.10's _sessionid) and no pid, no %MachineState{} and no session struct, so a processor cannot reach into the session through it. A key added to it later is additive for every processor already written: :opts, a {module, opts} registration's options, is one (see "Registration options" above).

Summary

Types

The plan context - see the moduledoc's "ctx" section.

What Statifier.Send.Processor.ioprocessors_entry/2 is handed beside the type string: the session's _sessionid and the registration's options ([] for a bare-module registration).

One instruction a planning callback returns - an element of Statifier.Session.Effects.instruction/0, typed opaquely here as Statifier.Invoke.Handler.instruction/0 is.

Callbacks

Plans the cancellation of the delayed sends this processor was handed under cancel.send_id (spec 6.3's cancel-them-all). Pure.

Whether this processor can serve the registration of the type string type with the registration's options opts ([] for a bare-module registration). Pure. Answers :ok, or {:error, reason} naming what is wrong, for example {:missing_option, :base_url}. A reason names what the registration cannot be served without, not only an option that is absent: Statifier.Send.BasicHTTP answers {:missing_option, :base_url} both for a registration with no :base_url and for one whose :base_url is present but not a string, and names no separate reason for a malformed value.

Plans the hand-off of one registered-type send. Pure. effect is the send the core produced and event the event built from it. For a delayed send the processor owns the delay.

The value of this processor's _ioprocessors entry (spec 5.10) for the registered type string type - for example a "location" a receiver can address this processor by. Pure and deterministic: the library calls it whenever it builds a registered set, which a replay and a resume do again. The map must be string-keyed at every level, as every datamodel value is. Optional: a processor that does not implement it gets an empty map.

The value of this processor's _ioprocessors entry for the registered type string type in the session context names (ADR-0075 decision 3): for example a "location" built from the session id and a base URL in the registration's options. Called once, when the session starts, and asked instead of Statifier.Send.Processor.ioprocessors_entry/1 when a processor implements both. Pure and deterministic, and string-keyed at every level, as Statifier.Send.Processor.ioprocessors_entry/1 is. A processor that cannot build its entry from context raises ArgumentError. A processor that also implements Statifier.Send.Processor.check_registration/2 refuses such a registration at a fresh start before this callback is asked, and the raise is then left to a direct caller. Optional.

Performs one {:handler, module, payload} instruction a planning callback returned - the impure half. MAY be called more than once for the same send (see the moduledoc's point 2). Optional: a processor whose planning callbacks return no such instruction needs none.

Types

ctx()

@type ctx() :: %{:session_id => String.t(), optional(atom()) => term()}

The plan context - see the moduledoc's "ctx" section.

entry_context()

@type entry_context() :: %{session_id: String.t(), opts: keyword()}

What Statifier.Send.Processor.ioprocessors_entry/2 is handed beside the type string: the session's _sessionid and the registration's options ([] for a bare-module registration).

instruction()

@type instruction() :: term()

One instruction a planning callback returns - an element of Statifier.Session.Effects.instruction/0, typed opaquely here as Statifier.Invoke.Handler.instruction/0 is.

Callbacks

cancel(cancel, ctx)

@callback cancel(cancel :: Statifier.Effect.Cancel.t(), ctx :: ctx()) ::
  {:ok, [instruction()]}

Plans the cancellation of the delayed sends this processor was handed under cancel.send_id (spec 6.3's cancel-them-all). Pure.

check_registration(type, opts)

(optional)
@callback check_registration(type :: String.t(), opts :: keyword()) ::
  :ok | {:error, reason :: term()}

Whether this processor can serve the registration of the type string type with the registration's options opts ([] for a bare-module registration). Pure. Answers :ok, or {:error, reason} naming what is wrong, for example {:missing_option, :base_url}. A reason names what the registration cannot be served without, not only an option that is absent: Statifier.Send.BasicHTTP answers {:missing_option, :base_url} both for a registration with no :base_url and for one whose :base_url is present but not a string, and names no separate reason for a malformed value.

Statifier.Session.start_link/2 asks it at a fresh start (no :resume option), in the caller, before any session process is spawned, for every registration whose module exports it, in the order of the type strings. The first {:error, reason} refuses the start with {:error, {:send_types, {:invalid_registration, type, reason}}} as start_link/2's own return; no process is spawned, so no {:proc_lib, :crash} report is produced. A resume does not ask it: the _ioprocessors entries a persisted position carries stand (ADR-0075 decision 3). A check that raises, throws, exits or answers anything else refuses nothing: the start goes ahead as if no check had been asked. Optional: a processor that does not implement it is not asked.

A check runs in its calling process. At a fresh start that is the process calling Statifier.Session.start_link/2, before the session starts. Under Statifier.start_session/2, or any supervisor child spec, that process is the supervisor, so a check that ends it ends every session that supervisor holds. On Statifier.MachineState.refresh_ioprocessors/1 it is the process calling that function; on a live Statifier.Session.refresh_ioprocessors/1 it is the session's own process. A check must not exit, link or signal its calling process. A raise, a throw or an exit/1 inside the check is caught, but an exit signal is not: one the check sends to its own process with the reason :kill ends that process, and one with the reason :normal ends it unless it traps exits.

deliver(effect, event, ctx)

@callback deliver(
  effect :: Statifier.Effect.Send.t() | Statifier.Effect.SendDelayed.t(),
  event :: Statifier.Event.t(),
  ctx :: ctx()
) :: {:ok, [instruction()]}

Plans the hand-off of one registered-type send. Pure. effect is the send the core produced and event the event built from it. For a delayed send the processor owns the delay.

ioprocessors_entry(type)

(optional)
@callback ioprocessors_entry(type :: String.t()) :: map()

The value of this processor's _ioprocessors entry (spec 5.10) for the registered type string type - for example a "location" a receiver can address this processor by. Pure and deterministic: the library calls it whenever it builds a registered set, which a replay and a resume do again. The map must be string-keyed at every level, as every datamodel value is. Optional: a processor that does not implement it gets an empty map.

ioprocessors_entry(type, context)

(optional)
@callback ioprocessors_entry(type :: String.t(), context :: entry_context()) :: map()

The value of this processor's _ioprocessors entry for the registered type string type in the session context names (ADR-0075 decision 3): for example a "location" built from the session id and a base URL in the registration's options. Called once, when the session starts, and asked instead of Statifier.Send.Processor.ioprocessors_entry/1 when a processor implements both. Pure and deterministic, and string-keyed at every level, as Statifier.Send.Processor.ioprocessors_entry/1 is. A processor that cannot build its entry from context raises ArgumentError. A processor that also implements Statifier.Send.Processor.check_registration/2 refuses such a registration at a fresh start before this callback is asked, and the raise is then left to a direct caller. Optional.

perform(payload, ctx)

(optional)
@callback perform(payload :: term(), ctx :: ctx()) :: :ok | {:error, term()}

Performs one {:handler, module, payload} instruction a planning callback returned - the impure half. MAY be called more than once for the same send (see the moduledoc's point 2). Optional: a processor whose planning callbacks return no such instruction needs none.