Repository: commanded/commanded (elixir)

Issue:
Process manager instance should log and ignore unexpected info messages
A process manager instance process will crash if it receives an unexpected info message. Instead of crashing, a catch-all `handle_info/2` callback should log the message and ignore it.

It is also worth considering whether to allow a process manager to handle info messages by implementing its own optional `handle_info/2` callback function.

Pull request:
Log and ignore unexpected messages received by event handlers and process manager instances
Implement catch-all `handle_info/2` callback function for event handler and process manager instance `GenServer` processes.

Fixes #332.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 26c5533..5af3e66 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,7 @@
 
 - Dynamic Commanded applications ([#324](https://github.com/commanded/commanded/pull/324)).
 - Command dispatch return ([#331](https://github.com/commanded/commanded/pull/331)).
+- Log and ignore unexpected messages received by event handlers and process manager instances ([#333](https://github.com/commanded/commanded/pull/333))
 
 ### Bug fixes
 
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index f3d49d0..96735c7 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -548,6 +548,14 @@ defmodule Commanded.Event.Handler do
     {:stop, reason, state}
   end
 
+  @doc false
+  @impl GenServer
+  def handle_info(message, state) do
+    Logger.error(fn -> describe(state) <> " received unexpected message: " <> inspect(message) end)
+
+    {:noreply, state}
+  end
+
   # Register this event handler as a subscription with the given consistency.
   defp register_subscription(%Handler{} = state) do
     %Handler{application: application, consistency: consistency, handler_name: name} = state
diff --git a/lib/commanded/process_managers/process_manager_instance.ex b/lib/commanded/process_managers/process_manager_instance.ex
index fbc0705..14c1ac8 100644
--- a/lib/commanded/process_managers/process_manager_instance.ex
+++ b/lib/commanded/process_managers/process_manager_instance.ex
@@ -144,6 +144,14 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
     {:stop, :normal, state}
   end
 
+  @doc false
+  @impl GenServer
+  def handle_info(message, state) do
+    Logger.error(fn -> describe(state) <> " received unexpected message: " <> inspect(message) end)
+
+    {:noreply, state}
+  end
+
   defp event_already_seen?(%RecordedEvent{}, %State{last_seen_event: nil}),
     do: false
 
diff --git a/lib/commanded/registration/local_registry.ex b/lib/commanded/registration/local_registry.ex
index c1351c8..ce4c55a 100644
--- a/lib/commanded/registration/local_registry.ex
+++ b/lib/commanded/registration/local_registry.ex
@@ -4,6 +4,8 @@ defmodule Commanded.Registration.LocalRegistry do
   `Registry` module.
   """
 
+  require Logger
+
   @behaviour Commanded.Registration.Adapter
 
   @doc """
@@ -108,7 +110,9 @@ defmodule Commanded.Registration.LocalRegistry do
   end
 
   @doc false
-  def handle_info(_msg, state) do
+  def handle_info(message, state) do
+    Logger.debug(fn -> "received unexpected message in handle_info/2: " <> inspect(message) end)
+
     {:noreply, state}
   end
 


Test patch:
diff --git a/config/test.exs b/config/test.exs
index 341e388..87afc8d 100644
--- a/config/test.exs
+++ b/config/test.exs
@@ -3,7 +3,7 @@ use Mix.Config
 alias Commanded.EventStore.Adapters.InMemory
 alias Commanded.Serialization.JsonSerializer
 
-config :logger, :console, level: :warn, format: "[$level] $message\n"
+config :logger, :console, level: :debug, format: "[$level] $message\n"
 
 config :ex_unit,
   capture_log: true,
diff --git a/test/aggregates/aggregate_test.exs b/test/aggregates/aggregate_test.exs
new file mode 100644
index 0000000..dd9ad70
--- /dev/null
+++ b/test/aggregates/aggregate_test.exs
@@ -0,0 +1,48 @@
+defmodule Commanded.Aggregates.AggregateTest do
+  use ExUnit.Case, async: true
+
+  alias Commanded.Aggregates.Aggregate
+  alias Commanded.DefaultApp
+
+  defmodule ExampleAggregate do
+    defstruct [:uuid]
+  end
+
+  setup do
+    start_supervised!(DefaultApp)
+    :ok
+  end
+
+  describe "aggregate process" do
+    setup do
+      aggregate_uuid = UUID.uuid4()
+
+      {:ok, pid} = start_aggregate(aggregate_uuid)
+
+      [aggregate_uuid: aggregate_uuid, pid: pid]
+    end
+
+    test "ignore unexpected messages", %{pid: pid} do
+      import ExUnit.CaptureLog
+
+      ref = Process.monitor(pid)
+
+      send_unexpected_mesage = fn ->
+        send(pid, :unexpected_message)
+
+        refute_receive {:DOWN, ^ref, :process, ^pid, _}
+      end
+
+      assert capture_log(send_unexpected_mesage) =~
+               "received unexpected message in handle_info/2: :unexpected_message"
+    end
+  end
+
+  def start_aggregate(aggregate_uuid) do
+    Aggregate.start_link([application: DefaultApp],
+      application: DefaultApp,
+      aggregate_module: ExampleAggregate,
+      aggregate_uuid: aggregate_uuid
+    )
+  end
+end
diff --git a/test/aggregates/execute_command_test.exs b/test/aggregates/execute_command_test.exs
index 9af0fca..d237c2b 100644
--- a/test/aggregates/execute_command_test.exs
+++ b/test/aggregates/execute_command_test.exs
@@ -125,7 +125,7 @@ defmodule Commanded.Aggregates.ExecuteCommandTest do
       assert_event_result(fn %Command{id: id} -> {:ok, %Event{id: id}} end)
     end
 
-    test "should allow `{:ok, event}` return value" do
+    test "should allow `{:ok, [event]}` tagged tuple return value" do
       assert_event_result(fn %Command{id: id} -> {:ok, [%Event{id: id}]} end)
     end
   end
diff --git a/test/event/handle_event_test.exs b/test/event/handle_event_test.exs
index d2cd4dc..3ba8ceb 100644
--- a/test/event/handle_event_test.exs
+++ b/test/event/handle_event_test.exs
@@ -62,6 +62,21 @@ defmodule Commanded.Event.HandleEventTest do
         assert AccountBalanceHandler.current_balance() == 1_050
       end)
     end
+
+    test "should ignore unexpected messages", %{handler: handler} do
+      import ExUnit.CaptureLog
+
+      ref = Process.monitor(handler)
+
+      send_unexpected_mesage = fn ->
+        send(handler, :unexpected_message)
+
+        refute_receive {:DOWN, ^ref, :process, ^handler, _}
+      end
+
+      assert capture_log(send_unexpected_mesage) =~
+               "Commanded.ExampleDomain.BankAccount.AccountBalanceHandler received unexpected message: :unexpected_message"
+    end
   end
 
   describe "reset event handler" do
diff --git a/test/process_managers/process_manager_instance_test.exs b/test/process_managers/process_manager_instance_test.exs
index ff81f67..0ba7a32 100644
--- a/test/process_managers/process_manager_instance_test.exs
+++ b/test/process_managers/process_manager_instance_test.exs
@@ -1,99 +1,154 @@
 defmodule Commanded.ProcessManagers.ProcessManagerInstanceTest do
-  use Commanded.StorageCase
+  use ExUnit.Case
 
-  alias Commanded.ExampleDomain.BankAccount.Commands.OpenAccount
-  alias Commanded.ExampleDomain.BankApp
-  alias Commanded.ExampleDomain.BankRouter
+  import Mox
+
+  alias Commanded.Application.Config
+  alias Commanded.Application.Mock, as: MockApplication
+  alias Commanded.ExampleDomain.BankAccount.Commands.WithdrawMoney
   alias Commanded.ExampleDomain.MoneyTransfer.Events.MoneyTransferRequested
   alias Commanded.ExampleDomain.TransferMoneyProcessManager
+  alias Commanded.EventStore.Adapters.Mock, as: MockEventStore
   alias Commanded.EventStore.RecordedEvent
+  alias Commanded.EventStore.SnapshotData
   alias Commanded.ProcessManagers.ProcessManagerInstance
-  alias Commanded.Helpers.CommandAuditMiddleware
-  alias Commanded.Helpers.ProcessHelper
 
-  setup do
-    start_supervised!(CommandAuditMiddleware)
-    start_supervised!(BankApp)
+  setup :set_mox_global
+  setup :verify_on_exit!
 
-    :ok
+  setup do
+    Config.associate(self(), MockApplication,
+      application: MockApplication,
+      event_store: {MockEventStore, %{}}
+    )
   end
 
-  test "process manager handles an event" do
-    transfer_uuid = UUID.uuid4()
-    account1_uuid = UUID.uuid4()
-    account2_uuid = UUID.uuid4()
-
-    :ok = open_account(account1_uuid, 1_000)
-    :ok = open_account(account2_uuid, 500)
-
-    {:ok, process_manager} =
-      ProcessManagerInstance.start_link(
-        application: BankApp,
-        idle_timeout: :infinity,
-        process_manager_name: "TransferMoneyProcessManager",
-        process_manager_module: TransferMoneyProcessManager,
-        process_router: self(),
-        process_uuid: transfer_uuid
-      )
-
-    event = %RecordedEvent{
-      event_number: 1,
-      stream_id: "stream-id",
-      stream_version: 1,
-      data: %MoneyTransferRequested{
-        transfer_uuid: transfer_uuid,
-        debit_account: account1_uuid,
-        credit_account: account2_uuid,
-        amount: 100
+  describe "process manager instance" do
+    test "handles an event and dispatches a command" do
+      transfer_uuid = UUID.uuid4()
+      debit_account = UUID.uuid4()
+      credit_account = UUID.uuid4()
+      expected_source_uuid = "\"TransferMoneyProcessManager\"-\"#{transfer_uuid}\""
+
+      expect(MockEventStore, :read_snapshot, fn _adapter_meta, ^expected_source_uuid ->
+        {:error, :snapshot_not_found}
+      end)
+
+      expect(MockEventStore, :record_snapshot, fn _adapter_meta, snapshot ->
+        assert %SnapshotData{
+                 data: %TransferMoneyProcessManager{
+                   amount: 100,
+                   credit_account: ^credit_account,
+                   debit_account: ^debit_account,
+                   status: :withdraw_money_from_debit_account,
+                   transfer_uuid: ^transfer_uuid
+                 },
+                 source_type: "Elixir.Commanded.ExampleDomain.TransferMoneyProcessManager",
+                 source_uuid: expected_source_uuid,
+                 source_version: 1
+               } = snapshot
+
+        :ok
+      end)
+
+      expect(MockApplication, :dispatch, fn command, _opts ->
+        assert %WithdrawMoney{
+                 account_number: ^debit_account,
+                 transfer_uuid: ^transfer_uuid,
+                 amount: 100
+               } = command
+
+        :ok
+      end)
+
+      {:ok, instance} = start_process_manager_instance(transfer_uuid)
+
+      event = %RecordedEvent{
+        event_number: 1,
+        stream_id: "stream-id",
+        stream_version: 1,
+        data: %MoneyTransferRequested{
+          transfer_uuid: transfer_uuid,
+          debit_account: debit_account,
+          credit_account: credit_account,
+          amount: 100
+        }
       }
-    }
 
-    :ok = ProcessManagerInstance.process_event(process_manager, event)
+      :ok = ProcessManagerInstance.process_event(instance, event)
 
-    # Should send ack to process router after processing event
-    assert_receive({:"$gen_cast", {:ack_event, ^event, _instance}}, 1_000)
+      # Should send ack to process router after processing event
+      assert_receive({:"$gen_cast", {:ack_event, ^event, _instance}}, 1_000)
+    end
 
-    ProcessHelper.shutdown(process_manager)
-  end
+    test "ignore unexpected messages" do
+      import ExUnit.CaptureLog
 
-  test "should provide `__name__/0` function" do
-    assert TransferMoneyProcessManager.__name__() ==
-             "Commanded.ExampleDomain.TransferMoneyProcessManager"
-  end
+      transfer_uuid = UUID.uuid4()
 
-  test "should ensure a process manager application is provided" do
-    assert_raise ArgumentError, "NoAppProcessManager expects :application option", fn ->
-      Code.eval_string("""
-        defmodule NoAppProcessManager do
-          use Commanded.ProcessManagers.ProcessManager, name: __MODULE__
-        end
-      """)
+      expect(MockEventStore, :read_snapshot, fn _adapter_meta, _source_uuid ->
+        {:error, :snapshot_not_found}
+      end)
+
+      {:ok, instance} = start_process_manager_instance(transfer_uuid)
+
+      ref = Process.monitor(instance)
+
+      send_unexpected_mesage = fn ->
+        send(instance, :unexpected_message)
+
+        refute_receive {:DOWN, ^ref, :process, ^instance, _}
+      end
+
+      assert capture_log(send_unexpected_mesage) =~
+               "Commanded.ExampleDomain.TransferMoneyProcessManager received unexpected message: :unexpected_message"
+    end
+
+    test "should provide `__name__/0` function" do
+      assert TransferMoneyProcessManager.__name__() ==
+               "Commanded.ExampleDomain.TransferMoneyProcessManager"
+    end
+
+    test "should ensure a process manager application is provided" do
+      assert_raise ArgumentError, "NoAppProcessManager expects :application option", fn ->
+        Code.eval_string("""
+          defmodule NoAppProcessManager do
+            use Commanded.ProcessManagers.ProcessManager, name: __MODULE__
+          end
+        """)
+      end
     end
-  end
 
-  test "should ensure a process manager name is provided" do
-    assert_raise ArgumentError, "UnnamedProcessManager expects :name option", fn ->
+    test "should ensure a process manager name is provided" do
+      assert_raise ArgumentError, "UnnamedProcessManager expects :name option", fn ->
+        Code.eval_string("""
+          defmodule UnnamedProcessManager do
+            use Commanded.ProcessManagers.ProcessManager, application: Commanded.DefaultApp
+          end
+        """)
+      end
+    end
+
+    test "should allow using process manager module as name" do
       Code.eval_string("""
-        defmodule UnnamedProcessManager do
-          use Commanded.ProcessManagers.ProcessManager, application: Commanded.DefaultApp
+        defmodule MyProcessManager do
+          use Commanded.ProcessManagers.ProcessManager,
+            application: Commanded.DefaultApp,
+            name: __MODULE__
         end
       """)
     end
   end
 
-  test "should allow using process manager module as name" do
-    Code.eval_string("""
-      defmodule MyProcessManager do
-        use Commanded.ProcessManagers.ProcessManager,
-          application: Commanded.DefaultApp,
-          name: __MODULE__
-      end
-    """)
-  end
-
-  defp open_account(account_number, initial_balance) do
-    command = %OpenAccount{account_number: account_number, initial_balance: initial_balance}
-
-    BankRouter.dispatch(command, application: BankApp)
+  defp start_process_manager_instance(transfer_uuid) do
+    start_supervised(
+      {ProcessManagerInstance,
+       application: MockApplication,
+       idle_timeout: :infinity,
+       process_manager_name: "TransferMoneyProcessManager",
+       process_manager_module: TransferMoneyProcessManager,
+       process_router: self(),
+       process_uuid: transfer_uuid}
+    )
   end
 end
diff --git a/test/support/mocks.ex b/test/support/mocks.ex
index abc03e9..6bfad3d 100644
--- a/test/support/mocks.ex
+++ b/test/support/mocks.ex
@@ -1,2 +1,3 @@
 Mox.defmock(Commanded.EventStore.Adapters.Mock, for: Commanded.EventStore.Adapter)
 Mox.defmock(Commanded.Commands.MockRouter, for: Commanded.Commands.Router)
+Mox.defmock(Commanded.Application.Mock, for: Commanded.Application)