evoq_consistency (evoq v1.30.1)
View SourceConsistency modes for command dispatch.
Supports: - eventual: Return immediately after events are persisted - strong: Wait for every event handler and projection subscribed to the appended events to have written them - {handlers, [atom()]}: Wait for those handlers only
Uses pg for handler acknowledgment tracking.
Strong Consistency Flow
1. The router OPENS a waiter for the stream, before the append (open/2): the caller's pid joins a pg group keyed by store and stream, so an ack can never arrive before anyone is listening. 2. The command is dispatched and its events are appended. 3. The router derives the required handlers from the appended events and the event type registry (required/1), which is the same lookup the event router delivers by. 4. Each handler and projection calls acknowledge/4 after its write, carrying the event's stream version. 5. await/4 returns when every required handler has acked a version at or past the one the dispatch produced, or {error, timeout}.
Acks that arrive before the wait (a fast handler, an append that delivered before the router got back) are in the caller's mailbox by the time await/4 runs, so they count. The version rides in the ack, so a handler that was already past the stream answers for it, and one that is behind does not.
What an ack says, and what a drop says
An ack is sent only after the handler's own write. A handler whose error policy dropped the event instead (skip or {dead_letter, _} from on_error/4) sends dropped/2, and the wait answers {error, {dropped, Handler}} at once, because that write will never happen. A projection that deliberately skips an event it does not want ({skip, State, ReadModel} from project/4) is not a drop: it decided, and it acks.
A handler on another node is not in the required set. The registry is a cluster-wide pg scope, so it names pids on every connected node, while the router delivers only to local ones.
Every waiter on a stream's group receives every ack for that stream, so close/1 takes the ones this wait did not consume out of the mailbox.
What strong does not wait for
A process manager instance. Which instances handle an event is decided at delivery time (correlate/2, which may start one), so the set a caller could wait for does not exist when it dispatches. A process manager that needs its own writes seen dispatches and waits itself.
A handler that dispatches must not wait for its own event's handlers
A cascade written as an evoq_event_handler (a module with interested_in/0, which the event type registry does hold, unlike a process manager instance) is itself in the required set of the dispatch that triggered it. Dispatching from its handle_event/4 with strong would wait for the read models of the events it just appended while holding the event it has not finished, and one of those read models may be the projection still handling that very event: nothing can end the wait.
The calling process is therefore never in its own required set (await/4 drops self()): a dispatcher cannot wait for an ack it owes, and it still waits for every other handler. What that does NOT fix is a cascade waiting for a projection that is still busy with the event that ran it, which no library can see: a cascade dispatches eventual, it does not read its own write, and the entry-point dispatch that triggered it is the one that asked for the guarantee.
Summary
Functions
Acknowledge a handler's or projection's write from the event it was handed.
Acknowledge that a handler has processed an event.
Wait until every required handler has acked the version IT was required for.
Leave the group a waiter joined. Safe to call twice.
Say that a handler did NOT write this event: its own error policy dropped it (skip or {dead_letter, _} from on_error/4).
Open a waiter for one stream, before its command is appended.
The handlers and projections subscribed to these appended events, each with the LAST version of the append it is required for.
Start the pg scope for consistency tracking.
Wait for exactly these handlers to ack this version.
Is this registry entry one this node can wait for?
Types
Functions
Acknowledge a handler's or projection's write from the event it was handed.
The routing metadata carries the store, the stream and the event's own stream version, so a handler that just wrote needs nothing else. Metadata without them (an event routed outside a store subscription) is not acknowledged: there is no stream for a waiter to be waiting on.
Acknowledge that a handler has processed an event.
Called by the handler and projection processes themselves, after their write, with the event's own stream version. Module and the calling pid are both offered as identities, so a waiter can name either.
-spec await(waiter(), [{pid() | atom(), integer()}], integer(), map()) -> ok | {error, timeout | {dropped, pid() | atom()}}.
Wait until every required handler has acked the version IT was required for.
Required pairs each handler (a pid or a module) with the last version of the append it was handed, as required/2 answers for a derived set. Waits for handlers and projections; a timeout answers as an error, never as a silent success. An empty required list has nothing to wait for and answers at once.
An ack means the handler wrote its effect. A handler whose own error policy dropped an event instead (skip, {dead_letter, _} from on_error/4) says so with dropped/2 rather than acknowledging, and this answers {error, {dropped, Handler}} at once: the write did not happen, and waiting out the deadline for it would say the same thing later and less clearly.
WHICH EVENTS ARE THIS DISPATCH'S is the append's own range of versions, {first_version, V} in the options (the caller that appended knows it; it is Version by default, the single event case). A drop is this call's business only inside it: a stream carries every command's appends, so a drop above it is the next command's (and a handler's messages arrive in order, so it can only be one) and a drop below it was answered for before this append began. Inside it, a drop fails the wait even when the same handler went on to write a later version of the append, because that is exactly the event the read model is missing.
-spec close(waiter()) -> ok.
Leave the group a waiter joined. Safe to call twice.
Every waiter in a stream's group receives every ack for that stream, so a caller that dispatches gets acks for handlers it never waited on, and a concurrent dispatcher's ack can arrive here too. await/4 takes out only what it needs; the rest are taken out here, so a caller that dispatches in a loop does not accumulate them in its mailbox. An ack already in flight when the process leaves the group can still land after close/1 has run: a caller that keeps its mailbox clean against that uses a catch-all handle_info.
Say that a handler did NOT write this event: its own error policy dropped it (skip or {dead_letter, _} from on_error/4).
The counterpart of acknowledge/2. It carries the same routing metadata and is deliberately not an ack: a waiter told ok about an event a handler dropped was told a write happened that did not.
Open a waiter for one stream, before its command is appended.
The calling process joins the pg group acks are published to, so an ack for this stream reaches it whatever happens between the append and the wait. Keyed by store and stream, never by version: the version is not known until the append, and the ack carries its own.
The handlers and projections subscribed to these appended events, each with the LAST version of the append it is required for.
Taken from the appended events and the event type registry, which is the set the event router delivers to: no list is kept anywhere. An event type no handler subscribes to contributes nothing, which is the truth.
FirstVersion is the version the append's first event got, so the events are handed FirstVersion, FirstVersion + 1, ... in the order they were appended (the router derives it the same way when it computes the range).
A handler answers for the version it was HANDED, not for the append's last one: a handler subscribed to one of two appended events writes that event and nothing else, so holding it to the second version is waiting for a write it was never given, and every multi-event dispatch would end at the deadline. A handler subscribed to both is required at the later one.
-spec start_pg() -> ok.
Start the pg scope for consistency tracking.
-spec wait_for(atom(), binary(), integer(), map()) -> ok | {error, timeout} | {error, all_needs_events}.
Wait for exactly these handlers to ack this version.
Kept for callers that name their handlers. all is REFUSED here rather than answered at once: this function sees no events, so it cannot know which handlers the dispatch would have to wait for, and ok would say "they have all written" about a set it never looked up. A dispatch asks for strong instead, which derives the set from the appended events.
It sees no events, so it also does not know the append's range of versions: only a drop of exactly Version fails this wait, and a drop of one of the append's earlier versions is not noticed. A caller that appended several events and wants the whole range calls open/2 and await/4 itself, with {first_version, V}.
Is this registry entry one this node can wait for?
The event type registry is a CLUSTER-WIDE pg scope, so get_handlers/1 names members of every connected node, while the router delivers only to local pids and skips a remote one ("belongs to another node"). Waiting for a remote pid is waiting for an ack that pid cannot send for this store: it is not subscribed here and never sees the event, so every dispatch would end in a consistency timeout. The required set is the delivered set, so a pid on another node is not in it.
An atom is not something the registry names: register_handler/2 refuses a module ({error, not_supported}), and the router's own module clause calls handle_event/4 without acknowledging anything. The clause answers true because a module the registry cannot hold never reaches here, and a caller that names one in {handlers, ...} is naming a handler with no process to deliver to; refusing it here would be a different, larger change.