Repository: commanded/commanded (elixir)

Issue:
ProcessManager locking Router dispatch (apparently)
The process manager is apparently locking the dispatch process of a command - I am getting a `:consistency_timeout` from the Router (timeout set to 15 seconds on test `env` that is more than enough for all listeners to finish). 

I've disabled all event listeners but the Process Manager - to make sure the issue wasn't any of the projectors - and the timeout persists. 

Can someone please take a look on my Process Manager and let me know if I am missing any important detail on the Manager configuration or if indeed it is a problem with the Manager?

The code below creates 4 stages in sequence. The last stage created has the field `next_stage_id: nil` and its event is not handled (it goes into the clause of the `interested?/1` function that should stop the manager).
```
defmodule MyApp.Methodologies.ProcessManagers.CreateGroupStagesProcess do
  use Commanded.ProcessManagers.ProcessManager,
    name: "Methodologies.ProcessManagers.CreateGroupStagesProcess",
    consistency: Application.get_env(:commanded, :consistency),
    router: MyApp.Methodologies.Router

  alias MyApp.Methodologies
  alias MyApp.Methodologies.ProcessManagers.CreateGroupStagesProcess
  alias MyApp.Methodologies.Commands.{
    CreateStage,
    SetNextStageId,
  }
  alias MyApp.Methodologies.Events.{
    GroupCreated,
    StageCreated
  }
  alias MyApp.Methodologies.Projections.Group

  defstruct [
    group_id: "",
    methodology: "",
    remaining_stage_types: [],
  ]

  def interested?(%GroupCreated{group_id: group_id}), do: {:start, group_id}
  def interested?(%StageCreated{group_id: group_id, next_stage_id: nil}), do: {:stop, group_id}
  def interested?(%StageCreated{group_id: group_id}), do: {:continue, group_id}
  def interested?(_event), do: false


  def handle(
    %CreateGroupStagesProcess{},
    %GroupCreated{group_id: group_id, methodology: methodology}
  ) do
    group = %Group{id: group_id, methodology: methodology}

    [first_stage | remaining_stage_types] = Methodologies.list_methodology_stages(group)
    next_stage_id = last_stage_id(remaining_stage_types)

    CreateStage.create(%{type: first_stage, previous_stage_id: nil, next_stage_id: next_stage_id}, group)
  end

  def handle(
    %CreateGroupStagesProcess{
      group_id: group_id, 
      methodology: methodology,
      remaining_stage_types: [next_stage_type | remaining_stage_types]}, 
    %StageCreated{group_id: group_id, stage_id: stage_id}
  ) do
    group = %Group{id: group_id, methodology: methodology}
    next_stage_id = last_stage_id(remaining_stage_types)

    create_stage = 
      CreateStage.create(%{type: next_stage_type, previous_stage_id: stage_id, next_stage_id: next_stage_id}, group)
    set_next_stage_id_of_previous_stage = 
      %SetNextStageId{stage_id: stage_id, next_stage_id: create_stage.stage_id}

    [create_stage, set_next_stage_id_of_previous_stage]
  end


  ## state mutators
  def apply(
    %CreateGroupStagesProcess{} = process, 
    %GroupCreated{group_id: group_id, methodology: methodology}
  ) do
    [_ | remaining_stage_types] = Methodologies.list_methodology_stages(methodology)
    %CreateGroupStagesProcess{process |
      group_id: group_id,
      methodology: methodology,
      remaining_stage_types: remaining_stage_types
    }
  end

  def apply(
    %CreateGroupStagesProcess{
      group_id: group_id,
      remaining_stage_types: [_ | remaining_stage_types]} = process, 
    %StageCreated{group_id: group_id}
  ) do
    %CreateGroupStagesProcess{process | remaining_stage_types: remaining_stage_types}
  end

  ## private helpers
  defp last_stage_id([]), do: nil
  defp last_stage_id([_|_]), do: ""
end
```

Pull request description:
Ensure process managers can be configured with `:strong` consistency
Fixes #170.

Patch:
diff --git a/.travis.yml b/.travis.yml
index abea2f1..9a833fe 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,7 +1,7 @@
 language: elixir
 
 elixir:
-  - 1.5.3
+  - 1.6.4
 
 otp_release:
   - 20.2
diff --git a/config/dev.exs b/config/dev.exs
index d2d855e..0ede7b5 100644
--- a/config/dev.exs
+++ b/config/dev.exs
@@ -1,1 +1,3 @@
 use Mix.Config
