Repository: commanded/commanded
Pull request #606 (2024-11-30 03:17:43)

Aggregate.handle_* now properly handles lifespans
Based on @bougueil's issue and work here:
https://github.com/commanded/commanded/pull/600

He found 3 cases where an agggregate's lifespan would be cancelled unintentionally.

Specifically
1) Calling `Aggregate.aggregate_state/3`
2) Calling `Aggregate.aggregate_version/3`
3) If the aggregate process receives any unhandled message

The work here is to add tests and provide a unified way of sending `:reply` or `:noreply` along with the lifetime info.


Patch:
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index 517137c..4ff141f 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -326,14 +326,13 @@ defmodule Commanded.Aggregates.Aggregate do
       if Snapshotting.snapshot_required?(snapshotting, aggregate_version) do
         :ok = GenServer.cast(self(), {:take_snapshot, lifespan_timeout})
 
+        # Don't reply with a lifetime because we just asked for a snapshot to
+        # be taken. When it finishes, it will set the timeout.
         {:reply, formatted_reply, state}
       else
         state = %Aggregate{state | lifespan_timeout: lifespan_timeout}
 
-        case lifespan_timeout do
-          {:stop, reason} -> {:stop, reason, formatted_reply, state}
-          lifespan_timeout -> {:reply, formatted_reply, state, lifespan_timeout}
-        end
+        reply_with_lifespan(formatted_reply, state)
       end
 
     telemetry_metadata = telemetry_metadata(context, from, state)
@@ -347,7 +346,7 @@ defmodule Commanded.Aggregates.Aggregate do
   def handle_call(:aggregate_state, _from, %Aggregate{} = state) do
     %Aggregate{aggregate_state: aggregate_state} = state
 
-    {:reply, aggregate_state, state}
+    reply_with_lifespan(aggregate_state, state)
   end
 
   @doc false
@@ -355,13 +354,13 @@ defmodule Commanded.Aggregates.Aggregate do
   def handle_call(:aggregate_version, _from, %Aggregate{} = state) do
     %Aggregate{aggregate_version: aggregate_version} = state
 
-    {:reply, aggregate_version, state}
+    reply_with_lifespan(aggregate_version, state)
   end
 
   @doc false
   @impl GenServer
   def handle_info({:events, events}, %Aggregate{} = state) do
-    %Aggregate{application: application, lifespan_timeout: lifespan_timeout} = state
+    %Aggregate{application: application} = state
 
     Logger.debug(describe(state) <> " received events: " <> inspect(events))
 
@@ -372,10 +371,7 @@ defmodule Commanded.Aggregates.Aggregate do
         |> Upcast.upcast_event_stream(additional_metadata: %{application: application})
         |> Enum.reduce(state, &handle_event/2)
 
-      case lifespan_timeout do
-        {:stop, reason} -> {:stop, reason, state}
-        lifespan_timeout -> {:noreply, state, lifespan_timeout}
-      end
+      noreply_with_lifespan(state)
     catch
       {:error, error} ->
         Logger.debug(describe(state) <> " stopping due to: " <> inspect(error))
@@ -393,6 +389,14 @@ defmodule Commanded.Aggregates.Aggregate do
     {:stop, :normal, state}
   end
 
+  @doc false
+  @impl GenServer
+  def handle_info(message, %Aggregate{} = state) do
+    Logger.debug("received unexpected message in handle_info/2: " <> inspect(message))
+
+    noreply_with_lifespan(state)
+  end
+
   defp event_already_seen?(%RecordedEvent{} = event, %Aggregate{} = state) do
     %RecordedEvent{stream_version: stream_version} = event
     %Aggregate{aggregate_version: aggregate_version} = state
@@ -582,7 +586,6 @@ defmodule Commanded.Aggregates.Aggregate do
     %Aggregate{
       aggregate_state: aggregate_state,
       aggregate_version: aggregate_version,
-      lifespan_timeout: lifespan_timeout,
       snapshotting: snapshotting
     } = state
 
@@ -599,10 +602,7 @@ defmodule Commanded.Aggregates.Aggregate do
           state
       end
 
