fix(subscriptions): stop a subscription delete from disconnecting its subscribers

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

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,25 +1472,28 @@ 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.
 
     - `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.

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 """

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

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}

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,

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,17 +41,57 @@ 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
       :undefined ->
         :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