perf(enactment): debounce snapshot writes via mailbox drain

Replace `GenServer.cast(self(), :take_snapshot)` with `send(self(),
:take_snapshot)` and a `handle_info` clause that drains queued
`:take_snapshot` messages via selective `receive` before writing. Bursts of
`complete_workitems` calls now collapse into a single storage write of the
latest state. Using a plain atom (rather than a cast envelope) avoids
coupling to OTP's private `:"$gen_cast"` shape.

Register a new `[:coloured_flow, :runner, :enactment, :take_snapshot]`
telemetry event for observability and update the default logger so it does
not crash on the new event. The integration test in `enactment_test.exs`
attaches to this event to assert that a burst of 5 snapshot requests yields
fewer than 5 storage writes.

Author: fahchen

lib/coloured_flow/runner/enactment/enactment.ex

@@ -339,7 +339,7 @@ defmodule ColouredFlow.Runner.Enactment do
     )
     |> case do
       {:ok, {{completed_workitems, new_state}, continue}} ->
-        :ok = GenServer.cast(self(), :take_snapshot)
+        send(self(), :take_snapshot)
 
         {:reply, {:ok, completed_workitems}, new_state, continue}
 
@@ -348,16 +348,6 @@ defmodule ColouredFlow.Runner.Enactment do
     end
   end
 
-  @impl GenServer
-  def handle_cast(:take_snapshot, %__MODULE__{} = state) do
-    Storage.take_enactment_snapshot(state.enactment_id, %Snapshot{
-      version: state.version,
-      markings: to_list(state.markings)
-    })
-
-    {:noreply, state, Lifespan.timeout(state)}
-  end
-
   # we peek at the first workitem to pre-check the state
   @spec preflight_completion(state(), [Workitem.id()]) ::
           {:ok, :complete_e | :complete, state()} | {:error, Exception.t()}
@@ -516,10 +506,35 @@ defmodule ColouredFlow.Runner.Enactment do
   end
 
   @impl GenServer
+  def handle_info(:take_snapshot, %__MODULE__{} = state) do
+    drain_take_snapshot_messages()
+
+    emit_event(:take_snapshot, state)
+
+    Storage.take_enactment_snapshot(state.enactment_id, %Snapshot{
+      version: state.version,
+      markings: to_list(state.markings)
+    })
+
+    {:noreply, state, Lifespan.timeout(state)}
+  end
+
   def handle_info(:timeout, state) do
     {:stop, {:shutdown, "Terminated due to inactivity"}, state}
   end
 
+  # Selectively drain any queued `:take_snapshot` messages in the mailbox so
+  # bursts of `complete_workitems` collapse into a single storage write of the
+  # latest state.
+  @spec drain_take_snapshot_messages() :: :ok
+  defp drain_take_snapshot_messages do
+    receive do
+      :take_snapshot -> drain_take_snapshot_messages()
+    after
+      0 -> :ok
+    end
+  end
+
   @impl GenServer
   def terminate({:shutdown, reason}, state) do
     emit_event(:stop, state, %{reason: reason})

lib/coloured_flow/runner/telemetry.ex

@@ -11,16 +11,18 @@ defmodule ColouredFlow.Runner.Telemetry do
   - `[:coloured_flow, :runner, :enactment, :stop]`
   - `[:coloured_flow, :runner, :enactment, :terminate]`
   - `[:coloured_flow, :runner, :enactment, :exception]`
+  - `[:coloured_flow, :runner, :enactment, :take_snapshot]`
 
   All enactment events share the same measurements and metadata, but their
   metadata are different.
 
-  | event        | measurements                      | metadata                                                                        |
-  | ------------ | --------------------------------- | ------------------------------------------------------------------------------- |
-  | `:start`     | `:system_time`, `:monotonic_time` | `:enactment_id`, `:enactment_state`                                             |
-  | `:stop`      | `:system_time`, `:monotonic_time` | `:enactment_id`, `:enactment_state`, `:reason`                                  |
-  | `:terminate` | `:system_time`, `:monotonic_time` | `:enactment_id`, `:enactment_state`, `termination_type`, `:termination_message` |
-  | `:exception` | `:system_time`, `:monotonic_time` | `:enactment_id`, `:enactment_state`, `:exception_reason`, `:exception`          |
+  | event            | measurements                      | metadata                                                                        |
+  | ---------------- | --------------------------------- | ------------------------------------------------------------------------------- |
+  | `:start`         | `:system_time`, `:monotonic_time` | `:enactment_id`, `:enactment_state`                                             |
+  | `:stop`          | `:system_time`, `:monotonic_time` | `:enactment_id`, `:enactment_state`, `:reason`                                  |
+  | `:terminate`     | `:system_time`, `:monotonic_time` | `:enactment_id`, `:enactment_state`, `termination_type`, `:termination_message` |
+  | `:exception`     | `:system_time`, `:monotonic_time` | `:enactment_id`, `:enactment_state`, `:exception_reason`, `:exception`          |
+  | `:take_snapshot` | `:system_time`, `:monotonic_time` | `:enactment_id`, `:enactment_state`                                             |
 
   Note that `:start` and `:stop` events will be emitted when the enactment server
   starts and stops respectively. So a `:stop` event will be emitted right after
