feat(runner/enactment): snapshot self-heal + crash circuit breaker

Wires the storage primitives from the prior commit into the GenServer
boot/teardown path:

* `init/1` queries
  `Storage.consecutive_crashes_since_progress/1`. Three or more
  consecutive `:crash` events without an intervening `Occurrence`
  flips `enactments.state` to `:exception` (reason `:restart_loop`)
  and aborts via `:ignore`, preventing the supervisor from spinning
  up a hot restart loop.
* `handle_continue(:populate_state, ...)` routes
  `{:error, {:snapshot_corrupt, _}}` through
  `load_or_recover_snapshot/1`. The runner records a non-fatal
  `:running` log row, deletes the bad snapshot, and replays from the
  enactment's initial markings; the next `:take_snapshot` rewrites the
  row.
* `terminate/2` calls `Storage.record_crash/2` on any non-`:normal`,
  non-`:shutdown` exit so the circuit breaker has data on the next
  boot. Normal and shutdown exits stay silent.

Tests in `test/coloured_flow/runner/enactment/error_handling_test.exs`
cover all three new flows.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: fahchen

lib/coloured_flow/runner/enactment/enactment.ex

@@ -54,6 +54,10 @@ defmodule ColouredFlow.Runner.Enactment do
   alias ColouredFlow.Runner.Storage
   alias ColouredFlow.Runner.Telemetry
 
+  require Logger
+
+  @consecutive_crash_threshold 3
+
   @typep enactment_id() :: Storage.enactment_id()
 
   @typedoc "The markings map of the enactment."
@@ -119,22 +123,28 @@ defmodule ColouredFlow.Runner.Enactment do
 
     state = struct(__MODULE__, options)
 
-    {:ok, state, {:continue, :populate_state}}
+    case Storage.consecutive_crashes_since_progress(state.enactment_id) do
+      count when count >= @consecutive_crash_threshold ->
+        exception =
+          Exceptions.RestartLoop.exception(
+            enactment_id: state.enactment_id,
+            count: count
+          )
+
+        :ok = Storage.exception_occurs(state.enactment_id, :restart_loop, exception)
+
+        Logger.warning(Exception.message(exception))
+
+        :ignore
+
+      _count ->
+        {:ok, state, {:continue, :populate_state}}
+    end
   end
 
   @impl GenServer
   def handle_continue(:populate_state, %__MODULE__{} = state) do
-    snapshot =
-      case Storage.read_enactment_snapshot(state.enactment_id) do
-        {:ok, snapshot} ->
-          snapshot
-
-        :error ->
-          %Snapshot{
-            version: 0,
-            markings: Storage.get_initial_markings(state.enactment_id)
-          }
-      end
+    snapshot = load_or_recover_snapshot(state)
 
     snapshot = catchup_snapshot(state.enactment_id, snapshot)
     Storage.take_enactment_snapshot(state.enactment_id, snapshot)
@@ -201,6 +211,43 @@ defmodule ColouredFlow.Runner.Enactment do
     %{snapshot | version: snapshot.version + steps, markings: markings}
   end
 
+  @spec load_or_recover_snapshot(state()) :: Snapshot.t()
+  defp load_or_recover_snapshot(%__MODULE__{} = state) do
+    case Storage.read_enactment_snapshot(state.enactment_id) do
+      {:ok, snapshot} ->
+        snapshot
+
+      :error ->
+        initial_snapshot(state)
+
+      {:error, {:snapshot_corrupt, underlying}} ->
+        exception =
+          Exceptions.SnapshotCorrupt.exception(
+            enactment_id: state.enactment_id,
+            underlying: underlying
+          )
+
+        :ok = Storage.recover_from_corrupt_snapshot(state.enactment_id, exception)
+
+        Logger.warning(Exception.message(exception))
+
+        emit_event(:exception, state, %{
+          exception_reason: :snapshot_corrupt,
+          exception: exception
+        })
+
+        initial_snapshot(state)
+    end
+  end
+
+  @spec initial_snapshot(state()) :: Snapshot.t()
+  defp initial_snapshot(%__MODULE__{} = state) do
+    %Snapshot{
+      version: 0,
+      markings: Storage.get_initial_markings(state.enactment_id)
+    }
+  end
+
   defp apply_calibration(%WorkitemCalibration{state: %__MODULE__{} = state} = calibration) do
     # We don't need to ensure `withdraw_workitems` and `produce_workitems` are atomic,
     # because the gen_server will restart if the process crashes.
@@ -573,8 +620,14 @@ defmodule ColouredFlow.Runner.Enactment do
     emit_event(:stop, state, %{reason: reason})
   end
 
-  def terminate(_reason, state) do
-    emit_event(:stop, state, %{reason: "unknown"})
+  def terminate(:normal, state) do
+    emit_event(:stop, state, %{reason: :normal})
+  end
+
+  def terminate(reason, state) do
+    exception = Exceptions.AbnormalExit.exception(reason: reason)
+    :ok = Storage.record_crash(state.enactment_id, exception)
+    emit_event(:stop, state, %{reason: reason})
   end
 
   @spec to_map(Enumerable.t(item)) :: %{Place.name() => item} when item: Marking.t()

test/coloured_flow/runner/enactment/error_handling_test.exs