+
+config :mix_test_watch, clear: true
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index 2f20670..6430e6f 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -28,7 +28,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
       supervisor: nil,
       subscription: nil,
       last_seen_event: nil,
-      pending_acks: [],
+      pending_acks: %{},
       pending_events: [],
     ]
   end
@@ -96,21 +96,25 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
   def handle_cast({:ack_event, event, instance}, %State{} = state) do
     %State{pending_acks: pending_acks} = state
+    %RecordedEvent{event_number: event_number} = event
 
-    case List.delete(pending_acks, instance) do
-      [] ->
-        # no pending acks so confirm receipt of event
-        state = confirm_receipt(event, %State{state | pending_acks: []})
+    state =
+      case pending_acks |> Map.get(event_number, []) |> List.delete(instance) do
+        [] ->
+          # Enqueue a message to continue processing any pending events
+          GenServer.cast(self(), :process_pending_events)
 
-        # continue processing any pending events
-        GenServer.cast(self(), :process_pending_events)
+          state = %State{state | pending_acks: Map.delete(pending_acks, event_number)}
 
-        {:noreply, state}
+          # no pending acks so confirm receipt of event
+          confirm_receipt(event, state)
 
-      pending ->
-        # pending acks, don't ack event but wait for outstanding instances
-        {:noreply, %State{state | pending_acks: pending}}
-    end
+        pending ->
+          # pending acks, don't ack event but wait for outstanding instances
+          %State{state | pending_acks: Map.put(pending_acks, event_number, pending)}
+      end
+
+    {:noreply, state}
   end
 
   @doc """
@@ -336,10 +340,15 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   # Delegate event to process instance who will ack event processing on success
   defp delegate_event(process_instance, %RecordedEvent{} = event, %State{} = state) do
     %State{pending_acks: pending_acks} = state
+    %RecordedEvent{event_number: event_number} = event
 
     :ok = ProcessManagerInstance.process_event(process_instance, event, self())
 
-    %State{state | pending_acks: [process_instance | pending_acks]}
+    %State{state |
+      pending_acks: Map.update(pending_acks, event_number, [process_instance], fn
+        pending -> [process_instance | pending]
+      end)
+    }
   end
 
   defp describe(%State{process_manager_module: process_manager_module}),


Test patch:
diff --git a/test/commands/dispatch_consistency_test.exs b/test/commands/dispatch_consistency_test.exs
index 77d9cef..d618fe6 100644
--- a/test/commands/dispatch_consistency_test.exs
+++ b/test/commands/dispatch_consistency_test.exs
@@ -9,6 +9,7 @@ defmodule Commanded.Commands.DispatchConsistencyTest do
     ExecutionResult,
     OptionalStronglyConsistentEventHandler,
     StronglyConsistentEventHandler,
+    StronglyConsistentProcessManager
   }
   alias Commanded.EventStore
   alias Commanded.Helpers.ProcessHelper
@@ -18,54 +19,38 @@ defmodule Commanded.Commands.DispatchConsistencyTest do
     RequestDispatchCommand,
   }
 
-  setup do
-    {:ok, handler1} = StronglyConsistentEventHandler.start_link()
-    {:ok, handler2} = EventuallyConsistentEventHandler.start_link()
+  describe "event handlers" do
+    setup :start_event_handlers
 
-    on_exit fn ->
-      ProcessHelper.shutdown(handler1)
-      ProcessHelper.shutdown(handler2)
+    test "should wait for strongly consistent event handler to handle event" do
+      command = %ConsistencyCommand{uuid: UUID.uuid4(), delay: 0}
+      assert :ok = ConsistencyRouter.dispatch(command, consistency: :strong)
     end
 
-    :ok
-  end
-
-  test "should wait for strongly consistent event handler to handle event" do
-    command = %ConsistencyCommand{uuid: UUID.uuid4(), delay: 0}
-    assert :ok = ConsistencyRouter.dispatch(command, consistency: :strong)
-  end
-
-  # default consistency timeout set to 100ms test config
-  test "should timeout waiting for strongly consistent event handler to handle event" do
-    command = %ConsistencyCommand{uuid: UUID.uuid4(), delay: 5_000}
-    assert {:error, :consistency_timeout} = ConsistencyRouter.dispatch(command, consistency: :strong)
-  end
+    # default consistency timeout set to 100ms test config
+    test "should timeout waiting for strongly consistent event handler to handle event" do
+      command = %ConsistencyCommand{uuid: UUID.uuid4(), delay: 5_000}
+      assert {:error, :consistency_timeout} = ConsistencyRouter.dispatch(command, consistency: :strong)
+    end
 
-  test "should not wait when command creates no events" do
-    command = %NoOpCommand{uuid: UUID.uuid4()}
-    assert :ok = ConsistencyRouter.dispatch(command, consistency: :strong)
-  end
+    test "should not wait when command creates no events" do
+      command = %NoOpCommand{uuid: UUID.uuid4()}
+      assert :ok = ConsistencyRouter.dispatch(command, consistency: :strong)
+    end
 
-  test "should allow strongly consistent event handler to dispatch a command" do
-    command = %RequestDispatchCommand{uuid: UUID.uuid4(), delay: 0}
-    assert :ok = ConsistencyRouter.dispatch(command, consistency: :strong)
-  end
+    test "should allow strongly consistent event handler to dispatch a command" do
+      command = %RequestDispatchCommand{uuid: UUID.uuid4(), delay: 0}
+      assert :ok = ConsistencyRouter.dispatch(command, consistency: :strong)
+    end
 
-  test "should timeout waiting for strongly consistent handler dispatching a command" do
-    command = %RequestDispatchCommand{uuid: UUID.uuid4(), delay: 5_000}
-    assert {:error, :consistency_timeout} = ConsistencyRouter.dispatch(command, consistency: :strong)
+    test "should timeout waiting for strongly consistent handler dispatching a command" do
+      command = %RequestDispatchCommand{uuid: UUID.uuid4(), delay: 5_000}
+      assert {:error, :consistency_timeout} = ConsistencyRouter.dispatch(command, consistency: :strong)
+    end
   end
 
   describe "optional `:strong` consistency" do
-    setup do
-      {:ok, handler3} = OptionalStronglyConsistentEventHandler.start_link()
-
-      on_exit fn ->
-        ProcessHelper.shutdown(handler3)
-      end
-
-      :ok
-    end
+    setup [:start_event_handlers, :start_optional_handler]
 
     test "should only wait for opt-in strongly consistent event handler to handle event" do
       command = %ConsistencyCommand{uuid: UUID.uuid4(), delay: 100}
@@ -84,6 +69,8 @@ defmodule Commanded.Commands.DispatchConsistencyTest do
   end
 
   describe "aggregate identity prefix" do
+    setup :start_event_handlers
+
     test "should wait for strongly consistent event handler to handle event" do
       uuid = UUID.uuid4()
       command = %ConsistencyCommand{uuid: uuid, delay: 0}
@@ -104,4 +91,46 @@ defmodule Commanded.Commands.DispatchConsistencyTest do
       assert length(recorded_events) == 1
     end
   end
+
+  describe "process manager consistency" do
+    setup :start_process_manager
+
+    test "should successfully dispatch command" do
+      command = %RequestDispatchCommand{uuid: UUID.uuid4(), delay: 5_000}
+
+      assert :ok = ConsistencyRouter.dispatch(command, consistency: :strong)
+    end
+  end
+
+  def start_event_handlers(_context) do
+    {:ok, handler1} = StronglyConsistentEventHandler.start_link()
+    {:ok, handler2} = EventuallyConsistentEventHandler.start_link()
+
+    on_exit fn ->
+      ProcessHelper.shutdown(handler1)
+      ProcessHelper.shutdown(handler2)
+    end
+
+    :ok
+  end
+
+  def start_optional_handler(_context) do
+    {:ok, handler3} = OptionalStronglyConsistentEventHandler.start_link()
+
+    on_exit fn ->
+      ProcessHelper.shutdown(handler3)
+    end
+
+    :ok
+  end
+
+  def start_process_manager(_context) do
+    {:ok, pm} = StronglyConsistentProcessManager.start_link()
+
+    on_exit fn ->
+      ProcessHelper.shutdown(pm)
+    end
+
+    :ok
+  end
 end
diff --git a/test/commands/support/consistency/consistency_aggregate_root.ex b/test/commands/support/consistency/consistency_aggregate_root.ex
index 423f3a3..64ccdb3 100644
--- a/test/commands/support/consistency/consistency_aggregate_root.ex
+++ b/test/commands/support/consistency/consistency_aggregate_root.ex
@@ -5,13 +5,13 @@ defmodule Commanded.Commands.ConsistencyAggregateRoot do
   defmodule ConsistencyCommand, do: defstruct [:uuid, :delay]
   defmodule NoOpCommand, do: defstruct [:uuid]
   defmodule RequestDispatchCommand, do: defstruct [:uuid, :delay]
-  defmodule ConsistencyEvent, do: defstruct [:delay]
+  defmodule ConsistencyEvent, do: defstruct [:uuid, :delay]
   defmodule DispatchRequestedEvent, do: defstruct [:uuid, :delay]
 
   alias Commanded.Commands.ConsistencyAggregateRoot
-  
-  def execute(%ConsistencyAggregateRoot{}, %ConsistencyCommand{delay: delay}) do
-    %ConsistencyEvent{delay: delay}
+
+  def execute(%ConsistencyAggregateRoot{}, %ConsistencyCommand{uuid: uuid, delay: delay}) do
+    %ConsistencyEvent{uuid: uuid, delay: delay}
   end
 
   def execute(%ConsistencyAggregateRoot{}, %NoOpCommand{}), do: []
diff --git a/test/commands/support/consistency/strongly_consistent_event_handler.ex b/test/commands/support/consistency/strongly_consistent_event_handler.ex
index e7f1876..00d2b04 100644
--- a/test/commands/support/consistency/strongly_consistent_event_handler.ex
+++ b/test/commands/support/consistency/strongly_consistent_event_handler.ex
@@ -5,22 +5,27 @@ defmodule Commanded.Commands.StronglyConsistentEventHandler do
 
   alias Commanded.Commands.{
     ConsistencyAggregateRoot,
-    ConsistencyRouter,
+    ConsistencyRouter
   }
+
   alias ConsistencyAggregateRoot.{
     ConsistencyCommand,
     ConsistencyEvent,
-    DispatchRequestedEvent,
+    DispatchRequestedEvent
   }
 
-  def handle(%ConsistencyEvent{delay: delay}, _metadata) do
+  # handle event by dispatching a command
+  def handle(%DispatchRequestedEvent{uuid: uuid, delay: delay}, _metadata) do
     :timer.sleep(delay)
-    :ok
+
+    command = %ConsistencyCommand{uuid: uuid, delay: delay}
+
+    ConsistencyRouter.dispatch(command, consistency: :strong)
   end
 
-  # handle event by dispatching a command
-  def handle(%DispatchRequestedEvent{uuid: uuid, delay: delay}, _metadata) do
+  def handle(%ConsistencyEvent{delay: delay}, _metadata) do
     :timer.sleep(delay)
-    ConsistencyRouter.dispatch(%ConsistencyCommand{uuid: uuid, delay: delay}, consistency: :strong)
+
+    :ok
   end
 end
diff --git a/test/commands/support/consistency/strongly_consistent_process_manager.ex b/test/commands/support/consistency/strongly_consistent_process_manager.ex
new file mode 100644
index 0000000..e639126
--- /dev/null
+++ b/test/commands/support/consistency/strongly_consistent_process_manager.ex
@@ -0,0 +1,42 @@
+defmodule Commanded.Commands.StronglyConsistentProcessManager do
+  alias Commanded.Commands.{ConsistencyRouter, StronglyConsistentProcessManager}
+
+  use Commanded.ProcessManagers.ProcessManager,
+    name: __MODULE__,
+    consistency: :strong,
+    router: ConsistencyRouter
+
+  defstruct [
+    :uuid
+  ]
+
+  alias Commanded.Commands.ConsistencyAggregateRoot.{
+    ConsistencyCommand,
+    ConsistencyEvent,
+    DispatchRequestedEvent,
+    NoOpCommand
+  }
+
+  def interested?(%DispatchRequestedEvent{uuid: uuid}), do: {:start, uuid}
+  def interested?(%ConsistencyEvent{uuid: uuid}), do: {:continue, uuid}
+
+  def handle(%StronglyConsistentProcessManager{}, %DispatchRequestedEvent{} = requested) do
+    %DispatchRequestedEvent{uuid: uuid, delay: delay} = requested
+    
+    %ConsistencyCommand{uuid: uuid, delay: delay}
+  end
+
+  def handle(%StronglyConsistentProcessManager{}, %ConsistencyEvent{uuid: uuid, delay: delay}) do
+    :timer.sleep(delay)
+
+    %NoOpCommand{uuid: uuid}
+  end
+
+  def apply(%StronglyConsistentProcessManager{}, %DispatchRequestedEvent{uuid: uuid}) do
+    %StronglyConsistentProcessManager{uuid: uuid}
+  end
+
+  def apply(%StronglyConsistentProcessManager{}, %ConsistencyEvent{uuid: uuid}) do
+    %StronglyConsistentProcessManager{uuid: uuid}
+  end
+end