Repository: commanded/commanded (elixir)

Issue:
Process manager stops if `interested?/1` returns an empty list
When a process manager receives `{:start, []}`, `{:continue, []}`, or `{:stop, process_uuid}` it should just acknowledge receipt of the event and continue since it has nothing to do.

Pull request description:
Process manager stops if `interested?/1` returns an empty list
Fix bug whereby an empty list returned by a process manager's `interested?/1` function causes the process router to hang indefinitely.

Fixes #185.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 63461cd..da6210a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,6 +13,7 @@
 ### Bug fixes
 
 - Fixes the typespec for command dispatch ([#325](https://github.com/commanded/commanded/pull/325)).
+- Process manager stops if `interested?/1` returns an empty list ([#335](https://github.com/commanded/commanded/pull/335)).
 
 ## v1.0.0
 
diff --git a/lib/commanded/process_managers/process_manager_instance.ex b/lib/commanded/process_managers/process_manager_instance.ex
index 14c1ac8..ff7cfb9 100644
--- a/lib/commanded/process_managers/process_manager_instance.ex
+++ b/lib/commanded/process_managers/process_manager_instance.ex
@@ -44,15 +44,15 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   @doc """
   Checks whether or not the process manager has already processed events
   """
-  def new?(process_manager) do
-    GenServer.call(process_manager, :new?)
+  def new?(instance) do
+    GenServer.call(instance, :new?)
   end
 
   @doc """
   Handle the given event by delegating to the process manager module
   """
-  def process_event(process_manager, %RecordedEvent{} = event) do
-    GenServer.cast(process_manager, {:process_event, event})
+  def process_event(instance, %RecordedEvent{} = event) do
+    GenServer.cast(instance, {:process_event, event})
   end
 
   @doc """
@@ -60,15 +60,15 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
 
   Typically called when it has reached its final state.
   """
-  def stop(process_manager) do
-    GenServer.call(process_manager, :stop)
+  def stop(instance) do
+    GenServer.call(instance, :stop)
   end
 
   @doc """
   Fetch the process state of this instance
   """
-  def process_state(process_manager) do
-    GenServer.call(process_manager, :process_state)
+  def process_state(instance) do
+    GenServer.call(instance, :process_state)
   end
 
   @doc false
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index 882f037..794683d 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -60,30 +60,24 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:ok, state, {:continue, :subscribe_to_events}}
   end
 
-  @doc """
-  Acknowledge successful handling of the given event by a process manager instance
-  """
+  # Acknowledge successful handling of the given event by a process manager instance.
   def ack_event(process_router, %RecordedEvent{} = event, instance) do
     GenServer.cast(process_router, {:ack_event, event, instance})
   end
 
-  @doc false
   def process_instance(process_router, process_uuid) do
     GenServer.call(process_router, {:process_instance, process_uuid})
   end
 
-  @doc false
   def process_instances(process_router) do
     GenServer.call(process_router, :process_instances)
   end
 
-  @doc false
   @impl GenServer
   def handle_continue(:subscribe_to_events, %State{} = state) do
     {:noreply, subscribe_to_events(state)}
   end
 
-  @doc false
   @impl GenServer
   def handle_call(:process_instances, _from, %State{} = state) do
     %State{process_managers: process_managers} = state
@@ -93,21 +87,13 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:reply, reply, state}
   end
 
-  @doc false
   @impl GenServer
   def handle_call({:process_instance, process_uuid}, _from, %State{} = state) do
-    %State{process_managers: process_managers} = state
-
-    reply =
-      case Map.get(process_managers, process_uuid) do
-        nil -> {:error, :process_manager_not_found}
-        process_manager -> process_manager
-      end
+    reply = get_process_manager(state, process_uuid)
 
     {:reply, reply, state}
   end
 
-  @doc false
   @impl GenServer
   def handle_cast({:ack_event, event, instance}, %State{} = state) do
     %State{pending_acks: pending_acks} = state
@@ -132,12 +118,10 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:noreply, state}
   end
 
-  @doc false
   @impl GenServer
   def handle_cast(:process_pending_events, %State{pending_events: []} = state),
     do: {:noreply, state}
 
-  @doc false
   @impl GenServer
   def handle_cast(:process_pending_events, %State{} = state) do
     %State{pending_events: [event | pending_events]} = state
@@ -159,7 +143,6 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     end
   end
 
-  @doc false
   # Subscription to event store has successfully subscribed, init process router
   @impl GenServer
   def handle_info({:subscribed, subscription}, %State{subscription: subscription} = state) do
@@ -170,13 +153,13 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:noreply, %State{state | supervisor: supervisor}}
   end
 
-  @doc false
   @impl GenServer
   def handle_info({:events, events}, %State{} = state) do
     %State{pending_events: pending_events} = state
 
     Logger.debug(fn -> describe(state) <> " received #{length(events)} event(s)" end)
 
