diff --git a/lib/event_store.ex b/lib/event_store.ex index 7c10c20e..3cbd0411 100644 --- a/lib/event_store.ex +++ b/lib/event_store.ex @@ -581,10 +581,7 @@ defmodule EventStore do name = name(opts) {conn, opts} = parse_opts(opts) - with :ok <- - Subscriptions.stop_subscription(name, stream_uuid, subscription_name, opts) do - Subscriptions.delete_subscription(conn, stream_uuid, subscription_name, opts) - end + Subscriptions.delete_subscription(name, conn, stream_uuid, subscription_name, opts) end ) end diff --git a/lib/event_store/sql/statements/subscription_ack.sql.eex b/lib/event_store/sql/statements/subscription_ack.sql.eex index 8b4e8cc3..c6da30f1 100644 --- a/lib/event_store/sql/statements/subscription_ack.sql.eex +++ b/lib/event_store/sql/statements/subscription_ack.sql.eex @@ -1,3 +1,3 @@ UPDATE "<%= schema %>".subscriptions -SET last_seen = $3 -WHERE stream_uuid = $1 AND subscription_name = $2; +SET last_seen = $2 +WHERE subscription_id = $1; diff --git a/lib/event_store/storage.ex b/lib/event_store/storage.ex index 0f2af5ac..668d88c2 100644 --- a/lib/event_store/storage.ex +++ b/lib/event_store/storage.ex @@ -74,8 +74,7 @@ defmodule EventStore.Storage do @doc """ Acknowledge receipt of an event by its number, for a single subscription. """ - defdelegate ack_last_seen_event(conn, stream_uuid, subscription_name, last_seen, opts), - to: Subscription + defdelegate ack_last_seen_event(conn, subscription_id, last_seen, opts), to: Subscription @doc """ Delete an existing named subscription to a stream. diff --git a/lib/event_store/storage/subscription.ex b/lib/event_store/storage/subscription.ex index 160294d0..c609a860 100644 --- a/lib/event_store/storage/subscription.ex +++ b/lib/event_store/storage/subscription.ex @@ -41,8 +41,8 @@ defmodule EventStore.Storage.Subscription do end end - def ack_last_seen_event(conn, stream_uuid, subscription_name, last_seen, opts) do - Subscription.Ack.execute(conn, stream_uuid, subscription_name, last_seen, opts) + def ack_last_seen_event(conn, subscription_id, last_seen, opts) do + Subscription.Ack.execute(conn, subscription_id, last_seen, opts) end def delete_subscription(conn, stream_uuid, subscription_name, opts), @@ -136,18 +136,28 @@ defmodule EventStore.Storage.Subscription do defmodule Ack do @moduledoc false - def execute(conn, stream_uuid, subscription_name, last_seen, opts) do + def execute(conn, subscription_id, last_seen, opts) do {schema, opts} = Keyword.pop(opts, :schema) query = Statements.subscription_ack(schema) - case Postgrex.query(conn, query, [stream_uuid, subscription_name, last_seen], opts) do + case Postgrex.query(conn, query, [subscription_id, last_seen], opts) do + # Naming the subscription by its identifier, rather than by the stream and name a later + # subscription can reuse, is what keeps this from writing a checkpoint onto whichever row + # holds that name now. + {:ok, %Postgrex.Result{num_rows: 0}} -> + Logger.warning( + "Failed to ack last seen event for subscription #{subscription_id} because it no longer exists" + ) + + {:error, :subscription_not_found} + {:ok, _result} -> :ok {:error, error} = reply -> Logger.warning( - "Failed to ack last seen event on stream \"#{stream_uuid}\" named \"#{subscription_name}\" due to: " <> + "Failed to ack last seen event for subscription #{subscription_id} due to: " <> inspect(error) ) diff --git a/lib/event_store/subscriptions.ex b/lib/event_store/subscriptions.ex index 844c17e7..08507fce 100644 --- a/lib/event_store/subscriptions.ex +++ b/lib/event_store/subscriptions.ex @@ -1,7 +1,6 @@ defmodule EventStore.Subscriptions do @moduledoc false - alias EventStore.Storage alias EventStore.Subscriptions.Subscription alias EventStore.Subscriptions.Supervisor, as: SubscriptionsSupervisor @@ -22,8 +21,9 @@ defmodule EventStore.Subscriptions do end defdelegate unsubscribe_from_stream(event_store, stream_uuid, name), to: SubscriptionsSupervisor - defdelegate stop_subscription(event_store, stream_uuid, name, opts), to: SubscriptionsSupervisor - defdelegate delete_subscription(conn, stream_uuid, subscription_name, opts), to: Storage + + defdelegate delete_subscription(event_store, conn, stream_uuid, name, opts), + to: SubscriptionsSupervisor @doc """ Get the delay between subscription retry attempts, in milliseconds, from the diff --git a/lib/event_store/subscriptions/subscription.ex b/lib/event_store/subscriptions/subscription.ex index bb2551a4..08497eac 100644 --- a/lib/event_store/subscriptions/subscription.ex +++ b/lib/event_store/subscriptions/subscription.ex @@ -89,14 +89,14 @@ defmodule EventStore.Subscriptions.Subscription do end @doc """ - Stop the subscription, unless a subscriber is still connected to it. + Delete the subscription, unless a subscriber is still connected to it. Accepts a `:timeout`, in milliseconds or `:infinity`. Whatever the subscription is busy with is storage work bounded by its own `:timeout`, so anything shorter than that gives up on a write that is still going to land. """ - def stop(subscription, opts \\ []) do - GenServer.call(subscription, :stop, Keyword.get(opts, :timeout, @default_timeout)) + def delete(subscription, opts \\ []) do + GenServer.call(subscription, :delete, Keyword.get(opts, :timeout, @default_timeout)) end @doc false @@ -187,6 +187,13 @@ defmodule EventStore.Subscriptions.Subscription do end end + @impl GenServer + def handle_info({:checkpoint_failed, reason}, %Subscription{} = state) do + Logger.warning(describe(state) <> " could not persist its checkpoint: #{inspect(reason)}") + + {:stop, reason, state} + end + @impl GenServer def handle_info({:EXIT, _from, reason}, %Subscription{} = state) do {:stop, reason, state} @@ -260,14 +267,27 @@ defmodule EventStore.Subscriptions.Subscription do end @impl GenServer - def handle_call(:stop, _from, %Subscription{} = state) do + def handle_call(:delete, _from, %Subscription{} = state) do %Subscription{ - subscription: %SubscriptionFsm{data: %SubscriptionState{subscribers: subscribers}} + subscription: + %SubscriptionFsm{data: %SubscriptionState{subscribers: subscribers}} = subscription } = state case map_size(subscribers) do - 0 -> {:stop, :shutdown, :ok, state} - _ -> {:reply, {:error, :subscribers_connected}, state} + 0 -> + # Holding the registered name for as long as this call runs is what keeps a replacement + # subscription from claiming the name and reading the row being deleted. + {reply, subscription} = SubscriptionFsm.delete(subscription) + + state = %Subscription{state | subscription: subscription} + + case reply do + :ok -> {:stop, :shutdown, :ok, state} + {:error, _error} -> {:reply, reply, state} + end + + _ -> + {:reply, {:error, :subscribers_connected}, state} end end diff --git a/lib/event_store/subscriptions/subscription_fsm.ex b/lib/event_store/subscriptions/subscription_fsm.ex index 441b06d2..b3941d46 100644 --- a/lib/event_store/subscriptions/subscription_fsm.ex +++ b/lib/event_store/subscriptions/subscription_fsm.ex @@ -210,6 +210,34 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do def checkpoint(%__MODULE__{data: %SubscriptionState{}} = fsm), do: fsm + # A subscription creates its own row when it subscribes, so deleting that row is the + # subscription's to do as well. + def delete(%__MODULE__{data: %SubscriptionState{} = data} = fsm) do + %SubscriptionState{ + conn: conn, + schema: schema, + stream_uuid: stream_uuid, + subscription_name: subscription_name, + query_timeout: query_timeout + } = data + + reply = + Storage.Subscription.delete_subscription(conn, stream_uuid, subscription_name, + schema: schema, + timeout: query_timeout + ) + + case reply do + # Nothing can remain pending once the row is gone, because a checkpoint has nothing left to + # be written to. A row that is still there keeps whatever was pending for it. + :ok -> + {:ok, %__MODULE__{fsm | data: %SubscriptionState{data | checkpoints_pending: 0}}} + + {:error, _error} -> + {reply, fsm} + end + end + # Notify events when subscribed def notify_events( %__MODULE__{state: :subscribed, data: %SubscriptionState{} = data} = fsm, @@ -735,6 +763,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do schema: schema, stream_uuid: stream_uuid, subscription_name: subscription_name, + subscription_id: subscription_id, last_ack: last_ack, query_timeout: query_timeout, checkpoints_pending: checkpoints_pending, @@ -749,12 +778,24 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do last_seen: last_ack }) - Telemetry.span(:subscription_checkpoint, metadata, fn -> - Storage.Subscription.ack_last_seen_event(conn, stream_uuid, subscription_name, last_ack, - schema: schema, - timeout: query_timeout - ) - end) + result = + Telemetry.span(:subscription_checkpoint, metadata, fn -> + Storage.Subscription.ack_last_seen_event(conn, subscription_id, last_ack, + schema: schema, + timeout: query_timeout + ) + end) + + # Reaching no row means the subscription being checkpointed has been deleted. Subscribing + # again would recreate it at the position the delete removed, so the subscription stops + # instead and leaves its subscribers to subscribe from wherever the delete left them. + case result do + {:error, :subscription_not_found} -> + send(self(), {:checkpoint_failed, :subscription_not_found}) + + _ -> + :ok + end end %SubscriptionState{data | checkpoints_pending: 0} diff --git a/lib/event_store/subscriptions/supervisor.ex b/lib/event_store/subscriptions/supervisor.ex index fef052be..3a589d23 100644 --- a/lib/event_store/subscriptions/supervisor.ex +++ b/lib/event_store/subscriptions/supervisor.ex @@ -5,6 +5,7 @@ defmodule EventStore.Subscriptions.Supervisor do use DynamicSupervisor + alias EventStore.Storage alias EventStore.Subscriptions alias EventStore.Subscriptions.Subscription @@ -41,12 +42,12 @@ defmodule EventStore.Subscriptions.Supervisor do end end - def stop_subscription(event_store, stream_uuid, subscription_name, opts \\ []) do + def delete_subscription(event_store, conn, stream_uuid, subscription_name, opts \\ []) do name = registry_name(event_store, stream_uuid, subscription_name) case Registry.whereis_name(name) do :undefined -> - :ok + Storage.delete_subscription(conn, stream_uuid, subscription_name, opts) subscription -> # Shaped after `:proc_lib.stop/3`, which is what `GenServer.stop/3` runs, except that only @@ -55,37 +56,49 @@ defmodule EventStore.Subscriptions.Supervisor do ref = Process.monitor(subscription) asked_at = System.monotonic_time(:millisecond) - case stop(subscription, opts) do - :ok -> - # Answering is not being gone: the reply is sent before `terminate/2`, where a - # subscription can still checkpoint. Waiting keeps a stale `last_seen` from landing on - # whatever row exists by the time it is written, and the caller keeps one deadline for - # both halves of that. - receive do - {:DOWN, ^ref, :process, ^subscription, _reason} -> :ok - after - remaining(opts, asked_at) -> - Process.demonitor(ref, [:flush]) + case delete(subscription, ref, opts) do + :gone -> + Process.demonitor(ref, [:flush]) - exit({:timeout, {__MODULE__, :stop_subscription, [subscription]}}) - end + Storage.delete_subscription(conn, stream_uuid, subscription_name, opts) {:error, _error} = error -> Process.demonitor(ref, [:flush]) error + + reply -> + # Answering is not being gone, and a caller that subscribes again under the same name + # the moment it is answered races a process that still holds it. One deadline covers + # being answered and being gone. + receive do + {:DOWN, ^ref, :process, ^subscription, _reason} -> reply + after + remaining(opts, asked_at) -> + Process.demonitor(ref, [:flush]) + + exit({:timeout, {__MODULE__, :delete_subscription, [subscription]}}) + end end end end - # A subscription that goes down while being asked to stop leaves nothing to stop. Its last - # subscriber has most likely just unsubscribed, which answers before the subscription it stops - # has terminated. A timeout is not that: it says the subscription may still be writing, so - # reading it as nothing left to stop would delete a row out from under a checkpoint. - defp stop(subscription, opts) do - Subscription.stop(subscription, opts) + # A subscription that goes down while being asked to delete itself has not deleted its row, and + # has given up the name that was keeping anything else from claiming it, which leaves the row to + # delete from here. A timeout is not that: it says the subscription may still be writing, so + # deleting the row from here would delete it out from under that write. + defp delete(subscription, ref, opts) do + Subscription.delete(subscription, opts) catch - :exit, {reason, {GenServer, :call, _args}} when reason != :timeout -> :ok + :exit, {reason, {GenServer, :call, _args}} when reason != :timeout -> + :gone + + :exit, reason -> + # Giving up here is still giving up on the monitor, which otherwise outlives the call and + # reaches whoever asked as a message they never set up to receive. + Process.demonitor(ref, [:flush]) + + exit(reason) end defp remaining(opts, asked_at) do diff --git a/test/storage/subscription_persistence_test.exs b/test/storage/subscription_persistence_test.exs index c438c88d..fd7ce5f5 100644 --- a/test/storage/subscription_persistence_test.exs +++ b/test/storage/subscription_persistence_test.exs @@ -69,10 +69,145 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do verify_subscription(subscription, 1) end + test "ack last seen event when the subscription no longer exists", context do + %{conn: conn, schema: schema} = context + + {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) + + :ok = delete_subscription(context) + + assert {:error, :subscription_not_found} = + Storage.ack_last_seen_event(conn, subscription_id, 1, schema: schema) + end + + test "ack last seen event only moves the subscription it names", context do + %{conn: conn, schema: schema} = context + + {:ok, %Storage.Subscription{subscription_id: acked}} = subscribe_to_stream(context) + + {:ok, %Storage.Subscription{subscription_id: untouched}} = + Storage.subscribe_to_stream(conn, @all_stream, "another_subscription", schema: schema) + + :ok = Storage.ack_last_seen_event(conn, acked, 1, schema: schema) + + assert {:ok, %Storage.Subscription{subscription_id: ^acked, last_seen: 1}} = + Storage.Subscription.subscription(conn, @all_stream, @subscription_name, + schema: schema + ) + + assert {:ok, %Storage.Subscription{subscription_id: ^untouched, last_seen: nil}} = + Storage.Subscription.subscription(conn, @all_stream, "another_subscription", + schema: schema + ) + end + + test "ack last seen event when the name now belongs to a different subscription", context do + %{conn: conn, schema: schema} = context + + {:ok, %Storage.Subscription{subscription_id: deleted}} = subscribe_to_stream(context) + + :ok = delete_subscription(context) + + {:ok, %Storage.Subscription{subscription_id: recreated}} = subscribe_to_stream(context) + + assert recreated != deleted + + assert {:error, :subscription_not_found} = + Storage.ack_last_seen_event(conn, deleted, 1, schema: schema) + + # The stream and name are the same as the deleted subscription's, so anything keyed by those + # would have credited this subscription with a position it never acknowledged. + assert {:ok, %Storage.Subscription{subscription_id: ^recreated, last_seen: nil}} = + Storage.Subscription.subscription(conn, @all_stream, @subscription_name, + schema: schema + ) + end + + test "ack last seen event reports a storage failure as itself, not as a missing subscription", + context do + %{conn: conn} = context + + {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) + + assert {:error, %Postgrex.Error{}} = + Storage.ack_last_seen_event(conn, subscription_id, 1, schema: "no_such_schema") + end + + test "ack last seen event for a subscription that never existed", context do + %{conn: conn, schema: schema} = context + + {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) + + assert {:error, :subscription_not_found} = + Storage.ack_last_seen_event(conn, subscription_id + 1_000, 1, schema: schema) + + assert {:ok, %Storage.Subscription{last_seen: nil}} = read_subscription(context) + end + + test "ack last seen event twice at the same position", context do + %{conn: conn, schema: schema} = context + + {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) + + assert :ok = Storage.ack_last_seen_event(conn, subscription_id, 3, schema: schema) + + # Postgres counts the rows it matched rather than the rows it changed, so an acknowledgement + # that moves nothing is still an acknowledgement its row accepted. + assert :ok = Storage.ack_last_seen_event(conn, subscription_id, 3, schema: schema) + + assert {:ok, %Storage.Subscription{last_seen: 3}} = read_subscription(context) + end + + test "ack last seen event at the position a subscription starts from", context do + %{conn: conn, schema: schema} = context + + {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) + + assert :ok = Storage.ack_last_seen_event(conn, subscription_id, 0, schema: schema) + + # Zero is a position that was acknowledged, where null is a subscription that never has. + assert {:ok, %Storage.Subscription{last_seen: 0}} = read_subscription(context) + end + + test "ack last seen event leaves everything but the position alone", context do + %{conn: conn, schema: schema} = context + + {:ok, %Storage.Subscription{} = before} = subscribe_to_stream(context) + + :ok = Storage.ack_last_seen_event(conn, before.subscription_id, 3, schema: schema) + + assert {:ok, %Storage.Subscription{} = after_ack} = read_subscription(context) + + assert after_ack == %Storage.Subscription{before | last_seen: 3} + end + + test "ack last seen event does not guard against a position going backwards", context do + %{conn: conn, schema: schema} = context + + {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context) + + :ok = Storage.ack_last_seen_event(conn, subscription_id, 5, schema: schema) + :ok = Storage.ack_last_seen_event(conn, subscription_id, 2, schema: schema) + + # Storage takes the position it is given. Nothing reaches here out of order, because the one + # process that owns a row acknowledges in order, and refusing the write would be indistinguish + # able from the row being gone. + assert {:ok, %Storage.Subscription{last_seen: 2}} = read_subscription(context) + end + + defp read_subscription(context) do + %{conn: conn, schema: schema} = context + + Storage.Subscription.subscription(conn, @all_stream, @subscription_name, schema: schema) + end + def ack_last_seen_event(context, last_seen) do %{conn: conn, schema: schema} = context - Storage.ack_last_seen_event(conn, @all_stream, @subscription_name, last_seen, schema: schema) + {:ok, %Storage.Subscription{subscription_id: subscription_id}} = + Storage.Subscription.subscription(conn, @all_stream, @subscription_name, schema: schema) + + Storage.ack_last_seen_event(conn, subscription_id, last_seen, schema: schema) end defp subscribe_to_stream(context) do diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs index 16337277..8a561bef 100644 --- a/test/subscriptions/subscribe_to_stream_test.exs +++ b/test/subscriptions/subscribe_to_stream_test.exs @@ -11,7 +11,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do Wait } - alias EventStore.Subscriptions.Subscription + alias EventStore.Subscriptions.{SlowLeaver, Subscription} alias EventStore.Support.CollectingSubscriber alias TestEventStore, as: EventStore @@ -880,8 +880,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do } do stream_uuid = UUID.uuid4() - name = - {Module.concat(@event_store, Subscriptions.Registry), {stream_uuid, subscription_name}} + name = registry_name(stream_uuid, subscription_name) subscriber = start_subscriber(stream_uuid, subscription_name) @@ -1171,6 +1170,141 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do assert {:ok, []} = Storage.subscriptions(@conn, schema: schema) end + test "should keep the subscription and its row when the delete fails in storage", + %{subscription_name: subscription_name, schema: schema} = context do + stream_uuid = UUID.uuid4() + + {:ok, %Storage.Subscription{}} = + Storage.subscribe_to_stream(@conn, stream_uuid, subscription_name, nil, schema: schema) + + subscription = + start_unconnected_subscription(context, stream_uuid, schema: "no_such_schema") + + assert {:error, %Postgrex.Error{}} = Subscription.delete(subscription) + + # Failing to delete the row is not having deleted it, so the subscription that owns it has + # to still be there to go on writing to it. + assert Process.alive?(subscription) + assert {:ok, [_subscription]} = Storage.subscriptions(@conn, schema: schema) + end + + test "should delete the row a subscription gave up on without deleting", + %{subscription_name: subscription_name, schema: schema} = context do + stream_uuid = UUID.uuid4() + + {:ok, %Storage.Subscription{}} = + Storage.subscribe_to_stream(@conn, stream_uuid, subscription_name, nil, schema: schema) + + subscription = start_unconnected_subscription(context, stream_uuid) + + # Suspended so the delete is still queued when the subscription dies, which is what leaves + # the row behind with nothing holding the name any more. + :ok = :sys.suspend(subscription) + + deleting = + Task.async(fn -> EventStore.delete_subscription(stream_uuid, subscription_name) end) + + assert_enqueued(subscription, 1) + + Process.exit(subscription, :kill) + + assert :ok = Task.await(deleting) + assert {:ok, []} = Storage.subscriptions(@conn, schema: schema) + end + + test "should wait without a deadline for a checkpoint when asked to", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + + subscriber = start_subscriber(stream_uuid, subscription_name) + + assert_receive {:subscriber, ^subscriber, subscription} + assert_receive {:subscribed, ^subscription} + + :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1)) + + assert_receive {:events, [%RecordedEvent{event_number: 1}]} + :ok = Subscription.ack(subscription, 1, subscriber) + + block_checkpoint() + + Process.exit(subscriber, :kill) + + assert_receive {:checkpointing, ^subscription} + + deleting = + Task.async(fn -> + EventStore.delete_subscription(stream_uuid, subscription_name, timeout: :infinity) + end) + + # Longer than the five seconds a `GenServer.call/2` waits by default, so a deadline that + # was not carried through as `:infinity` would have given up by now. + refute Task.yield(deleting, 6_000) + + send(subscription, :release_checkpoint) + + assert :ok = Task.await(deleting) + assert {:ok, []} = Storage.subscriptions(@conn, schema: schema) + end + + test "should not return until the subscription that answered it has released its name", %{ + subscription_name: subscription_name + } do + stream_uuid = UUID.uuid4() + + slow = start_slow_leaver(stream_uuid, subscription_name) + + assert :ok = EventStore.delete_subscription(stream_uuid, subscription_name) + + # Being answered says `terminate/2` has run, not that the name has been given up, which only + # happens once the process finishes exiting. + refute Process.alive?(slow) + assert :undefined == Registry.whereis_name(registry_name(stream_uuid, subscription_name)) + end + + test "should give up on a subscription that answered but will not go away", %{ + subscription_name: subscription_name + } do + stream_uuid = UUID.uuid4() + + slow = start_slow_leaver(stream_uuid, subscription_name) + + assert catch_exit( + EventStore.delete_subscription(stream_uuid, subscription_name, timeout: 100) + ) == {:timeout, {Subscriptions.Supervisor, :delete_subscription, [slow]}} + end + + test "should count the wait for an answer against the deadline it was given", %{ + subscription_name: subscription_name + } do + stream_uuid = UUID.uuid4() + + start_slow_leaver(stream_uuid, subscription_name, answer_after: 300, leaving_for: 400) + + # Being answered with 200ms of an 500ms deadline left is not 500ms to wait for the name, and + # reporting a delete that has not released the name is the failure being avoided. + assert {:timeout, _where} = + catch_exit( + EventStore.delete_subscription(stream_uuid, subscription_name, timeout: 500) + ) + end + + test "should wait out a subscription that answered when given no deadline", %{ + subscription_name: subscription_name + } do + stream_uuid = UUID.uuid4() + + slow = start_slow_leaver(stream_uuid, subscription_name) + + assert :ok = + EventStore.delete_subscription(stream_uuid, subscription_name, timeout: :infinity) + + refute Process.alive?(slow) + assert :undefined == Registry.whereis_name(registry_name(stream_uuid, subscription_name)) + end + test "should not disconnect a subscriber that connects while the delete is in flight", %{subscription_name: subscription_name} = context do stream_uuid = UUID.uuid4() @@ -1216,6 +1350,274 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do {:exit, _reason} -> refute Process.alive?(subscription) end end + + test "should delete its own row, rather than leave it to whoever asked", + %{subscription_name: subscription_name, schema: schema} = context do + stream_uuid = UUID.uuid4() + + {:ok, %Storage.Subscription{}} = + Storage.subscribe_to_stream(@conn, stream_uuid, subscription_name, nil, schema: schema) + + subscription = start_unconnected_subscription(context, stream_uuid) + + assert {:ok, [_subscription]} = Storage.subscriptions(@conn, schema: schema) + + ref = Process.monitor(subscription) + + # Asking the subscription directly, with nothing else able to delete on its behalf, is what + # shows the row going away while the subscription still holds its registered name. + assert :ok = Subscription.delete(subscription) + + assert {:ok, []} = Storage.subscriptions(@conn, schema: schema) + assert_receive {:DOWN, ^ref, :process, ^subscription, :shutdown} + end + + test "should stop when its checkpoint reaches no row of its own", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + + {:ok, subscription} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), + checkpoint_threshold: 100 + ) + + assert_receive {:subscribed, ^subscription} + + :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1)) + + assert_receive {:events, [%RecordedEvent{event_number: 1}] = events} + :ok = Subscription.ack(subscription, events) + + ref = Process.monitor(subscription) + + # Deleting the row behind the subscription's back is what a delete racing its name does. + :ok = Storage.delete_subscription(@conn, stream_uuid, subscription_name, schema: schema) + + send(subscription, :checkpoint) + + assert_receive {:DOWN, ^ref, :process, ^subscription, :subscription_not_found} + end + + test "should not write its checkpoint onto the subscription that took over its name", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + + {:ok, subscription} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), + checkpoint_threshold: 100 + ) + + assert_receive {:subscribed, ^subscription} + + :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1)) + + assert_receive {:events, [%RecordedEvent{event_number: 1}] = events} + :ok = Subscription.ack(subscription, events) + + :ok = Storage.delete_subscription(@conn, stream_uuid, subscription_name, schema: schema) + + {:ok, %Storage.Subscription{subscription_id: taken_over, last_seen: last_seen}} = + Storage.subscribe_to_stream(@conn, stream_uuid, subscription_name, nil, schema: schema) + + ref = Process.monitor(subscription) + + send(subscription, :checkpoint) + + assert_receive {:DOWN, ^ref, :process, ^subscription, :subscription_not_found} + + # A checkpoint names the row it belongs to, so the subscription holding the name now keeps + # the position it was created with instead of inheriting one it never acknowledged. + assert {:ok, [%Storage.Subscription{subscription_id: ^taken_over, last_seen: ^last_seen}]} = + Storage.subscriptions(@conn, schema: schema) + end + + test "should leave the subscription that took over its name every event to deliver", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + + {:ok, subscription} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), + checkpoint_threshold: 100 + ) + + assert_receive {:subscribed, ^subscription} + + :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1)) + + assert_receive {:events, [%RecordedEvent{event_number: 1}] = events} + :ok = Subscription.ack(subscription, events) + + :ok = Storage.delete_subscription(@conn, stream_uuid, subscription_name, schema: schema) + + {:ok, %Storage.Subscription{}} = + Storage.subscribe_to_stream(@conn, stream_uuid, subscription_name, nil, schema: schema) + + ref = Process.monitor(subscription) + + send(subscription, :checkpoint) + + assert_receive {:DOWN, ^ref, :process, ^subscription, :subscription_not_found} + + {:ok, successor} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + + assert_receive {:subscribed, ^successor} + + # Every other test here reads the position out of the row. This one reads it the way a + # subscriber does, because a position nobody reads back is not the thing that was at stake. + assert_receive {:events, [%RecordedEvent{event_number: 1}]} + end + + test "should leave no down message with the caller when it deletes a live subscription", + %{ + subscription_name: subscription_name, + schema: schema + } = context do + stream_uuid = UUID.uuid4() + + {:ok, %Storage.Subscription{}} = + Storage.subscribe_to_stream(@conn, stream_uuid, subscription_name, nil, schema: schema) + + _subscription = start_unconnected_subscription(context, stream_uuid) + + assert :ok = EventStore.delete_subscription(stream_uuid, subscription_name) + + # The delete watches the subscription from whichever process asked for it, so a monitor it + # does not clean up becomes an unexpected message in a caller that is a `GenServer`. + refute_received {:DOWN, _ref, :process, _pid, _reason} + end + + test "should leave no down message with the caller when there is nothing running", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + + {:ok, %Storage.Subscription{}} = + Storage.subscribe_to_stream(@conn, stream_uuid, subscription_name, nil, schema: schema) + + assert :ok = EventStore.delete_subscription(stream_uuid, subscription_name) + + refute_received {:DOWN, _ref, :process, _pid, _reason} + end + + test "should leave no down message with the caller when the delete fails in storage", + %{ + subscription_name: subscription_name, + schema: schema + } = context do + stream_uuid = UUID.uuid4() + + {:ok, %Storage.Subscription{}} = + Storage.subscribe_to_stream(@conn, stream_uuid, subscription_name, nil, schema: schema) + + _subscription = + start_unconnected_subscription(context, stream_uuid, schema: "no_such_schema") + + assert {:error, %Postgrex.Error{}} = + EventStore.delete_subscription(stream_uuid, subscription_name) + + refute_received {:DOWN, _ref, :process, _pid, _reason} + end + + test "should keep the subscription and its row when it is given no time at all", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + + {:ok, %Storage.Subscription{}} = + Storage.subscribe_to_stream(@conn, stream_uuid, subscription_name, nil, schema: schema) + + slow = start_slow_leaver(stream_uuid, subscription_name) + + assert {:timeout, _where} = + catch_exit( + EventStore.delete_subscription(stream_uuid, subscription_name, timeout: 0) + ) + + # Giving up is not being gone: the subscription may still be writing, and deleting its row + # from here would delete it out from under that write. + assert Process.alive?(slow) + assert {:ok, [%Storage.Subscription{}]} = Storage.subscriptions(@conn, schema: schema) + end + + test "should leave no down message with the caller when it gives up waiting for an answer", %{ + subscription_name: subscription_name + } do + stream_uuid = UUID.uuid4() + + slow = start_slow_leaver(stream_uuid, subscription_name) + + assert {:timeout, _where} = + catch_exit( + EventStore.delete_subscription(stream_uuid, subscription_name, timeout: 0) + ) + + ref = Process.monitor(slow) + + send(slow, :leave) + + # One down message for the monitor this test set up. A second would be the one the delete + # gave up on without cleaning up, arriving in a caller that never asked to watch anything. + assert_receive {:DOWN, _ref, :process, ^slow, :shutdown} + refute_received {:DOWN, _ref, :process, ^slow, _reason} + + Process.demonitor(ref, [:flush]) + end + + test "should leave the other subscriptions running when it stops", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + sibling_stream_uuid = UUID.uuid4() + sibling_name = UUID.uuid4() + + {:ok, subscription} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), + checkpoint_threshold: 100 + ) + + assert_receive {:subscribed, ^subscription} + + {:ok, sibling} = + EventStore.subscribe_to_stream(sibling_stream_uuid, sibling_name, self()) + + assert_receive {:subscribed, ^sibling} + + :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1)) + + assert_receive {:events, [%RecordedEvent{event_number: 1}] = events} + :ok = Subscription.ack(subscription, events) + + :ok = Storage.delete_subscription(@conn, stream_uuid, subscription_name, schema: schema) + + ref = Process.monitor(subscription) + + send(subscription, :checkpoint) + + assert_receive {:DOWN, ^ref, :process, ^subscription, :subscription_not_found} + + # Stopping over a row of its own is the subscription's own business, and a sibling that + # never shared that row has no reason to be restarted over it. + assert Process.alive?(sibling) + + :ok = EventStore.append_to_stream(sibling_stream_uuid, 0, EventFactory.create_events(1)) + + assert_receive {:events, [%RecordedEvent{event_number: 1}] = sibling_events} + :ok = Subscription.ack(sibling, sibling_events) + + Wait.until(fn -> + assert {:ok, [%Storage.Subscription{stream_uuid: ^sibling_stream_uuid, last_seen: 1}]} = + Storage.subscriptions(@conn, schema: schema) + end) + end end # OTP 28 reports the `gen_server` hibernation loop where earlier releases @@ -1301,9 +1703,27 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do end) end + # Holds the registered name, answers a delete, and only then takes its time going away, which is + # the gap between being answered and being gone. + defp start_slow_leaver(stream_uuid, subscription_name, opts \\ []) do + name = registry_name(stream_uuid, subscription_name) + + opts = Keyword.merge([name: {:via, Registry, name}, leaving_for: 500], opts) + + pid = start_supervised!({SlowLeaver, opts}) + + assert pid == Registry.whereis_name(name) + + pid + end + + defp registry_name(stream_uuid, subscription_name) do + {Module.concat([@event_store, Subscriptions.Registry]), {stream_uuid, subscription_name}} + end + # A started subscription with no subscriber connected to it yet, which is the only way a live # subscription has none: losing its last subscriber shuts it down. - defp start_unconnected_subscription(context, stream_uuid) do + defp start_unconnected_subscription(context, stream_uuid, overrides \\ []) do %{ conn: conn, schema: schema, @@ -1317,7 +1737,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do Subscriptions.Supervisor.start_subscription( event_store: @event_store, conn: conn, - schema: schema, + schema: Keyword.get(overrides, :schema, schema), serializer: serializer, correlation_id_type: correlation_id_type, causation_id_type: causation_id_type, diff --git a/test/subscriptions/subscription_fsm_test.exs b/test/subscriptions/subscription_fsm_test.exs new file mode 100644 index 00000000..4088c1a0 --- /dev/null +++ b/test/subscriptions/subscription_fsm_test.exs @@ -0,0 +1,357 @@ +defmodule EventStore.Subscriptions.SubscriptionFsmTest do + use EventStore.StorageCase + + alias EventStore.{Storage, UUID} + alias EventStore.Subscriptions.{SubscriptionFsm, SubscriptionState} + + # The states that own a stored position, and those that do not. + @persisting_states [:request_catch_up, :catching_up, :subscribed, :max_capacity] + @idle_states [:initial, :disconnected, :unsubscribed] + + setup context do + stream_uuid = UUID.uuid4() + subscription_name = UUID.uuid4() + + {:ok, %Storage.Subscription{subscription_id: subscription_id}} = + Storage.subscribe_to_stream(context.conn, stream_uuid, subscription_name, + schema: context.schema + ) + + [ + stream_uuid: stream_uuid, + subscription_name: subscription_name, + subscription_id: subscription_id + ] + end + + describe "delete/1" do + test "deletes the row the subscription created", context do + fsm = fsm(context) + + assert {:ok, %SubscriptionFsm{}} = SubscriptionFsm.delete(fsm) + + assert {:error, :subscription_not_found} = read_subscription(context) + end + + test "leaves nothing pending for a later checkpoint to write", context do + fsm = fsm(context, checkpoints_pending: 1, last_ack: 3) + + assert {:ok, %SubscriptionFsm{data: %SubscriptionState{checkpoints_pending: 0}}} = + SubscriptionFsm.delete(fsm) + end + + test "keeps what was pending when the row it would be written to is still there", context do + fsm = fsm(context, checkpoints_pending: 1, last_ack: 3, schema: "no_such_schema") + + assert {{:error, %Postgrex.Error{}}, + %SubscriptionFsm{data: %SubscriptionState{checkpoints_pending: 1}}} = + SubscriptionFsm.delete(fsm) + + assert {:ok, %Storage.Subscription{}} = read_subscription(context) + end + + test "deletes from every state the subscription can be in", context do + for state <- @persisting_states ++ @idle_states do + assert {:ok, %SubscriptionFsm{state: ^state}} = + SubscriptionFsm.delete(fsm(context, state: state)) + + assert {:error, :subscription_not_found} = read_subscription(context) + + recreate_row(context) + end + end + + test "reports success when the row is already gone", context do + :ok = + Storage.delete_subscription(context.conn, context.stream_uuid, context.subscription_name, + schema: context.schema + ) + + # A delete asks for the row to be absent, and it is. + assert {:ok, %SubscriptionFsm{}} = SubscriptionFsm.delete(fsm(context)) + end + + test "leaves no pending checkpoint behind whatever was pending", context do + for pending <- [0, 1, 5] do + assert {:ok, %SubscriptionFsm{data: %SubscriptionState{checkpoints_pending: 0}}} = + SubscriptionFsm.delete(fsm(context, checkpoints_pending: pending)) + + recreate_row(context) + end + end + + test "stops a subscription that goes on acknowledging after its row is deleted", context do + {:ok, %SubscriptionFsm{} = deleted} = SubscriptionFsm.delete(fsm(context)) + + # The row is gone but the subscription is still holding subscribers, so the next checkpoint + # is the moment it finds out. + SubscriptionFsm.checkpoint(%SubscriptionFsm{ + deleted + | data: %SubscriptionState{deleted.data | checkpoints_pending: 1, last_ack: 4} + }) + + assert_received {:checkpoint_failed, :subscription_not_found} + end + + test "removes the row left under the name of a subscription that keeps no position", + context do + # A transient subscription has no row of its own, so what it deletes is whatever the name + # refers to. + assert {:ok, %SubscriptionFsm{}} = SubscriptionFsm.delete(fsm(context, transient: true)) + + assert {:error, :subscription_not_found} = read_subscription(context) + end + + test "deletes only the subscription it names", context do + %{conn: conn, schema: schema} = context + + sibling_name = UUID.uuid4() + + {:ok, %Storage.Subscription{}} = + Storage.subscribe_to_stream(conn, context.stream_uuid, sibling_name, schema: schema) + + assert {:ok, %SubscriptionFsm{}} = SubscriptionFsm.delete(fsm(context)) + + assert {:ok, %Storage.Subscription{}} = + Storage.Subscription.subscription(conn, context.stream_uuid, sibling_name, + schema: schema + ) + end + end + + describe "checkpoint/1" do + test "writes the position of the subscription it names", context do + fsm = fsm(context, checkpoints_pending: 1, last_ack: 3) + + assert %SubscriptionFsm{data: %SubscriptionState{checkpoints_pending: 0}} = + SubscriptionFsm.checkpoint(fsm) + + assert {:ok, %Storage.Subscription{last_seen: 3}} = read_subscription(context) + + refute_received {:checkpoint_failed, _reason} + end + + test "writes nothing when no acknowledgement is pending", context do + fsm = fsm(context, checkpoints_pending: 0, last_ack: 3) + + assert %SubscriptionFsm{} = SubscriptionFsm.checkpoint(fsm) + + assert {:ok, %Storage.Subscription{last_seen: nil}} = read_subscription(context) + + refute_received {:checkpoint_failed, _reason} + end + + test "asks the subscription to stop once the row it names is gone", context do + fsm = fsm(context, checkpoints_pending: 1, last_ack: 3) + + :ok = + Storage.delete_subscription(context.conn, context.stream_uuid, context.subscription_name, + schema: context.schema + ) + + assert %SubscriptionFsm{data: %SubscriptionState{checkpoints_pending: 0}} = + SubscriptionFsm.checkpoint(fsm) + + assert_received {:checkpoint_failed, :subscription_not_found} + end + + test "does not write the position onto a subscription that reused the name", context do + %{conn: conn, schema: schema, stream_uuid: stream_uuid} = context + %{subscription_name: subscription_name, subscription_id: subscription_id} = context + + fsm = fsm(context, checkpoints_pending: 1, last_ack: 3) + + :ok = Storage.delete_subscription(conn, stream_uuid, subscription_name, schema: schema) + + {:ok, %Storage.Subscription{subscription_id: reused}} = + Storage.subscribe_to_stream(conn, stream_uuid, subscription_name, schema: schema) + + assert reused != subscription_id + + SubscriptionFsm.checkpoint(fsm) + + assert {:ok, %Storage.Subscription{subscription_id: ^reused, last_seen: nil}} = + read_subscription(context) + end + + test "keeps the subscription running when the write fails for a reason of its own", context do + fsm = fsm(context, checkpoints_pending: 1, last_ack: 3, schema: "no_such_schema") + + assert %SubscriptionFsm{} = SubscriptionFsm.checkpoint(fsm) + + # A write that failed is not a row that is gone, and stopping over one would turn every + # blip in storage into a subscription that has to be restarted. + refute_received {:checkpoint_failed, _reason} + end + + test "writes from every state that owns a position", context do + for state <- @persisting_states do + fsm = fsm(context, state: state, checkpoints_pending: 1, last_ack: 3) + + assert %SubscriptionFsm{state: ^state} = SubscriptionFsm.checkpoint(fsm) + + assert {:ok, %Storage.Subscription{last_seen: 3}} = read_subscription(context) + + reset_position(context) + end + end + + test "writes from no state that does not", context do + for state <- @idle_states do + fsm = fsm(context, state: state, checkpoints_pending: 1, last_ack: 3) + + assert %SubscriptionFsm{state: ^state} = SubscriptionFsm.checkpoint(fsm) + + assert {:ok, %Storage.Subscription{last_seen: nil}} = read_subscription(context) + end + + refute_received {:checkpoint_failed, _reason} + end + + test "leaves the state it was given alone", context do + for state <- @persisting_states ++ @idle_states do + fsm = fsm(context, state: state, checkpoints_pending: 1, last_ack: 3) + + assert %SubscriptionFsm{state: ^state} = SubscriptionFsm.checkpoint(fsm) + end + end + + test "writes nothing for a subscription that keeps no position", context do + fsm = fsm(context, transient: true, checkpoints_pending: 1, last_ack: 3) + + assert %SubscriptionFsm{data: %SubscriptionState{checkpoints_pending: 0}} = + SubscriptionFsm.checkpoint(fsm) + + # A transient subscription never created a row, so there is none to write to and none to + # find missing. + assert {:ok, %Storage.Subscription{last_seen: nil}} = read_subscription(context) + + refute_received {:checkpoint_failed, _reason} + end + + test "moves the position forward across successive checkpoints", context do + for position <- [1, 2, 7] do + fsm = fsm(context, checkpoints_pending: 1, last_ack: position) + + SubscriptionFsm.checkpoint(fsm) + + assert {:ok, %Storage.Subscription{last_seen: ^position}} = read_subscription(context) + end + + refute_received {:checkpoint_failed, _reason} + end + + test "writes the same position twice without reporting the row as missing", context do + fsm = fsm(context, checkpoints_pending: 1, last_ack: 3) + + SubscriptionFsm.checkpoint(fsm) + SubscriptionFsm.checkpoint(fsm) + + assert {:ok, %Storage.Subscription{last_seen: 3}} = read_subscription(context) + + # An acknowledgement that changes no row is still an acknowledgement the row accepted, and + # reading it as a missing subscription would stop a healthy subscription. + refute_received {:checkpoint_failed, _reason} + end + + test "clears what was pending once the row it names is gone", context do + fsm = fsm(context, checkpoints_pending: 1, last_ack: 3) + + :ok = + Storage.delete_subscription(context.conn, context.stream_uuid, context.subscription_name, + schema: context.schema + ) + + # Keeping it pending would have the stopping subscription try the same write again on its + # way out. + assert %SubscriptionFsm{data: %SubscriptionState{checkpoints_pending: 0}} = + SubscriptionFsm.checkpoint(fsm) + + assert_received {:checkpoint_failed, :subscription_not_found} + end + + test "writes to its own row when another stream reuses the name", context do + %{conn: conn, schema: schema, subscription_name: subscription_name} = context + + other_stream_uuid = UUID.uuid4() + + {:ok, %Storage.Subscription{subscription_id: other_id}} = + Storage.subscribe_to_stream(conn, other_stream_uuid, subscription_name, schema: schema) + + assert other_id != context.subscription_id + + SubscriptionFsm.checkpoint(fsm(context, checkpoints_pending: 1, last_ack: 3)) + + assert {:ok, %Storage.Subscription{last_seen: 3}} = read_subscription(context) + + # The name is only unique per stream, so a name on its own never identifies a row. + assert {:ok, %Storage.Subscription{last_seen: nil}} = + Storage.Subscription.subscription(conn, other_stream_uuid, subscription_name, + schema: schema + ) + end + + test "writes nothing before the subscription has a row of its own", context do + fsm = %SubscriptionFsm{ + fsm(context, checkpoints_pending: 1, last_ack: 3) + | state: :initial + } + + assert %SubscriptionFsm{} = SubscriptionFsm.checkpoint(fsm) + + assert {:ok, %Storage.Subscription{last_seen: nil}} = read_subscription(context) + + refute_received {:checkpoint_failed, _reason} + end + end + + defp fsm(context, overrides \\ []) do + %{conn: conn, schema: schema} = context + %{stream_uuid: stream_uuid, subscription_name: subscription_name} = context + %{subscription_id: subscription_id} = context + + data = %SubscriptionState{ + conn: conn, + schema: Keyword.get(overrides, :schema, schema), + stream_uuid: stream_uuid, + subscription_name: subscription_name, + subscription_id: subscription_id, + last_ack: Keyword.get(overrides, :last_ack, 0), + checkpoints_pending: Keyword.get(overrides, :checkpoints_pending, 0), + query_timeout: 15_000, + transient: Keyword.get(overrides, :transient, false) + } + + %SubscriptionFsm{state: Keyword.get(overrides, :state, :subscribed), data: data} + end + + defp reset_position(context) do + %{conn: conn, schema: schema, subscription_id: subscription_id} = context + + {:ok, _result} = + Postgrex.query( + conn, + ~s|UPDATE "#{schema}".subscriptions SET last_seen = NULL WHERE subscription_id = $1|, + [subscription_id] + ) + + :ok + end + + defp recreate_row(context) do + %{conn: conn, schema: schema} = context + %{stream_uuid: stream_uuid, subscription_name: subscription_name} = context + + {:ok, %Storage.Subscription{}} = + Storage.subscribe_to_stream(conn, stream_uuid, subscription_name, schema: schema) + + :ok + end + + defp read_subscription(context) do + %{conn: conn, schema: schema} = context + %{stream_uuid: stream_uuid, subscription_name: subscription_name} = context + + Storage.Subscription.subscription(conn, stream_uuid, subscription_name, schema: schema) + end +end diff --git a/test/subscriptions/subscription_locking_test.exs b/test/subscriptions/subscription_locking_test.exs index 758c097f..19300737 100644 --- a/test/subscriptions/subscription_locking_test.exs +++ b/test/subscriptions/subscription_locking_test.exs @@ -77,10 +77,10 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do :ok = disconnect(subscription) - :ok = - Storage.Subscription.ack_last_seen_event(@conn, "$all", subscription_name, 2, - schema: schema - ) + {:ok, %Storage.Subscription{subscription_id: subscription_id}} = + Storage.Subscription.subscription(@conn, "$all", subscription_name, schema: schema) + + :ok = Storage.Subscription.ack_last_seen_event(@conn, subscription_id, 2, schema: schema) :ok = reconnect(subscription) diff --git a/test/subscriptions/support/slow_leaver.ex b/test/subscriptions/support/slow_leaver.ex new file mode 100644 index 00000000..568d2cb1 --- /dev/null +++ b/test/subscriptions/support/slow_leaver.ex @@ -0,0 +1,43 @@ +defmodule EventStore.Subscriptions.SlowLeaver do + @moduledoc false + + # Answers a delete and only then takes its time going away, which is the gap a caller that reads + # being answered as being gone would fall into: the row is deleted and the registered name is + # still held. + + use GenServer, restart: :temporary + + def start_link(opts) do + {start_opts, opts} = Keyword.split(opts, [:name]) + + GenServer.start_link(__MODULE__, opts, start_opts) + end + + @impl GenServer + def init(opts) do + {:ok, + %{ + answer_after: Keyword.get(opts, :answer_after, 0), + leaving_for: Keyword.fetch!(opts, :leaving_for) + }} + end + + @impl GenServer + def handle_call(:delete, from, state) do + Process.send_after(self(), {:answer, from}, state.answer_after) + + {:noreply, state} + end + + @impl GenServer + def handle_info({:answer, from}, state) do + GenServer.reply(from, :ok) + + Process.send_after(self(), :leave, state.leaving_for) + + {:noreply, state} + end + + @impl GenServer + def handle_info(:leave, state), do: {:stop, :shutdown, state} +end