hecate_om_pubsub_subscriptions (hecate_om v0.23.0)

View Source

Reconciles the desired subscription set against hecate_om_pubsub_sup's actual running children (piece D, PLAN_HECATE_OM_MESH_WRAPPERS.md).

One supervised macula_subscriber per {Topic, HandlerMod, Args} not already running; any running one no longer desired is stopped. Same "declare it, reconcile on a timer" shape as hecate_om_capabilities's 30s republish -- and for the same reason: the desired set may be declared before the mesh is attached, so the tick also retries anything that couldn't start yet, and it's the backstop for the one scenario let-it-crash alone doesn't cover -- a full pool replacement (not just a link respawn) drops every subscription's Pool pid at once, and if the supervisor's own restart budget is exhausted before hecate_om_identity reconnects, nothing else would ever retry.

Reconnect handling that is deliberately NOT here: no macula_event_gone listening, no resubscribe-on-DOWN. Confirmed live (macula_link_respawn_replay_tests.erl) that an ordinary link respawn is invisible to a subscriber -- macula_client_replay already replays it one layer down. Building that in here would reintroduce the exact hand-rolled machinery this piece exists to make unnecessary.

Summary

Functions

The delta between what's desired and what's currently running: {ToStart, ToStop}. ToStart is the {Topic, HandlerMod, Args} triples whose Topic has no running child; ToStop is the running topics no longer in Desired.

Declare the desired subscription set. Safe to call repeatedly whenever it changes at runtime -- e.g. a federation_inbox-shaped service adding one topic per newly-registered entity -- diffs against what is currently running and touches only the delta.

Functions

diff(Desired, CurrentTopics)

-spec diff([{binary(), module(), term()}], [binary()]) -> {[{binary(), module(), term()}], [binary()]}.

The delta between what's desired and what's currently running: {ToStart, ToStop}. ToStart is the {Topic, HandlerMod, Args} triples whose Topic has no running child; ToStop is the running topics no longer in Desired.

ensure(Desired)

-spec ensure([{binary(), module(), term()}]) -> ok.

Declare the desired subscription set. Safe to call repeatedly whenever it changes at runtime -- e.g. a federation_inbox-shaped service adding one topic per newly-registered entity -- diffs against what is currently running and touches only the delta.

handle_call(Msg, From, S)

handle_cast(Msg, S)

handle_info(Other, State)

init(_)

start_link()

terminate(_, _)