fix(subscriptions): stop a deleted subscription from silently skipping its successor's events

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

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;

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.

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

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

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
 

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}

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