fix(dashboard): start_enactment + retry_enactment atomic with runner boot

`:start_enactment` persisted the row then started the runner. On runner boot
failure the row stayed behind as a phantom enactment. Now compensates by
deleting the just-inserted row when start fails so the operator never sees a
running row without a live process. `:retry_enactment` had the inverse bug —
it flipped state to `:running` before the runner restart, so a failed retry
"recovered" an enactment that still had no process. Now restart first, only
commit `:running` after success.

Also bounded `TelemetryFeedStore.last_seq`: previously grew one entry per
enactment id ever seen, leaking memory in long-lived /telemetry tabs. Prune
the tracker entry when its enactment ages fully out of the 500-event window.

Author: fahchen

dashboard/lib/coloured_flow_dashboard_web/stores/enactment_detail_store.ex

@@ -108,11 +108,11 @@ defmodule ColouredFlowDashboardWeb.Stores.EnactmentDetailStore do
       GenServer (same hot path the runner uses internally after completions).
       Reply codes: `:ok | :not_running | :runner_error`.
     * `:retry_enactment` — combines two public main-repo surfaces to recover an
-      `:exception`-state enactment: `Storage.retry_enactment/2` flips the DB
-      row back to `:running` and writes a `:retried` log entry; then
-      `Runner.Enactment.Supervisor.start_enactment/1` brings the GenServer
-      back online (the supervisor returns `{:ok, pid}` for both fresh starts
-      and `:already_started`). Reply codes:
+      `:exception`-state enactment: `Runner.Enactment.Supervisor.start_enactment/1`
+      first brings the GenServer back online (the supervisor returns `{:ok, pid}`
+      for both fresh starts and `:already_started`); only after a successful
+      boot does `Storage.retry_enactment/2` flip the DB row back to `:running`
+      and write a `:retried` log entry. Reply codes:
       `:ok | :not_exception | :already_terminated | :runner_error`. The
       command rejects when the dashboard's current `summary.state` is not
       `:exception` so an operator can't toggle a healthy enactment off the
@@ -302,6 +302,7 @@ defmodule ColouredFlowDashboardWeb.Stores.EnactmentDetailStore do
     pubsub = Map.get(params, "pubsub_name", @default_pubsub)
     topic_prefix = Map.get(params, "topic_prefix", @default_topic_prefix)
     flow_cache = Map.get(params, "flow_cache", @default_flow_cache)
+    runner_start_fun = Map.get(params, "runner_start_fun", &Runner.start_enactment/1)
     topic = Map.get(params, "topic", "#{topic_prefix}enactment:#{enactment_id}")
 
     :ok = Phoenix.PubSub.subscribe(pubsub, topic)
@@ -351,6 +352,7 @@ defmodule ColouredFlowDashboardWeb.Stores.EnactmentDetailStore do
       |> assign(:pubsub_name, pubsub)
       |> assign(:topic, topic)
       |> assign(:flow_cache, flow_cache)
+      |> assign(:runner_start_fun, runner_start_fun)
       |> assign(:summary, summary)
       |> assign(:transitions, transitions)
       |> assign(:diagram, diagram)
@@ -454,7 +456,8 @@ defmodule ColouredFlowDashboardWeb.Stores.EnactmentDetailStore do
   end
 
   def handle_command(:retry_enactment, _payload, socket) do
-    {:reply, retry_enactment_reply(socket.assigns.enactment_id), socket}
+    {:reply, retry_enactment_reply(socket.assigns.enactment_id, socket.assigns.runner_start_fun),
+     socket}
   end
 
   def handle_command(:inspect_transition, payload, socket) when is_map(payload) do
@@ -1741,21 +1744,23 @@ defmodule ColouredFlowDashboardWeb.Stores.EnactmentDetailStore do
   # lag behind a concurrent force-terminate; trusting it would let an
   # already-`:terminated` row be resurrected back to `:running` by
   # `Storage.retry_enactment/2`. Only an actual `:exception` row is retried.
-  defp retry_enactment_reply(enactment_id) when is_binary(enactment_id) do
+  defp retry_enactment_reply(enactment_id, runner_start_fun) when is_binary(enactment_id) do
     case authoritative_enactment_state(enactment_id) do
-      {:ok, :exception} -> do_retry_enactment(enactment_id)
+      {:ok, :exception} -> do_retry_enactment(enactment_id, runner_start_fun)
       {:ok, :running} -> %{code: :not_exception}
       {:ok, :terminated} -> %{code: :already_terminated}
       {:error, reason} -> %{code: :runner_error, message: inspect(reason)}
     end
   end
 