@@ -0,0 +1,139 @@
+defmodule ColouredFlow.Runner.Enactment.ErrorHandlingTest do
+  use ColouredFlow.RepoCase, async: true
+  use ColouredFlow.RunnerHelpers
+
+  alias ColouredFlow.Enactment.Marking
+  alias ColouredFlow.Runner.Enactment, as: EnactmentServer
+  alias ColouredFlow.Runner.Enactment.Snapshot
+  alias ColouredFlow.Runner.Exceptions
+  alias ColouredFlow.Runner.Storage
+
+  describe "snapshot_corrupt self-heal" do
+    setup :setup_flow
+    setup :setup_enactment
+
+    @describetag cpnet: :simple_sequence
+
+    @tag initial_markings: [%Marking{place: "input", tokens: ~MS[1]}]
+    test "deletes the corrupt snapshot row and replays from initial markings",
+         %{enactment: enactment} do
+      insert_corrupt_snapshot!(enactment.id, version: 1)
+
+      [enactment_server: enactment_server] = start_enactment(%{enactment: enactment})
+
+      :ok = wait_enactment_requests_handled!(enactment_server)
+
+      logs = Repo.all(Schemas.EnactmentLog, enactment_id: enactment.id)
+
+      assert Enum.any?(logs, fn log ->
+               log.state === :running and
+                 match?(%{reason: :snapshot_corrupt}, log.exception)
+             end)
+
+      [marking] = get_enactment_markings(enactment_server)
+      assert marking.place === "input"
+      assert marking.tokens === ~MS[1]
+    end
+
+    @tag initial_markings: [%Marking{place: "input", tokens: ~MS[1]}]
+    test "rewrites a fresh snapshot after self-heal", %{enactment: enactment} do
+      insert_corrupt_snapshot!(enactment.id, version: 5)
+
+      [enactment_server: enactment_server] = start_enactment(%{enactment: enactment})
+      :ok = wait_enactment_requests_handled!(enactment_server)
+
+      # The runner overwrote the corrupt row on boot. Subsequent reads succeed
+      # and reflect the replayed state (no occurrences exist yet, so version 0).
+      assert {:ok, %Snapshot{version: 0}} = Storage.read_enactment_snapshot(enactment.id)
+    end
+  end
+
+  describe "consecutive crash circuit breaker" do
+    setup :setup_flow
+    setup :setup_enactment
+
+    @describetag cpnet: :simple_sequence
+
+    @tag initial_markings: [%Marking{place: "input", tokens: ~MS[1]}]
+    test "init aborts via :ignore once crash threshold is exceeded",
+         %{enactment: enactment} do
+      for _i <- 1..3 do
+        :ok =
+          Storage.record_crash(
+            enactment.id,
+            Exceptions.AbnormalExit.exception(reason: :boom)
+          )
+      end
+
+      # `start_supervised` records `:ignore` as `{:ok, :undefined}`.
+      assert {:ok, :undefined} =
+               start_supervised(
+                 {EnactmentServer, [enactment_id: enactment.id]},
+                 id: enactment.id
+               )
+
+      schema = Repo.get(Schemas.Enactment, enactment.id)
+      assert schema.state === :exception
+
+      [%{exception: %{reason: :restart_loop}}] =
+        Schemas.EnactmentLog
+        |> Repo.all(enactment_id: enactment.id)
+        |> Enum.filter(&(&1.state === :exception))
+    end
+
+    @tag initial_markings: [%Marking{place: "input", tokens: ~MS[1]}]
+    test "starts normally below threshold", %{enactment: enactment} do
+      :ok =
+        Storage.record_crash(
+          enactment.id,
+          Exceptions.AbnormalExit.exception(reason: :boom)
+        )
+
+      [enactment_server: enactment_server] = start_enactment(%{enactment: enactment})
+
+      assert is_pid(enactment_server)
+      :ok = wait_enactment_requests_handled!(enactment_server)
+    end
+  end
+
+  describe "abnormal terminate logs :crash" do
+    setup :setup_flow
+    setup :setup_enactment
+
+    @describetag cpnet: :simple_sequence
+
+    @tag initial_markings: [%Marking{place: "input", tokens: ~MS[1]}]
+    test "non-shutdown stop triggers a :crash log row", %{enactment: enactment} do
+      [enactment_server: enactment_server] = start_enactment(%{enactment: enactment})
+      :ok = wait_enactment_requests_handled!(enactment_server)
+
+      ref = Process.monitor(enactment_server)
+      :ok = GenServer.stop(enactment_server, :crash_under_test, 500)
+
+      receive do
+        {:DOWN, ^ref, :process, ^enactment_server, _reason} -> :ok
+      after
+        500 -> ExUnit.Assertions.flunk("enactment server did not stop")
+      end
+
+      [crash_log] = Repo.all(Schemas.EnactmentLog, enactment_id: enactment.id)
+
+      assert crash_log.state === :running
+      assert crash_log.exception.reason === :crash
+    end
+  end
+
+  defp insert_corrupt_snapshot!(enactment_id, version: version) do
+    now = DateTime.utc_now()
+
+    Ecto.Adapters.SQL.query!(
+      Repo,
+      """
+      INSERT INTO coloured_flow.snapshots
+        (enactment_id, version, markings, inserted_at, updated_at)
+      VALUES ($1, $2, $3::jsonb, $4, $5)
+      """,
+      [Ecto.UUID.dump!(enactment_id), version, ~s(["not-a-marking-object"]), now, now]
+    )
+  end
+end