macula_pubsub_order (macula v9.1.1)
View SourcePer-subscription delivery ordering for pubsub events.
A publisher stamps every fact with a pool-monotonic seq (macula_client), contiguous within a publisher's lifetime (a refused publish does not burn a number). The mesh sends copies down several links at once, so a subscriber receives seqs out of order and macula_client dedups to the FIRST arrival — which scrambles a single publisher's stream. This module puts the deduped stream back into per-publisher order, offering three delivery modes a subscriber picks at subscribe time:
- ordered (default) — per-publisher FIFO by seq. Out-of-order arrivals are buffered and released in order; a genuinely missing seq is skipped after a timeout (see
flush/3), trading a bounded delay for the lost fact. - latest_only — deliver only if the seq exceeds the highest already delivered for that publisher (drop stale). No buffering, no head-of-line delay. For state-snapshot consumers that want freshness over completeness.
- as_arrives — deliver immediately in arrival order (the pre-1.x behaviour). Zero added latency; the consumer handles ordering itself.
seq re-bases to wall-clock microseconds when a publisher's pool restarts, so a large forward jump (> ?EPOCH_JUMP) is read as a restart: the old expected counter is abandoned rather than waited on.
Pure and side-effect-free: it returns the events to deliver now, and the caller does the sending. flush/3 is driven by the caller on a timer to release buffers whose gap has timed out.
Summary
Functions
Total events currently held in reorder buffers (introspection).
Release buffers whose head has waited past TimeoutMs for a missing seq: skip the gap up to the smallest buffered seq and drain. Returns the events to deliver now and the updated state (skip counter advanced by one per gap given up on). A no-op for non-ordered modes.
Offer an arrived (deduped) event. Returns the events to deliver now, in order, and the updated state. NowMs timestamps buffered arrivals for flush/3.
Types
-type event() :: term().
-type mode() :: ordered | latest_only | as_arrives.
-type seq() :: non_neg_integer().
-opaque t()
Functions
-spec buffered(t()) -> non_neg_integer().
Total events currently held in reorder buffers (introspection).
-spec flush(t(), integer(), non_neg_integer()) -> {[event()], t()}.
Release buffers whose head has waited past TimeoutMs for a missing seq: skip the gap up to the smallest buffered seq and drain. Returns the events to deliver now and the updated state (skip counter advanced by one per gap given up on). A no-op for non-ordered modes.
-spec new(mode(), pos_integer()) -> t().
Offer an arrived (deduped) event. Returns the events to deliver now, in order, and the updated state. NowMs timestamps buffered arrivals for flush/3.
-spec skips(t()) -> non_neg_integer().
Count of gaps skipped after timeout since new/1 (telemetry: the genuine per-publisher loss rate).