-  defp do_retry_enactment(enactment_id) do
-    :ok = Storage.retry_enactment(enactment_id, message: "operator-triggered")
+  defp do_retry_enactment(enactment_id, runner_start_fun) do
+    case runner_start_fun.(enactment_id) do
+      {:ok, _pid} ->
+        :ok = Storage.retry_enactment(enactment_id, message: "operator-triggered")
+        %{code: :ok}
 
-    case Runner.start_enactment(enactment_id) do
-      {:ok, _pid} -> %{code: :ok}
-      {:error, reason} -> %{code: :runner_error, message: inspect(reason)}
+      {:error, reason} ->
+        %{code: :runner_error, message: inspect(reason)}
     end
   rescue
     error -> %{code: :runner_error, message: Exception.message(error)}

dashboard/lib/coloured_flow_dashboard_web/stores/flow_catalog_store.ex

@@ -157,6 +157,7 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStore do
   def mount(params, socket) when is_map(params) do
     pubsub = Map.get(params, "pubsub_name", @default_pubsub)
     topic = Map.get(params, "topic", @default_topic)
+    runner_start_fun = Map.get(params, "runner_start_fun", &Runner.start_enactment/1)
 
     :ok = Phoenix.PubSub.subscribe(pubsub, topic)
 
@@ -166,6 +167,7 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStore do
       socket
       |> assign(:pubsub_name, pubsub)
       |> assign(:topic, topic)
+      |> assign(:runner_start_fun, runner_start_fun)
       |> assign(:counts, counts)
       |> assign(:flow_ids, ids)
       |> assign(:last_seq, %{})
@@ -182,7 +184,7 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStore do
   @impl Musubi.Store
   def handle_command(:start_enactment, payload, socket) when is_map(payload) do
     flow_id = Map.get(payload, "flow_id") || Map.get(payload, :flow_id)
-    {:reply, start_enactment_reply(flow_id), socket}
+    {:reply, start_enactment_reply(flow_id, socket.assigns.runner_start_fun), socket}
   end
 
   def handle_command(:refresh_catalog, _payload, socket) do
@@ -555,12 +557,13 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStore do
   # :start_enactment command
   # ---------------------------------------------------------------------------
 
-  defp start_enactment_reply(nil), do: %{code: :unknown_flow, enactment_id: nil}
+  defp start_enactment_reply(nil, _runner_start_fun),
+    do: %{code: :unknown_flow, enactment_id: nil}
 
-  defp start_enactment_reply(flow_id) when is_binary(flow_id) do
+  defp start_enactment_reply(flow_id, runner_start_fun) when is_binary(flow_id) do
     with {:ok, %{name: name} = flow} <- fetch_flow(flow_id),
          {:ok, seed} <- fetch_seed(name),
-         {:ok, enactment_id} <- insert_and_start(flow, seed) do
+         {:ok, enactment_id} <- insert_and_start(flow, seed, runner_start_fun) do
       %{code: :ok, enactment_id: enactment_id}
     else
       {:error, :unknown_flow} ->
@@ -577,7 +580,8 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStore do
     end
   end
 
-  defp start_enactment_reply(_other), do: %{code: :unknown_flow, enactment_id: nil}
+  defp start_enactment_reply(_other, _runner_start_fun),
+    do: %{code: :unknown_flow, enactment_id: nil}
 
   defp fetch_flow(flow_id) do
     case Storage.__storage__() do
@@ -618,10 +622,16 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStore do
     end
   end
 
-  defp insert_and_start(flow, seed) do
-    with {:ok, enactment_id} <- insert_enactment(flow, seed.initial_markings),
-         {:ok, _pid} <- start_runner(enactment_id) do
-      {:ok, enactment_id}
+  defp insert_and_start(flow, seed, runner_start_fun) do
+    with {:ok, enactment_id} <- insert_enactment(flow, seed.initial_markings) do
+      case start_runner(enactment_id, runner_start_fun) do
+        {:ok, _pid} ->
+          {:ok, enactment_id}
+
+        {:error, _reason} = error ->
+          _rollback = rollback_enactment_insert(flow, enactment_id)
+          error
+      end
     end
   end
 
@@ -643,8 +653,8 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStore do
     kind, reason -> {:error, {:storage, Exception.format(kind, reason)}}
   end
 
-  defp start_runner(enactment_id) do
-    case Runner.start_enactment(enactment_id) do
+  defp start_runner(enactment_id, runner_start_fun) do
+    case runner_start_fun.(enactment_id) do
       {:ok, pid} when is_pid(pid) -> {:ok, pid}
       {:ok, pid, _info} when is_pid(pid) -> {:ok, pid}
       :ignore -> {:error, {:runner, ":ignore"}}
@@ -656,6 +666,28 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStore do
     kind, reason -> {:error, {:runner, Exception.format(kind, reason)}}
   end
 
