fix(handler): prevent a reset from corrupting the handler read model

lib/commanded/event/handler.ex

@@ -1059,11 +1059,49 @@ defmodule Commanded.Event.Handler do
   end
 
   defp reset_subscription(%Handler{} = state) do
-    %Handler{subscription: subscription} = state
+    %Handler{subscription: %Subscription{subscription_pid: reset_subscription_pid} = subscription} =
+      state
 
     subscription = Subscription.reset(subscription)
 
-    %Handler{state | last_seen_event: nil, subscription: subscription, subscribe_timer: nil}
+    state =
+      state
+      |> discard_messages_from(reset_subscription_pid)
+      |> cancel_batch_timer()
+      |> cancel_subscribe_timer()
+
+    %Handler{state | last_seen_event: nil, subscription: subscription, batch_buffer: []}
+  end
+
+  # `Subscription.reset/1` stops the subscription's process before returning, and the replacement
+  # is not started until `subscribe_to_events/1` runs, so nothing can be delivered while this
+  # drains. `{:subscribed, pid}` identifies its sender and is matched against the subscription
+  # that was reset; `{:events, _}` does not carry one, so it can only be matched by shape.
+  defp discard_messages_from(%Handler{} = state, reset_subscription_pid) do
+    case discard_messages_from(reset_subscription_pid, 0) do
+      0 ->
+        state
+
+      discarded ->
+        Logger.debug(
+          describe(state) <>
+            " discarded #{discarded} message(s) queued by the subscription it reset"
+        )
+
+        state
+    end
+  end
+
+  defp discard_messages_from(reset_subscription_pid, discarded) do
+    receive do
+      {:events, _events} ->
+        discard_messages_from(reset_subscription_pid, discarded + 1)
+
+      {:subscribed, ^reset_subscription_pid} ->
+        discard_messages_from(reset_subscription_pid, discarded + 1)
+    after
+      0 -> discarded
+    end
   end
 
   defp subscribe_to_events(%Handler{} = state) do
@@ -1258,6 +1296,34 @@ defmodule Commanded.Event.Handler do
     end
   end
 
+  defp cancel_subscribe_timer(%Handler{subscribe_timer: nil} = state), do: state
+
+  defp cancel_subscribe_timer(%Handler{subscribe_timer: ref} = state) do
+    case Process.cancel_timer(ref) do
+      false ->
+        drain_subscribe_to_events_message()
+        %Handler{state | subscribe_timer: nil}
+
+      _remaining ->
+        %Handler{state | subscribe_timer: nil}
+    end
+  end
+
+  # `Process.cancel_timer/1` answers `false` once the timer has expired, and by then it has already
+  # delivered `:subscribe_to_events` into this handler's own mailbox, where cancelling can no longer
+  # reach it. Left queued, it outlives the reset and drives a second `subscribe_to_events/1` against
+  # the subscription the reset just established, which fails and re-arms the retry indefinitely.
+  #
+  # The name describes the mechanism rather than the reason; `discard_expired_subscribe_retry/0`
+  # would read better, and is only kept for symmetry with `drain_flush_batch_timeout_message/0`.
+  defp drain_subscribe_to_events_message do
+    receive do
+      :subscribe_to_events -> :ok
+    after
+      0 -> :ok
+    end
+  end
+
   defp handle_batch(events, context \\ %{}, handler)
 
   defp handle_batch(events, context, %Handler{last_seen_event: last_seen_event} = state)

lib/commanded/event_store/adapter.ex

@@ -94,13 +94,21 @@ defmodule Commanded.EventStore.Adapter do
 
   @doc """
   Delete an existing subscription.
+
+  Whether a subscription that still has subscribers can be deleted is adapter specific: an adapter
+  can refuse with `{:error, :subscribers_connected}` and leave them attached, or delete it and
+  disconnect them. A caller that must not disconnect a subscriber it does not own unsubscribes it
+  first.
   """
   @callback delete_subscription(
               adapter_meta,
               stream_uuid | :all,
               subscription_name
             ) ::
