macula_pubsub_order (macula v15.3.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. The seqs a skip gave up on are remembered (bounded per publisher): a copy that arrives afterwards is delivered late, wrapped {late, Event}, and counted -- it is no longer mistaken for a duplicate. Only true duplicates are dropped. 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 macula_realm_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 the station directory 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). The skipped seqs are remembered (bounded, see remember_skips/4) so a copy that arrives later can be delivered late. A no-op for non-ordered modes.

How long a publisher with nothing buffered may stay silent before it is dropped, when a new publisher arrives.

Count of late copies delivered since new/1: a fact whose gap was skipped and whose copy then arrived (telemetry: reorders a skip would previously have misread as loss).

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; a copy of a seq a skip gave up on is returned wrapped {late, Event}.

Count of arrivals dropped as already-past since new/1 -- true duplicates, not late copies of skipped gaps (those are delivered and counted by late_delivered/1).

How many publishers the state holds ordering state for (introspection).

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). The skipped seqs are remembered (bounded, see remember_skips/4) so a copy that arrives later can be delivered late. A no-op for non-ordered modes.

idle_prune_ms()

-spec idle_prune_ms() -> pos_integer().

How long a publisher with nothing buffered may stay silent before it is dropped, when a new publisher arrives.

late_delivered(_)

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

Count of late copies delivered since new/1: a fact whose gap was skipped and whose copy then arrived (telemetry: reorders a skip would previously have misread as loss).

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; a copy of a seq a skip gave up on is returned wrapped {late, Event}.

past_dropped(_)

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

Count of arrivals dropped as already-past since new/1 -- true duplicates, not late copies of skipped gaps (those are delivered and counted by late_delivered/1).

publishers(_)

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

How many publishers the state holds ordering state for (introspection).

skips(_)

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

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