From 813f0e78d6e7a330c3cf063ed2810543823c37a2 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 13:19:31 -0400 Subject: [PATCH 01/13] fix(handler): prevent a reset from corrupting the handler read model A reset that lands while events are in flight left the read model wrong and made the Postgres EventStore reject the resulting ack. Signed-off-by: Yordis Prieto --- lib/commanded/event/handler.ex | 22 +++++++++++++++++- test/event/reset_event_handler_test.exs | 30 +++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex index 12234713..e8a43da5 100644 --- a/lib/commanded/event/handler.ex +++ b/lib/commanded/event/handler.ex @@ -1063,7 +1063,27 @@ defmodule Commanded.Event.Handler do subscription = Subscription.reset(subscription) - %Handler{state | last_seen_event: nil, subscription: subscription, subscribe_timer: nil} + # The deleted subscription's process is stopped before `Subscription.reset/1` returns and the + # new subscription does not exist yet, so any queued events came from the subscription reset. + drain_stale_events() + + state = cancel_batch_timer(state) + + %Handler{ + state + | last_seen_event: nil, + subscription: subscription, + subscribe_timer: nil, + batch_buffer: [] + } + end + + defp drain_stale_events do + receive do + {:events, _events} -> drain_stale_events() + after + 0 -> :ok + end end defp subscribe_to_events(%Handler{} = state) do diff --git a/test/event/reset_event_handler_test.exs b/test/event/reset_event_handler_test.exs index a992fec7..2e4d50cc 100644 --- a/test/event/reset_event_handler_test.exs +++ b/test/event/reset_event_handler_test.exs @@ -38,6 +38,36 @@ defmodule Commanded.Event.ResetEventHandlerTest do end) end + 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 + @tag :skip test "should be reset when starting from `:current`" do stream_uuid = UUID.uuid4() From cae370d0f9d7563c6f0c716d8bbb449036198483 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 14:01:33 -0400 Subject: [PATCH 02/13] fix(handler): keep a reset from resurrecting its previous subscription A reset that arrives while the handler is still retrying, batching, or holding signals from the subscription it is replacing must not carry any of that state into the new subscription. Signed-off-by: Yordis Prieto --- lib/commanded/event/handler.ex | 41 +++-- lib/commanded/event_store/subscription.ex | 16 +- .../event/event_handler_subscription_test.exs | 96 +++++++++++ test/event/reset_batch_event_handler_test.exs | 43 +++++ test/event/reset_event_handler_test.exs | 161 ++++++++++++++++++ 5 files changed, 342 insertions(+), 15 deletions(-) diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex index e8a43da5..bc11a759 100644 --- a/lib/commanded/event/handler.ex +++ b/lib/commanded/event/handler.ex @@ -1064,23 +1064,19 @@ defmodule Commanded.Event.Handler do subscription = Subscription.reset(subscription) # The deleted subscription's process is stopped before `Subscription.reset/1` returns and the - # new subscription does not exist yet, so any queued events came from the subscription reset. - drain_stale_events() + # new subscription does not exist yet, so any queued message came from the deleted + # subscription and must not be attributed to its replacement. + drain_stale_subscription_messages() - state = cancel_batch_timer(state) + state = state |> cancel_batch_timer() |> cancel_subscribe_timer() - %Handler{ - state - | last_seen_event: nil, - subscription: subscription, - subscribe_timer: nil, - batch_buffer: [] - } + %Handler{state | last_seen_event: nil, subscription: subscription, batch_buffer: []} end - defp drain_stale_events do + defp drain_stale_subscription_messages do receive do - {:events, _events} -> drain_stale_events() + {:events, _events} -> drain_stale_subscription_messages() + {:subscribed, _subscription} -> drain_stale_subscription_messages() after 0 -> :ok end @@ -1278,6 +1274,27 @@ 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 + + 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/subscription.ex b/lib/commanded/event_store/subscription.ex index 76367976..ca5c7014 100644 --- a/lib/commanded/event_store/subscription.ex +++ b/lib/commanded/event_store/subscription.ex @@ -96,10 +96,20 @@ 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 + + if is_pid(subscription_pid) do + :ok = EventStore.unsubscribe(application, subscription_pid) + end - :ok = EventStore.unsubscribe(application, subscription_pid) - :ok = EventStore.delete_subscription(application, subscribe_to, subscription_name) + case EventStore.delete_subscription(application, subscribe_to, subscription_name) do + :ok -> :ok + {:error, :subscription_not_found} -> :ok + end %Subscription{ subscription 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..0d4e9b3d 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,47 @@ 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 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 2e4d50cc..331925cd 100644 --- a/test/event/reset_event_handler_test.exs +++ b/test/event/reset_event_handler_test.exs @@ -2,15 +2,24 @@ 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.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) @@ -68,6 +77,158 @@ defmodule Commanded.Event.ResetEventHandlerTest do 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 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) + + subscription_pid = + Wait.until(fn -> + assert %Handler{ + subscribe_timer: nil, + subscription: %Subscription{subscription_pid: subscription_pid} + } = :sys.get_state(handler) + + assert is_pid(subscription_pid) + + subscription_pid + end) + + assert Process.read_timer(subscribe_timer) == false + + # The first retry jitters within a second of backoff, so an uncancelled timer fires inside + # this window + Process.sleep(3_500) + + assert %Handler{ + subscribe_timer: nil, + subscription: %Subscription{subscription_pid: ^subscription_pid} + } = :sys.get_state(handler) + end + @tag :skip test "should be reset when starting from `:current`" do stream_uuid = UUID.uuid4() From a8e5e11f042955287b94f124fe2313ac2730f4a8 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 14:16:03 -0400 Subject: [PATCH 03/13] chore(handler): drop a sleep whose outcome is already asserted elsewhere Signed-off-by: Yordis Prieto --- test/event/reset_event_handler_test.exs | 26 +++++++------------------ 1 file changed, 7 insertions(+), 19 deletions(-) diff --git a/test/event/reset_event_handler_test.exs b/test/event/reset_event_handler_test.exs index 331925cd..8d49b3d8 100644 --- a/test/event/reset_event_handler_test.exs +++ b/test/event/reset_event_handler_test.exs @@ -205,28 +205,16 @@ defmodule Commanded.Event.ResetEventHandlerTest do send(handler, :reset) - subscription_pid = - Wait.until(fn -> - assert %Handler{ - subscribe_timer: nil, - subscription: %Subscription{subscription_pid: subscription_pid} - } = :sys.get_state(handler) - - assert is_pid(subscription_pid) + Wait.until(fn -> + assert %Handler{ + subscribe_timer: nil, + subscription: %Subscription{subscription_pid: subscription_pid} + } = :sys.get_state(handler) - subscription_pid - end) + assert is_pid(subscription_pid) + end) assert Process.read_timer(subscribe_timer) == false - - # The first retry jitters within a second of backoff, so an uncancelled timer fires inside - # this window - Process.sleep(3_500) - - assert %Handler{ - subscribe_timer: nil, - subscription: %Subscription{subscription_pid: ^subscription_pid} - } = :sys.get_state(handler) end @tag :skip From 07df654cf10c1a2ba39d92a0cfc2713c67ebc115 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 15:14:21 -0400 Subject: [PATCH 04/13] fix(handler): scope a reset's discarded messages to the subscription it reset A drain that matched any sender could silently swallow a live signal if the reset ever stopped being the only thing in flight, and left no trace that anything had been thrown away. Signed-off-by: Yordis Prieto --- lib/commanded/event/handler.ex | 44 ++++++++++++++++++------- test/event/reset_event_handler_test.exs | 35 ++++++++++++++++++++ 2 files changed, 68 insertions(+), 11 deletions(-) diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex index bc11a759..159b8bf5 100644 --- a/lib/commanded/event/handler.ex +++ b/lib/commanded/event/handler.ex @@ -1059,26 +1059,48 @@ 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) - # The deleted subscription's process is stopped before `Subscription.reset/1` returns and the - # new subscription does not exist yet, so any queued message came from the deleted - # subscription and must not be attributed to its replacement. - drain_stale_subscription_messages() - - state = state |> cancel_batch_timer() |> cancel_subscribe_timer() + 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 - defp drain_stale_subscription_messages do + # `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} -> drain_stale_subscription_messages() - {:subscribed, _subscription} -> drain_stale_subscription_messages() + {:events, _events} -> + discard_messages_from(reset_subscription_pid, discarded + 1) + + {:subscribed, ^reset_subscription_pid} -> + discard_messages_from(reset_subscription_pid, discarded + 1) after - 0 -> :ok + 0 -> discarded end end diff --git a/test/event/reset_event_handler_test.exs b/test/event/reset_event_handler_test.exs index 8d49b3d8..f2c086e3 100644 --- a/test/event/reset_event_handler_test.exs +++ b/test/event/reset_event_handler_test.exs @@ -152,6 +152,41 @@ defmodule Commanded.Event.ResetEventHandlerTest do 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 be reset while its subscription attempt is still being retried" do {:ok, competing_subscription} = EventStore.subscribe_to(BankApp, :all, "PendingSubscriptionHandler", self(), :origin, []) From eadbe8807c4eb055ad3e2dc7b0a7f965a9fc9bae Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 15:49:37 -0400 Subject: [PATCH 05/13] chore(handler): prove a reset cannot mistake another sender's messages for its own The reset's guarantees were only asserted end to end, so its scoping and its timer handling could regress without any test noticing. Signed-off-by: Yordis Prieto --- test/event/reset_batch_event_handler_test.exs | 39 +++++ test/event/reset_event_handler_test.exs | 146 +++++++++++++++++- 2 files changed, 184 insertions(+), 1 deletion(-) diff --git a/test/event/reset_batch_event_handler_test.exs b/test/event/reset_batch_event_handler_test.exs index 0d4e9b3d..eb5441ae 100644 --- a/test/event/reset_batch_event_handler_test.exs +++ b/test/event/reset_batch_event_handler_test.exs @@ -81,6 +81,45 @@ defmodule Commanded.Event.BatchResetEventHandlerTest do 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 f2c086e3..5a3843f7 100644 --- a/test/event/reset_event_handler_test.exs +++ b/test/event/reset_event_handler_test.exs @@ -187,6 +187,91 @@ defmodule Commanded.Event.ResetEventHandlerTest do 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, []) @@ -252,7 +337,6 @@ defmodule Commanded.Event.ResetEventHandlerTest do assert Process.read_timer(subscribe_timer) == false end - @tag :skip test "should be reset when starting from `:current`" do stream_uuid = UUID.uuid4() @@ -268,8 +352,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)) @@ -281,6 +378,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 From 850e08ee16b897df478987a1f99a4800ee5ea90b Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 16:06:57 -0400 Subject: [PATCH 06/13] chore(handler): cover the subscription retry that expires before a reset Cancelling that timer cannot recall a message it already delivered, and nothing asserted the difference. Signed-off-by: Yordis Prieto --- test/event/reset_event_handler_test.exs | 44 +++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/test/event/reset_event_handler_test.exs b/test/event/reset_event_handler_test.exs index 5a3843f7..34ee1068 100644 --- a/test/event/reset_event_handler_test.exs +++ b/test/event/reset_event_handler_test.exs @@ -337,6 +337,50 @@ defmodule Commanded.Event.ResetEventHandlerTest do assert Process.read_timer(subscribe_timer) == false 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() From 556616726229c7bd96ff3313fe5adfeb5627c09d Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 16:26:38 -0400 Subject: [PATCH 07/13] chore(handler): record why cancelling a subscribe retry is not enough on its own Signed-off-by: Yordis Prieto --- lib/commanded/event/handler.ex | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex index 159b8bf5..7a07acd1 100644 --- a/lib/commanded/event/handler.ex +++ b/lib/commanded/event/handler.ex @@ -1309,6 +1309,13 @@ defmodule Commanded.Event.Handler do 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 From e769f6abec4b25fe336dd23b06dff18862bd9e01 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 16:39:53 -0400 Subject: [PATCH 08/13] fix(in-memory): keep a reset from crashing the event store it shares A handler that resets while another subscriber still holds its subscription name has nothing of its own to delete, and taking the event store down with it loses every other subscriber too. Signed-off-by: Yordis Prieto --- lib/commanded/event_store/adapter.ex | 5 ++- .../event_store/adapters/in_memory.ex | 3 ++ lib/commanded/event_store/subscription.ex | 4 +++ test/event/reset_event_handler_test.exs | 34 +++++++++++++++++++ 4 files changed, 45 insertions(+), 1 deletion(-) diff --git a/lib/commanded/event_store/adapter.ex b/lib/commanded/event_store/adapter.ex index 50600340..3984a1ce 100644 --- a/lib/commanded/event_store/adapter.ex +++ b/lib/commanded/event_store/adapter.ex @@ -100,7 +100,10 @@ defmodule Commanded.EventStore.Adapter do stream_uuid | :all, subscription_name ) :: - :ok | {:error, :subscription_not_found} | {:error, error} + :ok + | {:error, :subscription_not_found} + | {:error, :subscription_has_subscribers} + | {: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..e72113d7 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, :subscription_has_subscribers}, 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 ca5c7014..3ea56ea5 100644 --- a/lib/commanded/event_store/subscription.ex +++ b/lib/commanded/event_store/subscription.ex @@ -106,9 +106,13 @@ defmodule Commanded.EventStore.Subscription 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, :subscription_has_subscribers} -> :ok end %Subscription{ diff --git a/test/event/reset_event_handler_test.exs b/test/event/reset_event_handler_test.exs index 34ee1068..2cbbbcc8 100644 --- a/test/event/reset_event_handler_test.exs +++ b/test/event/reset_event_handler_test.exs @@ -7,6 +7,7 @@ defmodule Commanded.Event.ResetEventHandlerTest do 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 @@ -337,6 +338,39 @@ defmodule Commanded.Event.ResetEventHandlerTest do 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, []) From 984e190bb82434259e77630042d56b7802de953e Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 16:52:19 -0400 Subject: [PATCH 09/13] fix(handler): keep one handler's reset from taking its concurrent siblings down The in-memory and EventStore adapters disagree on whether a subscription with subscribers can be deleted, and a concurrent handler reset hits that disagreement on every reset, so the divergence needs to be stated where the decision is made. Signed-off-by: Yordis Prieto --- lib/commanded/event_store/adapter.ex | 4 ++++ lib/commanded/event_store/subscription.ex | 6 +++++ test/event/event_handler_concurrency_test.exs | 22 +++++++++++++++++++ 3 files changed, 32 insertions(+) diff --git a/lib/commanded/event_store/adapter.ex b/lib/commanded/event_store/adapter.ex index 3984a1ce..39b96e22 100644 --- a/lib/commanded/event_store/adapter.ex +++ b/lib/commanded/event_store/adapter.ex @@ -94,6 +94,10 @@ defmodule Commanded.EventStore.Adapter do @doc """ Delete an existing subscription. + + Whether a subscription that still has subscribers can be deleted is adapter specific. The + in-memory adapter refuses with `{:error, :subscription_has_subscribers}` and leaves them + attached; the EventStore adapter deletes it and disconnects them. """ @callback delete_subscription( adapter_meta, diff --git a/lib/commanded/event_store/subscription.ex b/lib/commanded/event_store/subscription.ex index 3ea56ea5..a7a88a87 100644 --- a/lib/commanded/event_store/subscription.ex +++ b/lib/commanded/event_store/subscription.ex @@ -109,6 +109,12 @@ defmodule Commanded.EventStore.Subscription do # 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. + # + # TODO: the EventStore adapter answers `:ok` instead, having stopped the subscription process + # that every subscriber shares before deleting the checkpoint, so the same reset disconnects + # them rather than leaving them be. Settling which of the two is the contract means changing + # `EventStore.delete_subscription/3`, which exposes no way to ask whether a subscription has + # subscribers. case EventStore.delete_subscription(application, subscribe_to, subscription_name) do :ok -> :ok {:error, :subscription_not_found} -> :ok 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", From 0fc393dade27a9ff13bf7286f279973116228d55 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 22:16:12 -0400 Subject: [PATCH 10/13] chore(event-store): name the refusal after the state that causes it Signed-off-by: Yordis Prieto --- lib/commanded/event_store/adapter.ex | 6 +++--- lib/commanded/event_store/adapters/in_memory.ex | 2 +- lib/commanded/event_store/subscription.ex | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/lib/commanded/event_store/adapter.ex b/lib/commanded/event_store/adapter.ex index 39b96e22..516f13b6 100644 --- a/lib/commanded/event_store/adapter.ex +++ b/lib/commanded/event_store/adapter.ex @@ -96,8 +96,8 @@ defmodule Commanded.EventStore.Adapter do Delete an existing subscription. Whether a subscription that still has subscribers can be deleted is adapter specific. The - in-memory adapter refuses with `{:error, :subscription_has_subscribers}` and leaves them - attached; the EventStore adapter deletes it and disconnects them. + in-memory adapter refuses with `{:error, :still_subscribed}` and leaves them attached; the + EventStore adapter deletes it and disconnects them. """ @callback delete_subscription( adapter_meta, @@ -106,7 +106,7 @@ defmodule Commanded.EventStore.Adapter do ) :: :ok | {:error, :subscription_not_found} - | {:error, :subscription_has_subscribers} + | {:error, :still_subscribed} | {:error, error} @doc """ diff --git a/lib/commanded/event_store/adapters/in_memory.ex b/lib/commanded/event_store/adapters/in_memory.ex index e72113d7..77c67f94 100644 --- a/lib/commanded/event_store/adapters/in_memory.ex +++ b/lib/commanded/event_store/adapters/in_memory.ex @@ -289,7 +289,7 @@ defmodule Commanded.EventStore.Adapters.InMemory do {:ok, state} %PersistentSubscription{stream_uuid: ^stream_uuid} -> - {{:error, :subscription_has_subscribers}, state} + {{:error, :still_subscribed}, state} nil -> {{:error, :subscription_not_found}, state} diff --git a/lib/commanded/event_store/subscription.ex b/lib/commanded/event_store/subscription.ex index a7a88a87..30f0d2dc 100644 --- a/lib/commanded/event_store/subscription.ex +++ b/lib/commanded/event_store/subscription.ex @@ -118,7 +118,7 @@ defmodule Commanded.EventStore.Subscription do case EventStore.delete_subscription(application, subscribe_to, subscription_name) do :ok -> :ok {:error, :subscription_not_found} -> :ok - {:error, :subscription_has_subscribers} -> :ok + {:error, :still_subscribed} -> :ok end %Subscription{ From 8292480343796deddc7c651df014204391f2f7c5 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 22:44:27 -0400 Subject: [PATCH 11/13] chore(event-store): name the refusal after what the caller has to disconnect Signed-off-by: Yordis Prieto --- lib/commanded/event_store/adapter.ex | 4 ++-- lib/commanded/event_store/adapters/in_memory.ex | 2 +- lib/commanded/event_store/subscription.ex | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/lib/commanded/event_store/adapter.ex b/lib/commanded/event_store/adapter.ex index 516f13b6..beba9a65 100644 --- a/lib/commanded/event_store/adapter.ex +++ b/lib/commanded/event_store/adapter.ex @@ -96,7 +96,7 @@ defmodule Commanded.EventStore.Adapter do Delete an existing subscription. Whether a subscription that still has subscribers can be deleted is adapter specific. The - in-memory adapter refuses with `{:error, :still_subscribed}` and leaves them attached; the + in-memory adapter refuses with `{:error, :subscribers_connected}` and leaves them attached; the EventStore adapter deletes it and disconnects them. """ @callback delete_subscription( @@ -106,7 +106,7 @@ defmodule Commanded.EventStore.Adapter do ) :: :ok | {:error, :subscription_not_found} - | {:error, :still_subscribed} + | {:error, :subscribers_connected} | {:error, error} @doc """ diff --git a/lib/commanded/event_store/adapters/in_memory.ex b/lib/commanded/event_store/adapters/in_memory.ex index 77c67f94..cb16056a 100644 --- a/lib/commanded/event_store/adapters/in_memory.ex +++ b/lib/commanded/event_store/adapters/in_memory.ex @@ -289,7 +289,7 @@ defmodule Commanded.EventStore.Adapters.InMemory do {:ok, state} %PersistentSubscription{stream_uuid: ^stream_uuid} -> - {{:error, :still_subscribed}, state} + {{:error, :subscribers_connected}, state} nil -> {{:error, :subscription_not_found}, state} diff --git a/lib/commanded/event_store/subscription.ex b/lib/commanded/event_store/subscription.ex index 30f0d2dc..5e6d5c7f 100644 --- a/lib/commanded/event_store/subscription.ex +++ b/lib/commanded/event_store/subscription.ex @@ -118,7 +118,7 @@ defmodule Commanded.EventStore.Subscription do case EventStore.delete_subscription(application, subscribe_to, subscription_name) do :ok -> :ok {:error, :subscription_not_found} -> :ok - {:error, :still_subscribed} -> :ok + {:error, :subscribers_connected} -> :ok end %Subscription{ From db96a8acfc3dcec398ec272c821b5e72925671fd Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 21 Sep 2026 23:25:27 -0400 Subject: [PATCH 12/13] chore(event-store): stop promising which way an adapter settles a contested delete Signed-off-by: Yordis Prieto --- lib/commanded/event_store/adapter.ex | 7 ++++--- lib/commanded/event_store/subscription.ex | 6 ------ 2 files changed, 4 insertions(+), 9 deletions(-) diff --git a/lib/commanded/event_store/adapter.ex b/lib/commanded/event_store/adapter.ex index beba9a65..4cb1cf2f 100644 --- a/lib/commanded/event_store/adapter.ex +++ b/lib/commanded/event_store/adapter.ex @@ -95,9 +95,10 @@ defmodule Commanded.EventStore.Adapter do @doc """ Delete an existing subscription. - Whether a subscription that still has subscribers can be deleted is adapter specific. The - in-memory adapter refuses with `{:error, :subscribers_connected}` and leaves them attached; the - EventStore adapter deletes it and disconnects them. + 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, diff --git a/lib/commanded/event_store/subscription.ex b/lib/commanded/event_store/subscription.ex index 5e6d5c7f..b6eeb546 100644 --- a/lib/commanded/event_store/subscription.ex +++ b/lib/commanded/event_store/subscription.ex @@ -109,12 +109,6 @@ defmodule Commanded.EventStore.Subscription do # 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. - # - # TODO: the EventStore adapter answers `:ok` instead, having stopped the subscription process - # that every subscriber shares before deleting the checkpoint, so the same reset disconnects - # them rather than leaving them be. Settling which of the two is the contract means changing - # `EventStore.delete_subscription/3`, which exposes no way to ask whether a subscription has - # subscribers. case EventStore.delete_subscription(application, subscribe_to, subscription_name) do :ok -> :ok {:error, :subscription_not_found} -> :ok From 7d4c3c7f36ba7aac2a66c04730750f3e7884f0a7 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Wed, 23 Sep 2026 21:23:05 -0400 Subject: [PATCH 13/13] fix(event-store): stop a reset from crashing on a delete failure it cannot name An adapter is free to fail the delete for reasons this reset does not enumerate, and the caller loses the whole reset to a crash instead of hearing what went wrong. Signed-off-by: Yordis Prieto --- lib/commanded/event_store/subscription.ex | 25 ++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/lib/commanded/event_store/subscription.ex b/lib/commanded/event_store/subscription.ex index b6eeb546..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} @@ -110,9 +112,26 @@ defmodule Commanded.EventStore.Subscription do # 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 + :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{