+  defp rollback_enactment_insert(%{in_memory_record: _record}, enactment_id) do
+    case GenServer.whereis(InMemory) do
+      pid when is_pid(pid) ->
+        _result = GenServer.call(InMemory, {:delete, {:enactment, enactment_id}})
+        :ok
+
+      nil ->
+        :ok
+    end
+  end
+
+  defp rollback_enactment_insert(%{id: _flow_id}, enactment_id) do
+    case Repo.get(Schemas.Enactment, enactment_id) do
+      %Schemas.Enactment{} = row ->
+        _result = Repo.delete(row)
+        :ok
+
+      nil ->
+        :ok
+    end
+  end
+
   # ---------------------------------------------------------------------------
   # :fetch_flow_detail command
   # ---------------------------------------------------------------------------

dashboard/lib/coloured_flow_dashboard_web/stores/telemetry_feed_store.ex

@@ -87,6 +87,7 @@ defmodule ColouredFlowDashboardWeb.Stores.TelemetryFeedStore do
       |> assign(:oldest_seq, nil)
       |> assign(:newest_seq, nil)
       |> assign(:last_seq, %{})
+      |> assign(:entry_enactments, %{})
       # Ordered `[{id, seq}]` list — newest (highest seq) at head, matching
       # the stream's `at: 0` insertion semantics. Position-based insert keeps
       # the feed sorted by bridge seq even when the bridge's per-event Task
@@ -155,10 +156,12 @@ defmodule ColouredFlowDashboardWeb.Stores.TelemetryFeedStore do
     index = socket.assigns.entries_index
     position = insert_position(index, seq)
     next_index = List.insert_at(index, position, {id, seq})
+    next_entry_enactments = Map.put(socket.assigns.entry_enactments, id, event.enactment_id)
 
     socket
     |> stream_insert(:entries, entry, at: position)
     |> assign(:entries_index, next_index)
+    |> assign(:entry_enactments, next_entry_enactments)
     |> assign(:total_events, socket.assigns.total_events + 1)
     |> assign(:entries_in_window, length(next_index))
     |> assign(:newest_seq, head_seq(next_index))
@@ -183,17 +186,34 @@ defmodule ColouredFlowDashboardWeb.Stores.TelemetryFeedStore do
       socket
     else
       {{drop_id, _drop_seq}, next_index} = List.pop_at(index, -1)
+      drop_eid = Map.get(socket.assigns.entry_enactments, drop_id)
+      next_entry_enactments = Map.delete(socket.assigns.entry_enactments, drop_id)
+      next_last_seq = prune_last_seq(socket.assigns.last_seq, next_entry_enactments, drop_eid)
 
       socket
       |> stream_delete_by_item_key(:entries, drop_id)
       |> assign(:entries_index, next_index)
+      |> assign(:entry_enactments, next_entry_enactments)
+      |> assign(:last_seq, next_last_seq)
       |> assign(:entries_in_window, length(next_index))
       |> assign(:newest_seq, head_seq(next_index))
       |> assign(:oldest_seq, tail_seq(next_index))
       |> trim_window()
     end
   end
 
+  defp prune_last_seq(last_seq, _entry_enactments, nil), do: last_seq
+
+  defp prune_last_seq(last_seq, entry_enactments, enactment_id) do
+    if Enum.any?(entry_enactments, fn {_id, current_enactment_id} ->
+         current_enactment_id == enactment_id
+       end) do
+      last_seq
+    else
+      Map.delete(last_seq, enactment_id)
+    end
+  end
+
   defp head_seq([{_id, seq} | _rest]), do: seq
   defp head_seq([]), do: nil
 

dashboard/test/coloured_flow_dashboard_web/stores/enactment_detail_store_test.exs

@@ -8,6 +8,7 @@ defmodule ColouredFlowDashboardWeb.Stores.EnactmentDetailStoreTest do
   alias ColouredFlow.Enactment.BindingElement
   alias ColouredFlow.Runner.Enactment, as: RunnerEnactment
   alias ColouredFlow.Runner.Enactment.Workitem, as: RunnerWorkitem
+  alias ColouredFlow.Runner.Storage.Schemas
   alias ColouredFlowDashboard.Seed
   alias ColouredFlowDashboard.Seeds.ApprovalFlow
   alias ColouredFlowDashboard.TelemetryBridge
@@ -268,6 +269,24 @@ defmodule ColouredFlowDashboardWeb.Stores.EnactmentDetailStoreTest do
                Musubi.Testing.dispatch_command(page, :retry_enactment, %{})
     end
 
