hecate_pubsub_registry (macula v11.4.0)
View SourcePer-identity registry for hecate_pubsub_server processes.
Holds a RealmTag => pid() map and acts as the dispatch hub for inbound SUBSCRIBE / UNSUBSCRIBE / EVENT frames. The registry spawn-links one hecate_pubsub_server worker per realm and stores its pid. A linked worker that crashes delivers an `EXIT'' message which the registry traps + uses to clear the entry; a later register/3 or SUBSCRIBE yields a fresh server.
A realm's server lives while something holds it
A realm gets a server in two ways. register/3 materialises it for the station's own use, such as a realm the station publishes on; such a realm is pinned and stays until the registry stops. With a default_identity, a SUBSCRIBE for a realm without a server materialises one; that realm lives while a subscription holds it, and when an UNSUBSCRIBE or purge_subscriber/2 takes its last one, its server stops and its place frees. Nothing else starts a server: an UNSUBSCRIBE or EVENT for a realm without one gets {ok, []}, and relay_publish/3 builds its EVENT without one.
At most max_subscribed_realms realms (1000 by default) are materialised by SUBSCRIBE at once. A SUBSCRIBE that would take the registry past that gets {error, too_many_realms} and starts no server. Pinned realms do not count, so register/3 is for realms the station itself chooses, never for a realm a peer names.
Sprint A invariant
Realm tags are opaque 32-byte namespace keys. The registry does NOT validate authenticity — multi-tenancy is structural (one server per tag, no cross-realm leakage). Realm authority lives outside the station per PLAN_DEFERRED_WORK §6.
Multi-identity (PLAN_MULTI_IDENTITY_RELAY §Phase 2)
N identities run inside one BEAM. Each identity has its OWN pubsub_registry, owning its OWN per-realm pubsub_server pool. No cross-identity leakage — a realm tag X under identity A is a different overlay than the same realm tag X under identity B.
Phase 2 also folded the previous hecate_pubsub_server_sup (simple_one_for_one pool) into the registry: the registry spawn-links pubsub_servers itself. Equivalent semantics — they are temporary, the registry's monitor was already doing the bookkeeping that the supervisor would have — minus a module + the pid-passing coordination that splitting them required under per-identity supervision.
Summary
Functions
Route a SUBSCRIBE / UNSUBSCRIBE / EVENT frame for Realm to the matching pubsub_server. Returns the matched local subscribers (empty list for SUBSCRIBE / UNSUBSCRIBE).
Snapshot the realm tags currently materialised under RegistryPid. Used by status pages + tests.
Find the pubsub_server pid for Realm under RegistryPid, or report not_found.
Remove Sub (a subscriber pubkey) from every topic under every realm currently materialised on this registry — i.e. from every live hecate_pubsub_server it owns. A dead server pid (raced against its own EXIT cleanup, see handle_info/2) is skipped rather than treated as an error; the registry's own by_realm/by_pid bookkeeping self-heals on the pending EXIT. A realm a SUBSCRIBE materialised whose last subscription the purge took has its server stopped.
Idempotently start a pubsub_server for Realm under RegistryPid, signing with the key Identity returns. If a live server already exists, returns its pid; otherwise spawns a new one and records the mapping. A stale entry pointing at a dead pid is replaced transparently.
Relay an inbound PUBLISH frame for Realm to the matching pubsub_server. The server verifies the publication, builds an EVENT frame from its bytes and returns it together with the local subscribers that should receive it. The caller is responsible for sending EventFrame on each subscriber's peering connection.
Stop the registry. Uses reason shutdown (not the default normal) so the registry's spawn-linked pubsub_servers receive the exit signal and terminate alongside it. Without this, normal exit does not propagate to non-trapping linked workers.
Types
-type identity() :: fun(() -> macula_node_keys:node_key()).
-type opts() :: #{identity => identity(), identity_key => term(), max_subscribed_realms => pos_integer()}.
-type realm() :: <<_:256>>.
Functions
-spec dispatch_frame(pid(), realm(), <<_:256>>, macula_frame:frame()) -> {ok, [<<_:256>>]} | {error, not_found | too_many_realms}.
Route a SUBSCRIBE / UNSUBSCRIBE / EVENT frame for Realm to the matching pubsub_server. Returns the matched local subscribers (empty list for SUBSCRIBE / UNSUBSCRIBE).
For a realm without a server: when no default_identity was configured at start-up, every frame gets {error, not_found}. With one (the production path under macula_station_identity_sup), a SUBSCRIBE materialises the realm and is dispatched against the fresh server, or gets {error, too_many_realms} when SUBSCRIBE frames already hold max_subscribed_realms realms; an UNSUBSCRIBE or EVENT gets {ok, []} and starts nothing.
An UNSUBSCRIBE that takes the last subscription of a realm a SUBSCRIBE materialised stops that realm's server.
Snapshot the realm tags currently materialised under RegistryPid. Used by status pages + tests.
Find the pubsub_server pid for Realm under RegistryPid, or report not_found.
-spec purge_subscriber(pid(), <<_:256>>) -> ok.
Remove Sub (a subscriber pubkey) from every topic under every realm currently materialised on this registry — i.e. from every live hecate_pubsub_server it owns. A dead server pid (raced against its own EXIT cleanup, see handle_info/2) is skipped rather than treated as an error; the registry's own by_realm/by_pid bookkeeping self-heals on the pending EXIT. A realm a SUBSCRIBE materialised whose last subscription the purge took has its server stopped.
Intended caller: the station's peer/daemon connection-lifecycle path, once a NodeId is confirmed to have no remaining connection. Sub has no notion of "which realm" it was subscribed under, so this fans out to all of them rather than requiring the caller to know.
Idempotently start a pubsub_server for Realm under RegistryPid, signing with the key Identity returns. If a live server already exists, returns its pid; otherwise spawns a new one and records the mapping. A stale entry pointing at a dead pid is replaced transparently.
The realm is pinned: it stays when its last subscription leaves, and it does not count towards max_subscribed_realms. It stays pinned when its server stops, so the server the next register/3 or SUBSCRIBE starts for it is pinned too. Register only a realm the station itself chooses, such as one it publishes on, never a realm a peer names, or peers could grow the registry past its maximum. An identity given as the key itself, in place of a function that returns it, is refused as {error, {identity, not_a_loader}} before the registry is called.
-spec relay_publish(pid(), realm(), macula_frame:frame()) -> {ok, macula_frame:frame(), [<<_:256>>]} | {error, term()}.
Relay an inbound PUBLISH frame for Realm to the matching pubsub_server. The server verifies the publication, builds an EVENT frame from its bytes and returns it together with the local subscribers that should receive it. The caller is responsible for sending EventFrame on each subscriber's peering connection.
Returns {ok, EventFrame, [Subs]} on success, the publication's refusal when it does not verify, {error, realm_mismatch} when it names another realm, and {error, not_found} when no server is registered for the realm AND no default_identity was configured at start-up.
For a realm without a server and a default_identity set (the production path under macula_station_identity_sup), the registry verifies the publication under that identity's profile and still builds the EVENT from the same publication bytes, delivered direct, with no local subscribers. An EVENT carries no signature of its own, so the publisher's signature goes end to end. This keeps the EVENT available for publisher-side bloom-fan forwarding to peer stations that have the topic in their Bloom filter but no subscribe-on-peer chain terminating at us, and starts no process for a realm nobody here subscribes to. Tests that omit default_identity retain the strict {error, not_found} semantics.
-spec stop(pid()) -> ok.
Stop the registry. Uses reason shutdown (not the default normal) so the registry's spawn-linked pubsub_servers receive the exit signal and terminate alongside it. Without this, normal exit does not propagate to non-trapping linked workers.