From d1624a814dbd57973a74e014925f22c8dcfc5fee Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Mon, 21 Sep 2026 17:25:27 -0400
Subject: [PATCH 01/13] fix(subscriptions): stop a subscription delete from
 disconnecting its subscribers

Deleting a subscription tore down the process that every one of its subscribers shares, so a caller holding none of it could destroy a live subscription and erase the checkpoint it was working through, with the subscribers learning about it only from the resulting `:DOWN`.

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/event_store.ex                            | 12 ++--
 lib/event_store/subscriptions/subscription.ex | 16 +++++
 lib/event_store/subscriptions/supervisor.ex   | 13 +++-
 .../subscribe_to_stream_test.exs              | 71 +++++++++++++++++++
 test/telemetry_test.exs                       |  2 +
 5 files changed, 107 insertions(+), 7 deletions(-)

diff --git a/lib/event_store.ex b/lib/event_store.ex
index e4321da0..3ff3a218 100644
--- a/lib/event_store.ex
+++ b/lib/event_store.ex
@@ -1471,14 +1471,17 @@ defmodule EventStore do
     - `subscription_name` is used to identify the existing subscription to
       remove.
 
-  Returns `:ok` on success.
+  Returns `:ok` on success, or `{:error, :subscription_has_subscribers}` 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, :subscription_has_subscribers} | {: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, :subscription_has_subscribers}` when subscribers are
+  still connected to the subscription.
   """
   @callback delete_all_streams_subscription(subscription_name :: String.t(), opts :: options) ::
-              :ok | {:error, term}
+              :ok | {:error, :subscription_has_subscribers} | {:error, term}
 
   @doc """
   Read a snapshot, if available, for a given source.
diff --git a/lib/event_store/subscriptions/subscription.ex b/lib/event_store/subscriptions/subscription.ex
index 321b258b..348d5f82 100644
--- a/lib/event_store/subscriptions/subscription.ex
+++ b/lib/event_store/subscriptions/subscription.ex
@@ -84,6 +84,13 @@ defmodule EventStore.Subscriptions.Subscription do
     GenServer.call(subscription, {:unsubscribe, self()})
   end
 
+  @doc """
+  Whether any subscriber is still connected to the subscription.
+  """
+  def has_subscribers?(subscription) do
+    GenServer.call(subscription, :has_subscribers?)
+  end
+
   @doc false
   def last_seen(subscription) do
     GenServer.call(subscription, :last_seen)
@@ -244,6 +251,15 @@ defmodule EventStore.Subscriptions.Subscription do
     end
   end
 
+  @impl GenServer
+  def handle_call(:has_subscribers?, _from, %Subscription{} = state) do
+    %Subscription{
+      subscription: %SubscriptionFsm{data: %SubscriptionState{subscribers: subscribers}}
+    } = state
+
+    {:reply, subscribers != %{}, state}
+  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/supervisor.ex b/lib/event_store/subscriptions/supervisor.ex
index 71d30ed1..b6f64b3e 100644
--- a/lib/event_store/subscriptions/supervisor.ex
+++ b/lib/event_store/subscriptions/supervisor.ex
@@ -45,9 +45,16 @@ defmodule EventStore.Subscriptions.Supervisor do
         :ok
 
       subscription ->
-        supervisor = Module.concat(event_store, __MODULE__)
-
-        DynamicSupervisor.terminate_child(supervisor, subscription)
+        # Terminating through the supervisor rather than asking the subscription to stop itself,
+        # so that the checkpoint written while it terminates cannot race the caller deleting the
+        # subscription it belongs to.
+        if Subscription.has_subscribers?(subscription) do
+          {:error, :subscription_has_subscribers}
+        else
+          supervisor = Module.concat(event_store, __MODULE__)
+
+          DynamicSupervisor.terminate_child(supervisor, subscription)
+        end
     end
   end
 
diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs
index 3938c05d..f25ee892 100644
--- a/test/subscriptions/subscribe_to_stream_test.exs
+++ b/test/subscriptions/subscribe_to_stream_test.exs
@@ -797,11 +797,82 @@ 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, :subscription_has_subscribers} =
+               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, :subscription_has_subscribers} =
+               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
   end
 
   # OTP 28 reports the `gen_server` hibernation loop where earlier releases
diff --git a/test/telemetry_test.exs b/test/telemetry_test.exs
index 6d086774..9ca320b1 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)

From 425445df353c9ba916a7f8c2265811d18685a9ff Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Mon, 21 Sep 2026 18:09:35 -0400
Subject: [PATCH 02/13] fix(subscriptions): decide a delete and a competing
 connect in one step

Asking whether a subscription has subscribers and then stopping it were two separate calls, so a subscriber could connect in the gap between them and still be torn down by the very delete its presence should have refused.

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/event_store/subscriptions/subscription.ex | 14 ++--
 lib/event_store/subscriptions/supervisor.ex   | 24 +++---
 .../subscribe_to_stream_test.exs              | 81 +++++++++++++++++++
 3 files changed, 105 insertions(+), 14 deletions(-)

diff --git a/lib/event_store/subscriptions/subscription.ex b/lib/event_store/subscriptions/subscription.ex
index 348d5f82..13bc991b 100644
--- a/lib/event_store/subscriptions/subscription.ex
+++ b/lib/event_store/subscriptions/subscription.ex
@@ -85,10 +85,10 @@ defmodule EventStore.Subscriptions.Subscription do
   end
 
   @doc """
-  Whether any subscriber is still connected to the subscription.
+  Stop the subscription, unless a subscriber is still connected to it.
   """
-  def has_subscribers?(subscription) do
-    GenServer.call(subscription, :has_subscribers?)
+  def stop_unless_subscribed(subscription) do
+    GenServer.call(subscription, :stop_unless_subscribed)
   end
 
   @doc false
@@ -252,12 +252,16 @@ defmodule EventStore.Subscriptions.Subscription do
   end
 
   @impl GenServer
-  def handle_call(:has_subscribers?, _from, %Subscription{} = state) do
+  def handle_call(:stop_unless_subscribed, _from, %Subscription{} = state) do
     %Subscription{
       subscription: %SubscriptionFsm{data: %SubscriptionState{subscribers: subscribers}}
     } = state
 
-    {:reply, subscribers != %{}, state}
+    if subscribers == %{} do
+      {:stop, :shutdown, :ok, state}
+    else
+      {:reply, {:error, :subscription_has_subscribers}, state}
+    end
   end
 
   @impl GenServer
diff --git a/lib/event_store/subscriptions/supervisor.ex b/lib/event_store/subscriptions/supervisor.ex
index b6f64b3e..b85a931e 100644
--- a/lib/event_store/subscriptions/supervisor.ex
+++ b/lib/event_store/subscriptions/supervisor.ex
@@ -45,15 +45,21 @@ defmodule EventStore.Subscriptions.Supervisor do
         :ok
 
       subscription ->
-        # Terminating through the supervisor rather than asking the subscription to stop itself,
-        # so that the checkpoint written while it terminates cannot race the caller deleting the
-        # subscription it belongs to.
-        if Subscription.has_subscribers?(subscription) do
-          {:error, :subscription_has_subscribers}
-        else
-          supervisor = Module.concat(event_store, __MODULE__)
-
-          DynamicSupervisor.terminate_child(supervisor, subscription)
+        ref = Process.monitor(subscription)
+
+        # Letting the subscription itself decide keeps a subscriber that connects concurrently from
+        # being torn down, and waiting for it to go down keeps the checkpoint written while it
+        # terminates from racing the caller deleting the subscription it belongs to.
+        case Subscription.stop_unless_subscribed(subscription) do
+          :ok ->
+            receive do
+              {:DOWN, ^ref, :process, ^subscription, _reason} -> :ok
+            end
+
+          {:error, _error} = error ->
+            Process.demonitor(ref, [:flush])
+
+            error
         end
     end
   end
diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs
index f25ee892..8636ae42 100644
--- a/test/subscriptions/subscribe_to_stream_test.exs
+++ b/test/subscriptions/subscribe_to_stream_test.exs
@@ -873,12 +873,93 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
 
       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}]
 
+  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

From 886302f4eb8dee44cfffd12941420804a8816f2f Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Mon, 21 Sep 2026 18:31:38 -0400
Subject: [PATCH 03/13] fix(subscriptions): let a delete tolerate a
 subscription that is already going down

Unsubscribing answers before the subscription it stops has terminated, so a delete that follows it could reach a process that no longer answers calls and take the caller down with it.

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/event_store/subscriptions/supervisor.ex   | 11 +++++-
 .../subscribe_to_stream_test.exs              | 38 +++++++++++++++++++
 2 files changed, 48 insertions(+), 1 deletion(-)

diff --git a/lib/event_store/subscriptions/supervisor.ex b/lib/event_store/subscriptions/supervisor.ex
index b85a931e..06ef4ff2 100644
--- a/lib/event_store/subscriptions/supervisor.ex
+++ b/lib/event_store/subscriptions/supervisor.ex
@@ -50,7 +50,7 @@ defmodule EventStore.Subscriptions.Supervisor do
         # Letting the subscription itself decide keeps a subscriber that connects concurrently from
         # being torn down, and waiting for it to go down keeps the checkpoint written while it
         # terminates from racing the caller deleting the subscription it belongs to.
-        case Subscription.stop_unless_subscribed(subscription) do
+        case stop_unless_subscribed(subscription) do
           :ok ->
             receive do
               {:DOWN, ^ref, :process, ^subscription, _reason} -> :ok
@@ -64,6 +64,15 @@ defmodule EventStore.Subscriptions.Supervisor do
     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, and has to reach the caller.
+  defp stop_unless_subscribed(subscription) do
+    Subscription.stop_unless_subscribed(subscription)
+  catch
+    :exit, {reason, {GenServer, :call, _args}} when reason != :timeout -> :ok
+  end
+
   @impl DynamicSupervisor
   def init(_init_arg) do
     DynamicSupervisor.init(strategy: :one_for_one)
diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs
index 8636ae42..06093b81 100644
--- a/test/subscriptions/subscribe_to_stream_test.exs
+++ b/test/subscriptions/subscribe_to_stream_test.exs
@@ -24,6 +24,18 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
     {:ok, %{subscription_name: subscription_name}}
   end
 
+  # A subscription that has begun terminating answers no calls: it dies with the call still in its
+  # mailbox, which `GenServer.call` reports to its caller as an exit.
+  defmodule TerminatingSubscription do
+    use GenServer
+
+    @impl GenServer
+    def init(:ok), do: {:ok, :ok}
+
+    @impl GenServer
+    def handle_call(:stop_unless_subscribed, _from, state), do: {:stop, :shutdown, state}
+  end
+
   describe "single stream subscription" do
     setup [:append_events_to_another_stream]
 
@@ -874,6 +886,32 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
       assert {:ok, []} = Storage.subscriptions(@conn, schema: schema)
     end
 
+    test "should be deleted when its subscription is already terminating", %{
+      subscription_name: subscription_name,
+      schema: schema
+    } do
+      stream_uuid = UUID.uuid4()
+
+      name =
+        {Module.concat(@event_store, Subscriptions.Registry), {stream_uuid, subscription_name}}
+
+      {:ok, subscription} = EventStore.subscribe_to_stream(stream_uuid, subscription_name, self())
+
+      assert_receive {:subscribed, ^subscription}
+
+      :ok = EventStore.unsubscribe_from_stream(stream_uuid, subscription_name)
+
+      Wait.until(fn -> assert :undefined = Registry.whereis_name(name) end)
+
+      {:ok, terminating} =
+        GenServer.start(TerminatingSubscription, :ok, name: {:via, Registry, name})
+
+      assert :ok = EventStore.delete_subscription(stream_uuid, subscription_name)
+
+      refute Process.alive?(terminating)
+      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()

From e75ad727a19e54448f981aebb43324d8fa44385f Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Mon, 21 Sep 2026 21:58:29 -0400
Subject: [PATCH 04/13] chore(subscriptions): branch a subscription stop on a
 case

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/event_store/subscriptions/subscription.ex | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/lib/event_store/subscriptions/subscription.ex b/lib/event_store/subscriptions/subscription.ex
index 13bc991b..a88276dd 100644
--- a/lib/event_store/subscriptions/subscription.ex
+++ b/lib/event_store/subscriptions/subscription.ex
@@ -257,10 +257,9 @@ defmodule EventStore.Subscriptions.Subscription do
       subscription: %SubscriptionFsm{data: %SubscriptionState{subscribers: subscribers}}
     } = state
 
-    if subscribers == %{} do
-      {:stop, :shutdown, :ok, state}
-    else
-      {:reply, {:error, :subscription_has_subscribers}, state}
+    case map_size(subscribers) do
+      0 -> {:stop, :shutdown, :ok, state}
+      _ -> {:reply, {:error, :subscription_has_subscribers}, state}
     end
   end
 

From f4517a1f034185fc5b22cda1936767ef7903bbf1 Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Mon, 21 Sep 2026 22:16:00 -0400
Subject: [PATCH 05/13] chore(subscriptions): name the refusal after the state
 that causes it

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/event_store.ex                              | 15 +++++++--------
 lib/event_store/subscriptions/subscription.ex   |  2 +-
 test/subscriptions/subscribe_to_stream_test.exs |  4 ++--
 3 files changed, 10 insertions(+), 11 deletions(-)

diff --git a/lib/event_store.ex b/lib/event_store.ex
index 3ff3a218..447fdf9f 100644
--- a/lib/event_store.ex
+++ b/lib/event_store.ex
@@ -1471,17 +1471,16 @@ defmodule EventStore do
     - `subscription_name` is used to identify the existing subscription to
       remove.
 
-  Returns `:ok` on success, or `{:error, :subscription_has_subscribers}` 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.
+  Returns `:ok` on success, or `{:error, :still_subscribed}` 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, :subscription_has_subscribers} | {:error, term}
+              :ok | {:error, :still_subscribed} | {:error, term}
 
   @doc """
   Delete an existing persistent subscription to all streams.
@@ -1489,11 +1488,11 @@ defmodule EventStore do
     - `subscription_name` is used to identify the existing subscription to
       remove.
 
-  Returns `:ok` on success, or `{:error, :subscription_has_subscribers}` when subscribers are
-  still connected to the subscription.
+  Returns `:ok` on success, or `{:error, :still_subscribed}` when subscribers are still connected
+  to the subscription.
   """
   @callback delete_all_streams_subscription(subscription_name :: String.t(), opts :: options) ::
-              :ok | {:error, :subscription_has_subscribers} | {:error, term}
+              :ok | {:error, :still_subscribed} | {:error, term}
 
   @doc """
   Read a snapshot, if available, for a given source.
diff --git a/lib/event_store/subscriptions/subscription.ex b/lib/event_store/subscriptions/subscription.ex
index a88276dd..8f757be6 100644
--- a/lib/event_store/subscriptions/subscription.ex
+++ b/lib/event_store/subscriptions/subscription.ex
@@ -259,7 +259,7 @@ defmodule EventStore.Subscriptions.Subscription do
 
     case map_size(subscribers) do
       0 -> {:stop, :shutdown, :ok, state}
-      _ -> {:reply, {:error, :subscription_has_subscribers}, state}
+      _ -> {:reply, {:error, :still_subscribed}, state}
     end
   end
 
diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs
index 06093b81..5fbd219d 100644
--- a/test/subscriptions/subscribe_to_stream_test.exs
+++ b/test/subscriptions/subscribe_to_stream_test.exs
@@ -829,7 +829,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
 
       assert_receive {:subscribed, ^subscription}
 
-      assert {:error, :subscription_has_subscribers} =
+      assert {:error, :still_subscribed} =
                EventStore.delete_subscription(stream_uuid, subscription_name)
 
       assert Process.alive?(subscription)
@@ -867,7 +867,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
 
       :ok = EventStore.unsubscribe_from_stream(stream_uuid, subscription_name)
 
-      assert {:error, :subscription_has_subscribers} =
+      assert {:error, :still_subscribed} =
                EventStore.delete_subscription(stream_uuid, subscription_name)
 
       assert Process.alive?(subscription)

From d34b4fd32f208139a91a70deb584c8510294a58b Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Mon, 21 Sep 2026 22:44:26 -0400
Subject: [PATCH 06/13] chore(subscriptions): name the refusal after what the
 caller has to disconnect

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/event_store.ex                              | 14 +++++++-------
 lib/event_store/subscriptions/subscription.ex   |  2 +-
 lib/event_store/subscriptions/supervisor.ex     |  6 +++---
 test/subscriptions/subscribe_to_stream_test.exs |  9 +++++----
 4 files changed, 16 insertions(+), 15 deletions(-)

diff --git a/lib/event_store.ex b/lib/event_store.ex
index 447fdf9f..d933594d 100644
--- a/lib/event_store.ex
+++ b/lib/event_store.ex
@@ -1471,16 +1471,16 @@ defmodule EventStore do
     - `subscription_name` is used to identify the existing subscription to
       remove.
 
-  Returns `:ok` on success, or `{:error, :still_subscribed}` 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.
+  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, :still_subscribed} | {:error, term}
+              :ok | {:error, :subscribers_connected} | {:error, term}
 
   @doc """
   Delete an existing persistent subscription to all streams.
@@ -1488,11 +1488,11 @@ defmodule EventStore do
     - `subscription_name` is used to identify the existing subscription to
       remove.
 
-  Returns `:ok` on success, or `{:error, :still_subscribed}` when subscribers are still connected
-  to the subscription.
+  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, :still_subscribed} | {: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/subscription.ex b/lib/event_store/subscriptions/subscription.ex
index 8f757be6..d864c0b7 100644
--- a/lib/event_store/subscriptions/subscription.ex
+++ b/lib/event_store/subscriptions/subscription.ex
@@ -259,7 +259,7 @@ defmodule EventStore.Subscriptions.Subscription do
 
     case map_size(subscribers) do
       0 -> {:stop, :shutdown, :ok, state}
-      _ -> {:reply, {:error, :still_subscribed}, state}
+      _ -> {:reply, {:error, :subscribers_connected}, state}
     end
   end
 
diff --git a/lib/event_store/subscriptions/supervisor.ex b/lib/event_store/subscriptions/supervisor.ex
index 06ef4ff2..c55758b2 100644
--- a/lib/event_store/subscriptions/supervisor.ex
+++ b/lib/event_store/subscriptions/supervisor.ex
@@ -47,9 +47,9 @@ defmodule EventStore.Subscriptions.Supervisor do
       subscription ->
         ref = Process.monitor(subscription)
 
-        # Letting the subscription itself decide keeps a subscriber that connects concurrently from
-        # being torn down, and waiting for it to go down keeps the checkpoint written while it
-        # terminates from racing the caller deleting the subscription it belongs to.
+        # Letting the subscription itself decide keeps a subscriber that connects concurrently
+        # from being torn down, and waiting for it to go down keeps the checkpoint written while
+        # it terminates from racing the caller deleting the subscription it belongs to.
         case stop_unless_subscribed(subscription) do
           :ok ->
             receive do
diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs
index 5fbd219d..dcfd6d26 100644
--- a/test/subscriptions/subscribe_to_stream_test.exs
+++ b/test/subscriptions/subscribe_to_stream_test.exs
@@ -829,7 +829,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
 
       assert_receive {:subscribed, ^subscription}
 
-      assert {:error, :still_subscribed} =
+      assert {:error, :subscribers_connected} =
                EventStore.delete_subscription(stream_uuid, subscription_name)
 
       assert Process.alive?(subscription)
@@ -867,7 +867,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
 
       :ok = EventStore.unsubscribe_from_stream(stream_uuid, subscription_name)
 
-      assert {:error, :still_subscribed} =
+      assert {:error, :subscribers_connected} =
                EventStore.delete_subscription(stream_uuid, subscription_name)
 
       assert Process.alive?(subscription)
@@ -920,8 +920,9 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
       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.
+      # 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)
 

From 0672b7e91dbec2b6368a7a85d640536cce3f778a Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Mon, 21 Sep 2026 23:03:03 -0400
Subject: [PATCH 07/13] chore(subscriptions): let the return value carry the
 reason the stop refused

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/event_store/subscriptions/subscription.ex   | 6 +++---
 lib/event_store/subscriptions/supervisor.ex     | 6 +++---
 test/subscriptions/subscribe_to_stream_test.exs | 2 +-
 3 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/lib/event_store/subscriptions/subscription.ex b/lib/event_store/subscriptions/subscription.ex
index d864c0b7..d2bd1187 100644
--- a/lib/event_store/subscriptions/subscription.ex
+++ b/lib/event_store/subscriptions/subscription.ex
@@ -87,8 +87,8 @@ defmodule EventStore.Subscriptions.Subscription do
   @doc """
   Stop the subscription, unless a subscriber is still connected to it.
   """
-  def stop_unless_subscribed(subscription) do
-    GenServer.call(subscription, :stop_unless_subscribed)
+  def stop(subscription) do
+    GenServer.call(subscription, :stop)
   end
 
   @doc false
@@ -252,7 +252,7 @@ defmodule EventStore.Subscriptions.Subscription do
   end
 
   @impl GenServer
-  def handle_call(:stop_unless_subscribed, _from, %Subscription{} = state) do
+  def handle_call(:stop, _from, %Subscription{} = state) do
     %Subscription{
       subscription: %SubscriptionFsm{data: %SubscriptionState{subscribers: subscribers}}
     } = state
diff --git a/lib/event_store/subscriptions/supervisor.ex b/lib/event_store/subscriptions/supervisor.ex
index c55758b2..9971e21b 100644
--- a/lib/event_store/subscriptions/supervisor.ex
+++ b/lib/event_store/subscriptions/supervisor.ex
@@ -50,7 +50,7 @@ defmodule EventStore.Subscriptions.Supervisor do
         # Letting the subscription itself decide keeps a subscriber that connects concurrently
         # from being torn down, and waiting for it to go down keeps the checkpoint written while
         # it terminates from racing the caller deleting the subscription it belongs to.
-        case stop_unless_subscribed(subscription) do
+        case stop(subscription) do
           :ok ->
             receive do
               {:DOWN, ^ref, :process, ^subscription, _reason} -> :ok
@@ -67,8 +67,8 @@ defmodule EventStore.Subscriptions.Supervisor do
   # 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, and has to reach the caller.
-  defp stop_unless_subscribed(subscription) do
-    Subscription.stop_unless_subscribed(subscription)
+  defp stop(subscription) do
+    Subscription.stop(subscription)
   catch
     :exit, {reason, {GenServer, :call, _args}} when reason != :timeout -> :ok
   end
diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs
index dcfd6d26..beaa7ce4 100644
--- a/test/subscriptions/subscribe_to_stream_test.exs
+++ b/test/subscriptions/subscribe_to_stream_test.exs
@@ -33,7 +33,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
     def init(:ok), do: {:ok, :ok}
 
     @impl GenServer
-    def handle_call(:stop_unless_subscribed, _from, state), do: {:stop, :shutdown, state}
+    def handle_call(:stop, _from, state), do: {:stop, :shutdown, state}
   end
 
   describe "single stream subscription" do

From 3c6618da1f082bad43dbeba075c77ca1ba216c2a Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Mon, 21 Sep 2026 23:54:31 -0400
Subject: [PATCH 08/13] feat(subscriptions): make a checkpoint observable so
 callers can tell one is in flight

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 guides/Usage.md                               |  8 ++++++++
 .../subscriptions/subscription_fsm.ex         | 20 ++++++++++++++-----
 2 files changed, 23 insertions(+), 5 deletions(-)

diff --git a/guides/Usage.md b/guides/Usage.md
index b5de7a93..ce93d833 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,13 @@ 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. 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, and once more
+while it terminates.
+
 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/subscriptions/subscription_fsm.ex b/lib/event_store/subscriptions/subscription_fsm.ex
index 3604636d..13affacf 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}
 
@@ -731,6 +731,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
   defp persist_checkpoint(%SubscriptionState{transient: false} = data) do
     %SubscriptionState{
       conn: conn,
+      event_store: event_store,
       schema: schema,
       stream_uuid: stream_uuid,
       subscription_name: subscription_name,
@@ -740,10 +741,19 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
     } = 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 = %{
+        event_store: event_store,
+        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}

From ed03cc0fa5f81ab753fa9950e4e6a9559ec79d5d Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Mon, 21 Sep 2026 23:54:32 -0400
Subject: [PATCH 09/13] test(subscriptions): prove what a contested delete
 settles on instead of assuming it

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 .../subscribe_to_stream_test.exs              | 240 ++++++++++++++++--
 test/telemetry_test.exs                       |  66 +++++
 2 files changed, 289 insertions(+), 17 deletions(-)

diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs
index beaa7ce4..c7b5c004 100644
--- a/test/subscriptions/subscribe_to_stream_test.exs
+++ b/test/subscriptions/subscribe_to_stream_test.exs
@@ -24,18 +24,6 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
     {:ok, %{subscription_name: subscription_name}}
   end
 
-  # A subscription that has begun terminating answers no calls: it dies with the call still in its
-  # mailbox, which `GenServer.call` reports to its caller as an exit.
-  defmodule TerminatingSubscription do
-    use GenServer
-
-    @impl GenServer
-    def init(:ok), do: {:ok, :ok}
-
-    @impl GenServer
-    def handle_call(:stop, _from, state), do: {:stop, :shutdown, state}
-  end
-
   describe "single stream subscription" do
     setup [:append_events_to_another_stream]
 
@@ -886,7 +874,7 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
       assert {:ok, []} = Storage.subscriptions(@conn, schema: schema)
     end
 
-    test "should be deleted when its subscription is already terminating", %{
+    test "should be deleted while its subscription is still terminating", %{
       subscription_name: subscription_name,
       schema: schema
     } do
@@ -895,20 +883,186 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
       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 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 -> assert :undefined = Registry.whereis_name(name) end)
+      Wait.until(fn -> refute Process.alive?(subscription) end)
 
-      {:ok, terminating} =
-        GenServer.start(TerminatingSubscription, :ok, name: {:via, Registry, name})
+      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)
 
-      refute Process.alive?(terminating)
+      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
 
@@ -963,6 +1117,58 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
   # 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
+
+  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()
+
+    spawn(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)
diff --git a/test/telemetry_test.exs b/test/telemetry_test.exs
index 9ca320b1..9f3c314f 100644
--- a/test/telemetry_test.exs
+++ b/test/telemetry_test.exs
@@ -313,6 +313,72 @@ 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 "emits start and stop telemetry for paginate_streams" do
     stream_uuid = "telemetry-" <> UUID.uuid4()
 

From 739df40e61db13b6d5b1a0b64bff38e55ddea813 Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Tue, 22 Sep 2026 00:16:39 -0400
Subject: [PATCH 10/13] test(subscriptions): pin the checkpoint to the handler
 that still owns the row

A checkpoint deferred past the reply would have its stale last_seen land on whichever subscription holds the name next, and nothing else declares that it cannot be deferred.

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/event_store/subscriptions/supervisor.ex   |  9 ++-
 .../subscribe_to_stream_test.exs              | 66 ++++++++++++++++---
 2 files changed, 64 insertions(+), 11 deletions(-)

diff --git a/lib/event_store/subscriptions/supervisor.ex b/lib/event_store/subscriptions/supervisor.ex
index 9971e21b..91f46f38 100644
--- a/lib/event_store/subscriptions/supervisor.ex
+++ b/lib/event_store/subscriptions/supervisor.ex
@@ -45,13 +45,16 @@ defmodule EventStore.Subscriptions.Supervisor do
         :ok
 
       subscription ->
+        # 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)
 
-        # Letting the subscription itself decide keeps a subscriber that connects concurrently
-        # from being torn down, and waiting for it to go down keeps the checkpoint written while
-        # it terminates from racing the caller deleting the subscription it belongs to.
         case stop(subscription) 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. Unbounded, as `GenServer.stop/3` is.
             receive do
               {:DOWN, ^ref, :process, ^subscription, _reason} -> :ok
             end
diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs
index c7b5c004..f7735f2f 100644
--- a/test/subscriptions/subscribe_to_stream_test.exs
+++ b/test/subscriptions/subscribe_to_stream_test.exs
@@ -917,6 +917,35 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
       assert {:ok, []} = Storage.subscriptions(@conn, schema: schema)
     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
@@ -1126,6 +1155,24 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
     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])}"
 
@@ -1146,16 +1193,19 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
   defp start_subscriber(stream_uuid, subscription_name) do
     test = self()
 
-    spawn(fn ->
-      {:ok, subscription} =
-        EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(),
-          checkpoint_threshold: 100
-        )
+    start_supervised!(
+      {Task,
+       fn ->
+         {:ok, subscription} =
+           EventStore.subscribe_to_stream(stream_uuid, subscription_name, self(),
+             checkpoint_threshold: 100
+           )
 
-      send(test, {:subscriber, self(), subscription})
+         send(test, {:subscriber, self(), subscription})
 
-      forward_to_test(test, subscription)
-    end)
+         forward_to_test(test, subscription)
+       end}
+    )
   end
 
   defp forward_to_test(test, subscription) do

From 67d08a3494b7f0905daf4a60280a5038d5168d64 Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Tue, 22 Sep 2026 00:46:28 -0400
Subject: [PATCH 11/13] fix(subscriptions): stop abandoning a checkpoint the
 delete behind it has to wait for

A subscription is given fifteen seconds to write a checkpoint and the call asking it to stop waited five, so a slow write reached the caller as a crash rather than as the wait it is.

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/event_store/subscriptions/subscription.ex | 5 ++++-
 lib/event_store/subscriptions/supervisor.ex   | 3 ++-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/lib/event_store/subscriptions/subscription.ex b/lib/event_store/subscriptions/subscription.ex
index d2bd1187..c0ad4d9a 100644
--- a/lib/event_store/subscriptions/subscription.ex
+++ b/lib/event_store/subscriptions/subscription.ex
@@ -86,9 +86,12 @@ defmodule EventStore.Subscriptions.Subscription do
 
   @doc """
   Stop the subscription, unless a subscriber is still connected to it.
+
+  Waits without a deadline, because whatever the subscription is busy with is storage work bounded
+  by its own `query_timeout` and giving up early would answer before it is safe to.
   """
   def stop(subscription) do
-    GenServer.call(subscription, :stop)
+    GenServer.call(subscription, :stop, :infinity)
   end
 
   @doc false
diff --git a/lib/event_store/subscriptions/supervisor.ex b/lib/event_store/subscriptions/supervisor.ex
index 91f46f38..1f09f9f8 100644
--- a/lib/event_store/subscriptions/supervisor.ex
+++ b/lib/event_store/subscriptions/supervisor.ex
@@ -69,7 +69,8 @@ defmodule EventStore.Subscriptions.Supervisor do
 
   # 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, and has to reach the caller.
+  # 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) do
     Subscription.stop(subscription)
   catch

From 2aa0d16d2a9db5d54f0d27d075609d88864feac8 Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Tue, 22 Sep 2026 00:46:30 -0400
Subject: [PATCH 12/13] fix(subscriptions): let a checkpoint be correlated with
 the store that wrote it

A named instance reported its own name as the event store, so its checkpoints could not be lined up against any other operation it emitted.

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 guides/Usage.md                               | 12 +++---
 lib/event_store.ex                            |  1 +
 .../subscriptions/subscription_fsm.ex         | 17 +++++----
 .../subscriptions/subscription_state.ex       |  1 +
 .../subscribe_to_stream_test.exs              | 37 +++++++++++++++++++
 test/telemetry_test.exs                       | 33 +++++++++++++++++
 6 files changed, 88 insertions(+), 13 deletions(-)

diff --git a/guides/Usage.md b/guides/Usage.md
index ce93d833..9ae2d047 100644
--- a/guides/Usage.md
+++ b/guides/Usage.md
@@ -305,11 +305,13 @@ with `stream_uuid: "$all"` in the metadata, and
 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. 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, and once more
-while it terminates.
+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
diff --git a/lib/event_store.ex b/lib/event_store.ex
index d933594d..271cae8c 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,
diff --git a/lib/event_store/subscriptions/subscription_fsm.ex b/lib/event_store/subscriptions/subscription_fsm.ex
index 13affacf..441b06d2 100644
--- a/lib/event_store/subscriptions/subscription_fsm.ex
+++ b/lib/event_store/subscriptions/subscription_fsm.ex
@@ -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)
       }
@@ -731,22 +732,22 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
   defp persist_checkpoint(%SubscriptionState{transient: false} = data) do
     %SubscriptionState{
       conn: conn,
-      event_store: event_store,
       schema: schema,
       stream_uuid: stream_uuid,
       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
-      metadata = %{
-        event_store: event_store,
-        stream_uuid: stream_uuid,
-        subscription_name: subscription_name,
-        last_seen: last_ack
-      }
+      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,
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/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs
index f7735f2f..73129992 100644
--- a/test/subscriptions/subscribe_to_stream_test.exs
+++ b/test/subscriptions/subscribe_to_stream_test.exs
@@ -917,6 +917,43 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
       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 write its checkpoint while it still owns the row, not while terminating", %{
       subscription_name: subscription_name
     } do
diff --git a/test/telemetry_test.exs b/test/telemetry_test.exs
index 9f3c314f..38ff0e76 100644
--- a/test/telemetry_test.exs
+++ b/test/telemetry_test.exs
@@ -379,6 +379,39 @@ defmodule EventStore.TelemetryTest do
     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()
 

From d993b24695d638b0ede83a148490e34ad3bc20ce Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Tue, 22 Sep 2026 01:04:04 -0400
Subject: [PATCH 13/13] fix(subscriptions): let the caller of a delete decide
 how long it waits for a checkpoint

A deadline the caller cannot see or set is a deadline they cannot reason about, and the event store already takes a :timeout for exactly this.

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 lib/event_store.ex                            |  6 +--
 lib/event_store/subscriptions.ex              |  2 +-
 lib/event_store/subscriptions/subscription.ex | 13 +++++--
 lib/event_store/subscriptions/supervisor.ex   | 28 ++++++++++---
 .../subscribe_to_stream_test.exs              | 39 +++++++++++++++++++
 5 files changed, 75 insertions(+), 13 deletions(-)

diff --git a/lib/event_store.ex b/lib/event_store.ex
index 271cae8c..7c10c20e 100644
--- a/lib/event_store.ex
+++ b/lib/event_store.ex
@@ -579,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
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 c0ad4d9a..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,
@@ -87,11 +91,12 @@ defmodule EventStore.Subscriptions.Subscription do
   @doc """
   Stop the subscription, unless a subscriber is still connected to it.
 
-  Waits without a deadline, because whatever the subscription is busy with is storage work bounded
-  by its own `query_timeout` and giving up early would answer before it is safe to.
+  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) do
-    GenServer.call(subscription, :stop, :infinity)
+  def stop(subscription, opts \\ []) do
+    GenServer.call(subscription, :stop, Keyword.get(opts, :timeout, @default_timeout))
   end
 
   @doc false
diff --git a/lib/event_store/subscriptions/supervisor.ex b/lib/event_store/subscriptions/supervisor.ex
index 1f09f9f8..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
@@ -49,14 +53,21 @@ defmodule EventStore.Subscriptions.Supervisor do
         # 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) do
+        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. Unbounded, as `GenServer.stop/3` is.
+            # 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 ->
@@ -71,12 +82,19 @@ defmodule EventStore.Subscriptions.Supervisor do
   # 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) do
-    Subscription.stop(subscription)
+  defp stop(subscription, opts) do
+    Subscription.stop(subscription, opts)
   catch
     :exit, {reason, {GenServer, :call, _args}} when reason != :timeout -> :ok
   end
 
+  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
+
   @impl DynamicSupervisor
   def init(_init_arg) do
     DynamicSupervisor.init(strategy: :one_for_one)
diff --git a/test/subscriptions/subscribe_to_stream_test.exs b/test/subscriptions/subscribe_to_stream_test.exs
index 73129992..16337277 100644
--- a/test/subscriptions/subscribe_to_stream_test.exs
+++ b/test/subscriptions/subscribe_to_stream_test.exs
@@ -954,6 +954,45 @@ defmodule EventStore.Subscriptions.SubscribeToStreamTest do
       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