feat(runner): state-drift Tier 2 funnel via storage tuples

Detect divergence between the runner's in-memory workitem cache and
the database without rescuing exceptions in the GenServer.

Storage layer:
* `transition_workitems/2` and `complete_workitems/4` now return
  `{:error, {:state_drift, ctx}}` only when the Multi `:result` step
  rejects an `:unexpected_updated_rows` mismatch. Other Multi failures
  raise so they surface as crashes counted by the circuit breaker.
* `Storage.write_result()` and `Storage.state_drift_context()` types
  document the new contract; the keyword `ctx` carries `expected:` /
  `actual:` row counts that flow through to the persisted exception.

Enactment layer:
* `persist_or_drift!/3` funnels `{:error, {:state_drift, _}}` into
  `Storage.exception_occurs(:state_drift, _)` plus
  `exit({:shutdown, {:fatal, :state_drift}})`. The shutdown form keeps
  the supervisor's `max_restarts` counter untouched, so a single
  drifted enactment does not snowball into supervisor self-termination.
* `apply_calibration/1`, `start_workitems/2`, and `complete_workitems/3`
  pipe their storage writes through `persist_or_drift!`.

New `Exceptions.StateDrift` carries the action and ctx so operators
see the row-count discrepancy in `enactment_logs.exception`.

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

Author: fahchen

lib/coloured_flow/runner/enactment/enactment.ex

@@ -248,6 +248,31 @@ defmodule ColouredFlow.Runner.Enactment do
     }
   end
 
+  # Funnel storage writes that may report `:state_drift`. On drift, persist
+  # `:exception` state and exit via `{:shutdown, _}` so the supervisor does
+  # not restart and `terminate/2` does not record a `:crash`.
+  @spec persist_or_drift!(Storage.write_result(), state(), Workitem.transition_action()) ::
+          :ok | no_return()
+  defp persist_or_drift!(:ok, %__MODULE__{}, _action), do: :ok
+
+  defp persist_or_drift!({:error, {:state_drift, ctx}}, %__MODULE__{} = state, action) do
+    exception =
+      Exceptions.StateDrift.exception(
+        enactment_id: state.enactment_id,
+        action: action,
+        context: ctx
+      )
+
+    :ok = Storage.exception_occurs(state.enactment_id, :state_drift, exception)
+
+    emit_event(:exception, state, %{
+      exception_reason: :state_drift,
+      exception: exception
+    })
+
+    exit({:shutdown, {:fatal, :state_drift}})
+  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.
@@ -266,10 +291,14 @@ defmodule ColouredFlow.Runner.Enactment do
 
         Enum.each(grouped_wokitems, fn
           {:enabled, workitems} ->
-            :ok = Storage.withdraw_workitems(workitems, action: :withdraw)
+            workitems
+            |> Storage.withdraw_workitems(action: :withdraw)
+            |> persist_or_drift!(state, :withdraw)
 
           {:started, workitems} ->
-            :ok = Storage.withdraw_workitems(workitems, action: :withdraw_s)
+            workitems
+            |> Storage.withdraw_workitems(action: :withdraw_s)
+            |> persist_or_drift!(state, :withdraw_s)
         end)
 
         {:ok, workitems, %{workitems: workitems}}
@@ -349,7 +378,9 @@ defmodule ColouredFlow.Runner.Enactment do
     :start_workitems
     |> with_span(state, %{workitem_ids: workitem_ids}, fn ->
       with {:ok, {started_workitems, new_state}} <- start_workitems(state, workitem_ids) do
-        :ok = Storage.start_workitems(started_workitems, action: :start)
+        started_workitems
+        |> Storage.start_workitems(action: :start)
+        |> persist_or_drift!(state, :start)
 
         {:ok, {started_workitems, new_state}, %{workitems: started_workitems}}
       end
@@ -385,13 +416,13 @@ defmodule ColouredFlow.Runner.Enactment do
           {:ok, {workitem_occurrences, new_state, calibration_options}} <-
             complete_workitems(state, workitem_ids, workitem_id_and_outputs)
         ) do
-          :ok =
-            Storage.complete_workitems(
-              new_state.enactment_id,
-              new_state.version,
-              workitem_occurrences,
-              action: transition_action
-            )
+          new_state.enactment_id
+          |> Storage.complete_workitems(
+            new_state.version,
+            workitem_occurrences,
+            action: transition_action
+          )
+          |> persist_or_drift!(state, transition_action)
 
           completed_workitems = Enum.map(workitem_occurrences, &elem(&1, 0))
 

