Repository: commanded/commanded
Pull request #308 (2019-10-27 15:10:24)

Fix commanded aggregate race
address https://github.com/commanded/commanded/issues/279


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d1917a1..229fae4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,11 +1,5 @@
 # Changelog
 
-## v1.0.0-rc.1
-
-### Bug fixes
-
-- Fix process manager exception on start ([#307](https://github.com/commanded/commanded/pull/307)).
-
 ## v1.0.0-rc.0
 
 ### Breaking changes
diff --git a/README.md b/README.md
index 9eca5e0..5bea842 100644
--- a/README.md
+++ b/README.md
@@ -19,15 +19,13 @@ You can use Commanded with one of the following event stores for persistence:
 
 Please refer to the [CHANGELOG](CHANGELOG.md) for features, bug fixes, and any upgrade advice included for each release.
 
-Requires Erlang/OTP v21.0 and Elixir v1.6 or later.
-
 ---
 
 ### Supporting Commanded
 
 You can help support Commanded by helping to fund its ongoing development, new features, and releases.
 
-- [Become a GitHub sponsor](https://github.com/sponsors/slashdotdash).
+- [Become a backer or sponsor on OpenCollective](https://opencollective.com/commanded).
 - [View sponsors & backers](BACKERS.md)
 
 ---
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index 2db274e..7c3fe13 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -95,6 +95,14 @@ defmodule Commanded.Aggregates.Aggregate do
       when is_atom(application) and is_atom(aggregate_module) and is_binary(aggregate_uuid),
       do: {application, aggregate_module, aggregate_uuid}
 
+  @doc false
+  @impl GenServer
+  def init(%Aggregate{} = state) do
+    # Initial aggregate state is populated by loading its state snapshot and/or
+    # events from the event store.
+    {:ok, state, {:continue, :populate_aggregate_state}}
+  end
+
   @doc """
   Execute the given command against the aggregate.
 
@@ -126,7 +134,13 @@ defmodule Commanded.Aggregates.Aggregate do
       when is_atom(aggregate_module) and is_binary(aggregate_uuid) and
              (is_number(timeout) or timeout == :infinity) do
     name = via_name(application, aggregate_module, aggregate_uuid)
-    GenServer.call(name, {:execute_command, context}, timeout)
+
+    try do
+      GenServer.call(name, {:execute_command, context}, timeout)
+    catch
+      :exit, {:normal, {GenServer, :call, [^name, {:execute_command, ^context}, ^timeout]}} ->
+        {:exit, :normal, :aggregate_stopped}
+    end
   end
 
   @doc false
@@ -154,15 +168,6 @@ defmodule Commanded.Aggregates.Aggregate do
   end
 
   @doc false
-  @impl GenServer
-  def init(%Aggregate{} = state) do
-    # Initial aggregate state is populated by loading its state snapshot and/or
-    # events from the event store.
-    {:ok, state, {:continue, :populate_aggregate_state}}
-  end
-
-  @doc false
-  @impl GenServer
   def handle_continue(:populate_aggregate_state, %Aggregate{} = state) do
     # Subscribe to aggregate's events to catch any events appended to its stream
     # by another process, such as directly appended to the event store.
diff --git a/lib/commanded/commands/dispatcher.ex b/lib/commanded/commands/dispatcher.ex
index dfcd57f..a035b19 100644
--- a/lib/commanded/commands/dispatcher.ex
+++ b/lib/commanded/commands/dispatcher.ex
@@ -89,6 +89,7 @@ defmodule Commanded.Commands.Dispatcher do
     result =
       case Task.yield(task, timeout) || Task.shutdown(task) do
         {:ok, reply} -> reply
+        {:exit, :normal, :aggregate_stopped} = error -> error
         {:exit, error} -> {:error, :aggregate_execution_failed, error}
         nil -> {:error, :aggregate_execution_timeout}
       end
@@ -101,6 +102,9 @@ defmodule Commanded.Commands.Dispatcher do
         |> after_dispatch(payload)
         |> respond_with_success(payload, events)
 
+      {:exit, :normal, :aggregate_stopped} ->
+        execute(pipeline, payload)
+
       {:error, error} ->
         pipeline
         |> Pipeline.respond({:error, error})
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index 457d9c0..41a4ebf 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -404,21 +404,14 @@ defmodule Commanded.Event.Handler do
   def name(application, handler_name), do: {application, __MODULE__, handler_name}
 
   @doc false
-  @impl GenServer
   def init(%Handler{} = state) do
     :ok = register_subscription(state)
+    :ok = GenServer.cast(self(), :subscribe_to_events)
 
-    {:ok, state, {:continue, :subscribe_to_events}}
+    {:ok, state}
   end
 
   @doc false
-  @impl GenServer
-  def handle_continue(:subscribe_to_events, %Handler{} = state) do
-    {:noreply, subscribe_to_events(state)}
-  end
-
-  @doc false
-  @impl GenServer
   def handle_call(:last_seen_event, _from, %Handler{} = state) do
     %Handler{last_seen_event: last_seen_event} = state
 
@@ -426,7 +419,6 @@ defmodule Commanded.Event.Handler do
   end
 
   @doc false
-  @impl GenServer
   def handle_call(:config, _from, %Handler{} = state) do
     %Handler{consistency: consistency, subscribe_from: subscribe_from, subscribe_to: subscribe_to} =
       state
@@ -437,7 +429,11 @@ defmodule Commanded.Event.Handler do
   end
 
   @doc false
-  @impl GenServer
+  def handle_cast(:subscribe_to_events, %Handler{} = state) do
+    {:noreply, subscribe_to_events(state)}
+  end
+
+  @doc false
   def handle_info(:reset, %Handler{} = state) do
     %Handler{handler_module: handler_module} = state
 
@@ -464,7 +460,6 @@ defmodule Commanded.Event.Handler do
 
   @doc false
   # Subscription to event store has successfully subscribed, init event handler
-  @impl GenServer
   def handle_info({:subscribed, subscription}, %Handler{subscription: subscription} = state) do
     Logger.debug(fn -> describe(state) <> " has successfully subscribed to event store" end)
 
@@ -482,7 +477,6 @@ defmodule Commanded.Event.Handler do
   end
 
   @doc false
-  @impl GenServer
   def handle_info({:events, events}, %Handler{} = state) do
     Logger.debug(fn -> describe(state) <> " received events: #{inspect(events)}" end)
 
@@ -501,7 +495,6 @@ defmodule Commanded.Event.Handler do
   end
 
   @doc false
-  @impl GenServer
   def handle_info({:DOWN, ref, :process, _pid, reason}, %Handler{subscription_ref: ref} = state) do
     Logger.debug(fn -> describe(state) <> " subscription DOWN due to: #{inspect(reason)}" end)
 
@@ -509,7 +502,7 @@ defmodule Commanded.Event.Handler do
     {:stop, reason, state}
   end
 
-  # Register this event handler as a subscription with the given consistency.
+  # 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 08f8521..36965c4 100644
--- a/lib/commanded/process_managers/process_manager_instance.ex
+++ b/lib/commanded/process_managers/process_manager_instance.ex
@@ -40,6 +40,10 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
     GenServer.start_link(__MODULE__, state)
   end
 
+  def init(%State{} = state) do
+    {:ok, state, {:continue, :fetch_state}}
+  end
+
   @doc """
   Checks whether or not the process manager has already processed events
   """
@@ -70,16 +74,9 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
     GenServer.call(process_manager, :process_state)
   end
 
-  @doc false
-  @impl GenServer
-  def init(%State{} = state) do
-    {:ok, state, {:continue, :fetch_state}}
-  end
-
   @doc """
   Attempt to fetch intial process state from snapshot storage.
   """
-  @impl GenServer
   def handle_continue(:fetch_state, %State{} = state) do
     %State{application: application} = state
 
@@ -100,7 +97,6 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   end
 
   @doc false
-  @impl GenServer
   def handle_call(:stop, _from, %State{} = state) do
     :ok = delete_state(state)
 
@@ -109,7 +105,6 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   end
 
   @doc false
-  @impl GenServer
   def handle_call(:process_state, _from, %State{} = state) do
     %State{idle_timeout: idle_timeout, process_state: process_state} = state
 
@@ -117,7 +112,6 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   end
 
   @doc false
-  @impl GenServer
   def handle_call(:new?, _from, %State{} = state) do
     %State{idle_timeout: idle_timeout, last_seen_event: last_seen_event} = state
 
@@ -127,7 +121,6 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   @doc """
   Handle the given event, using the process manager module, against the current process state
   """
-  @impl GenServer
   def handle_cast({:process_event, event}, %State{} = state) do
     case event_already_seen?(event, state) do
       true -> process_seen_event(event, state)
@@ -136,7 +129,6 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   end
 
   @doc false
-  @impl GenServer
   def handle_info(:timeout, %State{} = state) do
     Logger.debug(fn -> describe(state) <> " stopping due to inactivity timeout" end)
 
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index 882f037..273c437 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -37,13 +37,13 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     ]
   end
 
-  def start_link(application, process_name, process_module, opts \\ []) do
-    name = {application, ProcessRouter, process_name}
+  def start_link(application, name, module, opts \\ []) do
+    name = {application, ProcessRouter, name}
 
     state = %State{
       application: application,
-      process_manager_name: process_name,
-      process_manager_module: process_module,
+      process_manager_name: name,
+      process_manager_module: module,
       consistency: Keyword.get(opts, :consistency, :eventual),
       subscribe_from: Keyword.get(opts, :start_from, :origin),
       event_timeout: Keyword.get(opts, :event_timeout),
@@ -256,7 +256,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:stop, reason, state}
   end
 
-  # Register this process manager as a subscription with the given consistency.
+  # Register this process manager as a subscription with the given consistency
   defp register_subscription(%State{} = state) do
     %State{application: application, consistency: consistency, process_manager_name: name} = state
 


Test patch:
diff --git a/test/aggregates/aggregate_concurrency_test.exs b/test/aggregates/aggregate_concurrency_test.exs
index 0b170a9..004a1da 100644
--- a/test/aggregates/aggregate_concurrency_test.exs
+++ b/test/aggregates/aggregate_concurrency_test.exs
@@ -10,16 +10,11 @@ defmodule Commanded.Aggregates.AggregateConcurrencyTest do
 
   setup do
     expect(MockEventStore, :subscribe_to, fn
-      {MockedApp.EventStore, _config}, stream_uuid, handler_name, handler, _subscribe_from ->
-        assert is_binary(stream_uuid)
-        assert is_binary(handler_name)
-
+      {MockedApp.EventStore, _config}, _stream_uuid, _handler_name, handler, _subscribe_from ->
         {:ok, handler}
     end)
 
-    expect(MockEventStore, :subscribe, fn {MockedApp.EventStore, _config}, aggregate_uuid ->
-      assert is_binary(aggregate_uuid)
-
+    expect(MockEventStore, :subscribe, fn {MockedApp.EventStore, _config}, _aggregate_uuid ->
       :ok
     end)
 
diff --git a/test/aggregates/aggregate_lifespan_test.exs b/test/aggregates/aggregate_lifespan_test.exs
index 3919f18..167e5e6 100644
--- a/test/aggregates/aggregate_lifespan_test.exs
+++ b/test/aggregates/aggregate_lifespan_test.exs
@@ -194,6 +194,24 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do
       assert_receive {:DOWN, ^ref, :process, _, :normal}
     end
 
+    test "should reopen the open_aggregate to execute command if it is stopped by the previous command", %{
+      aggregate_uuid: aggregate_uuid,
+      ref: ref
+    } do
+      command = %Command{uuid: aggregate_uuid, action: :noop, lifespan: :stop}
+
+      tasks =
+        for _i <- 1..3 do
+          Task.async(fn ->
+            :ok = LifespanRouter.dispatch(command, application: DefaultApp)
+          end)
+        end
+
+      Task.yield_many(tasks)
+
+      assert_receive {:DOWN, ^ref, :process, _, :normal}
+    end
+
     test "should stop process when requested", %{
       aggregate_uuid: aggregate_uuid,
       ref: ref
diff --git a/test/event/handler_init_test.exs b/test/event/handler_init_test.exs
index 5a58e06..7a1629e 100644
--- a/test/event/handler_init_test.exs
+++ b/test/event/handler_init_test.exs
@@ -9,9 +9,7 @@ defmodule Commanded.Event.HandlerInitTest do
   setup do
     reply_to = self()
 
-    subscribe_to = fn _event_store, :all, handler_name, handler, _subscribe_from ->
-      assert is_binary(handler_name)
-
+    subscribe_to = fn _event_store, :all, _handler_name, handler, _subscribe_from ->
       {:ok, handler}
     end
 
diff --git a/test/process_managers/process_manager_routing_test.exs b/test/process_managers/process_manager_routing_test.exs
index e36e917..b618c24 100644
--- a/test/process_managers/process_manager_routing_test.exs
+++ b/test/process_managers/process_manager_routing_test.exs
@@ -12,10 +12,7 @@ defmodule Commanded.ProcessManagers.ProcessManagerRoutingTest do
 
   setup do
     expect(MockEventStore, :subscribe_to, fn
-      _event_store, :all, name, pid, :origin ->
-        assert is_binary(name)
-        assert is_pid(pid)
-
+      _event_store, :all, _name, pid, :origin ->
         send(pid, {:subscribed, self()})
 
         {:ok, self()}
diff --git a/test/process_managers/process_manager_subscription_test.exs b/test/process_managers/process_manager_subscription_test.exs
index 3863ddf..ac5f26f 100644
--- a/test/process_managers/process_manager_subscription_test.exs
+++ b/test/process_managers/process_manager_subscription_test.exs
@@ -50,9 +50,11 @@ defmodule Commanded.ProcessManagers.ProcessManagerSubscriptionTest do
   defp start_process_manager(subscription) do
     reply_to = self()
 
-    expect(MockEventStore, :subscribe_to, fn {Commanded.MockedApp.EventStore, _config},
+    expect(MockEventStore, :subscribe_to, fn _event_store,
                                              :all,
-                                             "ExampleProcessManager",
+                                             {Commanded.MockedApp,
+                                              Commanded.ProcessManagers.ProcessRouter,
+                                              "ExampleProcessManager"},
                                              pm,
                                              :origin ->
       send(pm, {:subscribed, subscription})