-              :ok | {:error, :subscription_not_found} | {:error, error}
+              :ok
+              | {:error, :subscription_not_found}
+              | {:error, :subscribers_connected}
+              | {:error, error}
 
   @doc """
   Read a snapshot, if available, for a given source.

lib/commanded/event_store/adapters/in_memory.ex

@@ -288,6 +288,9 @@ defmodule Commanded.EventStore.Adapters.InMemory do
 
           {:ok, state}
 
+        %PersistentSubscription{stream_uuid: ^stream_uuid} ->
+          {{:error, :subscribers_connected}, state}
+
         nil ->
           {{:error, :subscription_not_found}, state}
       end

lib/commanded/event_store/subscription.ex

@@ -1,6 +1,8 @@
 defmodule Commanded.EventStore.Subscription do
   @moduledoc false
 
+  require Logger
+
   alias Commanded.EventStore
   alias Commanded.EventStore.{RecordedEvent, Subscription}
 
@@ -96,10 +98,41 @@ defmodule Commanded.EventStore.Subscription do
       subscription_ref: subscription_ref
     } = subscription
 
-    Process.demonitor(subscription_ref)
+    # A reset can arrive while a subscribe retry is still pending, in which case nothing has been
+    # monitored, subscribed, or persisted yet.
+    if is_reference(subscription_ref) do
+      Process.demonitor(subscription_ref, [:flush])
+    end
 
-    :ok = EventStore.unsubscribe(application, subscription_pid)
-    :ok = EventStore.delete_subscription(application, subscribe_to, subscription_name)
+    if is_pid(subscription_pid) do
+      :ok = EventStore.unsubscribe(application, subscription_pid)
+    end
+
+    # Another subscriber can still hold the subscription name, either a sibling of a concurrent
+    # handler or an unrelated one this handler has been losing a race against. Its checkpoint is
+    # not this handler's to discard, and the reset has to go ahead regardless.
+    case EventStore.delete_subscription(application, subscribe_to, subscription_name) do
+      :ok ->
+        :ok
+
+      {:error, :subscription_not_found} ->
+        :ok
+
+      {:error, :subscribers_connected} ->
+        :ok
+
+      # TODO: an adapter is allowed to fail the delete for reasons that are neither of the above,
+      # and a reset that carries on leaves the handler reading from a checkpoint it meant to drop.
+      # Decide whether that should fail the reset instead of being reported and continuing.
+      {:error, error} ->
+        Logger.warning(fn ->
+          "Subscription " <>
+            inspect(subscription_name) <>
+            " could not be deleted for reset: " <> inspect(error)
+        end)
+
+        :ok
+    end
 
     %Subscription{
       subscription

test/event/event_handler_concurrency_test.exs

@@ -100,6 +100,28 @@ defmodule Commanded.Event.EventHandlerConcurrencyTest do
       assert length(unique_pids) == 5
     end
 
+    test "should reset one handler without disturbing the others", %{supervisor: supervisor} do
+      for _ <- 1..5, do: assert_receive({:init, _pid})
+
+      [{_, handler, _, _} | _] = Supervisor.which_children(supervisor)
+
+      event_store = Process.whereis(Module.concat([DefaultApp, "EventStore"]))
+      event_store_ref = Process.monitor(event_store)
+      handler_ref = Process.monitor(handler)
+
+      send(handler, :reset)
+
+      refute_receive {:DOWN, ^event_store_ref, :process, ^event_store, _reason}
+      refute_receive {:DOWN, ^handler_ref, :process, ^handler, _reason}
+
+      assert %{active: 5, specs: 5, supervisors: 0, workers: 5} =
+               Supervisor.count_children(supervisor)
+
+      append_events_to_stream("stream1", count: 1)
+
+      assert_receive {:event, "stream1", _pid}
+    end
+
     test "should error when handler started with `:strong` consistency" do
       assert_raise ArgumentError,
                    "cannot use `:strong` consistency with concurrency",

test/event/event_handler_subscription_test.exs

@@ -2,6 +2,7 @@ defmodule Commanded.Event.EventHandlerSubscriptionTest do
   use Commanded.MockEventStoreCase
 
   alias Commanded.Event.Handler
+  alias Commanded.Helpers.Wait
 
   defmodule ExampleHandler do
     use Commanded.Event.Handler,
@@ -59,6 +60,101 @@ defmodule Commanded.Event.EventHandlerSubscriptionTest do
 
       assert_receive {:subscribed, ^subscription}
     end
+
+    test "should reset while a subscription retry is pending" do
+      reply_to = self()
+
+      # Both the initial subscription attempt and the attempt made by the reset fail
+      expect(MockEventStore, :subscribe_to, 2, fn
+        _event_store_meta, :all, "ExampleHandler", handler, :origin, _opts ->
+          send(reply_to, {:subscribe_to, handler})
+
+          {:error, :subscription_already_exists}
+      end)
+
+      {:ok, handler} = ExampleHandler.start_link()
+
+      assert_receive {:subscribe_to, ^handler}
+      assert_handler_subscription_timer(handler, 1..3_000)
+
+      Process.unlink(handler)
+      ref = Process.monitor(handler)
+
+      send(handler, :reset)
+
+      assert_receive {:subscribe_to, ^handler}
+      refute_receive {:DOWN, ^ref, :process, ^handler, _reason}
+    end
+
+    test "should reset when the subscription to delete does not exist" do
+      reply_to = self()
+
+      expect(MockEventStore, :subscribe_to, 2, fn
+        _event_store_meta, :all, "ExampleHandler", handler, :origin, _opts ->
+          send(reply_to, {:subscribe_to, handler})
+
+          {:error, :subscription_already_exists}
+      end)
+
+      expect(MockEventStore, :delete_subscription, fn _event_store_meta, :all, "ExampleHandler" ->
+        {:error, :subscription_not_found}
+      end)
+
+      {:ok, handler} = ExampleHandler.start_link()
+
+      assert_receive {:subscribe_to, ^handler}
+      assert_handler_subscription_timer(handler, 1..3_000)
+
+      Process.unlink(handler)
+      ref = Process.monitor(handler)
+
+      send(handler, :reset)
+
+      assert_receive {:subscribe_to, ^handler}
+      refute_receive {:DOWN, ^ref, :process, ^handler, _reason}
+    end
+
+    test "should not subscribe twice when reset while a subscription retry is pending" do
+      reply_to = self()
+
+      expect(MockEventStore, :subscribe_to, fn
+        _event_store_meta, :all, "ExampleHandler", handler, :origin, _opts ->
+          send(reply_to, {:subscribe_to, handler})
+
+          {:error, :subscription_already_exists}
+      end)
+
+      {:ok, handler} = ExampleHandler.start_link()
+
+      assert_receive {:subscribe_to, ^handler}
+
+      %Handler{subscribe_timer: subscribe_timer} = :sys.get_state(handler)
+
+      {:ok, subscription} = start_subscription()
+
+      expect_subscribe_to(subscription)
+
+      stub(MockEventStore, :subscribe_to, fn
+        _event_store_meta, :all, "ExampleHandler", handler, :origin, _opts ->
+          send(reply_to, {:resubscribe_to, handler})
+
+          {:error, :subscription_already_exists}
+      end)
+
+      send(handler, :reset)
+
+      assert_receive {:subscribed, ^subscription}
+
+      Wait.until(4_000, fn ->
+        refute Process.read_timer(subscribe_timer)
+      end)
+
+      # Processed in mailbox order, so a retry that fired before the timer was read has already
+      # been handled by the time this returns
+      :sys.get_state(handler)
+
+      refute_received {:resubscribe_to, ^handler}
+    end
   end
 
   defp assert_handler_subscription_timer(handler, expected_timer_range) do