macula_pubsub_order (macula v11.1.0)

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. A publisher's first facts, and its first facts after a restart, are held for up to one timeout (or until the buffer cap), and its order starts at the lowest seq held, so a lower seq that arrives second is delivered rather than lost.
  • 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

Types

The key ordering state is kept under: the publisher, or {unverified, Publisher} for events whose publisher signature did not verify, so the two never share state.

t/0

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. A publisher whose first facts are still held starts its order at the lowest held seq instead, without a skip. 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. Pub is the publisher, or {unverified, Publisher} for an event whose publisher signature did not verify; each key has its own ordering state. In ordered mode a publisher's first facts are held until its first order timeout (see flush/3) or the buffer cap.

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

Types

event/0

-type event() :: term().

The key ordering state is kept under: the publisher, or {unverified, Publisher} for events whose publisher signature did not verify, so the two never share state.

mode/0

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

publisher_key/0

-type publisher_key() :: binary() | {unverified, binary()}.

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. A publisher whose first facts are still held starts its order at the lowest held seq instead, without a skip. 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(), publisher_key(), 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. Pub is the publisher, or {unverified, Publisher} for an event whose publisher signature did not verify; each key has its own ordering state. In ordered mode a publisher's first facts are held until its first order timeout (see flush/3) or the buffer cap.

skips(_)

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

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