diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex index 12234713..7a07acd1 100644 --- a/lib/commanded/event/handler.ex +++ b/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) diff --git a/lib/commanded/event_store/adapter.ex b/lib/commanded/event_store/adapter.ex index 50600340..4cb1cf2f 100644 --- a/lib/commanded/event_store/adapter.ex +++ b/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. diff --git a/lib/commanded/event_store/adapters/in_memory.ex b/lib/commanded/event_store/adapters/in_memory.ex index 4c43c504..cb16056a 100644 --- a/lib/commanded/event_store/adapters/in_memory.ex +++ b/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 diff --git a/lib/commanded/event_store/subscription.ex b/lib/commanded/event_store/subscription.ex index 76367976..962334ec 100644 --- a/lib/commanded/event_store/subscription.ex +++ b/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 diff --git a/test/event/event_handler_concurrency_test.exs b/test/event/event_handler_concurrency_test.exs index 2de6ed1d..79814249 100644 --- a/test/event/event_handler_concurrency_test.exs +++ b/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", diff --git a/test/event/event_handler_subscription_test.exs b/test/event/event_handler_subscription_test.exs index af420374..577ee805 100644 --- a/test/event/event_handler_subscription_test.exs +++ b/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 diff --git a/test/event/reset_batch_event_handler_test.exs b/test/event/reset_batch_event_handler_test.exs index 1bf8dcbd..eb5441ae 100644 --- a/test/event/reset_batch_event_handler_test.exs +++ b/test/event/reset_batch_event_handler_test.exs @@ -3,8 +3,10 @@ defmodule Commanded.Event.BatchResetEventHandlerTest do import Commanded.Assertions.EventAssertions + alias Commanded.Event.Handler alias Commanded.Event.Mapper alias Commanded.EventStore + alias Commanded.EventStore.Subscription alias Commanded.ExampleDomain.BankAccount.BankAccountBatchHandler alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened alias Commanded.ExampleDomain.BankApp @@ -38,6 +40,86 @@ defmodule Commanded.Event.BatchResetEventHandlerTest do end) end + test "should discard events buffered for a batch before the reset" do + stream_uuid = UUID.uuid4() + + handler = + start_supervised!({BankAccountBatchHandler, start_from: :current, batch_timeout: 500}) + + Wait.until(fn -> + assert BankAccountBatchHandler.subscribed?() + end) + + %Handler{subscription: %Subscription{subscription_pid: subscription_pid}} = + :sys.get_state(handler) + + buffered_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}] + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(buffered_events)) + + Wait.until(fn -> + assert %Handler{batch_buffer: [_buffered_event]} = :sys.get_state(handler) + end) + + :ok = BankAccountBatchHandler.change_prefix("PREF_") + + send(handler, :reset) + + Wait.until(fn -> + assert %Handler{subscription: %Subscription{subscription_pid: pid}} = + :sys.get_state(handler) + + refute pid in [nil, subscription_pid] + end) + + events_after_reset = [%BankAccountOpened{account_number: "ACC456", initial_balance: 2_000}] + + :ok = + EventStore.append_to_stream(BankApp, stream_uuid, 1, to_event_data(events_after_reset)) + + Wait.until(2_000, fn -> + assert BankAccountBatchHandler.current_accounts() == ["PREF_ACC456"] + end) + end + + test "should rearm the batch timer when its timeout expired before the reset" do + stream_uuid = UUID.uuid4() + + handler = start_supervised!({BankAccountBatchHandler, batch_timeout: 1_000}) + + Wait.until(fn -> + assert BankAccountBatchHandler.subscribed?() + end) + + buffered_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}] + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(buffered_events)) + + Wait.until(fn -> + assert %Handler{batch_buffer: [_buffered_event], batch_timer_ref: batch_timer_ref} = + :sys.get_state(handler) + + assert is_reference(batch_timer_ref) + end) + + :ok = BankAccountBatchHandler.change_prefix("PREF_") + + :ok = :sys.suspend(handler) + + send(handler, :reset) + + # Letting the batch timeout expire while suspended queues its message behind the reset, so + # the reset must clear it: a `batch_timer_ref` left behind keeps the replayed batch from + # ever arming a timer of its own. + Wait.until(3_000, fn -> + assert {:messages, [:reset, :flush_batch_timeout]} = Process.info(handler, :messages) + end) + + :ok = :sys.resume(handler) + + Wait.until(5_000, fn -> + assert BankAccountBatchHandler.current_accounts() == ["PREF_ACC123"] + end) + end + test "should be reset when starting from `:current`" do stream_uuid = UUID.uuid4() diff --git a/test/event/reset_event_handler_test.exs b/test/event/reset_event_handler_test.exs index a992fec7..2cbbbcc8 100644 --- a/test/event/reset_event_handler_test.exs +++ b/test/event/reset_event_handler_test.exs @@ -2,15 +2,25 @@ defmodule Commanded.Event.ResetEventHandlerTest do use ExUnit.Case import Commanded.Assertions.EventAssertions + import ExUnit.CaptureLog + alias Commanded.Event.Handler alias Commanded.Event.Mapper alias Commanded.EventStore + alias Commanded.EventStore.RecordedEvent + alias Commanded.EventStore.Subscription alias Commanded.ExampleDomain.BankAccount.BankAccountHandler alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened alias Commanded.ExampleDomain.BankApp alias Commanded.Helpers.Wait alias Commanded.UUID + defmodule PendingSubscriptionHandler do + use Commanded.Event.Handler, + application: Commanded.ExampleDomain.BankApp, + name: "PendingSubscriptionHandler" + end + describe "reset event handler" do setup do start_supervised!(BankApp) @@ -38,7 +48,373 @@ defmodule Commanded.Event.ResetEventHandlerTest do end) end - @tag :skip + test "should discard events delivered by the subscription before the reset" do + stream_uuid = UUID.uuid4() + initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}] + + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(initial_events)) + + handler = start_supervised!(BankAccountHandler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["ACC123"] + end) + + :ok = :sys.suspend(handler) + + send(handler, :reset) + + stale_events = [%BankAccountOpened{account_number: "ACC456", initial_balance: 1_000}] + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 1, to_event_data(stale_events)) + + Wait.until(fn -> + assert {:messages, [:reset, {:events, [_event]}]} = Process.info(handler, :messages) + end) + + :ok = :sys.resume(handler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["ACC123", "ACC456"] + end) + end + + test "should discard the `DOWN` message of a subscription that died before the reset" do + stream_uuid = UUID.uuid4() + initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}] + + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(initial_events)) + + handler = start_supervised!(BankAccountHandler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["ACC123"] + end) + + %Handler{subscription: %Subscription{subscription_pid: subscription_pid}} = + :sys.get_state(handler) + + :ok = BankAccountHandler.change_prefix("PREF_") + + :ok = :sys.suspend(handler) + + send(handler, :reset) + + :ok = EventStore.unsubscribe(BankApp, subscription_pid) + + Wait.until(fn -> + assert {:messages, [:reset, {:DOWN, _ref, :process, ^subscription_pid, _reason}]} = + Process.info(handler, :messages) + end) + + log = + capture_log(fn -> + :ok = :sys.resume(handler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["PREF_ACC123"] + end) + end) + + refute log =~ "received unexpected message" + end + + test "should discard the `subscribed` message of a subscription deleted by a later reset" do + stream_uuid = UUID.uuid4() + initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}] + + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(initial_events)) + + handler = start_supervised!(BankAccountHandler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["ACC123"] + end) + + :ok = BankAccountHandler.change_prefix("PREF_") + + :ok = :sys.suspend(handler) + + send(handler, :reset) + send(handler, :reset) + + Wait.until(fn -> + assert {:messages, [:reset, :reset]} = Process.info(handler, :messages) + end) + + log = + capture_log(fn -> + :ok = :sys.resume(handler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["PREF_ACC123"] + end) + end) + + refute log =~ "received unexpected message" + end + + test "should keep a `subscribed` message sent by anything other than the reset subscription" do + stream_uuid = UUID.uuid4() + initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}] + + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(initial_events)) + + handler = start_supervised!(BankAccountHandler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["ACC123"] + end) + + :ok = BankAccountHandler.change_prefix("PREF_") + + :ok = :sys.suspend(handler) + + send(handler, :reset) + send(handler, {:subscribed, self()}) + + Wait.until(fn -> + assert {:messages, [:reset, {:subscribed, _}]} = Process.info(handler, :messages) + end) + + log = + capture_log(fn -> + :ok = :sys.resume(handler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["PREF_ACC123"] + end) + end) + + assert log =~ "received unexpected message: {:subscribed, #{inspect(self())}}" + end + + test "should keep messages unrelated to the subscription it reset" do + stream_uuid = UUID.uuid4() + initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}] + + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(initial_events)) + + handler = start_supervised!(BankAccountHandler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["ACC123"] + end) + + :ok = BankAccountHandler.change_prefix("PREF_") + + :ok = :sys.suspend(handler) + + send(handler, :reset) + send(handler, :first_unrelated_message) + send(handler, {:second_unrelated_message, self()}) + + Wait.until(fn -> + assert {:messages, [:reset, :first_unrelated_message, {:second_unrelated_message, _}]} = + Process.info(handler, :messages) + end) + + log = + capture_log(fn -> + :ok = :sys.resume(handler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["PREF_ACC123"] + end) + end) + + assert log =~ + ~r/received unexpected message: :first_unrelated_message.*received unexpected message: \{:second_unrelated_message/s + end + + test "should discard every message queued by the subscription it reset" do + level = Logger.level() + Logger.configure(level: :debug) + on_exit(fn -> Logger.configure(level: level) end) + + stream_uuid = UUID.uuid4() + initial_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}] + + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(initial_events)) + + handler = start_supervised!(BankAccountHandler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["ACC123"] + end) + + %Handler{subscription: %Subscription{subscription_pid: subscription_pid}} = + :sys.get_state(handler) + + :ok = BankAccountHandler.change_prefix("PREF_") + + :ok = :sys.suspend(handler) + + send(handler, :reset) + send(handler, {:events, []}) + send(handler, {:events, []}) + send(handler, {:subscribed, subscription_pid}) + + Wait.until(fn -> + assert {:messages, + [:reset, {:events, []}, {:events, []}, {:subscribed, ^subscription_pid}]} = + Process.info(handler, :messages) + end) + + log = + capture_log(fn -> + :ok = :sys.resume(handler) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["PREF_ACC123"] + end) + end) + + assert log =~ "discarded 3 message(s) queued by the subscription it reset" + refute log =~ "received unexpected message" + end + + test "should be reset while its subscription attempt is still being retried" do + {:ok, competing_subscription} = + EventStore.subscribe_to(BankApp, :all, "PendingSubscriptionHandler", self(), :origin, []) + + handler = start_supervised!(PendingSubscriptionHandler) + + Wait.until(fn -> + assert %Handler{ + subscribe_timer: subscribe_timer, + subscription: %Subscription{subscription_pid: nil} + } = :sys.get_state(handler) + + assert is_reference(subscribe_timer) + end) + + :ok = EventStore.unsubscribe(BankApp, competing_subscription) + + ref = Process.monitor(handler) + + send(handler, :reset) + + refute_receive {:DOWN, ^ref, :process, ^handler, _reason} + + Wait.until(fn -> + assert %Handler{subscription: %Subscription{subscription_pid: subscription_pid}} = + :sys.get_state(handler) + + assert is_pid(subscription_pid) + end) + end + + test "should cancel a pending subscription retry when reset" do + {:ok, competing_subscription} = + EventStore.subscribe_to(BankApp, :all, "PendingSubscriptionHandler", self(), :origin, []) + + handler = start_supervised!(PendingSubscriptionHandler) + + subscribe_timer = + Wait.until(fn -> + assert %Handler{ + subscribe_timer: subscribe_timer, + subscription: %Subscription{subscription_pid: nil} + } = :sys.get_state(handler) + + assert is_reference(subscribe_timer) + + subscribe_timer + end) + + :ok = EventStore.unsubscribe(BankApp, competing_subscription) + + send(handler, :reset) + + Wait.until(fn -> + assert %Handler{ + subscribe_timer: nil, + subscription: %Subscription{subscription_pid: subscription_pid} + } = :sys.get_state(handler) + + assert is_pid(subscription_pid) + end) + + assert Process.read_timer(subscribe_timer) == false + end + + test "should be reset while another subscriber still holds its subscription name" do + {:ok, competing_subscription} = + EventStore.subscribe_to(BankApp, :all, "PendingSubscriptionHandler", self(), :origin, []) + + assert_receive {:subscribed, ^competing_subscription} + + handler = start_supervised!(PendingSubscriptionHandler) + + Wait.until(fn -> + assert %Handler{ + subscribe_timer: subscribe_timer, + subscription: %Subscription{subscription_pid: nil} + } = :sys.get_state(handler) + + assert is_reference(subscribe_timer) + end) + + event_store_pid = Process.whereis(Module.concat([BankApp, "EventStore"])) + handler_ref = Process.monitor(handler) + event_store_ref = Process.monitor(event_store_pid) + + send(handler, :reset) + + refute_receive {:DOWN, ^handler_ref, :process, ^handler, _reason} + refute_receive {:DOWN, ^event_store_ref, :process, ^event_store_pid, _reason} + + stream_uuid = UUID.uuid4() + events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}] + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(events)) + + assert_receive {:events, [%RecordedEvent{}]} + end + + test "should discard a subscription retry that expired before the reset" do + {:ok, competing_subscription} = + EventStore.subscribe_to(BankApp, :all, "PendingSubscriptionHandler", self(), :origin, []) + + handler = start_supervised!(PendingSubscriptionHandler) + + Wait.until(fn -> + assert %Handler{ + subscribe_timer: subscribe_timer, + subscription: %Subscription{subscription_pid: nil} + } = :sys.get_state(handler) + + assert is_reference(subscribe_timer) + end) + + :ok = :sys.suspend(handler) + + send(handler, :reset) + + # The retry has to expire while the handler is suspended, so its message is already queued + # behind the reset by the time cancelling the timer can no longer recall it. + Wait.until(3_000, fn -> + assert {:messages, [:reset, :subscribe_to_events]} = Process.info(handler, :messages) + end) + + :ok = EventStore.unsubscribe(BankApp, competing_subscription) + + log = + capture_log(fn -> + :ok = :sys.resume(handler) + + Wait.until(fn -> + assert %Handler{ + subscribe_timer: nil, + subscription: %Subscription{subscription_pid: subscription_pid} + } = :sys.get_state(handler) + + assert is_pid(subscription_pid) + end) + end) + + refute log =~ "failed to subscribe to event store" + end + test "should be reset when starting from `:current`" do stream_uuid = UUID.uuid4() @@ -54,8 +430,21 @@ defmodule Commanded.Event.ResetEventHandlerTest do :ok = BankAccountHandler.change_prefix("PREF_") + %Handler{subscription: %Subscription{subscription_pid: subscription_pid}} = + :sys.get_state(handler) + send(handler, :reset) + # Subscribing from `:current` resolves the checkpoint when the replacement subscription is + # created, so an event appended before that lands behind it and is never delivered. + Wait.until(fn -> + assert %Handler{subscription: %Subscription{subscription_pid: reset_subscription_pid}} = + :sys.get_state(handler) + + assert is_pid(reset_subscription_pid) + refute reset_subscription_pid == subscription_pid + end) + new_event = [%BankAccountOpened{account_number: "ACC1234", initial_balance: 1_000}] :ok = EventStore.append_to_stream(BankApp, stream_uuid, 1, to_event_data(new_event)) @@ -67,6 +456,53 @@ defmodule Commanded.Event.ResetEventHandlerTest do assert BankAccountHandler.current_accounts() == ["PREF_ACC1234"] end) end + + test "should not apply an event delivered before a reset that starts from `:current`" do + stream_uuid = UUID.uuid4() + + # Ignored initial events + ignored_events = [%BankAccountOpened{account_number: "ACC123", initial_balance: 1_000}] + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 0, to_event_data(ignored_events)) + + handler = start_supervised!({BankAccountHandler, start_from: :current}) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == [] + end) + + %Handler{subscription: %Subscription{subscription_pid: subscription_pid}} = + :sys.get_state(handler) + + :ok = BankAccountHandler.change_prefix("PREF_") + + :ok = :sys.suspend(handler) + + send(handler, :reset) + + in_flight_events = [%BankAccountOpened{account_number: "ACC456", initial_balance: 1_000}] + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 1, to_event_data(in_flight_events)) + + Wait.until(fn -> + assert {:messages, [:reset, {:events, [_event]}]} = Process.info(handler, :messages) + end) + + :ok = :sys.resume(handler) + + Wait.until(fn -> + assert %Handler{subscription: %Subscription{subscription_pid: reset_subscription_pid}} = + :sys.get_state(handler) + + assert is_pid(reset_subscription_pid) + refute reset_subscription_pid == subscription_pid + end) + + new_events = [%BankAccountOpened{account_number: "ACC789", initial_balance: 1_000}] + :ok = EventStore.append_to_stream(BankApp, stream_uuid, 2, to_event_data(new_events)) + + Wait.until(fn -> + assert BankAccountHandler.current_accounts() == ["PREF_ACC789"] + end) + end end defp to_event_data(events) do