Macula SDK — Pub/Sub Guide

Topic-based messaging over the relay mesh.

Audience: Erlang/OTP applications publishing or subscribing to events. Since v3.11.0 (V2 surface). For the V1 (pre-3.11.0) single-connection surface, see ../migrations/V1_TO_V2_PUBSUB.md.

TL;DR

%% 1. Connect to the mesh — see CONNECTING_GUIDE.md.
{ok, Pool} = macula:connect(Seeds, #{}).

%% 2. Subscribe.
Topic = macula_topic:app_fact(Realm, Org, App,
                              <<"orders">>, <<"placed">>, 1),
{ok, Sub} = macula:subscribe(Pool, Realm, Topic, self()).

%% 3. Receive events.
receive
    {macula_event, Sub, Topic, Payload, Meta} ->
        handle(Topic, Payload, Meta)
end.

%% 4. Publish.
ok = macula:publish(Pool, Realm, Topic,
                    #{order_id => <<"ord-123">>, total => 4999}).

%% 5. Drop the sub.
ok = macula:unsubscribe(Pool, Sub).

Three core ideas

1. Realm-per-call

Every publish and every subscribe carries an explicit 32-byte realm tag. There is no connect-time default realm. Realms are how the mesh isolates traffic — a subscriber on Realm A never receives an event published to Realm B, even if the topic strings match exactly.

%% Same topic, different realms — totally separate streams.
ok = macula:publish(Pool, RealmA, Topic, PayloadA),
ok = macula:publish(Pool, RealmB, Topic, PayloadB).

Realms are 32-byte binaries. Use macula_realm:id/1 (SHA-256 of the human-readable realm name) or your own realm registry to derive them.

2. Topics describe event types, not entities

Non-negotiable for scalable pub/sub.

Topic   = WHAT happened (event type, immutable)
Payload = WHO/WHERE/WHEN it happened (entity details)

Wrong (topic explosion):

%% DO NOT DO THIS — entity ID baked into topic
Topic = macula_topic:app_fact(Realm, Org, App,
                              <<"weather">>,
                              <<"manchester_main_street_wind_measured">>, 1),
macula:publish(Pool, Realm, Topic, #{speed => 42.5}).

Right (IDs in payload):

Topic = macula_topic:app_fact(Realm, Org, App,
                              <<"weather">>, <<"wind_measured">>, 1),
macula:publish(Pool, Realm, Topic, #{
    station_id => <<"manchester-main-street">>,
    speed => 42.5,
    unit => <<"km/h">>,
    timestamp => erlang:system_time(millisecond)
}).

3. Topics are built — never hand-typed

%% Always
Topic = macula_topic:app_fact(Realm, Org, App, Domain, Name, Version),

%% Never
Topic = <<"my.realm/my.org/my.app/orders/placed_v1">>.

macula_topic returns a canonical binary that the SDK and stations agree on. Hand-rolled strings are rejected.

See TOPIC_NAMING_GUIDE.md for the canonical specification of the five-segment shape and the realm_fact / org_fact / app_fact tier choice.

Subscribing

{ok, SubRef} = macula:subscribe(Pool, Realm, Topic, Subscriber).

The subscriber receives:

{macula_event, SubRef, Topic, Payload, Meta}

Meta is a map carrying delivery context:

{publisher, seq} is the dedup key. The pool guarantees you see each (Realm, Publisher, Seq) tuple at most once, even when the same EVENT arrives via multiple links (e.g. with replication_factor > 1).

When the subscription ends

The only way a live subscription produces a terminal message today is the pool closing:

{macula_event_gone, SubRef, pool_closed}

A link dying does not send this — the pool logs _macula.client.link_down, schedules a respawn, and silently re-issues the subscription against the new link once it's up (see Connecting Guide). A subscriber sees no gap-signaling message for that case, only a possible gap in delivery itself, which ordered mode's order_timeout_ms skip handles the same way it handles any other loss.

After event_gone arrives, no further events come for that SubRef.

Subscribing with options — delivery ordering

{ok, SubRef} = macula:subscribe(Pool, Realm, Topic, Subscriber, Opts).

The delivery option chooses how a single publisher's stream is ordered on the way to your subscriber. This matters because the mesh sends copies of a fact down several links at once, and the pool dedups to the first arrival — so without ordering, a single publisher's seq 1, 2, 3 can reach you as 1, 3, 2. Each publisher's seq is pool-monotonic and contiguous, which is exactly what makes ordered delivery possible.

%% default — per-publisher FIFO
{ok, R1} = macula:subscribe(Pool, Realm, Topic, self()),

%% newest-wins, drop stale
{ok, R2} = macula:subscribe(Pool, Realm, Topic, self(),
                            #{delivery => latest_only}),

%% raw arrival order (the pre-8.8 behaviour)
{ok, R3} = macula:subscribe(Pool, Realm, Topic, self(),
                            #{delivery => as_arrives}).

Ordered mode and loss. ordered trades a bounded delay for order: if seq 2 never arrives, the buffer holds 3, 4, … only until order_timeout_ms elapses, then skips the gap and releases them. That skip is the accepted "order-not-guaranteed delivery" trade for a lost fact — a reorder buffer cannot invent a message the mesh dropped. Design mesh facts to be idempotent and version-stamped so an occasional skip washes out.

Total order is not offered, by design. ordered restores a single publisher's order (cheap, over ordered transport). It does not impose a total order across different publishers — that would need a single sequencer (a consensus log) that the mesh deliberately does not have. Cross-publisher order is not something a decentralised broadcast can give you; carry a version or timestamp in the fact if a consumer needs to relate two publishers' events.

Pool-level tuning (connect/2 options)

Telemetry — is loss real?

macula:status/1 reports pubsub_gap_skips: the number of per-publisher gaps given up on after the timeout, i.e. the genuine loss rate an ordered subscriber could not fill. A near-zero value means the mesh is delivering and ordered costs you almost nothing; a rising value is the signal to look at delivery, not ordering.

{ok, #{pubsub_gap_skips := Skips}} = macula:status(Pool).

(publisher, seq) and dedup

{publisher, seq} is also the dedup key. The pool sees each (Realm, Publisher, Seq) tuple at most once, even when the same EVENT arrives via multiple links (e.g. replication_factor > 1). In ordered and latest_only modes the delivery layer additionally uses the seq to order or drop; in as_arrives the dedup layer is the only filter.

Subscribing in a callback module

A common pattern: a gen_server subscribes in init/1, handles events in handle_info/2.

-module(my_orders_listener).
-behaviour(gen_server).

init(_Args) ->
    Pool  = my_app_mesh:pool(),
    Realm = my_app_mesh:realm(),
    Topic = macula_topic:app_fact(Realm, my_org, my_app,
                                  <<"orders">>, <<"placed">>, 1),
    {ok, Sub} = macula:subscribe(Pool, Realm, Topic, self()),
    {ok, #{sub => Sub}}.

handle_info({macula_event, Sub, _Topic, Payload, _Meta},
            #{sub := Sub} = S) ->
    on_order_placed(Payload),
    {noreply, S};
handle_info({macula_event_gone, Sub, Reason},
            #{sub := Sub} = S) ->
    %% Pool went away — supervisor will restart us.
    {stop, {pool_gone, Reason}, S}.

Pattern-match the Sub reference into the function head — that keeps a process subscribing to multiple topics readable.

Publishing

ok = macula:publish(Pool, Realm, Topic, Payload).

Returns:

Partial success counts as success. With replication_factor > 1, publish/4 returns ok as soon as the first selected link accepts the frame. Subsequent links are best-effort.

Publishing with options

ok = macula:publish(Pool, Realm, Topic, Payload, #{timeout_ms => 1000}).

Delivery guarantees

• At-most-once — fire and forget. No publisher-visible ack from subscribers.
• Per-publisher delivery order — ordered by default at the subscriber (see Subscribing with options above): out-of-order arrivals are buffered and released in seq order, with a genuinely missing seq skipped after order_timeout_ms. The mesh itself does not guarantee arrival order — a relay spreads one publisher's burst across concurrent verify workers, and a receiver may admit an event by more than one path — the subscriber-side ordered buffer is what turns that into in-order delivery. Opt into as_arrives if you'd rather see raw arrival order and reorder yourself.
• Cross-publisher ordering — none, by design. Two publishers' events arrive in arbitrary interleaving; see "Total order is not offered, by design" above.
• Cross-link dedup — the pool dedupes by (Realm, Publisher, Seq) over a 60-second window (configurable; see dedup_window_ms in CONNECTING_GUIDE.md).
• Cross-station gossip — default since 4.5.0. A daemon connected to station A and a daemon connected to station B see each other's publishes once subscription interest and the fact itself have gossiped between the stations; publisher-end-to-end signatures plus (publisher, seq) dedup at each hop is what makes this safe past one hop.

Unsubscribing

ok = macula:unsubscribe(Pool, SubRef).

Idempotent — unknown SubRef is a no-op. The subscriber pid does not receive a event_gone message for an explicit unsubscribe — event_gone is reserved for involuntary termination.

The wire-level subscription against the link persists for the pool's lifetime. One wire sub per (Realm, Topic) is multiplexed across local consumers; the pool drops the topic from its index when the last local consumer leaves, but does not currently send UNSUBSCRIBE on the wire (Phase 4 will tighten).

If the subscriber pid dies before calling unsubscribe/2, the pool detects the 'DOWN' and drops the sub spec automatically.

Topic naming reference

Quick reference. Full specification: TOPIC_NAMING_GUIDE.md.

Every topic is exactly five slash-separated segments:

{realm}/{publisher_org}/{publisher_app}/{domain}/{name}_v{N}

Pick a tier based on who owns the topic:

Past tense for facts (order_placed, wind_measured, user_registered). Present tense for hopes (order_place, payment_authorize).

System topics (_mesh.*, _macula.*) are infrastructure-owned and dot-separated. Do not publish to them from app code.

Patterns

Re-subscribe after pool restart

If your supervisor restarts the pool, your subscribers are not automatically re-attached. Either:

• Restart your subscriber processes alongside the pool (one supervisor with rest_for_one strategy), or
• Watch for {macula_event_gone, _, pool_closed} and re-subscribe to the new pool.

Multiple subscribers on one pool

A single pool can have arbitrarily many local subscribers. The pool issues exactly one wire-level SUBSCRIBE per (Realm, Topic), multiplexes inbound events to every local subscriber for that pair, and dedupes across links.

Backpressure

Events are delivered as Erlang messages. If a subscriber is slow, its mailbox grows. The pool itself never blocks. Apply your usual mailbox-flow-control patterns (process throttling, batching, etc.).

Idempotent handlers

Even though the pool dedupes by (Realm, Publisher, Seq), network weirdness across long restarts can theoretically allow a duplicate sneak through after the dedup window expires. Make handlers idempotent — match on a payload-level key (order id, sensor id + timestamp, etc.) when correctness matters.

Best practices

1. IDs in payloads, not topics. Always.
2. Past-tense facts, present-tense hopes. No CRUD verbs (created, updated, deleted).
3. Build topics via macula_topic — never inline strings.
4. Include a timestamp — erlang:system_time(millisecond) in every payload.
5. Pattern-match the SubRef in handlers when listening to multiple topics.
6. Keep handlers fast — spawn workers for heavy processing.
7. Make handlers idempotent — (publisher, seq) is a strong dedup key, but cross-restart edge cases exist; don't rely on exactly-once.

Diagnostics

These come through macula_diagnostics:event/2; wire them into your observability layer.

How it works (relay side)

Routing behind the relay (DHT-based subscriber discovery, peering, cross-station gossip, bloom filters, sticky-routing) is the relay's concern. See macula-station for the current relay implementation.

From the SDK side, you publish and subscribe; the pool handles the rest.

See also

• Connecting Guide — pool model, lifecycle, options
• Topic Naming Guide — canonical topic shape
• V1 → V2 Migration — what changed in 3.11.0
• Authorization Guide — UCAN/DID identity
• macula_pubsub — slice module
• macula_client — pool implementation