From 972d9b3bf0a25efebe5dff3263dc083216782d06 Mon Sep 17 00:00:00 2001 From: Santiago Tortosa Date: Tue, 8 Apr 2025 15:01:56 +0200 Subject: [PATCH 1/2] Monitor Notifications.Listener listen_to process (cherry picked from commit 97aa48402087e1c13189ba0ff635708643a6cdac) Signed-off-by: Yordis Prieto --- lib/event_store/notifications/listener.ex | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/lib/event_store/notifications/listener.ex b/lib/event_store/notifications/listener.ex index cfaea7dd..7aaab0c8 100644 --- a/lib/event_store/notifications/listener.ex +++ b/lib/event_store/notifications/listener.ex @@ -44,6 +44,14 @@ defmodule EventStore.Notifications.Listener do dispatch_events([], state) end + def handle_info({:DOWN, _ref, :process, listen_to, reason}, %{listen_to: listen_to} = state) do + {:stop, reason, state} + end + + def handle_info({:DOWN, _ref, :process, _pid, _reason}, state) do + {:noreply, state} + end + def handle_demand(incoming_demand, %Listener{} = state) do %Listener{demand: pending_demand} = state @@ -63,6 +71,8 @@ defmodule EventStore.Notifications.Listener do {:eventually, ref} -> ref end + Process.monitor(listen_to) + %Listener{state | ref: ref} end From d12cca0d394a8b9198bc8834299af309d9552de2 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 15 Sep 2026 08:22:52 -0400 Subject: [PATCH 2/2] fix(notifications): react to the listener connection going down The monitor added for that purpose never matched, so recovery depended on an unrelated crash path and surfaced a misleading exit reason. Signed-off-by: Yordis Prieto --- lib/event_store/notifications/listener.ex | 20 ++++-- test/notifications/listener_test.exs | 83 +++++++++++++++++++++++ 2 files changed, 97 insertions(+), 6 deletions(-) create mode 100644 test/notifications/listener_test.exs diff --git a/lib/event_store/notifications/listener.ex b/lib/event_store/notifications/listener.ex index 7aaab0c8..77094990 100644 --- a/lib/event_store/notifications/listener.ex +++ b/lib/event_store/notifications/listener.ex @@ -13,7 +13,15 @@ defmodule EventStore.Notifications.Listener do alias EventStore.Notifications.{Listener, Notification} - defstruct [:listen_to, :query_timeout, :schema, :ref, demand: 0, queue: :queue.new()] + defstruct [ + :listen_to, + :monitor_ref, + :query_timeout, + :schema, + :ref, + demand: 0, + queue: :queue.new() + ] def start_link(opts) do {start_opts, listener_opts} = @@ -44,12 +52,12 @@ defmodule EventStore.Notifications.Listener do dispatch_events([], state) end - def handle_info({:DOWN, _ref, :process, listen_to, reason}, %{listen_to: listen_to} = state) do + def handle_info({:DOWN, ref, :process, _object, reason}, %Listener{monitor_ref: ref} = state) do {:stop, reason, state} end - def handle_info({:DOWN, _ref, :process, _pid, _reason}, state) do - {:noreply, state} + def handle_info({:DOWN, _ref, :process, _object, _reason}, %Listener{} = state) do + {:noreply, [], state} end def handle_demand(incoming_demand, %Listener{} = state) do @@ -71,9 +79,9 @@ defmodule EventStore.Notifications.Listener do {:eventually, ref} -> ref end - Process.monitor(listen_to) + monitor_ref = Process.monitor(listen_to) - %Listener{state | ref: ref} + %Listener{state | ref: ref, monitor_ref: monitor_ref} end defp dispatch_events(events, %Listener{demand: 0} = state) do diff --git a/test/notifications/listener_test.exs b/test/notifications/listener_test.exs new file mode 100644 index 00000000..ecc2f3c1 --- /dev/null +++ b/test/notifications/listener_test.exs @@ -0,0 +1,83 @@ +defmodule EventStore.Notifications.ListenerTest do + use EventStore.StorageCase + + @moduletag :capture_log + + alias EventStore.{EventFactory, PubSub, Wait} + + @listener TestEventStore.EventStore.Notifications.Listener + @listen_to TestEventStore.Postgrex.Notifications + + describe "listen_to process down" do + test "keeps delivering events after the listen_to process dies" do + stream_uuid = "example-stream" + + :ok = PubSub.subscribe(TestEventStore, stream_uuid) + + listener = whereis(@listener) + + kill(@listen_to) + wait_until_listening(listener) + + :ok = append_events(stream_uuid, 3) + + assert_receive {:events, events}, 5_000 + assert length(events) == 3 + end + + test "stops with the reason reported for the listen_to process" do + listener = whereis(@listener) + ref = Process.monitor(listener) + + kill(@listen_to) + + assert_receive {:DOWN, ^ref, :process, ^listener, reason}, 5_000 + assert reason == :killed + end + + test "ignores a down message for an unrelated process" do + stream_uuid = "example-stream" + + :ok = PubSub.subscribe(TestEventStore, stream_uuid) + + listener = whereis(@listener) + + send(listener, {:DOWN, make_ref(), :process, spawn(fn -> :ok end), :normal}) + + :ok = append_events(stream_uuid, 3) + + assert_receive {:events, events}, 5_000 + assert length(events) == 3 + assert Process.alive?(listener) + end + end + + defp whereis(name) do + pid = Process.whereis(name) + assert is_pid(pid) + pid + end + + defp kill(name) do + pid = whereis(name) + Process.exit(pid, :kill) + end + + # The supervisor restarts the notifications connection and the listener. Wait + # for a replacement listener, and for a connection that is actually online: + # `listen/3` answers `{:eventually, ref}` while it is still reconnecting. + defp wait_until_listening(previous_listener) do + Wait.until(10_000, fn -> + listener = Process.whereis(@listener) + + assert is_pid(listener) + assert listener != previous_listener + assert {:ok, ref} = Postgrex.Notifications.listen(@listen_to, "listener_test_probe") + assert is_reference(ref) + end) + end + + defp append_events(stream_uuid, count) do + TestEventStore.append_to_stream(stream_uuid, :any_version, EventFactory.create_events(count)) + end +end