macula_pubsub_order (macula v10.21.0)
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. A large BACKWARD jump is read the same way. A publisher that restarts with a counter re-seeded from zero instead of wall-clock (a macula-station's own hecate_pubsub_server before 10.17.0) would otherwise have every fact after the restart dropped as "past" until the counter climbed back over the old watermark -- silently, with the link, the wire subscription and dedup all healthy. That is how hecate-stations went deaf for 10+ hours after a fleet rollout on 2026-09-02. A backstep within the threshold is still a late duplicate.
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).