+    test "keeps the storage row in :exception when runner restart fails",
+         %{topic_prefix: topic_prefix, flow_cache: flow_cache} do
+      enactment = insert_enactment_with_state!(:exception)
+
+      page =
+        mount_store(enactment.id, topic_prefix, flow_cache,
+          runner_start_fun: fn enactment_id ->
+            assert enactment_id == enactment.id
+            {:error, :runner_down}
+          end
+        )
+
+      assert {:ok, %{code: :runner_error}} =
+               Musubi.Testing.dispatch_command(page, :retry_enactment, %{})
+
+      assert %Schemas.Enactment{state: :exception} = Repo.get!(Schemas.Enactment, enactment.id)
+    end
+
     # Race pin (P20 HIGH): another operator force-terminated the row, but
     # this page's cached `summary.state` is still `:exception` because the
     # `:enactment_terminate` event has not been broadcast/applied yet. The
@@ -1422,12 +1441,20 @@ defmodule ColouredFlowDashboardWeb.Stores.EnactmentDetailStoreTest do
   # credo:disable-for-next-line Credo.Check.Warning.UnsafeToAtom
   defp unique_cache_atom(name) when is_binary(name), do: String.to_atom(name)
 
-  defp mount_store(enactment_id, topic_prefix, flow_cache) do
-    Musubi.Testing.mount(EnactmentDetailStore, %{
+  defp mount_store(enactment_id, topic_prefix, flow_cache, opts \\ []) do
+    params = %{
       "id" => enactment_id,
       "topic_prefix" => topic_prefix,
       "flow_cache" => flow_cache
-    })
+    }
+
+    params =
+      case Keyword.fetch(opts, :runner_start_fun) do
+        {:ok, runner_start_fun} -> Map.put(params, "runner_start_fun", runner_start_fun)
+        :error -> params
+      end
+
+    Musubi.Testing.mount(EnactmentDetailStore, params)
   end
 
   # Inserts a real `enactments` row in the requested state so the

dashboard/test/coloured_flow_dashboard_web/stores/flow_catalog_store_test.exs

@@ -178,6 +178,18 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStoreTest do
       assert {:ok, %{code: :unknown_flow, enactment_id: nil}} =
                Musubi.Testing.dispatch_command(page, :start_enactment, %{flow_id: bogus})
     end
+
+    test "rolls back the inserted row when Runner.start_enactment/1 fails", %{topic: topic} do
+      flow_record = InMemory.insert_flow!(ApprovalFlow.cpnet())
+      flow_id = InMemory.flow(flow_record, :id)
+      before_ids = current_enactment_ids()
+      page = mount_store(topic, fn _enactment_id -> {:error, :runner_down} end)
+
+      assert {:ok, %{code: :runner_error, enactment_id: nil}} =
+               Musubi.Testing.dispatch_command(page, :start_enactment, %{flow_id: flow_id})
+
+      assert current_enactment_ids() == before_ids
+    end
   end
 
   describe ":refresh_catalog command" do
@@ -290,8 +302,11 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStoreTest do
   defp discriminator(context),
     do: Integer.to_string(:erlang.phash2({context.module, context.test}))
 
-  defp mount_store(topic) do
-    Musubi.Testing.mount(FlowCatalogStore, %{"topic" => topic})
+  defp mount_store(topic, runner_start_fun \\ &ColouredFlow.Runner.start_enactment/1) do
+    Musubi.Testing.mount(FlowCatalogStore, %{
+      "topic" => topic,
+      "runner_start_fun" => runner_start_fun
+    })
   end
 
   defp broadcast!(topic, %Event{} = event) do
@@ -310,6 +325,16 @@ defmodule ColouredFlowDashboardWeb.Stores.FlowCatalogStoreTest do
     _error -> 0
   end
 
+  defp current_enactment_ids do
+    InMemory
+    |> Module.safe_concat("Enactment")
+    |> :ets.tab2list()
+    |> Enum.map(&InMemory.enactment(&1, :id))
+    |> MapSet.new()
+  rescue
+    _error -> MapSet.new()
+  end
+
   # Reads the FlowSummary item out of the mount-time patch envelope's
   # stream_ops. The catalog stream emits an `insert` op per flow with the
   # native FlowSummary struct as the `item` field. Mailbox is drained in

dashboard/test/coloured_flow_dashboard_web/stores/telemetry_feed_store_test.exs

@@ -132,6 +132,17 @@ defmodule ColouredFlowDashboardWeb.Stores.TelemetryFeedStoreTest do
       assert assigns.newest_seq == 5
     end
 
+    test "bounds last_seq under distinct-enactment churn", %{topic: topic, page: page} do
+      for seq <- 1..1_200 do
+        broadcast!(topic, build_event(:produce_workitems_stop, Ecto.UUID.generate(), seq))
+      end
+
+      assigns = Musubi.Testing.assigns(page)
+
+      assert assigns.entries_in_window == 5
+      assert map_size(assigns.last_seq) <= 5
+    end
+
     test "two enactments interleave without stale-drop crosstalk",
          %{topic: topic, page: page} do
       eid_a = Ecto.UUID.generate()