hecate_pubsub (macula v10.11.0)
View SourceRealm-scoped PubSub state + dispatch (Part 6 §6).
Holds the topic-to-subscriber index for one realm and converts incoming SUBSCRIBE / UNSUBSCRIBE / EVENT frames into local state mutations + delivery instructions. The wire layer that actually transmits frames lives elsewhere — typically hecate_plumtree for intra-realm fan-out.
Pipeline
- Local subscribe —
subscribe/3adds the subscriber to the topic's set. The wrapper builds a SUBSCRIBE frame for upstream propagation if needed. - Local publish —
build_event/3takes a PUBLISH spec and produces a signed EVENT frame the wrapper hands to Plumtree for fan-out. The publisher signs the EVENT once; intermediate hops do NOT re-sign, so every subscriber can verify authenticity end-to-end (Part 6 §6.4). - Receive EVENT —
deliver_event/2returns the list of local subscribers whose subscription matches the event's topic + realm. The wrapper notifies each via the application channel. - Receive SUBSCRIBE / UNSUBSCRIBE —
process/3updates local state.
State is per realm: an instance handles one realm's topics only. Cross-realm leakage is impossible — the realm is baked into the state and every dispatch checks it.
Reference: plans/PLAN_MACULA_V2_PART6_PROTOCOL.md §6; plans/PLAN_PHASE_5_BREAKDOWN.md Session 5.5.
Summary
Functions
Match an incoming EVENT frame to local subscribers. Returns an empty list if the realm doesn't match (defensive — the transport should already route by realm) or no-one is subscribed.
Remove Sub from every topic in this realm, dropping any topic whose subscriber set becomes empty as a result — the same drop_or_keep/3 rule unsubscribe/3 applies to one topic, fanned out across all of them in one pass.
Types
-type state() :: #{realm := <<_:256>>, subscriptions := #{topic() => sets:set(subscriber())}}.
-type subscriber() :: macula_identity:pubkey().
-type topic() :: binary().
Functions
-spec build_event(state(), macula_frame:publish_spec(), macula_identity:key_pair()) -> macula_frame:frame().
-spec deliver_event(state(), macula_frame:frame()) -> [subscriber()].
Match an incoming EVENT frame to local subscribers. Returns an empty list if the realm doesn't match (defensive — the transport should already route by realm) or no-one is subscribed.
-spec is_subscribed(state(), topic(), subscriber()) -> boolean().
-spec new(<<_:256>>) -> state().
-spec process(state(), macula_identity:pubkey(), macula_frame:frame()) -> {state(), [subscriber()]}.
-spec purge_subscriber(state(), subscriber()) -> state().
Remove Sub from every topic in this realm, dropping any topic whose subscriber set becomes empty as a result — the same drop_or_keep/3 rule unsubscribe/3 applies to one topic, fanned out across all of them in one pass.
For a peer or daemon that disconnects without sending UNSUBSCRIBE for everything it held: without this, a topic whose only subscriber was that departed connection never empties, so it never leaves topics/1 — and macula_station_peering_router (which treats every entry in topics/1 as local interest worth re-subscribing on every peer, regardless of whether the original subscriber was a peer-sourced entry) keeps re-propagating it mesh-wide forever. See macula-station/plans/DESIGN_SUBSCRIPTION_LIFECYCLE_GC.md.
-spec subscribe(state(), topic(), subscriber()) -> state().
-spec subscriber_count(state()) -> non_neg_integer().
-spec subscribers(state(), topic()) -> [subscriber()].
-spec topic_count(state()) -> non_neg_integer().
-spec unsubscribe(state(), topic(), subscriber()) -> state().