+    # Exclude already seen events
     unseen_events =
       events
       |> Enum.reject(&event_already_seen?(&1, state))
@@ -185,24 +168,23 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     state =
       case {pending_events, unseen_events} do
         {[], []} ->
-          # no pending or unseen events, so state is unmodified
+          # No pending or unseen events, so state is unmodified
           state
 
         {[], _} ->
-          # no pending events, but some unseen events so start processing them
+          # No pending events, but some unseen events so start processing them
           GenServer.cast(self(), :process_pending_events)
 
           %State{state | pending_events: unseen_events}
 
         {_, _} ->
-          # already processing pending events, append the unseen events so they are processed afterwards
+          # Already processing pending events, append the unseen events so they are processed afterwards
           %State{state | pending_events: pending_events ++ unseen_events}
       end
 
     {:noreply, state}
   end
 
-  @doc false
   # Shutdown process manager when processing an event has taken too long.
   @impl GenServer
   def handle_info({:event_timeout, event_number}, %State{} = state) do
@@ -224,7 +206,6 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     end
   end
 
-  @doc false
   # Stop process manager when event store subscription process terminates.
   @impl GenServer
   def handle_info(
@@ -236,7 +217,6 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:stop, reason, state}
   end
 
-  @doc false
   # Remove a process manager instance that has stopped with a normal exit reason.
   @impl GenServer
   def handle_info({:DOWN, _ref, :process, pid, :normal}, %State{} = state) do
@@ -247,7 +227,6 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:noreply, state}
   end
 
-  @doc false
   # Stop process router when a process manager instance terminates abnormally.
   @impl GenServer
   def handle_info({:DOWN, _ref, :process, _pid, reason}, %State{} = state) do
@@ -278,7 +257,6 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     %State{state | subscription: subscription, subscription_ref: subscription_ref}
   end
 
-  # Ignore already seen event
   defp event_already_seen?(
          %RecordedEvent{event_number: event_number},
          %State{last_seen_event: last_seen_event}
@@ -292,6 +270,9 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
     try do
       case process_manager_module.interested?(data) do
+        {:start, []} ->
+          ack_and_continue(event, state)
+
         {:start, process_uuid} ->
           Logger.debug(fn -> describe(state) <> " is interested in event " <> describe(event) end)
 
@@ -303,6 +284,9 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
             delegate_event(process_instance, event, state)
           end)
 
+        {:start!, []} ->
+          ack_and_continue(event, state)
+
         {:start!, process_uuid} ->
           Logger.debug(fn -> describe(state) <> " is interested in event " <> describe(event) end)
 
@@ -315,13 +299,10 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
               if ProcessManagerInstance.new?(process_instance) do
                 {state, [process_instance | process_instances]}
               else
-                throw(
-                  handle_routing_error(
-                    {:error, {:start!, :process_already_started}},
-                    event,
-                    state
-                  )
-                )
+                error = {:error, {:start!, :process_already_started}}
+                reply = handle_routing_error(error, event, state)
+
+                throw(reply)
               end
             end)
 
@@ -329,6 +310,9 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
           |> Enum.reverse()
           |> Enum.reduce(state, &delegate_event(&1, event, &2))
 
+        {:continue, []} ->
+          ack_and_continue(event, state)
+
         {:continue, process_uuid} ->
           Logger.debug(fn -> describe(state) <> " is interested in event " <> describe(event) end)
 
@@ -340,6 +324,9 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
             delegate_event(process_instance, event, state)
           end)
 
+        {:continue!, []} ->
+          ack_and_continue(event, state)
+
         {:continue!, process_uuid} ->
           Logger.debug(fn -> describe(state) <> " is interested in event " <> describe(event) end)
 
@@ -350,13 +337,10 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
               {process_instance, state} = start_or_continue_process_manager(process_uuid, state)
 
               if ProcessManagerInstance.new?(process_instance) do
-                throw(
-                  handle_routing_error(
-                    {:error, {:continue!, :process_not_started}},
-                    event,
-                    state
-                  )
-                )
+                error = {:error, {:continue!, :process_not_started}}
+                reply = handle_routing_error(error, event, state)
+
+                throw(reply)
               else
                 {state, [process_instance | process_instances]}
               end
@@ -438,13 +422,11 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   end
 
   defp start_or_continue_process_manager(process_uuid, %State{} = state) do
-    %State{process_managers: process_managers} = state
-
-    case Map.get(process_managers, process_uuid) do
-      process_manager when is_pid(process_manager) ->
+    case get_process_manager(state, process_uuid) do
+      {:ok, process_manager} ->
         {process_manager, state}
 