lib/coloured_flow/runner/exceptions/state_drift.ex

@@ -0,0 +1,36 @@
+defmodule ColouredFlow.Runner.Exceptions.StateDrift do
+  @moduledoc """
+  The runner detected divergence between its in-memory state and what the storage
+  layer believed: a Multi rolled back, or the number of rows updated did not match
+  the number of workitems we expected to transition.
+
+  Recovery is unsafe in-process: the workitem cache is stale. The runner records
+  the event, marks `enactments.state = :exception`, and exits with
+  `{:shutdown, {:fatal, :state_drift}}`. Operators can clear it via reoffer.
+  """
+
+  use TypedStructor
+
+  alias ColouredFlow.Runner.Enactment.Workitem
+  alias ColouredFlow.Runner.Storage
+
+  typed_structor definer: :defexception, enforce: true do
+    field :enactment_id, Storage.enactment_id()
+    field :action, Workitem.transition_action()
+    field :context, Storage.state_drift_context(), default: []
+  end
+
+  @impl Exception
+  def exception(arguments) do
+    struct!(__MODULE__, arguments)
+  end
+
+  @impl Exception
+  def message(%__MODULE__{} = exception) do
+    "State drift detected while handling action #{inspect(exception.action)} " <>
+      "in enactment #{inspect(exception.enactment_id)}: " <>
+      "the storage rejected the transition because the in-memory workitem cache " <>
+      "is out of sync with the database. " <>
+      "Context: #{inspect(exception.context)}."
+  end
+end

lib/coloured_flow/runner/storage/default.ex

@@ -19,19 +19,6 @@ defmodule ColouredFlow.Runner.Storage.Default do
   @type enactment_id() :: ColouredFlow.Runner.Storage.enactment_id()
   @type flow_id() :: ColouredFlow.Runner.Storage.flow_id()
 
-  # Exception types raised by `Codec.Marking.decode/2` when it encounters a
-  # payload it cannot parse. Listed explicitly so that connection errors
-  # (`DBConnection.ConnectionError`, `Postgrex.Error`, etc.) are NOT
-  # reclassified as `:snapshot_corrupt`.
-  @snapshot_corrupt_exceptions [
-    ArgumentError,
-    FunctionClauseError,
-    KeyError,
-    MatchError,
-    Protocol.UndefinedError,
-    RuntimeError
-  ]
-
   @impl ColouredFlow.Runner.Storage
   def get_flow_by_enactment(enactment_id) do
     Schemas.Flow
@@ -242,12 +229,13 @@ defmodule ColouredFlow.Runner.Storage.Default do
   end
 
   @typep transition_option() :: ColouredFlow.Runner.Storage.transition_option()
+  @typep write_result() :: ColouredFlow.Runner.Storage.write_result()
 
   @doc """
   Transition a workitem from one state to another. This is a shortcut for
   `ColouredFlow.Runner.Default.transition_workitems/2`.
   """
-  @spec transition_workitem(Workitem.t(), [transition_option()]) :: :ok
+  @spec transition_workitem(Workitem.t(), [transition_option()]) :: write_result()
   def transition_workitem(workitem, options) do
     workitem
     |> List.wrap()
