feat(dashboard): /telemetry backfills recent events from bridge ring buffer

TelemetryFeedStore was per-connection only — fresh /telemetry mounts started
empty because events fired while the user was on other pages never had a
subscriber on cf:telemetry. Bridge now writes every telemetry-feed event to
a 500-entry ETS ring buffer inside its existing Task.Supervisor child, and
the store replays that buffer on mount through the same stale-aware
ingest_event path before live PubSub takes over.

Author: fahchen

dashboard/lib/coloured_flow_dashboard/telemetry_bridge.ex

@@ -60,6 +60,15 @@ defmodule ColouredFlowDashboard.TelemetryBridge do
   cache on miss, so the first row construction for an unseen enactment
   takes the storage hit; subsequent rows are served from ETS.
 
+  ## Events buffer
+
+  The bridge also owns a second ETS table (`:set`, `:public`, `:named_table`)
+  keyed by telemetry-feed `seq` and valued by the `%Event{topic: :telemetry}`
+  emitted for that sequence. It acts as a small ring buffer so fresh
+  `/telemetry` mounts can backfill recent events before continuing with live
+  PubSub updates. Inserts and trimming run inside the existing task body so the
+  runner never pays that maintenance cost inline.
+
   ## Catalog drift
 
   The list returned by `events/0` is the bridge's authoritative event catalog.
@@ -84,7 +93,9 @@ defmodule ColouredFlowDashboard.TelemetryBridge do
   @default_pubsub :coloured_flow_dashboard_pubsub
   @default_task_supervisor ColouredFlowDashboard.TaskSupervisor
   @default_flow_cache :coloured_flow_dashboard_telemetry_bridge_flow_cache
+  @default_events_buffer :coloured_flow_dashboard_telemetry_bridge_events_buffer
   @default_topic_prefix "cf:"
+  @window 500
 
   @enactment_lifecycle_events ~w[start stop terminate exception take_snapshot]a
   @workitem_operations ~w[produce_workitems start_workitems withdraw_workitems complete_workitems]a
@@ -99,6 +110,7 @@ defmodule ColouredFlowDashboard.TelemetryBridge do
           | {:task_supervisor, atom()}
           | {:topic_prefix, String.t()}
           | {:flow_cache, atom()}
+          | {:events_buffer, atom()}
           | {:broadcast_fn, broadcast_fn()}
 
   @spec start_link([option()]) :: GenServer.on_start()
@@ -136,19 +148,22 @@ defmodule ColouredFlowDashboard.TelemetryBridge do
     task_supervisor = Keyword.get(opts, :task_supervisor, @default_task_supervisor)
     topic_prefix = Keyword.get(opts, :topic_prefix, @default_topic_prefix)
     flow_cache = Keyword.get(opts, :flow_cache, @default_flow_cache)
+    events_buffer = Keyword.get(opts, :events_buffer, @default_events_buffer)
     broadcast_fn = Keyword.get(opts, :broadcast_fn, &Phoenix.PubSub.broadcast/3)
 
     # GenServer owns the ETS table so it dies with us (test cleanup is free).
     # `:public` lets the Task.Supervisor child write to it without going
     # through the GenServer.
     flow_cache = ensure_table(flow_cache)
+    events_buffer = ensure_table(events_buffer)
 
     config = %{
       handler_id: handler_id,
       pubsub: pubsub,
       task_supervisor: task_supervisor,
       topic_prefix: topic_prefix,
       flow_cache: flow_cache,
+      events_buffer: events_buffer,
       broadcast_fn: broadcast_fn
     }
 
@@ -159,7 +174,7 @@ defmodule ColouredFlowDashboard.TelemetryBridge do
 
     :ok = :telemetry.attach_many(handler_id, events(), &__MODULE__.handle_event/4, config)
 
-    {:ok, %{handler_id: handler_id, flow_cache: flow_cache}}
+    {:ok, %{handler_id: handler_id, flow_cache: flow_cache, events_buffer: events_buffer}}
   end
 
   defp ensure_table(name) do
@@ -246,8 +261,12 @@ defmodule ColouredFlowDashboard.TelemetryBridge do
 
   defp dispatch(event_name, measurements, metadata, seq, config) do
     case build_broadcasts(event_name, measurements, metadata, seq, config) do
