defmodule Vela.Distributed.Cluster do @moduledoc """ Monitors cluster membership and maintains a list of connected nodes. Subscribes to Erlang node events via :net_kernel.monitor_nodes/2. When nodes join or leave, broadcasts to all processes that joined the :vela_cluster_events pg group. This is a global singleton — one per BEAM node, shared by all caches. """ use GenServer @pg_scope :vela_pg @pg_group :vela_cluster_events # ---- Public API ---- def start_link(opts \\ []) do GenServer.start_link(__MODULE__, opts, name: __MODULE__) end @doc "Returns all currently connected nodes including the local node." def nodes do GenServer.call(__MODULE__, :nodes) end @doc """ Subscribe the calling process to cluster events. The process will receive: {:vela_cluster_event, :node_up, node} {:vela_cluster_event, :node_down, node} """ def subscribe do :pg.join(@pg_scope, @pg_group, self()) end @doc "Unsubscribe the calling process from cluster events." def unsubscribe do :pg.leave(@pg_scope, @pg_group, self()) end # ---- GenServer Callbacks ---- @impl true def init(_opts) do # Subscribe to node join/leave events from the BEAM :net_kernel.monitor_nodes(true, node_type: :visible) # Start with all currently connected nodes plus ourselves connected = [node() | Node.list()] {:ok, %{nodes: MapSet.new(connected)}} end @impl true def handle_call(:nodes, _from, state) do {:reply, MapSet.to_list(state.nodes), state} end @impl true def handle_info({:nodeup, new_node, _info}, state) do new_nodes = MapSet.put(state.nodes, new_node) :telemetry.execute( [:vela, :cache, :node, :up], %{}, %{node: new_node} ) broadcast(:node_up, new_node) {:noreply, %{state | nodes: new_nodes}} end @impl true def handle_info({:nodedown, lost_node, _info}, state) do new_nodes = MapSet.delete(state.nodes, lost_node) :telemetry.execute( [:vela, :cache, :node, :down], %{}, %{node: lost_node} ) broadcast(:node_down, lost_node) {:noreply, %{state | nodes: new_nodes}} end defp broadcast(event, node) do members = :pg.get_local_members(@pg_scope, @pg_group) Enum.each(members, &send(&1, {:vela_cluster_event, event, node})) end end