@@ -258,7 +246,7 @@ defmodule ColouredFlow.Runner.Storage.Default do
   Transition the workitems from one state to another in accordance with the state
   machine (See `t:ColouredFlow.Runner.Enactment.Workitem.state/0`).
   """
-  @spec transition_workitems([Workitem.t()], [transition_option()]) :: :ok
+  @spec transition_workitems([Workitem.t()], [transition_option()]) :: write_result()
   def transition_workitems([], _options), do: :ok
 
   def transition_workitems(workitems, options) when is_list(workitems) do
@@ -271,13 +259,18 @@ defmodule ColouredFlow.Runner.Storage.Default do
       {:ok, _changes} ->
         :ok
 
-      {
-        :error,
-        :result,
-        {:unexpected_updated_rows, exception_ctx},
-        _changes_so_far
-      } ->
-        unexpected_updated_rows!(workitems, action, exception_ctx)
+      {:error, :result, {:unexpected_updated_rows, ctx}, _changes} ->
+        {:error, {:state_drift, ctx}}
+
+      {:error, op, value, changes} ->
+        raise """
+        Workitem transition failed unexpectedly.
+
+        Action: #{inspect(action)}
+        Operation: #{inspect(op)}
+        Value: #{inspect(value)}
+        Changes: #{inspect(changes)}
+        """
     end
   end
 
@@ -346,21 +339,6 @@ defmodule ColouredFlow.Runner.Storage.Default do
     defp get_from_state(unquote(to), unquote(action)), do: unquote(from)
   end
 
-  defp unexpected_updated_rows!(workitems, transition, options) do
-    # When the actual number is not equal to the expected number,
-    # it means the workitems in the gen_server are not consistent with the database.
-    # So we just raise an error to crash the process, and let the supervisor
-    # restart the gen_server and retry.
-    expected = Keyword.fetch!(options, :expected)
-    actual = Keyword.fetch!(options, :actual)
-
-    raise """
-    The number of workitems to #{transition} is not equal to the actual number.
-    Expected: #{expected}, Actual: #{actual}
-    Workitems: #{Enum.map_join(workitems, ", ", & &1.id)}
-    """
-  end
-
   @impl ColouredFlow.Runner.Storage
   def start_workitems(workitems, options) do
     transition_workitems(workitems, options)
@@ -395,13 +373,18 @@ defmodule ColouredFlow.Runner.Storage.Default do
       {:ok, _result} ->
         :ok
 
-      {
-        :error,
-        :result,
-        {:unexpected_updated_rows, exception_ctx},
-        _changes_so_far
-      } ->
-        unexpected_updated_rows!(completed_workitems, action, exception_ctx)
+      {:error, :result, {:unexpected_updated_rows, ctx}, _changes} ->
+        {:error, {:state_drift, ctx}}
+
+      {:error, op, value, changes} ->
+        raise """
+        Workitem completion failed unexpectedly.
+
+        Action: #{inspect(action)}
+        Operation: #{inspect(op)}
+        Value: #{inspect(value)}
+        Changes: #{inspect(changes)}
+        """
     end
   end
 
@@ -447,6 +430,19 @@ defmodule ColouredFlow.Runner.Storage.Default do
     :ok
   end
 
+  # Exception types raised by `Codec.Marking.decode/2` when it encounters a
+  # payload it cannot parse. Listed explicitly so that connection errors
+  # (`DBConnection.ConnectionError`, `Postgrex.Error`, etc.) are NOT
+  # reclassified as `:snapshot_corrupt`.
+  @snapshot_corrupt_exceptions [
+    ArgumentError,
+    FunctionClauseError,
+    KeyError,
+    MatchError,
+    Protocol.UndefinedError,
+    RuntimeError
+  ]
+
   @impl ColouredFlow.Runner.Storage
   def read_enactment_snapshot(enactment_id) do
     # Codec decoding runs inside `Repo.get_by/2` when Ecto loads the

lib/coloured_flow/runner/storage/storage.ex

@@ -25,6 +25,16 @@ defmodule ColouredFlow.Runner.Storage do
   @type enactment_id() :: Ecto.UUID.t()
   @type flow_id() :: Ecto.UUID.t()
 
+  @typedoc """
+  Mutating workitem operations may return `{:error, {:state_drift, ctx}}` when the
+  in-memory state diverges from storage (Multi `:result` step rejecting an
+  `:unexpected_updated_rows` mismatch). Callers funnel this into a Tier 2 fatal
+  exit; the keyword `ctx` carries `expected:` / `actual:` row counts so operators
+  see the discrepancy in the persisted exception.
+  """
+  @type state_drift_context() :: [expected: pos_integer(), actual: non_neg_integer()]
+  @type write_result() :: :ok | {:error, {:state_drift, state_drift_context()}}
+
   @doc """
   Get the flow of an enactment.
   """
@@ -101,21 +111,21 @@ defmodule ColouredFlow.Runner.Storage do
   @callback start_workitems(
               started_workitems :: [Workitem.t(:started)],
               options :: [transition_option()]
-            ) :: :ok
+            ) :: write_result()
 
   @doc group: :workitem
   @callback withdraw_workitems(
               withdrawn_workitems :: [Workitem.t(:withdrawn)],
               options :: [transition_option()]
-            ) :: :ok
+            ) :: write_result()
 
   @doc group: :workitem
   @callback complete_workitems(
               enactment_id(),
               current_version :: non_neg_integer(),
               workitem_occurrences :: [{Workitem.t(:completed), Occurrence.t()}],
               options :: [transition_option()]
-            ) :: :ok
+            ) :: write_result()
 
   @doc """
   Takes a snapshot of the given enactment.