-      [] -> :ok
-      broadcasts -> deliver_broadcasts(config, broadcasts)
+      [] ->
+        :ok
+
+      broadcasts ->
+        buffer_recent_event(config.events_buffer, broadcasts)
+        deliver_broadcasts(config, broadcasts)
     end
   end
 
@@ -257,6 +276,33 @@ defmodule ColouredFlowDashboard.TelemetryBridge do
     end)
   end
 
+  defp buffer_recent_event(events_buffer, broadcasts) when is_atom(events_buffer) do
+    case Enum.find_value(broadcasts, fn
+           {_topic, %Event{topic: :telemetry, seq: seq} = event}
+           when is_integer(seq) and seq > 0 ->
+             event
+
+           _other ->
+             nil
+         end) do
+      %Event{} = event ->
+        :ets.insert(events_buffer, {event.seq, event})
+        trim_events_buffer(events_buffer, event.seq)
+
+      nil ->
+        :ok
+    end
+  end
+
+  defp trim_events_buffer(events_buffer, newest_seq) when newest_seq > @window do
+    cutoff = newest_seq - @window
+    spec = [{{:"$1", :_}, [{:<, :"$1", cutoff}], [true]}]
+    :ets.select_delete(events_buffer, spec)
+    :ok
+  end
+
+  defp trim_events_buffer(_events_buffer, _newest_seq), do: :ok
+
   @spec build_broadcasts(
           :telemetry.event_name(),
           :telemetry.event_measurements(),
@@ -377,6 +423,32 @@ defmodule ColouredFlowDashboard.TelemetryBridge do
 
   defp enactments_topic(_prefix, _kind, _common), do: []
 
+  @doc """
+  Returns buffered telemetry-feed events sorted by ascending sequence.
+
+  Returns `[]` when the table does not exist.
+  """
+  @spec recent_events() :: [Event.t()]
+  def recent_events, do: recent_events(@default_events_buffer)
+
+  @doc """
+  Returns buffered telemetry-feed events sorted by ascending sequence from the
+  given ETS table.
+  """
+  @spec recent_events(atom()) :: [Event.t()]
+  def recent_events(events_buffer) when is_atom(events_buffer) do
+    case :ets.whereis(events_buffer) do
+      :undefined ->
+        []
+
+      _table ->
+        events_buffer
+        |> :ets.tab2list()
+        |> Enum.map(fn {_seq, %Event{} = event} -> event end)
+        |> Enum.sort_by(& &1.seq)
+    end
+  end
+
   @doc """
   Looks up (or resolves and caches) the flow topic id for an enactment.
 

dashboard/lib/coloured_flow_dashboard_web/stores/telemetry_feed_store.ex

@@ -30,6 +30,11 @@ defmodule ColouredFlowDashboardWeb.Stores.TelemetryFeedStore do
   Per-enactment seq is still honoured via `SeqTracker` so a late event
   cannot leapfrog a newer one for the same enactment id.
 
+  Mount subscribes first, then backfills from the bridge-owned ETS events
+  buffer through the same `ingest_event/2` path used for live PubSub events.
+  That keeps ordering, stale-drop, and window trimming identical across the
+  replay and live phases.
+
   Read-only — no commands. The operator inspects the feed; submitting
   outputs and other state mutation happens on the per-enactment surfaces.
   """
@@ -46,6 +51,7 @@ defmodule ColouredFlowDashboardWeb.Stores.TelemetryFeedStore do
   @default_pubsub :coloured_flow_dashboard_pubsub
   @default_topic "cf:telemetry"
   @default_flow_cache :coloured_flow_dashboard_telemetry_bridge_flow_cache
+  @default_events_buffer :coloured_flow_dashboard_telemetry_bridge_events_buffer
   @window 500
 
   state do
@@ -64,6 +70,7 @@ defmodule ColouredFlowDashboardWeb.Stores.TelemetryFeedStore do
     pubsub = Map.get(params, "pubsub_name", @default_pubsub)
     topic = Map.get(params, "topic", @default_topic)
     flow_cache = Map.get(params, "flow_cache", @default_flow_cache)
+    events_buffer = Map.get(params, "events_buffer", @default_events_buffer)
     window = Map.get(params, "window", @window)
 
     :ok = Phoenix.PubSub.subscribe(pubsub, topic)
@@ -73,6 +80,7 @@ defmodule ColouredFlowDashboardWeb.Stores.TelemetryFeedStore do
       |> assign(:pubsub_name, pubsub)
       |> assign(:topic, topic)
       |> assign(:flow_cache, flow_cache)
+      |> assign(:events_buffer, events_buffer)
       |> assign(:window, window)
       |> assign(:total_events, 0)
       |> assign(:entries_in_window, 0)
@@ -88,6 +96,17 @@ defmodule ColouredFlowDashboardWeb.Stores.TelemetryFeedStore do
       |> assign(:entries_index, [])
       |> stream(:entries, [], reset: true)
 
+    socket =
+      Enum.reduce(TelemetryBridge.recent_events(events_buffer), socket, fn event, acc ->
+        if SeqTracker.stale?(event, acc.assigns.last_seq) do
+          acc
+        else
+          acc
+          |> assign(:last_seq, SeqTracker.bump(acc.assigns.last_seq, event))
+          |> ingest_event(event)
+        end
+      end)
+
     {:ok, socket}
   end
 

dashboard/test/coloured_flow_dashboard/telemetry_bridge_test.exs

@@ -19,6 +19,16 @@ defmodule ColouredFlowDashboard.TelemetryBridgeTest do
 
   @pubsub :coloured_flow_dashboard_pubsub
   @task_supervisor ColouredFlowDashboard.TaskSupervisor
+  @bridge_table_names [
+    {:cf_dashboard_bridge_test_flow_cache_1, :cf_dashboard_bridge_test_events_buffer_1,
+     :cf_dashboard_bridge_test_bridge_1},
+    {:cf_dashboard_bridge_test_flow_cache_2, :cf_dashboard_bridge_test_events_buffer_2,
+     :cf_dashboard_bridge_test_bridge_2},
+    {:cf_dashboard_bridge_test_flow_cache_3, :cf_dashboard_bridge_test_events_buffer_3,
+     :cf_dashboard_bridge_test_bridge_3},
+    {:cf_dashboard_bridge_test_flow_cache_4, :cf_dashboard_bridge_test_events_buffer_4,
+     :cf_dashboard_bridge_test_bridge_4}
+  ]
 
   defp unique_token, do: Integer.to_string(System.unique_integer([:positive, :monotonic]))
 
@@ -56,10 +66,7 @@ defmodule ColouredFlowDashboard.TelemetryBridgeTest do
     token = unique_token()
     prefix = "cf-#{context.test_token}-#{token}:"
     handler_id = {context.module, context.test, token}
-    # credo:disable-for-next-line Credo.Check.Warning.UnsafeToAtom
-    flow_cache = String.to_atom("#{context.module}.FlowCache.#{context.test_token}.#{token}")
-    # credo:disable-for-next-line Credo.Check.Warning.UnsafeToAtom
-    bridge_name = String.to_atom("#{context.module}.Bridge.#{context.test_token}.#{token}")
+    {flow_cache, events_buffer, bridge_name} = next_bridge_names()
 
     bridge_opts =
       [
@@ -68,7 +75,8 @@ defmodule ColouredFlowDashboard.TelemetryBridgeTest do
         pubsub: @pubsub,
         task_supervisor: @task_supervisor,
         topic_prefix: prefix,
-        flow_cache: flow_cache
+        flow_cache: flow_cache,
+        events_buffer: events_buffer
       ] ++ opts
 
     # Explicit `:id` so multiple bridges with different names can coexist
@@ -77,7 +85,13 @@ defmodule ColouredFlowDashboard.TelemetryBridgeTest do
     spec = Supervisor.child_spec({TelemetryBridge, bridge_opts}, id: bridge_name)
     pid = start_supervised!(spec)
 
-    %{prefix: prefix, handler_id: handler_id, flow_cache: flow_cache, bridge: pid}
+    %{
+      prefix: prefix,
+      handler_id: handler_id,
+      flow_cache: flow_cache,
+      events_buffer: events_buffer,
+      bridge: pid
+    }
   end
 
   defp subscribe_topics(prefix, enactment_id) do
@@ -325,6 +339,83 @@ defmodule ColouredFlowDashboard.TelemetryBridgeTest do
     end
   end
 
+  describe "events buffer" do
+    setup context, do: start_bridge(context)
+
+    test "records each broadcast event and is readable via recent_events/1", %{
+      prefix: prefix,
+      handler_id: handler_id,
+      events_buffer: events_buffer
+    } do
+      eid_a = unique_id("-buffer-a")
+      eid_b = unique_id("-buffer-b")
+      :ok = Phoenix.PubSub.subscribe(@pubsub, "#{prefix}telemetry")
+
+      execute_via_handler(
+        handler_id,
+        [:coloured_flow, :runner, :enactment, :start],
+        %{system_time: System.system_time()},
+        %{enactment_id: eid_a, enactment_state: build_state(eid_a)}
+      )
+
+      execute_via_handler(
+        handler_id,
+        [:coloured_flow, :runner, :enactment, :start],
+        %{system_time: System.system_time()},
+        %{enactment_id: eid_b, enactment_state: build_state(eid_b)}
+      )
+
+      assert_receive {:cf_event, %Event{enactment_id: ^eid_a, topic: :telemetry}}, 1_000
+      assert_receive {:cf_event, %Event{enactment_id: ^eid_b, topic: :telemetry}}, 1_000
+
+      events = TelemetryBridge.recent_events(events_buffer)
+
+      assert [
+               %Event{enactment_id: ^eid_a, topic: :telemetry, seq: seq_a},
+               %Event{enactment_id: ^eid_b, topic: :telemetry, seq: seq_b}
+             ] = events
+
+      assert seq_a > 0
+      assert seq_b > seq_a
+    end
+
+    test "drops entries older than the window", %{
+      prefix: prefix,
+      handler_id: handler_id,
+      events_buffer: events_buffer
+    } do
+      total = 503
+      :ok = Phoenix.PubSub.subscribe(@pubsub, "#{prefix}telemetry")
+
+      for n <- 1..total do
+        eid = unique_id("-trim-#{n}")
+
+        execute_via_handler(
+          handler_id,
+          [:coloured_flow, :runner, :enactment, :start],
+          %{system_time: System.system_time()},
+          %{enactment_id: eid, enactment_state: build_state(eid)}
+        )
+      end
+
+      received_seqs =
+        for _idx <- 1..total do
+          assert_receive {:cf_event, %Event{topic: :telemetry, seq: seq}}, 1_000
+          seq
+        end
+
+      events = TelemetryBridge.recent_events(events_buffer)
+      seqs = Enum.map(events, & &1.seq)
+      newest_seq = Enum.max(received_seqs)
+      expected = received_seqs |> Enum.filter(&(&1 >= newest_seq - 500)) |> Enum.sort()
+      oldest_received_seq = received_seqs |> Enum.sort() |> hd()
+
+      assert seqs == expected
+      assert Enum.all?(seqs, &(&1 >= newest_seq - 500))
+      refute oldest_received_seq in seqs
+    end
+  end
+
   describe "topic isolation" do
     test "parallel bridges on different prefixes do not cross-pollute", context do
       %{prefix: prefix_a, handler_id: handler_a} = start_bridge(context)
@@ -454,7 +545,9 @@ defmodule ColouredFlowDashboard.TelemetryBridgeTest do
 
       max_pre = Enum.max(pre_seqs)
       assert max_pre > 0
-      assert pre_seqs == Enum.sort(pre_seqs), "pre-restart seqs must be monotonic"
+
+      assert Enum.sort(pre_seqs) == Enum.uniq(Enum.sort(pre_seqs)),
+             "pre-restart seqs must be unique and monotonic as values"
 
       # Kill the bridge and wait for the supervisor to surface the exit so
       # `start_bridge/2` below opens a fresh ETS / new GenServer state.
@@ -483,7 +576,9 @@ defmodule ColouredFlowDashboard.TelemetryBridgeTest do
         end
 
       min_post = Enum.min(post_seqs)
-      assert post_seqs == Enum.sort(post_seqs), "post-restart seqs must be monotonic"
+
+      assert Enum.sort(post_seqs) == Enum.uniq(Enum.sort(post_seqs)),
+             "post-restart seqs must be unique and monotonic as values"
 
       assert min_post > max_pre,
              "post-restart seqs must STRICTLY exceed pre-restart max " <>
@@ -590,4 +685,10 @@ defmodule ColouredFlowDashboard.TelemetryBridgeTest do
     do: InMemory.enactment(record, :id)
 
   defp runner_enactment_id(%{id: id}) when is_binary(id), do: id
+
+  defp next_bridge_names do
+    index = Process.get(:bridge_names_index, 0)
+    Process.put(:bridge_names_index, index + 1)
+    Enum.fetch!(@bridge_table_names, index)
+  end
 end