Repository: commanded/commanded
Pull request #272 (2019-05-06 19:43:45)

Register event handler and process manager subscriptions on process start
Use an ETS table to track `:strong` event store subscriptions. Previously subscriptions were tracked using the configured pub/sub provider, `Registry` for local deployment and Phoenix pubsub for distributed. 

The Phoenix pubsub provider when used with Redis occasionally failed to correctly track processes and took a few seconds during application start up. This pull request uses a local ETS table to track subscriptions for more reliability.


Patch:
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index 94b245d..91bded2 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -344,6 +344,7 @@ defmodule Commanded.Event.Handler do
 
   @doc false
   def init(%Handler{} = state) do
+    :ok = register_subscription(state)
     :ok = GenServer.cast(self(), :subscribe_to_events)
 
     {:ok, state}
@@ -373,17 +374,10 @@ defmodule Commanded.Event.Handler do
   def handle_info({:subscribed, subscription}, %Handler{subscription: subscription} = state) do
     Logger.debug(fn -> describe(state) <> " has successfully subscribed to event store" end)
 
-    %Handler{
-      consistency: consistency,
-      handler_module: handler_module,
-      handler_name: handler_name
-    } = state
+    %Handler{handler_module: handler_module} = state
 
     case handler_module.init() do
       :ok ->
-        # register this event handler as a subscription with the given consistency
-        :ok = Subscriptions.register(handler_name, consistency)
-
         {:noreply, state}
 
       {:stop, reason} ->
@@ -408,6 +402,13 @@ defmodule Commanded.Event.Handler do
     end
   end
 
+  # Register this event handler as a subscription with the given consistency
+  defp register_subscription(%Handler{} = state) do
+    %Handler{consistency: consistency, handler_name: name} = state
+
+    Subscriptions.register(name, consistency)
+  end
+
   defp subscribe_to_all_streams(%Handler{} = state) do
     %Handler{
       handler_name: handler_name,
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index 47eada0..c7dffc2 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -47,6 +47,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   end
 
   def init(%State{} = state) do
+    :ok = register_subscription(state)
     :ok = GenServer.cast(self(), :subscribe_to_events)
 
     {:ok, state}
@@ -142,14 +143,7 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
   def handle_info({:subscribed, subscription}, %State{subscription: subscription} = state) do
     Logger.debug(fn -> describe(state) <> " has successfully subscribed to event store" end)
 
-    %State{
-      consistency: consistency,
-      command_dispatcher: command_dispatcher,
-      process_manager_name: process_manager_name
-    } = state
-
-    # Register this process manager as a subscription with the given consistency
-    :ok = Subscriptions.register(process_manager_name, consistency)
+    %State{command_dispatcher: command_dispatcher} = state
 
     {:ok, supervisor} = Supervisor.start_link(command_dispatcher)
 
@@ -192,6 +186,13 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
     {:stop, reason, state}
   end
 
+  # Register this process manager as a subscription with the given consistency
+  defp register_subscription(%State{} = state) do
+    %State{consistency: consistency, process_manager_name: name} = state
+
+    Subscriptions.register(name, consistency)
+  end
+
   defp subscribe_to_all_streams(%State{} = state) do
     %State{process_manager_name: process_manager_name, subscribe_from: subscribe_from} = state
 
diff --git a/lib/commanded/subscriptions.ex b/lib/commanded/subscriptions.ex
index bc98b3c..e70fc80 100644
--- a/lib/commanded/subscriptions.ex
+++ b/lib/commanded/subscriptions.ex
@@ -4,25 +4,23 @@ defmodule Commanded.Subscriptions do
   use GenServer
 
   alias Commanded.EventStore.RecordedEvent
-  alias Commanded.{PubSub, Subscriptions}
+  alias Commanded.PubSub
+  alias Commanded.Subscriptions
 
-  @subscriptions_topic "subscriptions"
   @ack_topic "ack_event"
 
-  defstruct streams_table: nil,
-            started_at: nil,
-            subscribers: []
+  defstruct [
+    :streams_table,
+    :started_at,
+    subscribers: []
+  ]
 
   def start_link(arg) do
     GenServer.start_link(__MODULE__, arg, name: __MODULE__)
   end
 
-  @doc """
-  Register an event store subscription with the given consistency guarantee.
-  """
-  def register(name, consistency)
-  def register(_name, :eventual), do: :ok
-  def register(name, :strong), do: PubSub.track(@subscriptions_topic, name)
+  defdelegate register(name, consistency), to: Subscriptions.Registry
+  defdelegate all(), to: Subscriptions.Registry
 
   @doc """
   Acknowledge receipt and sucessful processing of the given event by the named
@@ -39,12 +37,7 @@ defmodule Commanded.Subscriptions do
   end
 
   @doc false
-  def all, do: subscriptions()
-
-  @doc false
-  def reset do
-    GenServer.call(__MODULE__, :reset)
-  end
+  def reset, do: GenServer.call(__MODULE__, :reset)
 
   @doc """
   Have all the registered handlers processed the given event?
@@ -86,7 +79,9 @@ defmodule Commanded.Subscriptions do
     {:ok, initial_state()}
   end
 
-  def handle_call(:reset, _from, %Subscriptions{streams_table: streams_table}) do
+  def handle_call(:reset, _from, %Subscriptions{} = state) do
+    %Subscriptions{streams_table: streams_table} = state
+
     :ets.delete(streams_table)
 
     {:reply, :ok, initial_state()}
@@ -162,7 +157,9 @@ defmodule Commanded.Subscriptions do
     {:noreply, state}
   end
 
-  def handle_info({:DOWN, _ref, :process, pid, _reason}, %Subscriptions{subscribers: subscribers} = state) do
+  def handle_info({:DOWN, _ref, :process, pid, _reason}, %Subscriptions{} = state) do
+    %Subscriptions{subscribers: subscribers} = state
+
     state = %Subscriptions{state |
       subscribers: remove_by_pid(subscribers, pid),
     }
@@ -177,11 +174,9 @@ defmodule Commanded.Subscriptions do
     }
   end
 
