macula_pubsub_order (macula v9.1.1)

View Source

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

Count of gaps skipped after timeout since new/1 (telemetry: the genuine per-publisher loss rate).

Types

event/0

-type event() :: term().

mode/0

-type mode() :: ordered | latest_only | as_arrives.

seq/0

-type seq() :: non_neg_integer().

t/0

-opaque t()

Functions

buffered(_)

-spec buffered(t()) -> non_neg_integer().

Total events currently held in reorder buffers (introspection).

flush(S, Now, Timeout)

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

new(Mode)

-spec new(mode()) -> t().

new(Mode, Max)

-spec new(mode(), pos_integer()) -> t().

offer(S, Pub, Seq, Ev, Now)

-spec offer(t(), binary(), seq(), event(), integer()) -> {[event()], 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.

skips(_)

-spec skips(t()) -> non_neg_integer().

Count of gaps skipped after timeout since new/1 (telemetry: the genuine per-publisher loss rate).