@@ -219,13 +229,13 @@ defmodule ColouredFlow.Runner.Storage do
   end
 
   @doc false
-  @spec start_workitems([Workitem.t(:started)], [transition_option()]) :: :ok
+  @spec start_workitems([Workitem.t(:started)], [transition_option()]) :: write_result()
   def start_workitems(workitems, options) do
     __storage__().start_workitems(workitems, options)
   end
 
   @doc false
-  @spec withdraw_workitems([Workitem.t(:withdrawn)], [transition_option()]) :: :ok
+  @spec withdraw_workitems([Workitem.t(:withdrawn)], [transition_option()]) :: write_result()
   def withdraw_workitems(workitems, options) do
     __storage__().withdraw_workitems(workitems, options)
   end
@@ -236,7 +246,7 @@ defmodule ColouredFlow.Runner.Storage do
           current_version :: non_neg_integer(),
           workitem_occurrences :: [{Workitem.t(:completed), Occurrence.t()}],
           [transition_option()]
-        ) :: :ok
+        ) :: write_result()
   def complete_workitems(enactment_id, current_version, workitem_occurrences, options) do
     __storage__().complete_workitems(enactment_id, current_version, workitem_occurrences, options)
   end

test/coloured_flow/runner/enactment/error_handling_test.exs

@@ -2,6 +2,8 @@ defmodule ColouredFlow.Runner.Enactment.ErrorHandlingTest do
   use ColouredFlow.RepoCase, async: true
   use ColouredFlow.RunnerHelpers
 
+  import Ecto.Query, only: [from: 2]
+
   alias ColouredFlow.Enactment.Marking
   alias ColouredFlow.Runner.Enactment, as: EnactmentServer
   alias ColouredFlow.Runner.Enactment.Snapshot
@@ -96,6 +98,66 @@ defmodule ColouredFlow.Runner.Enactment.ErrorHandlingTest do
     end
   end
 
+  describe "state_drift fatal funnel" do
+    setup :setup_flow
+    setup :setup_enactment
+
+    @describetag cpnet: :simple_sequence
+
+    @tag initial_markings: [%Marking{place: "input", tokens: ~MS[1]}]
+    test "start_workitems drift exits with :shutdown and persists :state_drift",
+         %{enactment: enactment} do
+      [enactment_server: enactment_server] = start_enactment(%{enactment: enactment})
+
+      [%Enactment.Workitem{state: :enabled} = workitem] =
+        get_enactment_workitems(enactment_server)
+
+      # Delete the workitem row out from under the GenServer to simulate drift.
+      delete_query = from(w in Schemas.Workitem, where: w.id == ^workitem.id)
+      Repo.delete_all(delete_query)
+
+      ref = Process.monitor(enactment_server)
+      catch_exit(GenServer.call(enactment_server, {:start_workitems, [workitem.id]}))
+
+      assert_receive {:DOWN, ^ref, :process, ^enactment_server,
+                      {:shutdown, {:fatal, :state_drift}}},
+                     500
+
+      schema = Repo.get(Schemas.Enactment, enactment.id)
+      assert schema.state === :exception
+
+      [exception_log] =
+        Schemas.EnactmentLog
+        |> Repo.all(enactment_id: enactment.id)
+        |> Enum.filter(&(&1.state === :exception))
+
+      assert exception_log.exception.reason === :state_drift
+    end
+
+    @tag initial_markings: [%Marking{place: "input", tokens: ~MS[1]}]
+    test "complete_workitems drift exits with :shutdown and persists :state_drift",
+         %{enactment: enactment} do
+      [enactment_server: enactment_server] = start_enactment(%{enactment: enactment})
+
+      [workitem] = get_enactment_workitems(enactment_server)
+      workitem = start_workitem(workitem, enactment_server)
+
+      # Delete the started workitem row out from under the GenServer.
+      delete_query = from(w in Schemas.Workitem, where: w.id == ^workitem.id)
+      Repo.delete_all(delete_query)
+
+      ref = Process.monitor(enactment_server)
+      catch_exit(GenServer.call(enactment_server, {:complete_workitems, %{workitem.id => []}}))
+
+      assert_receive {:DOWN, ^ref, :process, ^enactment_server,
+                      {:shutdown, {:fatal, :state_drift}}},
+                     500
+
+      schema = Repo.get(Schemas.Enactment, enactment.id)
+      assert schema.state === :exception
+    end
+  end
+
   describe "abnormal terminate logs :crash" do
     setup :setup_flow
     setup :setup_enactment