-      nil ->
+      {:error, :process_manager_not_found} ->
         start_process_manager(process_uuid, state)
     end
   end
@@ -483,14 +465,14 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   defp stop_process_manager(process_uuid, %State{} = state) do
     %State{process_managers: process_managers} = state
 
-    case Map.get(process_managers, process_uuid) do
-      nil ->
-        state
-
-      process_manager ->
+    case get_process_manager(state, process_uuid) do
+      {:ok, process_manager} ->
         :ok = ProcessManagerInstance.stop(process_manager)
 
         %State{state | process_managers: Map.delete(process_managers, process_uuid)}
+
+      {:error, :process_manager_not_found} ->
+        state
     end
   end
 
@@ -504,6 +486,15 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     end)
   end
 
+  defp get_process_manager(%State{} = state, process_uuid) do
+    %State{process_managers: process_managers} = state
+
+    case Map.get(process_managers, process_uuid) do
+      process_manager when is_pid(process_manager) -> {:ok, process_manager}
+      nil -> {:error, :process_manager_not_found}
+    end
+  end
+
   defp do_ack_event(event, %State{} = state) do
     %State{
       application: application,


Test patch:
diff --git a/test/process_managers/dynamic_process_manager_application_test.exs b/test/process_managers/dynamic_process_manager_application_test.exs
index abe97a9..d65cf0a 100644
--- a/test/process_managers/dynamic_process_manager_application_test.exs
+++ b/test/process_managers/dynamic_process_manager_application_test.exs
@@ -67,9 +67,8 @@ defmodule Commanded.ProcessManager.DynamicProcessManagerApplicationTest do
 
   defp wait_for_process_instance(process_router, aggregate_uuid) do
     Wait.until(fn ->
-      process_instance = ProcessRouter.process_instance(process_router, aggregate_uuid)
-
-      assert is_pid(process_instance)
+      assert {:ok, process_instance} =
+               ProcessRouter.process_instance(process_router, aggregate_uuid)
 
       process_instance
     end)
diff --git a/test/process_managers/process_manager_idle_timeout_test.exs b/test/process_managers/process_manager_idle_timeout_test.exs
index 6b1e431..7e9c568 100644
--- a/test/process_managers/process_manager_idle_timeout_test.exs
+++ b/test/process_managers/process_manager_idle_timeout_test.exs
@@ -79,13 +79,11 @@ defmodule Commanded.ProcessManagers.ProcessManagerIdleTimeoutTest do
 
   defp wait_for_process_instance(pm, process_uuid) do
     Wait.until(fn ->
-      with instance <- ProcessRouter.process_instance(pm, process_uuid) do
-        assert is_pid(instance)
+      assert {:ok, instance} = ProcessRouter.process_instance(pm, process_uuid)
 
-        ref = Process.monitor(instance)
+      ref = Process.monitor(instance)
 
-        {instance, ref}
-      end
+      {instance, ref}
     end)
   end
 end
diff --git a/test/process_managers/process_manager_routing_test.exs b/test/process_managers/process_manager_routing_test.exs
index e36e917..23440ff 100644
--- a/test/process_managers/process_manager_routing_test.exs
+++ b/test/process_managers/process_manager_routing_test.exs
@@ -11,22 +11,7 @@ defmodule Commanded.ProcessManagers.ProcessManagerRoutingTest do
   alias Commanded.ProcessManagers.RoutingProcessManager.Stopped
 
   setup do
-    expect(MockEventStore, :subscribe_to, fn
-      _event_store, :all, name, pid, :origin ->
-        assert is_binary(name)
-        assert is_pid(pid)
-
-        send(pid, {:subscribed, self()})
-
-        {:ok, self()}
-    end)
-
-    stub(MockEventStore, :read_snapshot, fn _event_store, _snapshot_uuid ->
-      {:error, :snapshot_not_found}
-    end)
-
-    stub(MockEventStore, :record_snapshot, fn _event_store, _snapshot -> :ok end)
-    stub(MockEventStore, :ack_event, fn _event_store, _pid, _event -> :ok end)
+    mock_event_store()
 
     {:ok, pid} = RoutingProcessManager.start_link()
 
@@ -88,6 +73,21 @@ defmodule Commanded.ProcessManagers.ProcessManagerRoutingTest do
       assert_receive {:started, ^instance}
       assert_receive {:DOWN, ^ref, :process, ^instance, :normal}
     end
+
+    test "should ignore an empty list returned on `:start`", %{
+      pid: pid,
+      process_uuid: process_uuid
+    } do
+      send_events(pid, [
+        %Started{process_uuid: [], reply_to: self()},
+        %Started{process_uuid: process_uuid, reply_to: self()}
+      ])
+
+      instance = wait_for_instance(pid, process_uuid)
+
+      assert_receive {:started, ^instance}
+      refute_receive {:started, _instance}
+    end
   end
 
   describe "process manager strict routing" do
@@ -179,6 +179,25 @@ defmodule Commanded.ProcessManagers.ProcessManagerRoutingTest do
     end
   end
 
+  defp mock_event_store do
+    expect(MockEventStore, :subscribe_to, fn
+      _event_store, :all, name, pid, :origin ->
+        assert is_binary(name)
+        assert is_pid(pid)
+
+        send(pid, {:subscribed, self()})
+
+        {:ok, self()}
+    end)
+
+    stub(MockEventStore, :read_snapshot, fn _event_store, _snapshot_uuid ->
+      {:error, :snapshot_not_found}
+    end)
+
+    stub(MockEventStore, :record_snapshot, fn _event_store, _snapshot -> :ok end)
+    stub(MockEventStore, :ack_event, fn _event_store, _pid, _event -> :ok end)
+  end
+
   defp send_events(pid, events, initial_event_number \\ 1) do
     recorded_events = EventFactory.map_to_recorded_events(events, initial_event_number)
 
@@ -187,9 +206,7 @@ defmodule Commanded.ProcessManagers.ProcessManagerRoutingTest do
 
   defp wait_for_instance(pid, process_uuid) do
     Wait.until(fn ->
-      instance = ProcessRouter.process_instance(pid, process_uuid)
-
-      assert is_pid(instance)
+      assert {:ok, instance} = ProcessRouter.process_instance(pid, process_uuid)
 
       instance
     end)
diff --git a/test/process_managers/process_manager_timeout_test.exs b/test/process_managers/process_manager_timeout_test.exs
index dd878a0..95fde5a 100644
--- a/test/process_managers/process_manager_timeout_test.exs
+++ b/test/process_managers/process_manager_timeout_test.exs
@@ -55,11 +55,9 @@ defmodule Commanded.ProcessManagers.ProcessManagerTimeoutTest do
 
   defp wait_for_process_instance(process_router, aggregate_uuid) do
     Wait.until(fn ->
-      process_instance = ProcessRouter.process_instance(process_router, aggregate_uuid)
+      assert {:ok, instance} = ProcessRouter.process_instance(process_router, aggregate_uuid)
 
-      assert is_pid(process_instance)
-
-      process_instance
+      instance
     end)
   end
 end
diff --git a/test/process_managers/process_router_process_pending_events_test.exs b/test/process_managers/process_router_process_pending_events_test.exs
index f23885c..95559fb 100644
--- a/test/process_managers/process_router_process_pending_events_test.exs
+++ b/test/process_managers/process_router_process_pending_events_test.exs
@@ -180,10 +180,11 @@ defmodule Commanded.ProcessManagers.ProcessRouterProcessPendingEventsTest do
     wait_for_event(ExampleApp, Interested, fn event -> event.index == 6 end)
 
     Wait.until(fn ->
-      process_instance = ProcessRouter.process_instance(process_router, aggregate_uuid)
-      refute process_instance == {:error, :process_manager_not_found}
+      assert {:ok, process_instance} =
+               ProcessRouter.process_instance(process_router, aggregate_uuid)
 
       %{items: items} = ProcessManagerInstance.process_state(process_instance)
+
       assert items == [1, 2, 3, 4, 5, 6]
     end)
   end
diff --git a/test/process_managers/resume_process_manager_test.exs b/test/process_managers/resume_process_manager_test.exs
index 71b7da5..db8a05b 100644
--- a/test/process_managers/resume_process_manager_test.exs
+++ b/test/process_managers/resume_process_manager_test.exs
@@ -31,9 +31,9 @@ defmodule Commanded.ProcessManagers.ResumeProcessManagerTest do
 
     # wait for process instance to receive event
     Wait.until(fn ->
-      process_instance = ProcessRouter.process_instance(process_router, process_uuid)
+      assert {:ok, process_instance} =
+               ProcessRouter.process_instance(process_router, process_uuid)
 
-      assert process_instance != {:error, :process_manager_not_found}
       assert %{status_history: ["start"]} = ProcessManagerInstance.process_state(process_instance)
     end)
 
@@ -51,8 +51,8 @@ defmodule Commanded.ProcessManagers.ResumeProcessManagerTest do
     wait_for_event(ResumeApp, ProcessResumed, fn event -> event.process_uuid == process_uuid end)
 
     Wait.until(fn ->
-      process_instance = ProcessRouter.process_instance(process_router, process_uuid)
-      assert process_instance != {:error, :process_manager_not_found}
+      assert {:ok, process_instance} =
+               ProcessRouter.process_instance(process_router, process_uuid)
 
       state = ProcessManagerInstance.process_state(process_instance)
       assert state.status_history == ["start", "resume"]