feat: add group lookup to admin muster dashboard (#2062)

Author: edgurgel

forum/README.md

@@ -82,6 +82,7 @@ Forum.Muster.join(scope, group, pid)        # :ok | {:error, :rpc_failed | :not_
 Forum.Muster.leave(scope, group, pid)       # :ok | {:error, term}
 Forum.Muster.router(scope, group)           # {:ok, node} | {:rebalancing, [node]}
 Forum.Muster.targets(scope, group, sender_view_hash)  # (on router) {:ok, [node]} | {:error, :flood}
+Forum.Muster.occupancy(scope, group)        # (on router) raw held [node], NOT barrier-gated
 Forum.Muster.members(scope)                 # [node]
 Forum.Muster.status(scope)                   # :ready | :converging | :rebalancing | :unknown
 Forum.Muster.local_members(scope, group)    # [pid]

forum/lib/forum/muster.ex

@@ -355,6 +355,19 @@ defmodule Forum.Muster do
     end
   end
 
+  @doc """
+  Returns the source nodes this node's router-role occupancy table currently
+  holds for `group` in `scope` — the raw table content.
+
+  Unlike `targets/3` this is **not** barrier-gated: it does not consult
+  `can_decide?/2`, so it reflects what this node believes about `group` even when
+  a broadcast would flood. Intended for inspection/debugging
+  """
+  @spec occupancy(atom, group) :: [node]
+  def occupancy(scope, group) when is_atom(scope) do
+    Scope.occupancy(scope, group)
+  end
+
   defp ring_name(scope), do: :"#{scope}_muster_ring"
 
   @doc "List local pids registered to `group` in `scope`."

forum/test/forum/muster_test.exs

@@ -462,6 +462,13 @@ defmodule Forum.MusterTest do
       assert :src@nowhere in Scope.occupancy(scope, :rg1)
     end
 
+    test "Muster.occupancy/2 exposes the raw router-held rows", %{scope: scope} do
+      assert :ok = Scope.occupied(scope, :rg_pub, :src@nowhere, 1, fake_pid())
+      # Public, non-barrier-gated read; delegates to Scope.occupancy/2.
+      assert Muster.occupancy(scope, :rg_pub) == Scope.occupancy(scope, :rg_pub)
+      assert :src@nowhere in Muster.occupancy(scope, :rg_pub)
+    end
+
     test "vacant_batch/4 deletes multiple {group, source_node} rows", %{scope: scope} do
       src = fake_pid()
       :ok = Scope.occupied(scope, :rg2a, :src@nowhere, 1, src)

lib/realtime_web/dashboard/muster.ex

@@ -8,9 +8,18 @@ defmodule RealtimeWeb.Dashboard.Muster do
   node's health at a glance. Use the node picker (top-right) to switch nodes;
   the page reads the selected node from `page.node` and RPCs it for the snapshot
   (the LiveView itself always runs on the serving node).
+
+  It also offers a **group lookup**: given a group id, it resolves the group's
+  router on the inspected node (`Forum.Muster.router/2`), RPCs that router for
+  the barrier-gated fan-out targets (`Forum.Muster.targets/3`), and shows both
+  the inspected node's local member count and the raw occupancy the router holds
+  for the group. This makes it easy to see, for one group, who would receive a
+  broadcast and why (targets vs. flood).
   """
   use Phoenix.LiveDashboard.PageBuilder
 
+  alias Realtime.GenRpc
+
   # Lifecycle status -> Bootstrap contextual color.
   @status_colors %{
     ready: "success",
@@ -26,19 +35,48 @@ defmodule RealtimeWeb.Dashboard.Muster do
 
   @impl true
   def mount(_params, _session, socket) do
-    {:ok, assign(socket, node_data: node_data(socket.assigns.page.node))}
+    {:ok, assign(socket, group_query: nil) |> refresh_data()}
   end
 
   @impl true
   def handle_event("refresh", _params, socket) do
-    {:noreply, assign(socket, node_data: node_data(socket.assigns.page.node))}
+    {:noreply, refresh_data(socket)}
+  end
+
+  @impl true
+  def handle_event("lookup_group", %{"group" => group}, socket) do
+    query = normalize_group(group)
+    {:noreply, socket |> assign(group_query: query) |> refresh_data()}
+  end
+
+  @impl true
+  def handle_event("clear_group", _params, socket) do
+    {:noreply, socket |> assign(group_query: nil) |> refresh_data()}
   end
 
   # Invoked by LiveDashboard's automatic refresh timer (enabled by default). Without
   # this, the timer would tick and re-render but never re-fetch the snapshot.
   @impl true
   def handle_refresh(socket) do
-    {:noreply, assign(socket, node_data: node_data(socket.assigns.page.node))}
+    {:noreply, refresh_data(socket)}
+  end
+
+  # Re-fetches the node snapshot and (if a group is queried) re-runs the group
+  # lookup, so both stay live across the manual and automatic refresh paths.
+  defp refresh_data(socket) do
+    node = socket.assigns.page.node
+
+    assign(socket,
+      node_data: node_data(node),
+      group_data: group_data(node, socket.assigns.group_query)
+    )
+  end
+
+  defp normalize_group(group) do
+    case String.trim(group) do
+      "" -> nil
+      trimmed -> trimmed
+    end
   end
 
   @impl true
@@ -102,12 +140,95 @@ defmodule RealtimeWeb.Dashboard.Muster do
           </div>
         <% end %>
       </div>
+
+      <div class="card mb-3">
+        <div class="card-header"><strong>Group lookup</strong></div>
+        <div class="card-body">
+          <form phx-submit="lookup_group" class="d-flex gap-2 align-items-center mb-0">
+            <input
+              type="text"
+              name="group"
+              value={@group_query}
+              placeholder="group id (e.g. tenant id)"
+              class="form-control form-control-sm flex-grow-1"
+              autocomplete="off"
+            />
+            <button type="submit" class="btn btn-sm btn-primary text-nowrap">Look up</button>
+            <%= if @group_query do %>
+              <button type="button" phx-click="clear_group" class="btn btn-sm btn-outline-secondary text-nowrap">Clear</button>
+            <% end %>
+          </form>
+
+          <%= if @group_data do %>
+            <%= if @group_data.error do %>
+              <div class="text-danger small mt-3"><%= @group_data.error %></div>
+            <% else %>
+              <table class="table table-sm table-hover mb-0 mt-3" style="table-layout: fixed">
+                <colgroup>
+                  <col style="width: 16rem" />
+                  <col />
+                </colgroup>
+                <tbody>
+                  <tr><td class="ps-3">Group</td><td class="text-break"><code><%= inspect(@group_data.group) %></code></td></tr>
+                  <tr>
+                    <td class="ps-3">Inspected node</td>
+                    <td class="text-break"><%= @group_data.node %></td>
+                  </tr>
+                  <tr>
+                    <td class="ps-3">Local members here</td>
+                    <td><%= @group_data.local_member_count %></td>
+                  </tr>
+                  <tr>
+                    <td class="ps-3">Sender view hash</td>
+                    <td class="text-break"><code><%= inspect(@group_data.view_hash) %></code></td>
+                  </tr>
+                  <tr>
+                    <td class="ps-3">Router (<code>router/2</code>)</td>
+                    <td class="text-break"><%= fmt_router(@group_data.router) %></td>
+                  </tr>
+                  <tr>
+                    <td class="ps-3">Targets via router (<code>targets/3</code>)</td>
+                    <td class="text-break"><%= fmt_targets(@group_data.targets) %></td>
+                  </tr>
+                  <tr>
+                    <td class="ps-3">Router occupancy (raw, count)</td>
+                    <td class="text-break"><%= fmt_occupancy(@group_data.router_occupancy) %></td>
+                  </tr>
+                </tbody>
+              </table>
+            <% end %>
+          <% end %>
+        </div>
+      </div>
     </div>
     """
   end
 
   defp status_color(status), do: Map.get(@status_colors, status, "secondary")
 
+  defp fmt_router({:ok, node}), do: to_string(node)
+  defp fmt_router({:rebalancing, nodes}), do: "rebalancing — fan out to #{inspect(nodes)}"
+
+  defp fmt_targets({:ok, nodes}),
+    do: "#{inspect(Enum.sort(nodes))} (#{length(nodes)} node#{plural(nodes)})"
+
+  defp fmt_targets({:error, :flood}),
+    do: "flood — barrier not satisfied (converging, or sender/router view mismatch)"
+
+  defp fmt_targets({:rebalancing, nodes}),
+    do: "rebalancing — no single router; fan out to #{inspect(nodes)}"
+
+  defp fmt_targets({:error, :rpc_error, reason}), do: "router RPC failed: #{inspect(reason)}"
+
+  defp fmt_occupancy(nodes) when is_list(nodes),
+    do: "#{inspect(Enum.sort(nodes))} (#{length(nodes)})"
+
+  defp fmt_occupancy({:rebalancing, _nodes}), do: "—"
+  defp fmt_occupancy({:error, :rpc_error, reason}), do: "router RPC failed: #{inspect(reason)}"
+
+  defp plural([_]), do: ""
+  defp plural(_), do: "s"
+
   # The page LiveView always runs on the serving node; `page.node` is the node
   # chosen in the picker (a URL param). So we gather locally only when that node
   # IS us, and RPC to gather_local_info/0 on the selected node otherwise.
@@ -152,6 +273,99 @@ defmodule RealtimeWeb.Dashboard.Muster do
     end
   end
 
+  # Group lookup: reads the inspected node's local routing state, then (on the
+  # serving node) asks that state's router node for its targets/occupancy — the
+  # two hops kept flat rather than nested. nil group => no lookup.
+  defp group_data(_target, nil), do: nil
+
+  defp group_data(target, group) do
+    base = %{group: group, node: target, error: nil}
+
+    # Two independent hops, orchestrated here on the serving node so neither RPC
+    # is nested inside the other: first read the inspected node's local routing
+    # state, then ask that state's router node for its targets/occupancy.
+    case GenRpc.call(target, __MODULE__, :gather_local_group_info, [group], []) do
+      {:ok, local} ->
+        {targets, occupancy} = router_view(local.scope, group, local.view_hash, local.router)
+
+        Map.merge(base, %{
+          local_member_count: local.local_member_count,
+          view_hash: local.view_hash,
+          router: local.router,
+          targets: targets,
+          router_occupancy: occupancy
+        })
+
+      {:error, :rpc_error, reason} ->
+        Map.put(base, :error, "RPC failed: #{inspect(reason)}")
+
+      {:error, msg} ->
+        Map.put(base, :error, msg)
+    end
+  end
+
+  @doc """
+  Reads the inspected node's local routing state for `group`: the configured
+  scope, this node's cluster-view hash (the sender hash used for `targets/3`),
+  the group's router as this node's ring sees it (`Forum.Muster.router/2`), and
+  the local member count here.
+
+  Runs on the node being inspected so `router/2` and the view hash reflect that
+  node. `group_data/2` then asks the returned router node for `targets/3` and its
+  raw occupancy, keeping the two hops flat rather than nesting one RPC in another.
+
+  Returns `{:error, message}` when no scope is configured or the coordinator has
+  not published its state yet, so the page can surface a friendly error.
+  """
+  def gather_local_group_info(group) do
+    case Application.get_env(:realtime, :muster_scope) do
+      nil ->
+        {:error, "No Muster scope configured on this node"}
+
+      scope ->
+        try do
+          {:ok,
+           %{
+             scope: scope,
+             view_hash: Forum.Muster.view_hash(scope),
+             router: Forum.Muster.router(scope, group),
+             local_member_count: Forum.Muster.local_member_count(scope, group)
+           }}
+        rescue
+          ArgumentError ->
+            {:error, "Muster scope #{inspect(scope)} has no published state on this node yet"}
+        catch
+          :exit, reason -> {:error, "Muster group lookup failed: #{inspect(reason)}"}
+        end
+    end
+  end
+
+  # With a settled ring, ask the router node for its targets/3 and raw occupancy
+  # in a single RPC; while rebalancing there is no single router, so surface the
+  # fan-out list. GenRpc.call handles router_node == node() itself; on RPC failure
+  # both slots carry the {:error, :rpc_error, _} for the fmt_* helpers to render.
+  defp router_view(scope, group, view_hash, {:ok, router_node}) do
+    case GenRpc.call(router_node, __MODULE__, :router_group_info, [scope, group, view_hash], []) do
+      {:error, :rpc_error, _} = error -> {error, error}
+      {targets, occupancy} -> {targets, occupancy}
+    end
+  end
+
+  defp router_view(_scope, _group, _view_hash, {:rebalancing, nodes} = rebalancing) do
+    {rebalancing, {:rebalancing, nodes}}
+  end
+
+  @doc """
+  Reads the router-role view of `group` on this (router) node in one hop: the
+  barrier-gated fan-out targets (`Forum.Muster.targets/3`) and the raw occupancy
+  the router currently holds (`Forum.Muster.Scope.occupancy/2`). Returned as
+  `{targets_result, occupancy}` so `group_data/2` fetches both with a single RPC
+  to the router.
+  """
+  def router_group_info(scope, group, view_hash) do
+    {Forum.Muster.targets(scope, group, view_hash), Forum.Muster.occupancy(scope, group)}
+  end
+
   defp to_view(s) do
     counts = s.group_state_counts
     group_counts = Enum.map(@group_state_order, fn state -> {state, Map.get(counts, state, 0)} end)