diff --git a/guides/Usage.md b/guides/Usage.md index b5de7a93..9ae2d047 100644 --- a/guides/Usage.md +++ b/guides/Usage.md @@ -292,6 +292,7 @@ The first pass covers these operations: - `:paginate_streams` - `:subscribe_to_stream` - `:delete_subscription` +- `:subscription_checkpoint` - `:read_snapshot` - `:record_snapshot` - `:delete_snapshot` @@ -303,6 +304,15 @@ with `stream_uuid: "$all"` in the metadata, and `subscribe_to_all_streams/3` emits `[:eventstore, :subscribe_to_stream, ...]` with the same stream identifier. +`:subscription_checkpoint` is emitted by the subscription process itself, once +per checkpoint written to storage, rather than by a caller. Alongside the same +`:event_store` and `:name` every other operation carries, its metadata includes +`:stream_uuid`, `:subscription_name`, and the `:last_seen` event number being +persisted. A subscription checkpoints when its pending acknowledgements reach +`checkpoint_threshold`, when `checkpoint_after` elapses, when its last +subscriber leaves, and again while terminating with acknowledgements still +pending. + Stop metadata includes a normalized `:result` for all instrumented operations. Operations that return `:ok` emit `result: :ok`. Operations that return `{:ok, value}` also emit `result: :ok` so telemetry does not copy returned diff --git a/lib/event_store.ex b/lib/event_store.ex index e4321da0..7c10c20e 100644 --- a/lib/event_store.ex +++ b/lib/event_store.ex @@ -535,6 +535,7 @@ defmodule EventStore do |> Keyword.merge( conn: conn, event_store: name, + telemetry_metadata: Telemetry.metadata(__MODULE__, opts), query_timeout: query_timeout, schema: schema, serializer: serializer, @@ -578,10 +579,10 @@ defmodule EventStore do %{stream_uuid: stream_uuid, subscription_name: subscription_name}, fn -> name = name(opts) + {conn, opts} = parse_opts(opts) - with :ok <- Subscriptions.stop_subscription(name, stream_uuid, subscription_name) do - {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 end @@ -1471,14 +1472,16 @@ defmodule EventStore do - `subscription_name` is used to identify the existing subscription to remove. - Returns `:ok` on success. + Returns `:ok` on success, or `{:error, :subscribers_connected}` when subscribers are still + connected to the subscription. Unsubscribe them first, with `c:unsubscribe_from_stream/3`, so + that deleting a subscription never disconnects a subscriber that did not ask for it. """ @callback delete_subscription( stream_uuid :: String.t(), subscription_name :: String.t(), opts :: options ) :: - :ok | {:error, term} + :ok | {:error, :subscribers_connected} | {:error, term} @doc """ Delete an existing persistent subscription to all streams. @@ -1486,10 +1489,11 @@ defmodule EventStore do - `subscription_name` is used to identify the existing subscription to remove. - Returns `:ok` on success. + Returns `:ok` on success, or `{:error, :subscribers_connected}` when subscribers are still + connected to the subscription. """ @callback delete_all_streams_subscription(subscription_name :: String.t(), opts :: options) :: - :ok | {:error, term} + :ok | {:error, :subscribers_connected} | {:error, term} @doc """ Read a snapshot, if available, for a given source. diff --git a/lib/event_store/subscriptions.ex b/lib/event_store/subscriptions.ex index 299d3ed3..844c17e7 100644 --- a/lib/event_store/subscriptions.ex +++ b/lib/event_store/subscriptions.ex @@ -22,7 +22,7 @@ defmodule EventStore.Subscriptions do end defdelegate unsubscribe_from_stream(event_store, stream_uuid, name), to: SubscriptionsSupervisor - defdelegate stop_subscription(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 @doc """ diff --git a/lib/event_store/subscriptions/subscription.ex b/lib/event_store/subscriptions/subscription.ex index 321b258b..bb2551a4 100644 --- a/lib/event_store/subscriptions/subscription.ex +++ b/lib/event_store/subscriptions/subscription.ex @@ -14,6 +14,10 @@ defmodule EventStore.Subscriptions.Subscription do alias EventStore.RecordedEvent alias EventStore.Subscriptions.{SubscriptionFsm, Subscription, SubscriptionState} + # Mirrors the storage `:timeout` a subscription gives its own queries, for the internal callers + # that have no configured event store to read one from. + @default_timeout 15_000 + defstruct [ :stream_uuid, :subscription_name, @@ -84,6 +88,17 @@ defmodule EventStore.Subscriptions.Subscription do GenServer.call(subscription, {:unsubscribe, self()}) end + @doc """ + Stop 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)) + end + @doc false def last_seen(subscription) do GenServer.call(subscription, :last_seen) @@ -244,6 +259,18 @@ defmodule EventStore.Subscriptions.Subscription do end end + @impl GenServer + def handle_call(:stop, _from, %Subscription{} = state) do + %Subscription{ + subscription: %SubscriptionFsm{data: %SubscriptionState{subscribers: subscribers}} + } = state + + case map_size(subscribers) do + 0 -> {:stop, :shutdown, :ok, state} + _ -> {:reply, {:error, :subscribers_connected}, state} + end + end + @impl GenServer def handle_call(:last_seen, _from, %Subscription{subscription: subscription} = state) do %SubscriptionFsm{data: %SubscriptionState{last_ack: last_seen}} = subscription diff --git a/lib/event_store/subscriptions/subscription_fsm.ex b/lib/event_store/subscriptions/subscription_fsm.ex index 3604636d..441b06d2 100644 --- a/lib/event_store/subscriptions/subscription_fsm.ex +++ b/lib/event_store/subscriptions/subscription_fsm.ex @@ -1,7 +1,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do @moduledoc false - alias EventStore.{AdvisoryLocks, PubSub, RecordedEvent, Storage} + alias EventStore.{AdvisoryLocks, PubSub, RecordedEvent, Storage, Telemetry} alias EventStore.Streams.Stream alias EventStore.Subscriptions.{SubscriptionState, Subscriber} @@ -28,6 +28,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do checkpoint_after: opts[:checkpoint_after] || 0, checkpoint_threshold: opts[:checkpoint_threshold] || 1, query_timeout: opts[:query_timeout] || 15_000, + telemetry_metadata: Keyword.get(opts, :telemetry_metadata, %{}), max_size: opts[:max_size] || 1_000, transient: Keyword.get(opts, :transient, false) } @@ -736,14 +737,24 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do subscription_name: subscription_name, last_ack: last_ack, query_timeout: query_timeout, - checkpoints_pending: checkpoints_pending + checkpoints_pending: checkpoints_pending, + telemetry_metadata: telemetry_metadata } = data if checkpoints_pending > 0 do - Storage.Subscription.ack_last_seen_event(conn, stream_uuid, subscription_name, last_ack, - schema: schema, - timeout: query_timeout - ) + metadata = + Map.merge(telemetry_metadata, %{ + stream_uuid: stream_uuid, + subscription_name: subscription_name, + 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) end %SubscriptionState{data | checkpoints_pending: 0} diff --git a/lib/event_store/subscriptions/subscription_state.ex b/lib/event_store/subscriptions/subscription_state.ex index 44a1b909..20ae04e1 100644 --- a/lib/event_store/subscriptions/subscription_state.ex +++ b/lib/event_store/subscriptions/subscription_state.ex @@ -19,6 +19,7 @@ defmodule EventStore.Subscriptions.SubscriptionState do :partition_by, :query_timeout, :lock_ref, + telemetry_metadata: %{}, correlation_id_type: "uuid", causation_id_type: "uuid", last_received: 0, diff --git a/lib/event_store/subscriptions/supervisor.ex b/lib/event_store/subscriptions/supervisor.ex index 71d30ed1..fef052be 100644 --- a/lib/event_store/subscriptions/supervisor.ex +++ b/lib/event_store/subscriptions/supervisor.ex @@ -8,6 +8,10 @@ defmodule EventStore.Subscriptions.Supervisor do alias EventStore.Subscriptions alias EventStore.Subscriptions.Subscription + # Mirrors the storage `:timeout` a subscription gives its own queries, for the internal callers + # that have no configured event store to read one from. + @default_timeout 15_000 + def start_link(opts) do DynamicSupervisor.start_link(__MODULE__, [], opts) end @@ -37,7 +41,7 @@ defmodule EventStore.Subscriptions.Supervisor do end end - def stop_subscription(event_store, stream_uuid, subscription_name) do + def stop_subscription(event_store, stream_uuid, subscription_name, opts \\ []) do name = registry_name(event_store, stream_uuid, subscription_name) case Registry.whereis_name(name) do @@ -45,9 +49,49 @@ defmodule EventStore.Subscriptions.Supervisor do :ok subscription -> - supervisor = Module.concat(event_store, __MODULE__) + # Shaped after `:proc_lib.stop/3`, which is what `GenServer.stop/3` runs, except that only + # the subscription can decide whether a subscriber is still connected to it, so asking has + # to be a call of our own. + 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]) + + exit({:timeout, {__MODULE__, :stop_subscription, [subscription]}}) + end + + {:error, _error} = error -> + Process.demonitor(ref, [:flush]) + + error + 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) + catch + :exit, {reason, {GenServer, :call, _args}} when reason != :timeout -> :ok + end - DynamicSupervisor.terminate_child(supervisor, subscription) + defp remaining(opts, asked_at) do + case Keyword.get(opts, :timeout, @default_timeout) do + :infinity -> :infinity + timeout -> max(timeout - (System.monotonic_time(:millisecond) - asked_at), 0) end end diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs index 3938c05d..16337277 100644 --- a/test/subscriptions/subscribe_to_stream_test.exs +++ b/test/subscriptions/subscribe_to_stream_test.exs @@ -797,17 +797,540 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + assert_receive {:subscribed, ^subscription} + + :ok = EventStore.unsubscribe_from_stream(stream_uuid, subscription_name) + assert :ok = EventStore.delete_subscription(stream_uuid, subscription_name) refute Process.alive?(subscription) assert {:ok, []} = Storage.subscriptions(@conn, schema: schema) end + + test "should not be deleted while a subscriber is connected", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + + {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + + assert_receive {:subscribed, ^subscription} + + assert {:error, :subscribers_connected} = + EventStore.delete_subscription(stream_uuid, subscription_name) + + assert Process.alive?(subscription) + assert {:ok, [_subscription]} = Storage.subscriptions(@conn, schema: schema) + end + + test "should not be deleted while another of its concurrent subscribers is connected", %{ + subscription_name: subscription_name + } do + stream_uuid = UUID.uuid4() + test = self() + + {:ok, subscription} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), + concurrency_limit: 2 + ) + + assert_receive {:subscribed, ^subscription} + + sibling = + spawn_link(fn -> + {:ok, ^subscription} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), + concurrency_limit: 2 + ) + + send(test, :sibling_subscribed) + + receive do + :stop -> :ok + end + end) + + assert_receive :sibling_subscribed + + :ok = EventStore.unsubscribe_from_stream(stream_uuid, subscription_name) + + assert {:error, :subscribers_connected} = + EventStore.delete_subscription(stream_uuid, subscription_name) + + assert Process.alive?(subscription) + + send(sibling, :stop) + end + + test "should be deleted when it has no subscribers", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + + assert :ok = EventStore.delete_subscription(stream_uuid, subscription_name) + + assert {:ok, []} = Storage.subscriptions(@conn, schema: schema) + end + + test "should be deleted while its subscription is still terminating", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + + name = + {Module.concat(@event_store, Subscriptions.Registry), {stream_uuid, subscription_name}} + + 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() + + send(subscriber, :unsubscribe) + + # Losing its last subscriber makes the subscription write its checkpoint and then stop, so + # holding that write holds it registered and answering nothing. + assert_receive {:checkpointing, ^subscription} + assert ^subscription = Registry.whereis_name(name) + + deleting = + Task.async(fn -> EventStore.delete_subscription(stream_uuid, subscription_name) end) + + refute Task.yield(deleting, 100) + assert {:ok, [_subscription]} = Storage.subscriptions(@conn, schema: schema) + + send(subscription, :release_checkpoint) + + assert_receive {:unsubscribed, :ok} + + assert :ok = Task.await(deleting) + refute Process.alive?(subscription) + assert {:ok, []} = Storage.subscriptions(@conn, schema: schema) + end + + test "should wait for a checkpoint that outlasts the default call timeout", %{ + 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() + + # Losing its subscriber outright, rather than being unsubscribed, leaves nobody waiting on a + # call of their own, so the delete brings the only deadline still in play. + Process.exit(subscriber, :kill) + + assert_receive {:checkpointing, ^subscription} + + deleting = + Task.async(fn -> EventStore.delete_subscription(stream_uuid, subscription_name) end) + + # Longer than the five seconds a `GenServer.call/2` waits by default, and shorter than the + # fifteen a checkpoint write is given, so giving up would be giving up on a live write. + 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 keep the row when it gives up waiting for a checkpoint", %{ + 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 -> + try do + EventStore.delete_subscription(stream_uuid, subscription_name, timeout: 100) + catch + :exit, reason -> {:exit, reason} + end + end) + + assert {:ok, {:exit, {:timeout, _where}}} = Task.yield(deleting, 2_000) + + # Giving up says the subscription may still be writing, so the row it writes to has to stay. + assert {:ok, [_subscription]} = Storage.subscriptions(@conn, schema: schema) + + send(subscription, :release_checkpoint) + end + + test "should write its checkpoint while it still owns the row, not while terminating", %{ + subscription_name: subscription_name + } 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) + + record_checkpoints() + + send(subscriber, :unsubscribe) + + assert_receive {:unsubscribed, :ok} + assert_receive {:checkpoint, %{last_seen: 1}} + + assert :ok = EventStore.delete_subscription(stream_uuid, subscription_name) + + # A checkpoint written any later has no row of its own left to write to, and its `last_seen` + # would land on whichever subscription holds the name next. + refute_receive {:checkpoint, _metadata} + end + + test "should keep serving its subscriber after refusing to be deleted", %{ + subscription_name: subscription_name, + schema: schema + } do + stream_uuid = UUID.uuid4() + + {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + + assert_receive {:subscribed, ^subscription} + + :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1)) + + assert_receive {:events, [%RecordedEvent{event_number: 1}] = first} + :ok = Subscription.ack(subscription, first) + + assert {:error, :subscribers_connected} = + EventStore.delete_subscription(stream_uuid, subscription_name) + + :ok = EventStore.append_to_stream(stream_uuid, 1, EventFactory.create_events(1)) + + assert_receive {:events, [%RecordedEvent{event_number: 2}] = second} + :ok = Subscription.ack(subscription, second) + + Wait.until(fn -> + assert {:ok, [%Storage.Subscription{last_seen: 2}]} = + Storage.subscriptions(@conn, schema: schema) + end) + end + + test "should leave no monitor behind when it refuses", %{ + subscription_name: subscription_name + } do + stream_uuid = UUID.uuid4() + + {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + + assert_receive {:subscribed, ^subscription} + + assert {:error, :subscribers_connected} = + EventStore.delete_subscription(stream_uuid, subscription_name) + + :ok = EventStore.unsubscribe_from_stream(stream_uuid, subscription_name) + + Wait.until(fn -> refute Process.alive?(subscription) end) + + refute_receive {:DOWN, _ref, :process, ^subscription, _reason} + end + + test "should be deleted when it was persisted without being started", %{ + 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, [_subscription]} = Storage.subscriptions(@conn, schema: schema) + + assert :ok = EventStore.delete_subscription(stream_uuid, subscription_name) + + assert {:ok, []} = Storage.subscriptions(@conn, schema: schema) + end + + test "should be deleted once when several callers delete it at the same time", + %{subscription_name: subscription_name} = context do + stream_uuid = UUID.uuid4() + + subscription = start_unconnected_subscription(context, stream_uuid) + + results = + 1..5 + |> Task.async_stream( + fn _ -> EventStore.delete_subscription(stream_uuid, subscription_name) end, + max_concurrency: 5, + ordered: false + ) + |> Enum.map(fn {:ok, result} -> result end) + + assert results == [:ok, :ok, :ok, :ok, :ok] + refute Process.alive?(subscription) + end + + test "should let the same subscription name be used again after being deleted", %{ + subscription_name: subscription_name + } do + stream_uuid = UUID.uuid4() + + :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1)) + + {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + + assert_receive {:subscribed, ^subscription} + assert_receive {:events, [%RecordedEvent{event_number: 1}] = events} + + :ok = Subscription.ack(subscription, events) + :ok = EventStore.unsubscribe_from_stream(stream_uuid, subscription_name) + + assert :ok = EventStore.delete_subscription(stream_uuid, subscription_name) + + resubscribed = + Wait.until(fn -> + assert {:ok, resubscribed} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + + resubscribed + end) + + assert resubscribed != subscription + + assert_receive {:subscribed, ^resubscribed} + + # The deleted checkpoint is what makes the already acknowledged event arrive again. + assert_receive {:events, [%RecordedEvent{event_number: 1}]} + end + + test "should not delete an all streams subscription while a subscriber is connected", %{ + subscription_name: subscription_name, + schema: schema + } do + {:ok, subscription} = EventStore.subscribe_to_all_streams(subscription_name, self()) + + assert_receive {:subscribed, ^subscription} + + assert {:error, :subscribers_connected} = + EventStore.delete_all_streams_subscription(subscription_name) + + assert Process.alive?(subscription) + + assert {:ok, [%Storage.Subscription{stream_uuid: "$all"}]} = + Storage.subscriptions(@conn, schema: schema) + end + + test "should delete an all streams subscription once its subscriber has gone", %{ + subscription_name: subscription_name, + schema: schema + } do + {:ok, subscription} = EventStore.subscribe_to_all_streams(subscription_name, self()) + + assert_receive {:subscribed, ^subscription} + + :ok = EventStore.unsubscribe_from_all_streams(subscription_name) + + assert :ok = EventStore.delete_all_streams_subscription(subscription_name) + + refute Process.alive?(subscription) + assert {:ok, []} = Storage.subscriptions(@conn, schema: schema) + 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() + test = self() + + subscription = start_unconnected_subscription(context, stream_uuid) + supervisor = Module.concat(@event_store, Subscriptions.Supervisor) + + # Suspended so that a stop split back into a check and a terminate cannot reach the + # terminate before the connect it is racing has been served, which is the interleaving this + # guards. + :ok = :sys.suspend(supervisor) + :ok = :sys.suspend(subscription) + + deleting = + Task.async(fn -> EventStore.delete_subscription(stream_uuid, subscription_name) end) + + assert_enqueued(subscription, 1) + + spawn(fn -> + result = + try do + Subscription.connect(subscription, self(), []) + catch + :exit, reason -> {:exit, reason} + end + + send(test, {:connect, result}) + end) + + assert_enqueued(subscription, 2) + + :ok = :sys.resume(subscription) + + assert_receive {:connect, connect_result} + + :ok = :sys.resume(supervisor) + + assert :ok = Task.await(deleting) + + case connect_result do + {:ok, ^subscription} -> assert Process.alive?(subscription) + {:exit, _reason} -> refute Process.alive?(subscription) + end + end end # OTP 28 reports the `gen_server` hibernation loop where earlier releases # reported `:erlang.hibernate/3`. @hibernated_functions [{:erlang, :hibernate, 3}, {:gen_server, :loop_hibernate, 4}] + # Runs in the subscription process, so a checkpoint write holds it wherever it was written from. + def block_checkpoint_event(_event, _measurements, _metadata, test) do + send(test, {:checkpointing, self()}) + + receive do + :release_checkpoint -> :ok + end + end + + def record_checkpoint_event(_event, _measurements, metadata, test) do + send(test, {:checkpoint, metadata}) + end + + defp record_checkpoints do + handler_id = "#{inspect(__MODULE__)}-checkpoints-#{System.unique_integer([:positive])}" + + :ok = + :telemetry.attach( + handler_id, + [:eventstore, :subscription_checkpoint, :stop], + &__MODULE__.record_checkpoint_event/4, + self() + ) + + on_exit(fn -> :telemetry.detach(handler_id) end) + end + + defp block_checkpoint do + handler_id = "#{inspect(__MODULE__)}-checkpoint-#{System.unique_integer([:positive])}" + + :ok = + :telemetry.attach( + handler_id, + [:eventstore, :subscription_checkpoint, :start], + &__MODULE__.block_checkpoint_event/4, + self() + ) + + on_exit(fn -> :telemetry.detach(handler_id) end) + end + + # Subscribes from its own process, so that an unsubscribe the subscription cannot answer yet + # blocks the subscriber rather than the test. The threshold keeps an acknowledgement from + # writing the checkpoint that unsubscribing then has to write. + defp start_subscriber(stream_uuid, subscription_name) do + test = self() + + start_supervised!( + {Task, + fn -> + {:ok, subscription} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), + checkpoint_threshold: 100 + ) + + send(test, {:subscriber, self(), subscription}) + + forward_to_test(test, subscription) + end} + ) + end + + defp forward_to_test(test, subscription) do + receive do + :unsubscribe -> + send(test, {:unsubscribed, Subscription.unsubscribe(subscription)}) + + message -> + send(test, message) + forward_to_test(test, subscription) + end + end + + defp assert_enqueued(pid, count) do + Wait.until(fn -> + assert {:message_queue_len, ^count} = Process.info(pid, :message_queue_len) + end) + 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 + %{ + conn: conn, + schema: schema, + serializer: serializer, + correlation_id_type: correlation_id_type, + causation_id_type: causation_id_type, + subscription_name: subscription_name + } = context + + {:ok, subscription} = + Subscriptions.Supervisor.start_subscription( + event_store: @event_store, + conn: conn, + schema: schema, + serializer: serializer, + correlation_id_type: correlation_id_type, + causation_id_type: causation_id_type, + retry_interval: 60_000, + stream_uuid: stream_uuid, + subscription_name: subscription_name, + buffer_size: 1, + start_from: 0 + ) + + subscription + end + defp assert_hibernated(pid) do assert {:current_function, current_function} = Process.info(pid, :current_function) assert current_function in @hibernated_functions diff --git a/test/telemetry_test.exs b/test/telemetry_test.exs index 6d086774..38ff0e76 100644 --- a/test/telemetry_test.exs +++ b/test/telemetry_test.exs @@ -290,6 +290,8 @@ defmodule EventStore.TelemetryTest do assert_receive {:subscribed, ^subscription} + :ok = EventStore.unsubscribe_from_stream(stream_uuid, subscription_name) + attach_telemetry(:delete_subscription) assert :ok = EventStore.delete_subscription(stream_uuid, subscription_name) @@ -311,6 +313,105 @@ defmodule EventStore.TelemetryTest do refute Process.alive?(subscription) end + test "emits stop telemetry for a delete_subscription its subscribers refuse" do + stream_uuid = UUID.uuid4() + subscription_name = "telemetry-" <> UUID.uuid4() + + assert {:ok, subscription} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + + assert_receive {:subscribed, ^subscription} + + attach_telemetry(:delete_subscription) + + assert {:error, :subscribers_connected} = + EventStore.delete_subscription(stream_uuid, subscription_name) + + assert_start_event(:delete_subscription, + event_store: EventStore, + stream_uuid: stream_uuid, + subscription_name: subscription_name + ) + + assert_stop_event(:delete_subscription, + event_store: EventStore, + result: {:error, :subscribers_connected}, + stream_uuid: stream_uuid, + subscription_name: subscription_name + ) + + refute_exception_event(:delete_subscription) + assert Process.alive?(subscription) + assert :ok = Subscription.unsubscribe(subscription) + end + + test "emits start and stop telemetry for a subscription checkpoint" do + stream_uuid = UUID.uuid4() + subscription_name = "telemetry-" <> UUID.uuid4() + + assert {:ok, subscription} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self()) + + assert_receive {:subscribed, ^subscription} + + attach_telemetry(:subscription_checkpoint) + + assert :ok = EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1)) + + assert_receive {:events, events} + assert :ok = Subscription.ack(subscription, events) + + assert_start_event(:subscription_checkpoint, + event_store: EventStore, + last_seen: 1, + stream_uuid: stream_uuid, + subscription_name: subscription_name + ) + + assert_stop_event(:subscription_checkpoint, + event_store: EventStore, + last_seen: 1, + result: :ok, + stream_uuid: stream_uuid, + subscription_name: subscription_name + ) + + refute_exception_event(:subscription_checkpoint) + end + + test "names a named instance the same way every other checkpoint metadata does" do + start_supervised!({TestEventStore, name: :eventstore1, schema: "public"}) + + stream_uuid = UUID.uuid4() + subscription_name = "telemetry-" <> UUID.uuid4() + + assert {:ok, subscription} = + EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(), + name: :eventstore1 + ) + + assert_receive {:subscribed, ^subscription} + + attach_telemetry(:subscription_checkpoint) + + assert :ok = + EventStore.append_to_stream(stream_uuid, 0, EventFactory.create_events(1), + name: :eventstore1 + ) + + assert_receive {:events, events} + assert :ok = Subscription.ack(subscription, events) + + assert_stop_event(:subscription_checkpoint, + event_store: EventStore, + last_seen: 1, + name: :eventstore1, + result: :ok, + stream_uuid: stream_uuid, + subscription_name: subscription_name + ) + end + test "emits start and stop telemetry for paginate_streams" do stream_uuid = "telemetry-" <> UUID.uuid4()