hecate_pubsub_server (macula v11.5.0)
View SourcePubSub gen_server wrapping hecate_pubsub state for one realm namespace.
Activates the dormant hecate_pubsub pure-state module by giving it a process identity. One server instance owns the topic-to-subscriber index for a single realm tag; the realm is an opaque 32-byte namespace key, not validated against any authority. Multi-tenancy comes from running multiple servers under different realm tags — the station does not arbitrate which realm tags are "real" (Sprint A: realm identity lives outside infrastructure).
Phase 1 scope (this commit)
State mutations + frame processing only. The publish path signs a PUBLISH with the server's node identity key, builds its EVENT and returns the matched LOCAL subscribers, but does NOT fan out across the cluster — that requires the Plumtree wire layer (hecate_plumtree) and the DHT topic-discovery integration which land in subsequent commits.
Sequencing
- This commit: server in isolation, no integration with station listener or DHT.
- Next: per-realm-namespace registry under
hecate_overlay_supso the listener can route inbound SUBSCRIBE / UNSUBSCRIBE / EVENT frames to the right server. - Then: Plumtree fan-out for cross-station delivery.
- Then: DHT integration for topic-mesh discovery.
Summary
Functions
Process an inbound EVENT frame received from the wire. Its publication is verified first; one that does not verify matches no one. Returns the matched local subscribers; the caller delivers.
Run an identity loader and check the key it returns, as a pubsub server requires of its identity and a registry of its default identity. A loader that raises is refused as {identity, loader_failed}, as a pool's is, and the error it raised, which can hold the key, goes nowhere. A key that is not an identity key in the node's configured crypto profile is refused by name. Returns the profile and the key.
Generic frame dispatch — handles subscribe / unsubscribe / event uniformly. Returns the matched subscribers for event frames, empty list for subscribe / unsubscribe.
Sign a PUBLISH for Topic/Payload with the server's node identity key, build its EVENT, and return the EVENT together with the set of LOCAL subscribers that match. The caller is responsible for handing the frame to the cross-station delivery layer (Plumtree, future commit) and for delivering to the matched local subscribers via the application channel.
Remove Sub from every topic this server holds, dropping any topic that empties out as a result. See hecate_pubsub:purge_subscriber/2.
Relay an inbound PUBLISH frame from a remote daemon. The server verifies its publication once, under the server's profile, and builds the EVENT from the same publication bytes, so the publisher's signature goes end to end and no hop re-signs it (D17). Returns the EVENT and the matched local subscribers. The caller (typically the peer observer) is responsible for sending EventFrame on each subscriber's peering connection.
Start the server for a realm. A server whose identity does not return an identity key in the node's configured crypto profile does not start, so one node never runs two profiles, and one given the key itself in place of a function that returns it does not start either, with {error, {identity, not_a_loader}}.
Types
-type opts() :: #{realm := <<_:256>>, identity := fun(() -> macula_node_keys:node_key())}.
Functions
-spec deliver_event(pid(), macula_frame:frame()) -> [<<_:256>>].
Process an inbound EVENT frame received from the wire. Its publication is verified first; one that does not verify matches no one. Returns the matched local subscribers; the caller delivers.
-spec identity_loaded(fun(() -> macula_node_keys:node_key())) -> {ok, macula_crypto_profile:profile(), macula_node_keys:node_key()} | {error, term()}.
Run an identity loader and check the key it returns, as a pubsub server requires of its identity and a registry of its default identity. A loader that raises is refused as {identity, loader_failed}, as a pool's is, and the error it raised, which can hold the key, goes nowhere. A key that is not an identity key in the node's configured crypto profile is refused by name. Returns the profile and the key.
-spec process_frame(pid(), <<_:256>>, macula_frame:frame()) -> [<<_:256>>].
Generic frame dispatch — handles subscribe / unsubscribe / event uniformly. Returns the matched subscribers for event frames, empty list for subscribe / unsubscribe.
-spec publish(pid(), binary(), binary()) -> {macula_frame:frame(), [<<_:256>>]}.
Sign a PUBLISH for Topic/Payload with the server's node identity key, build its EVENT, and return the EVENT together with the set of LOCAL subscribers that match. The caller is responsible for handing the frame to the cross-station delivery layer (Plumtree, future commit) and for delivering to the matched local subscribers via the application channel.
-spec purge_subscriber(pid(), <<_:256>>) -> ok.
Remove Sub from every topic this server holds, dropping any topic that empties out as a result. See hecate_pubsub:purge_subscriber/2.
-spec realm(pid()) -> <<_:256>>.
-spec relay_publish(pid(), macula_frame:frame()) -> {macula_frame:frame(), [<<_:256>>]} | {error, term()}.
Relay an inbound PUBLISH frame from a remote daemon. The server verifies its publication once, under the server's profile, and builds the EVENT from the same publication bytes, so the publisher's signature goes end to end and no hop re-signs it (D17). Returns the EVENT and the matched local subscribers. The caller (typically the peer observer) is responsible for sending EventFrame on each subscriber's peering connection.
Returns the publication's refusal when it does not verify, and {error, realm_mismatch} when its realm is not this server's (the registry routes by realm, so that is a defensive check).
Start the server for a realm. A server whose identity does not return an identity key in the node's configured crypto profile does not start, so one node never runs two profiles, and one given the key itself in place of a function that returns it does not start either, with {error, {identity, not_a_loader}}.
-spec stop(pid()) -> ok.
-spec subscriber_count(pid()) -> non_neg_integer().
-spec topic_count(pid()) -> non_neg_integer().