@@ -234,6 +236,7 @@ defmodule ColouredFlow.Runner.Telemetry do
     [:coloured_flow, :runner, :enactment, :stop],
     [:coloured_flow, :runner, :enactment, :terminate],
     [:coloured_flow, :runner, :enactment, :exception],
+    [:coloured_flow, :runner, :enactment, :take_snapshot],
     [:coloured_flow, :runner, :enactment, :produce_workitems, :start],
     [:coloured_flow, :runner, :enactment, :produce_workitems, :stop],
     [:coloured_flow, :runner, :enactment, :produce_workitems, :exception],

lib/coloured_flow/runner/telemetry/default_logger.ex

@@ -11,7 +11,7 @@ defmodule ColouredFlow.Runner.Telemetry.DefaultLogger do
         metadata,
         opts
       )
-      when transition in [:start, :stop, :terminate, :exception] do
+      when transition in [:start, :stop, :terminate, :exception, :take_snapshot] do
     log(:enactment, opts, fn ->
       basic =
         case transition do
@@ -43,6 +43,12 @@ defmodule ColouredFlow.Runner.Telemetry.DefaultLogger do
               exception_reason: metadata.exception_reason,
               error: Exception.format_banner(:error, metadata.exception)
             }
+
+          :take_snapshot ->
+            %{
+              event: "enactment:take_snapshot",
+              system_time: convert_system_time(measurements.system_time)
+            }
         end
 
       enactment = build_enactment_info(metadata.enactment_state)

test/coloured_flow/runner/enactment/enactment_test.exs

@@ -255,4 +255,98 @@ defmodule ColouredFlow.Runner.EnactmentTest do
                Enum.map(current_workitems, & &1.id) -- Enum.map(previous_workitems, & &1.id)
     end
   end
+
+  describe "take_snapshot debounce" do
+    test "drains queued :take_snapshot messages before writing", %{enactment: enactment} do
+      enactment_id = enactment.id
+
+      state = %Enactment{
+        enactment_id: enactment_id,
+        version: 1,
+        markings: %{},
+        workitems: %{}
+      }
+
+      # queue extra :take_snapshot messages before invoking the handler;
+      # the first invocation should drain all of them in one shot.
+      for _i <- 1..4, do: send(self(), :take_snapshot)
+
+      assert {:noreply, ^state, _timeout} =
+               Enactment.handle_info(:take_snapshot, state)
+
+      # mailbox should now be empty of :take_snapshot messages because the
+      # debounce drained them; no further snapshot writes are needed.
+      refute_received :take_snapshot
+
+      # only one snapshot should have been persisted for the burst.
+      assert %Schemas.Snapshot{version: 1} =
+               Schemas.Snapshot |> from() |> where(enactment_id: ^enactment_id) |> Repo.one()
+    end
+
+    test "single :take_snapshot message still writes one snapshot", %{enactment: enactment} do
+      enactment_id = enactment.id
+
+      state = %Enactment{
+        enactment_id: enactment_id,
+        version: 2,
+        markings: %{},
+        workitems: %{}
+      }
+
+      assert {:noreply, ^state, _timeout} =
+               Enactment.handle_info(:take_snapshot, state)
+
+      refute_received :take_snapshot
+
+      assert %Schemas.Snapshot{version: 2} =
+               Schemas.Snapshot |> from() |> where(enactment_id: ^enactment_id) |> Repo.one()
+    end
+
+    test "coalesces a burst of :take_snapshot messages into fewer storage writes",
+         %{enactment: enactment} do
+      [enactment_server: enactment_server] = start_enactment(%{enactment: enactment})
+
+      # Wait until the boot-time snapshot from `handle_continue` has been emitted
+      # so it is not counted in the burst measurement.
+      :ok = wait_enactment_requests_handled!(enactment_server)
+
+      self_pid = self()
+      handler_id = "snapshot-debounce-test-#{System.unique_integer([:positive])}"
+
+      :ok =
+        :telemetry.attach(
+          handler_id,
+          [:coloured_flow, :runner, :enactment, :take_snapshot],
+          fn _event, _measurements, _metadata, _config ->
+            send(self_pid, :snapshot_taken)
+          end,
+          nil
+        )
+
+      on_exit(fn -> :telemetry.detach(handler_id) end)
+
+      burst_size = 5
+      for _i <- 1..burst_size, do: send(enactment_server, :take_snapshot)
+
+      # Block until the GenServer has processed the queued messages.
+      :ok = wait_enactment_requests_handled!(enactment_server)
+
+      snapshot_stream =
+        Stream.repeatedly(fn ->
+          receive do
+            :snapshot_taken -> :ok
+          after
+            50 -> :end_of_stream
+          end
+        end)
+
+      snapshot_count =
+        snapshot_stream
+        |> Enum.take_while(&(&1 == :ok))
+        |> length()
+
+      assert snapshot_count >= 1
+      assert snapshot_count < burst_size
+    end
+  end
 end