-  defp subscriptions, do: PubSub.list(@subscriptions_topic)
-
   # Have all subscriptions handled the event for the given stream and version
   defp handled_by_all?(stream_uuid, stream_version, consistency, exclude, %Subscriptions{} = state) do
-    subscriptions()
+    Subscriptions.Registry.all()
     |> Enum.reject(fn {_name, pid} -> MapSet.member?(exclude, pid) end)
     |> Enum.filter(fn {name, _pid} ->
       # Optionally filter subscriptions to those provided by the `consistency` option
@@ -194,7 +189,9 @@ defmodule Commanded.Subscriptions do
   end
 
   # Has the named subscription handled the event for the given stream and version
-  defp handled_by?(name, stream_uuid, stream_version, %Subscriptions{streams_table: streams_table}) do
+  defp handled_by?(name, stream_uuid, stream_version, %Subscriptions{} = state) do
+    %Subscriptions{streams_table: streams_table} = state
+
     case :ets.lookup(streams_table, {name, stream_uuid}) do
       [{{^name, ^stream_uuid}, last_seen, _inserted_at}] when last_seen >= stream_version -> true
       _ -> false
@@ -209,7 +206,9 @@ defmodule Commanded.Subscriptions do
   end
 
   # Notify any subscribers waiting on a given stream if it is at the expected version
-  defp notify_subscribers(stream_uuid, %Subscriptions{subscribers: subscribers} = state) do
+  defp notify_subscribers(stream_uuid, %Subscriptions{} = state) do
+    %Subscriptions{subscribers: subscribers} = state
+
     Enum.reduce(subscribers, subscribers, fn
       ({pid, ^stream_uuid, expected_stream_version, consistency, exclude} = subscriber, subscribers) ->
         case handled_by_all?(stream_uuid, expected_stream_version, consistency, exclude, state) do
@@ -237,7 +236,9 @@ defmodule Commanded.Subscriptions do
   end
 
   # Delete subscription ack's that are older than the configured ttl
-  defp purge_expired_streams(ttl, %Subscriptions{streams_table: streams_table, started_at: started_at}) do
+  defp purge_expired_streams(ttl, %Subscriptions{} = state) do
+     %Subscriptions{streams_table: streams_table, started_at: started_at} = state
+
     stale_epoch = monotonic_time() - started_at - (ttl / 1_000)
 
     streams_table
diff --git a/lib/commanded/subscriptions/registry.ex b/lib/commanded/subscriptions/registry.ex
new file mode 100644
index 0000000..7560822
--- /dev/null
+++ b/lib/commanded/subscriptions/registry.ex
@@ -0,0 +1,37 @@
+defmodule Commanded.Subscriptions.Registry do
+  @moduledoc false
+
+  # Provides read/write access to a public ETS table used to track event store
+  # subscriptions. This process' only use is as the owner of the ETS table and
+  # should never crash.
+
+  use GenServer
+
+  def start_link(arg) do
+    GenServer.start_link(__MODULE__, arg, name: __MODULE__)
+  end
+
+  @doc """
+  Register an event store subscription with the given consistency guarantee.
+  """
+  def register(name, consistency)
+  def register(_name, :eventual), do: :ok
+  def register(name, :strong) do
+    true = :ets.insert(__MODULE__, {name, self()})
+
+    :ok
+  end
+
+  @doc """
+  Get all registered subscriptions.
+  """
+  def all do
+    :ets.tab2list(__MODULE__)
+  end
+
+  def init(_arg) do
+    table = :ets.new(__MODULE__, [:set, :public, :named_table])
+
+    {:ok, table}
+  end
+end
diff --git a/lib/commanded/supervisor.ex b/lib/commanded/supervisor.ex
index bd8e1fd..04a8f75 100644
--- a/lib/commanded/supervisor.ex
+++ b/lib/commanded/supervisor.ex
@@ -13,8 +13,9 @@ defmodule Commanded.Supervisor do
       Registration.child_spec() ++ PubSub.child_spec() ++
       [
         {Task.Supervisor, name: Commanded.Commands.TaskDispatcher},
-        {Commanded.Aggregates.Supervisor, []},
-        {Commanded.Subscriptions, []}
+        Commanded.Aggregates.Supervisor,
+        Commanded.Subscriptions.Registry,
+        Commanded.Subscriptions
       ]
 
     Supervisor.init(children, strategy: :one_for_one)


Test patch:
diff --git a/test/subscriptions/subscriptions_test.exs b/test/subscriptions/subscriptions_test.exs
index ea67436..9240105 100644
--- a/test/subscriptions/subscriptions_test.exs
+++ b/test/subscriptions/subscriptions_test.exs
@@ -10,7 +10,40 @@ defmodule Commanded.SubscriptionsTest do
       :ok = Subscriptions.register("handler2", :eventual)
       :ok = Subscriptions.register("handler3", :strong)
 
-      assert Subscriptions.all() |> Enum.sort() == [{"handler1", self()}, {"handler3", self()}]
+      assert Subscriptions.all() ==[{"handler3", self()}, {"handler1", self()}]
+    end
+
+    test "should not remove PID when process terminates" do
+      pid = spawn_link(fn ->
+        :ok = Subscriptions.register("handler1", :strong)
+      end)
+
+      ref = Process.monitor(pid)
+      assert_receive {:DOWN, ^ref, :process, _, :normal}
+      
+      assert Subscriptions.all() ==[{"handler1", pid}]
+    end
+
+    test "should replace registered subscription PID" do
+      reply_to = self()
+
+      pid1 = spawn_link(fn ->
+        :ok = Subscriptions.register("handler1", :strong)
+
+        send(reply_to,{:handler, self()})
+      end)
+
+      assert_receive {:handler, ^pid1}
+      assert Subscriptions.all() ==[{"handler1", pid1}]
+
+      pid2 = spawn_link(fn ->
+        :ok = Subscriptions.register("handler1", :strong)
+
+        send(reply_to, {:handler, self()})
+      end)
+
+      assert_receive {:handler, ^pid2}
+      assert Subscriptions.all() ==[{"handler1", pid2}]
     end
 
     test "should ack event" do