-    case lifespan_timeout do
-      {:stop, reason} -> {:stop, reason, state}
-      lifespan_timeout -> {:noreply, state, lifespan_timeout}
-    end
+    noreply_with_lifespan(state)
   end
 
   defp telemetry_start(telemetry_metadata) do
@@ -666,4 +666,22 @@ defmodule Commanded.Aggregates.Aggregate do
 
     "#{inspect(aggregate_module)}<#{aggregate_uuid}@#{aggregate_version}>"
   end
+
+  defp reply_with_lifespan(reply, state) do
+    %Aggregate{lifespan_timeout: lifespan_timeout} = state
+
+    case lifespan_timeout do
+      {:stop, reason} -> {:stop, reason, reply, state}
+      lifespan_timeout -> {:reply, reply, state, lifespan_timeout}
+    end
+  end
+
+  defp noreply_with_lifespan(state) do
+    %Aggregate{lifespan_timeout: lifespan_timeout} = state
+
+    case lifespan_timeout do
+      {:stop, reason} -> {:stop, reason, state}
+      lifespan_timeout -> {:noreply, state, lifespan_timeout}
+    end
+  end
 end


Test patch:
diff --git a/test/aggregates/aggregate_lifespan_test.exs b/test/aggregates/aggregate_lifespan_test.exs
index 7309ae8..f19e4a4 100644
--- a/test/aggregates/aggregate_lifespan_test.exs
+++ b/test/aggregates/aggregate_lifespan_test.exs
@@ -1,7 +1,7 @@
 defmodule Commanded.Aggregates.AggregateLifespanTest do
   use ExUnit.Case
 
-  alias Commanded.Aggregates.{DefaultLifespanRouter, LifespanAggregate, LifespanRouter}
+  alias Commanded.Aggregates.{Aggregate, DefaultLifespanRouter, LifespanAggregate, LifespanRouter}
   alias Commanded.Aggregates.LifespanAggregate.{Command, Event}
   alias Commanded.{DefaultApp, EventStore}
   alias Commanded.EventStore.RecordedEvent
@@ -99,6 +99,64 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do
       assert_receive {:DOWN, ^ref, :process, _pid, :normal}
     end
 
+    test "honours lifespan after a call to aggregate_state", %{
+      aggregate_uuid: aggregate_uuid,
+      ref: ref,
+      reply_to: reply_to
+    } do
+      command = %Command{
+        uuid: aggregate_uuid,
+        reply_to: reply_to,
+        action: :event,
+        lifespan: 500
+      }
+
+      :ok = LifespanRouter.dispatch(command, application: DefaultApp)
+
+      %{lifespan: 500} = Aggregate.aggregate_state(DefaultApp, LifespanAggregate, aggregate_uuid)
+
+      assert_receive {:DOWN, ^ref, :process, _pid, :normal}
+    end
+
+    test "honours lifespan after a call to aggregate_version", %{
+      aggregate_uuid: aggregate_uuid,
+      ref: ref,
+      reply_to: reply_to
+    } do
+      command = %Command{
+        uuid: aggregate_uuid,
+        reply_to: reply_to,
+        action: :event,
+        lifespan: 500
+      }
+
+      :ok = LifespanRouter.dispatch(command, application: DefaultApp)
+
+      1 = Aggregate.aggregate_version(DefaultApp, LifespanAggregate, aggregate_uuid)
+
+      assert_receive {:DOWN, ^ref, :process, _pid, :normal}
+    end
+
+    test "honours lifespan after an info message is handled", %{
+      aggregate_uuid: aggregate_uuid,
+      pid: pid,
+      ref: ref,
+      reply_to: reply_to
+    } do
+      command = %Command{
+        uuid: aggregate_uuid,
+        reply_to: reply_to,
+        action: :event,
+        lifespan: 500
+      }
+
+      :ok = LifespanRouter.dispatch(command, application: DefaultApp)
+
+      send(pid, :unexpected_message)
+
+      assert_receive {:DOWN, ^ref, :process, _pid, :normal}
+    end
+
     test "should call `after_command/1` callback function when no domain events", %{
       aggregate_uuid: aggregate_uuid,
       reply_to: reply_to