diff --git a/guides/Architecture.md b/guides/Architecture.md
index 49824631..d4fda6ac 100644
--- a/guides/Architecture.md
+++ b/guides/Architecture.md
@@ -13,3 +13,5 @@ The pooled connection uses `:exp` back-off. However the other connections use `:
 #### Why are these connections stopped on error?
 
 Related processes need to be notified when the connection exits or is restarted. The [postgrex](https://hexdocs.pm/postgrex/) library doesn't provide a way of being notified when the connection terminates. As an example, EventStore uses [Postgres' advisory locks](https://www.postgresql.org/docs/current/static/explicit-locking.html#ADVISORY-LOCKS) to guarantee only one instance of a subscription runs, regardless of how many nodes are running (or whether they are clustered). These advisory locks are tied to a database connection and are released when the connection terminates. When this happens, EventStore attempts to reacquire the locks. The `EventStore.MonitoredServer` module provides this process monitoring and `after_exit` and `after_restart` callback functions.
+
+Losing the lock is detected after the fact, so progress writes are also guarded by a per-subscription lock generation recorded in the `subscriptions` table. The generation moves on with each lock acquisition and a checkpoint from a previous generation is rejected, which is how a node that no longer holds the lock finds out.
diff --git a/guides/Cluster.md b/guides/Cluster.md
index 46569808..dc42e6b0 100644
--- a/guides/Cluster.md
+++ b/guides/Cluster.md
@@ -10,6 +10,10 @@ PostgreSQL's `LISTEN` / `NOTIFY` is used to pub/sub event notifications. A liste
 
 PostgreSQL's [advisory locks](https://www.postgresql.org/docs/current/static/explicit-locking.html#ADVISORY-LOCKS) are used to limit each uniquely named subscription to run at most once. This prevents multiple instances of a subscription from running on different nodes. Advisory locks are faster than table locks, are stored in memory to avoid table bloat, and are automatically cleaned up by the server at the end of the session.
 
+The lock decides which node may record the subscription's progress, not which node may run its handler. A node only learns that its lock session ended after the fact, so for a short time two nodes can both process the same events. Each time the lock is acquired the subscription row records a new lock generation, and every checkpoint is written against the generation its owner holds. A node whose lock was taken over has its checkpoints rejected, drops the events it still had in flight, and tries to subscribe again later. The node that took over resumes from the last checkpoint committed before the hand-off, and a checkpoint can never move the subscription's position backwards.
+
+Subscribers must therefore be idempotent, as they are for any at-least-once delivery. This protection takes effect once every node runs a version that records lock generations; until then, nodes on an older version behave as before.
+
 ## Automatic cluster formation
 
 You can use [libcluster](https://github.com/bitwalker/libcluster) to automatically form clusters of Erlang nodes, with either static or dynamic node membership.
diff --git a/guides/Upgrades.md b/guides/Upgrades.md
index dd8ade25..bc0675ce 100644
--- a/guides/Upgrades.md
+++ b/guides/Upgrades.md
@@ -2,6 +2,10 @@
 
 The [CHANGELOG](https://github.com/commanded/eventstore/blob/master/CHANGELOG.md) is used to indicate when a schema migration is required for a given version of the EventStore.
 
+## Migrate before rolling nodes
+
+Run the migration before starting nodes on the new version. A node on a version that expects a newer schema cannot acquire subscription locks against an unmigrated database and keeps retrying until the migration has run. Nodes on the previous version keep working against the migrated schema, so a rolling upgrade is safe once the migration is in place.
+
 ## Migrating a database with `mix`
 
 You can upgrade an existing EventStore database using the following mix task:
diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex
index 97b284d8..e5bfea68 100644
--- a/lib/event_store/advisory_locks.ex
+++ b/lib/event_store/advisory_locks.ex
@@ -18,6 +18,7 @@ defmodule EventStore.AdvisoryLocks do
     defstruct [
       :conn,
       :connection,
+      :monitor,
       :query_timeout,
       :ref,
       :schema,
@@ -28,28 +29,31 @@ defmodule EventStore.AdvisoryLocks do
 
     def new(opts) do
       conn = Keyword.fetch!(opts, :conn)
+      monitor = Keyword.fetch!(opts, :monitor)
       query_timeout = Keyword.fetch!(opts, :query_timeout)
       schema = Keyword.fetch!(opts, :schema)
 
-      %State{conn: conn, query_timeout: query_timeout, schema: schema}
+      %State{conn: conn, monitor: monitor, query_timeout: query_timeout, schema: schema}
     end
   end
 
   defmodule Lock do
     @moduledoc false
 
+    alias EventStore.Storage.Lock.Lease, as: StorageLease
     alias EventStore.Storage.Session
 
     @type t :: %Lock{
             key: non_neg_integer(),
             owner: pid(),
             ref: reference(),
-            session: Session.t()
+            session: Session.t(),
+            lease: StorageLease.t() | nil
           }
-    defstruct [:key, :owner, :ref, :session]
+    defstruct [:key, :owner, :ref, :session, :lease]
 
-    def new(key, owner, %Session{} = session) do
-      %Lock{key: key, owner: owner, ref: make_ref(), session: session}
+    def new(key, owner, %Session{} = session, lease \\ nil) do
+      %Lock{key: key, owner: owner, ref: make_ref(), session: session, lease: lease}
     end
 
     def ref(%Lock{ref: ref}), do: ref
@@ -57,6 +61,28 @@ defmodule EventStore.AdvisoryLocks do
     def held_by?(%Lock{session: session}, %Session{} = current), do: session == current
   end
 
+  defmodule Lease do
+    @moduledoc false
+
+    alias EventStore.Storage.Lock.Lease, as: StorageLease
+    alias EventStore.Storage.Session
+    alias EventStore.Subscriptions.FencingToken
+
+    @type t :: %Lease{
+            ref: reference(),
+            token: FencingToken.t(),
+            last_seen: non_neg_integer() | nil,
+            session: Session.t()
+          }
+    defstruct [:ref, :token, :last_seen, :session]
+
+    def new(ref, %StorageLease{} = lease) do
+      %StorageLease{token: token, last_seen: last_seen, session: session} = lease
+
+      %Lease{ref: ref, token: token, last_seen: last_seen, session: session}
+    end
+  end
+
   defmodule Connection do
     @moduledoc false
 
@@ -76,7 +102,8 @@ defmodule EventStore.AdvisoryLocks do
     def current?(%Connection{}, _pid), do: false
   end
 
-  alias EventStore.AdvisoryLocks.{Connection, Lock, State}
+  alias DBConnection.ConnectionError
+  alias EventStore.AdvisoryLocks.{Connection, Lease, Lock, State}
   alias EventStore.Storage
   alias EventStore.Storage.Session
 
@@ -92,13 +119,14 @@ defmodule EventStore.AdvisoryLocks do
   def init(%State{} = state) do
     Process.flag(:trap_exit, true)
 
-    %State{conn: conn} = state
+    %State{conn: conn, monitor: monitor} = state
 
-    {:ok, ref} = MonitoredServer.monitor(conn)
+    {:ok, ref} = MonitoredServer.monitor(monitor)
 
     # A connection made before this server was registered announced itself to
-    # nobody, so it is replaced by one whose notices can be received.
-    :ok = DBConnection.disconnect_all(conn, 0)
+    # nobody, so it is replaced by one whose notices can be received. A pool that
+    # is down has none, so the replacement is skipped instead of crashing startup.
+    replace_unannounced_connection(conn)
 
     {:ok, %State{state | ref: ref}}
   end
@@ -126,28 +154,138 @@ defmodule EventStore.AdvisoryLocks do
           {:ok, reference()} | {:error, :lock_already_taken} | {:error, term}
   def try_advisory_lock(server, key, timeout \\ 5_000) when is_integer(key) do
     GenServer.call(server, {:try_advisory_lock, key, self(), timeout}, timeout)
+  catch
+    :exit, {:timeout, _call} -> {:error, :timeout}
   end
 
-  def handle_call({:try_advisory_lock, key, owner, timeout}, _from, %State{} = state) do
-    case try_acquire_exclusive_lock(key, owner, timeout, state) do
-      {:ok, %Lock{} = lock} ->
+  @doc """
+  Attempt to obtain the advisory lock of a subscription, recording the new lock
+  owner on the subscription row.
+
+  Returns `{:ok, lease}` when the lock is acquired, `{:error, :lock_already_taken}`
+  if it cannot be acquired immediately, or `{:error, :subscription_not_found}`
+  when the subscription no longer exists. The lock is released and lost the same
+  way as the one returned by `try_advisory_lock/3`, identified by `lease.ref`.
+  """
+  @spec try_subscription_lock(server :: GenServer.server(), pos_integer(), timeout()) ::
+          {:ok, Lease.t()}
+          | {:error, :lock_already_taken}
+          | {:error, :subscription_not_found}
+          | {:error, term}
+  def try_subscription_lock(server, subscription_id, timeout \\ 5_000)
+      when is_integer(subscription_id) do
+    GenServer.call(server, {:try_subscription_lock, subscription_id, self(), timeout}, timeout)
+  catch
+    :exit, {:timeout, _call} -> {:error, :timeout}
+  end
+
+  @doc """
+  Release a lock acquired by `try_advisory_lock/3` or `try_subscription_lock/3`
+  before its owner terminates.
+
+  The owner is not sent a `lock_released` message. Releasing a reference that is
+  no longer held succeeds.
+
+  Returns as soon as the request is queued so a caller is never held up by a
+  lock connection that has stopped answering.
+  """
+  @spec release(server :: GenServer.server(), reference()) :: :ok
+  def release(server, ref) when is_reference(ref) do
+    GenServer.cast(server, {:release, ref})
+  end
+
+  def handle_cast({:release, ref}, %State{} = state) do
+    %State{locks: locks} = state
+
+    state =
+      case Enum.find(locks, fn {_owner_ref, lock} -> Lock.ref(lock) == ref end) do
+        nil ->
+          state
+
+        {owner_ref, %Lock{} = lock} ->
+          Process.demonitor(owner_ref, [:flush])
+
+          release_held_lock(lock, %State{state | locks: Map.delete(locks, owner_ref)})
+      end
+
+    {:noreply, state}
+  end
+
+  def handle_call(
+        {:try_subscription_lock, subscription_id, owner, timeout},
+        _from,
+        %State{} = state
+      ) do
+    case {held_lock(state, subscription_id, owner),
+          claim_subscription_lock(subscription_id, timeout, state)} do
+      {nil, {:ok, %Storage.Lock.Lease{session: session} = lease}} ->
+        lock = Lock.new(subscription_id, owner, session, lease)
         state = monitor_acquired_lock(lock, state)
 
-        {:reply, {:ok, Lock.ref(lock)}, state}
+        {:reply, {:ok, Lease.new(Lock.ref(lock), lease)}, state}
+
+      {{owner_ref, %Lock{} = held}, {:ok, %Storage.Lock.Lease{session: session} = lease}} ->
+        %Lock{key: key, ref: ref} = held
+        %State{locks: locks} = state
 
-      reply ->
+        if Lock.held_by?(held, session), do: release_lock(key, state)
+
+        lock = %Lock{held | session: session, lease: lease}
+
+        {:reply, {:ok, Lease.new(ref, lease)},
+         %State{state | locks: Map.put(locks, owner_ref, lock)}}
+
+      {_held, reply} ->
         {:reply, reply, state}
     end
   end
 
+  def handle_call({:try_advisory_lock, key, owner, timeout}, _from, %State{} = state) do
+    case held_lock(state, key, owner) do
+      {_owner_ref, %Lock{} = lock} ->
+        {:reply, {:ok, Lock.ref(lock)}, state}
+
+      nil ->
+        case try_acquire_exclusive_lock(key, owner, timeout, state) do
+          {:ok, %Lock{} = lock} ->
+            state = monitor_acquired_lock(lock, state)
+
+            {:reply, {:ok, Lock.ref(lock)}, state}
+
+          reply ->
+            {:reply, reply, state}
+        end
+    end
+  end
+
+  defp claim_subscription_lock(subscription_id, timeout, %State{} = state) do
+    %State{schema: schema} = state
+
+    query(state, fn conn ->
+      Storage.Lock.claim_subscription_lock(conn, subscription_id,
+        schema: schema,
+        timeout: timeout
+      )
+    end)
+  end
+
+  defp held_lock(%State{locks: locks}, key, owner) do
+    Enum.find(locks, fn
+      {_owner_ref, %Lock{key: ^key, owner: ^owner}} -> true
+      _other -> false
+    end)
+  end
+
   defp try_acquire_exclusive_lock(key, owner, timeout, %State{} = state) do
-    %State{conn: conn, schema: schema} = state
+    %State{schema: schema} = state
 
     with {:ok, %Session{} = session} <-
-           Storage.Lock.try_acquire_exclusive_session_lock(conn, key,
-             schema: schema,
-             timeout: timeout
-           ) do
+           query(state, fn conn ->
+             Storage.Lock.try_acquire_exclusive_session_lock(conn, key,
+               schema: schema,
+               timeout: timeout
+             )
+           end) do
       {:ok, Lock.new(key, owner, session)}
     end
   end
@@ -246,9 +384,9 @@ defmodule EventStore.AdvisoryLocks do
     do: state
 
   defp verify_sessions(%State{} = state) do
-    %State{conn: conn, locks: locks, query_timeout: query_timeout, released: released} = state
+    %State{locks: locks, query_timeout: query_timeout, released: released} = state
 
-    case Session.current(conn, timeout: query_timeout) do
+    case query(state, &Session.current(&1, timeout: query_timeout)) do
       {:ok, %Session{} = session} ->
         {held, lost} = Map.split_with(locks, fn {_ref, lock} -> Lock.held_by?(lock, session) end)
 
@@ -300,9 +438,30 @@ defmodule EventStore.AdvisoryLocks do
     :ok
   end
 
+  # An owner holds a key once, so releasing it unlocks that owner's hold and leaves
+  # the holds of every other owner in place.
+  defp release_held_lock(%Lock{} = lock, %State{released: released} = state) do
+    verify_sessions(%State{state | released: [lock | released]})
+  end
+
   defp release_lock(key, %State{} = state) do
-    %State{conn: conn, query_timeout: query_timeout, schema: schema} = state
+    %State{query_timeout: query_timeout, schema: schema} = state
+
+    query(state, &Storage.Lock.unlock(&1, key, schema: schema, timeout: query_timeout))
+  end
 
-    Storage.Lock.unlock(conn, key, schema: schema, timeout: query_timeout)
+  defp query(%State{conn: conn}, fun) do
+    fun.(conn)
+  catch
+    :exit, reason ->
+      {:error,
+       ConnectionError.exception("advisory lock connection pool exited: " <> inspect(reason))}
+  end
+
+  defp replace_unannounced_connection(conn) do
+    DBConnection.disconnect_all(conn, 0)
+    :ok
+  catch
+    :exit, _reason -> :ok
   end
 end
diff --git a/lib/event_store/sql/init.ex b/lib/event_store/sql/init.ex
index 99081d30..18f5a97b 100644
--- a/lib/event_store/sql/init.ex
+++ b/lib/event_store/sql/init.ex
@@ -233,7 +233,8 @@ defmodule EventStore.Sql.Init do
         stream_uuid text NOT NULL,
         subscription_name text NOT NULL,
         last_seen bigint NULL,
-        created_at timestamp with time zone DEFAULT NOW() NOT NULL
+        created_at timestamp with time zone DEFAULT NOW() NOT NULL,
+        lock_generation bigint DEFAULT 0 NOT NULL
     );
     """
   end
@@ -276,7 +277,7 @@ defmodule EventStore.Sql.Init do
   defp record_event_store_schema_version do
     """
     INSERT INTO schema_migrations (major_version, minor_version, patch_version)
-    VALUES (1, 3, 2);
+    VALUES (1, 5, 0);
     """
   end
 end
diff --git a/lib/event_store/sql/statements.ex b/lib/event_store/sql/statements.ex
index 6d53011d..ce2a1fdc 100644
--- a/lib/event_store/sql/statements.ex
+++ b/lib/event_store/sql/statements.ex
@@ -21,6 +21,7 @@ defmodule EventStore.Sql.Statements do
         {:try_advisory_lock, [:schema]},
         {:try_advisory_session_lock, [:schema]},
         {:advisory_unlock, [:schema]},
+        {:claim_subscription_lock, [:schema]},
         {:subscription_ack, [:schema]},
         {:insert_snapshot, [:schema]},
         {:delete_snapshot, [:schema]},
diff --git a/lib/event_store/sql/statements/claim_subscription_lock.sql.eex b/lib/event_store/sql/statements/claim_subscription_lock.sql.eex
new file mode 100644
index 00000000..798fbaea
--- /dev/null
+++ b/lib/event_store/sql/statements/claim_subscription_lock.sql.eex
@@ -0,0 +1,16 @@
+WITH lock AS (
+  SELECT pg_try_advisory_lock(
+    '"<%= schema %>".subscriptions'::regclass::oid::int,
+    (CASE WHEN $1 > 2147483647 THEN mod($1, 2147483647) ELSE $1 END)::int
+  ) AS acquired, pg_backend_pid() AS pid, COALESCE(
+    NULLIF(current_setting('eventstore.session_id', true), ''),
+    set_config('eventstore.session_id', md5(random()::text || clock_timestamp()::text), false)
+  ) AS session_id
+), lease AS (
+  UPDATE "<%= schema %>".subscriptions
+  SET lock_generation = lock_generation + 1
+  WHERE subscription_id = $1 AND (SELECT acquired FROM lock)
+  RETURNING lock_generation, last_seen
+)
+SELECT lock.acquired, lock.pid, lock.session_id, lease.lock_generation, lease.last_seen
+FROM lock LEFT JOIN lease ON true;
diff --git a/lib/event_store/sql/statements/subscription_ack.sql.eex b/lib/event_store/sql/statements/subscription_ack.sql.eex
index c6da30f1..ee3f9517 100644
--- a/lib/event_store/sql/statements/subscription_ack.sql.eex
+++ b/lib/event_store/sql/statements/subscription_ack.sql.eex
@@ -1,3 +1,8 @@
-UPDATE "<%= schema %>".subscriptions
-SET last_seen = $2
-WHERE subscription_id = $1;
+WITH progress AS (
+  UPDATE "<%= schema %>".subscriptions
+  SET last_seen = GREATEST(last_seen, $2)
+  WHERE subscription_id = $1 AND lock_generation = $3
+  RETURNING subscription_id
+)
+SELECT EXISTS (SELECT 1 FROM progress) AS acknowledged,
+       EXISTS (SELECT 1 FROM "<%= schema %>".subscriptions WHERE subscription_id = $1) AS present;
diff --git a/lib/event_store/storage.ex b/lib/event_store/storage.ex
index 668d88c2..7e95d43d 100644
--- a/lib/event_store/storage.ex
+++ b/lib/event_store/storage.ex
@@ -74,7 +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, subscription_id, last_seen, opts), to: Subscription
+  defdelegate ack_last_seen_event(conn, token, last_seen, opts), to: Subscription
 
   @doc """
   Delete an existing named subscription to a stream.
diff --git a/lib/event_store/storage/lock.ex b/lib/event_store/storage/lock.ex
index 9fd609bc..0c4301a3 100644
--- a/lib/event_store/storage/lock.ex
+++ b/lib/event_store/storage/lock.ex
@@ -2,7 +2,9 @@ defmodule EventStore.Storage.Lock do
   @moduledoc false
 
   alias EventStore.Sql.Statements
+  alias EventStore.Storage.Lock.Lease
   alias EventStore.Storage.Session
+  alias EventStore.Subscriptions.FencingToken
 
   @doc """
   Use two 32-bit key values for advisory locks where the first key acts as the
@@ -49,6 +51,40 @@ defmodule EventStore.Storage.Lock do
     end
   end
 
+  @doc """
+  Takes the advisory lock of a subscription and bumps its `lock_generation` in
+  the same statement, returning the lease that fences the subscription's
+  progress writes to this lock owner.
+  """
+  def claim_subscription_lock(conn, subscription_id, opts) do
+    {schema, opts} = Keyword.pop(opts, :schema)
+
+    query = Statements.claim_subscription_lock(schema)
+
+    case Postgrex.query(conn, query, [subscription_id], opts) do
+      {:ok, %Postgrex.Result{rows: [[false, _backend_pid, _session_id, _generation, _last_seen]]}} ->
+        {:error, :lock_already_taken}
+
+      {:ok, %Postgrex.Result{rows: [[true, _backend_pid, _session_id, nil, nil]]}} ->
+        with :ok <- unlock(conn, subscription_id, [schema: schema] ++ opts) do
+          {:error, :subscription_not_found}
+        end
+
+      {:ok, %Postgrex.Result{rows: [[true, backend_pid, session_id, generation, last_seen]]}} ->
+        token = FencingToken.new(subscription_id, generation)
+
+        {:ok, Lease.new(token, last_seen, Session.new(backend_pid, session_id))}
+
+      {:error, %Postgrex.Error{} = error} ->
+        _ = unlock(conn, subscription_id, [schema: schema] ++ opts)
+
+        {:error, error}
+
+      {:error, %DBConnection.ConnectionError{}} = reply ->
+        reply
+    end
+  end
+
   def unlock(conn, key, opts) do
     {schema, opts} = Keyword.pop(opts, :schema)
 
diff --git a/lib/event_store/storage/lock/lease.ex b/lib/event_store/storage/lock/lease.ex
new file mode 100644
index 00000000..7ecac724
--- /dev/null
+++ b/lib/event_store/storage/lock/lease.ex
@@ -0,0 +1,19 @@
+defmodule EventStore.Storage.Lock.Lease do
+  @moduledoc false
+
+  alias EventStore.Storage.Session
+  alias EventStore.Subscriptions.FencingToken
+
+  @type t :: %__MODULE__{
+          token: FencingToken.t(),
+          last_seen: non_neg_integer() | nil,
+          session: Session.t()
+        }
+
+  @enforce_keys [:token, :last_seen, :session]
+  defstruct [:token, :last_seen, :session]
+
+  def new(%FencingToken{} = token, last_seen, %Session{} = session) do
+    %__MODULE__{token: token, last_seen: last_seen, session: session}
+  end
+end
diff --git a/lib/event_store/storage/subscription.ex b/lib/event_store/storage/subscription.ex
index c609a860..2d41f1b6 100644
--- a/lib/event_store/storage/subscription.ex
+++ b/lib/event_store/storage/subscription.ex
@@ -5,6 +5,7 @@ defmodule EventStore.Storage.Subscription do
 
   alias EventStore.Sql.Statements
   alias EventStore.Storage.Subscription
+  alias EventStore.Subscriptions.FencingToken
 
   alias EventStore.Storage.Subscription.{
     CreateSubscription,
@@ -41,8 +42,8 @@ defmodule EventStore.Storage.Subscription do
     end
   end
 
-  def ack_last_seen_event(conn, subscription_id, last_seen, opts) do
-    Subscription.Ack.execute(conn, subscription_id, last_seen, opts)
+  def ack_last_seen_event(conn, %FencingToken{} = token, last_seen, opts) do
+    Subscription.Ack.execute(conn, token, last_seen, opts)
   end
 
   def delete_subscription(conn, stream_uuid, subscription_name, opts),
@@ -136,25 +137,34 @@ defmodule EventStore.Storage.Subscription do
   defmodule Ack do
     @moduledoc false
 
-    def execute(conn, subscription_id, last_seen, opts) do
+    def execute(conn, %FencingToken{} = token, last_seen, opts) do
+      %FencingToken{subscription_id: subscription_id, generation: generation} = token
+
       {schema, opts} = Keyword.pop(opts, :schema)
 
       query = Statements.subscription_ack(schema)
 
-      case Postgrex.query(conn, query, [subscription_id, last_seen], opts) do
+      case Postgrex.query(conn, query, [subscription_id, last_seen, generation], 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}} ->
+        {:ok, %Postgrex.Result{rows: [[true, _present]]}} ->
+          :ok
+
+        {:ok, %Postgrex.Result{rows: [[false, true]]}} ->
+          Logger.warning(
+            "Failed to ack last seen event for subscription #{subscription_id} because another lock owner has claimed it"
+          )
+
+          {:error, :fenced}
+
+        {:ok, %Postgrex.Result{rows: [[false, false]]}} ->
           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 for subscription #{subscription_id} due to: " <>
diff --git a/lib/event_store/subscriptions/fencing_token.ex b/lib/event_store/subscriptions/fencing_token.ex
new file mode 100644
index 00000000..b72f7d6c
--- /dev/null
+++ b/lib/event_store/subscriptions/fencing_token.ex
@@ -0,0 +1,16 @@
+defmodule EventStore.Subscriptions.FencingToken do
+  @moduledoc false
+
+  @type t :: %__MODULE__{
+          subscription_id: pos_integer(),
+          generation: pos_integer()
+        }
+
+  @enforce_keys [:subscription_id, :generation]
+  defstruct [:subscription_id, :generation]
+
+  def new(subscription_id, generation)
+      when is_integer(subscription_id) and is_integer(generation) and generation > 0 do
+    %__MODULE__{subscription_id: subscription_id, generation: generation}
+  end
+end
diff --git a/lib/event_store/subscriptions/subscription.ex b/lib/event_store/subscriptions/subscription.ex
index 08497eac..61497a3d 100644
--- a/lib/event_store/subscriptions/subscription.ex
+++ b/lib/event_store/subscriptions/subscription.ex
@@ -11,7 +11,7 @@ defmodule EventStore.Subscriptions.Subscription do
 
   require Logger
 
-  alias EventStore.RecordedEvent
+  alias EventStore.{AdvisoryLocks, RecordedEvent}
   alias EventStore.Subscriptions.{SubscriptionFsm, Subscription, SubscriptionState}
 
   # Mirrors the storage `:timeout` a subscription gives its own queries, for the internal callers
@@ -169,6 +169,27 @@ defmodule EventStore.Subscriptions.Subscription do
     {:noreply, state}
   end
 
+  @impl GenServer
+  def handle_info({:fenced, lock_ref}, %Subscription{} = state) do
+    %Subscription{
+      subscription: %SubscriptionFsm{data: %SubscriptionState{} = data} = subscription
+    } =
+      state
+
+    %SubscriptionState{event_store: event_store} = data
+
+    Logger.warning(describe(state) <> " lost its lock to another owner, disconnecting")
+
+    :ok = AdvisoryLocks.release(Module.concat(event_store, AdvisoryLocks), lock_ref)
+
+    state =
+      subscription
+      |> SubscriptionFsm.disconnect(lock_ref)
+      |> apply_subscription_to_state(state)
+
+    {:noreply, state}
+  end
+
   @impl GenServer
   def handle_info({:DOWN, _ref, :process, pid, reason}, %Subscription{} = state) do
     %Subscription{subscription: subscription} = state
diff --git a/lib/event_store/subscriptions/subscription_fsm.ex b/lib/event_store/subscriptions/subscription_fsm.ex
index b3941d46..64905061 100644
--- a/lib/event_store/subscriptions/subscription_fsm.ex
+++ b/lib/event_store/subscriptions/subscription_fsm.ex
@@ -72,10 +72,11 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
     data = SubscriptionState.reset_event_tracking(data)
 
     with {:ok, subscription} <- create_subscription(data),
-         {:ok, lock_ref} <- try_acquire_exclusive_lock(data, subscription),
+         {:ok, %AdvisoryLocks.Lease{} = lease} <-
+           try_acquire_subscription_lock(data, subscription),
          :ok <- subscribe_to_events(data) do
-      %Storage.Subscription{subscription_id: subscription_id, last_seen: last_seen} =
-        subscription
+      %Storage.Subscription{subscription_id: subscription_id} = subscription
+      %AdvisoryLocks.Lease{ref: lock_ref, token: token, last_seen: last_seen} = lease
 
       last_seen = last_seen || 0
 
@@ -83,6 +84,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
         data
         | subscription_id: subscription_id,
           lock_ref: lock_ref,
+          fencing_token: token,
           last_received: last_seen,
           last_sent: last_seen,
           last_ack: last_seen
@@ -101,18 +103,20 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
   # Attempt to subscribe
   def subscribe(%__MODULE__{state: :disconnected, data: %SubscriptionState{} = data} = fsm) do
     with {:ok, subscription} <- create_subscription(data),
-         {:ok, lock_ref} <- try_acquire_exclusive_lock(data, subscription) do
-      %Storage.Subscription{
-        subscription_id: subscription_id,
-        last_seen: last_seen
-      } = subscription
+         {:ok, %AdvisoryLocks.Lease{} = lease} <-
+           try_acquire_subscription_lock(data, subscription) do
+      %Storage.Subscription{subscription_id: subscription_id} = subscription
+      %AdvisoryLocks.Lease{ref: lock_ref, token: token, last_seen: last_seen} = lease
 
       last_ack = last_seen || 0
+      %SubscriptionState{last_received: last_received} = data
 
       data = %SubscriptionState{
         data
         | subscription_id: subscription_id,
           lock_ref: lock_ref,
+          fencing_token: token,
+          last_received: max(last_received, last_ack),
           last_sent: last_ack,
           last_ack: last_ack
       }
@@ -306,13 +310,15 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
         lock_ref
       ) do
     data =
-      %SubscriptionState{data | lock_ref: nil}
+      %SubscriptionState{data | lock_ref: nil, fencing_token: nil}
       |> SubscriptionState.reset_event_tracking()
       |> purge_in_flight_events()
 
     %__MODULE__{fsm | state: :disconnected, data: data}
   end
 
+  def disconnect(%__MODULE__{} = fsm, _lock_ref), do: fsm
+
   def unsubscribe(
         %__MODULE__{state: :unsubscribed, data: %SubscriptionState{}} = fsm,
         _subscriber
@@ -355,7 +361,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
     )
   end
 
-  defp try_acquire_exclusive_lock(
+  defp try_acquire_subscription_lock(
          %SubscriptionState{} = data,
          %Storage.Subscription{} = subscription
        ) do
@@ -364,7 +370,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
 
     server = Module.concat(event_store, AdvisoryLocks)
 
-    AdvisoryLocks.try_advisory_lock(server, subscription_id, timeout)
+    AdvisoryLocks.try_subscription_lock(server, subscription_id, timeout)
   end
 
   defp subscribe_to_events(%SubscriptionState{} = data) do
@@ -763,14 +769,15 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
       schema: schema,
       stream_uuid: stream_uuid,
       subscription_name: subscription_name,
-      subscription_id: subscription_id,
+      fencing_token: fencing_token,
+      lock_ref: lock_ref,
       last_ack: last_ack,
       query_timeout: query_timeout,
       checkpoints_pending: checkpoints_pending,
       telemetry_metadata: telemetry_metadata
     } = data
 
-    if checkpoints_pending > 0 do
+    if checkpoints_pending > 0 and fencing_token != nil do
       metadata =
         Map.merge(telemetry_metadata, %{
           stream_uuid: stream_uuid,
@@ -780,7 +787,7 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
 
       result =
         Telemetry.span(:subscription_checkpoint, metadata, fn ->
-          Storage.Subscription.ack_last_seen_event(conn, subscription_id, last_ack,
+          Storage.Subscription.ack_last_seen_event(conn, fencing_token, last_ack,
             schema: schema,
             timeout: query_timeout
           )
@@ -793,6 +800,9 @@ defmodule EventStore.Subscriptions.SubscriptionFsm do
         {:error, :subscription_not_found} ->
           send(self(), {:checkpoint_failed, :subscription_not_found})
 
+        {:error, :fenced} ->
+          send(self(), {:fenced, lock_ref})
+
         _ ->
           :ok
       end
diff --git a/lib/event_store/subscriptions/subscription_state.ex b/lib/event_store/subscriptions/subscription_state.ex
index 20ae04e1..7a0a5c70 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,
+    :fencing_token,
     telemetry_metadata: %{},
     correlation_id_type: "uuid",
     causation_id_type: "uuid",
diff --git a/lib/event_store/supervisor.ex b/lib/event_store/supervisor.ex
index d2da6dae..369a73f1 100644
--- a/lib/event_store/supervisor.ex
+++ b/lib/event_store/supervisor.ex
@@ -59,9 +59,13 @@ defmodule EventStore.Supervisor do
         advisory_locks_name = Module.concat([name, AdvisoryLocks])
         advisory_locks_postgrex_conn = Module.concat([advisory_locks_name, Postgrex])
 
+        advisory_locks_postgrex_monitor =
+          Module.concat([advisory_locks_name, Postgrex, MonitoredServer])
+
         advisory_locks_config =
           config
           |> Config.advisory_locks_postgrex_opts()
+          |> Keyword.put(:name, advisory_locks_postgrex_conn)
           |> Config.add_connection_listener(advisory_locks_name)
 
         subscriptions_name = Module.concat([name, Subscriptions.Supervisor])
@@ -81,12 +85,13 @@ defmodule EventStore.Supervisor do
             Supervisor.child_spec(
               {MonitoredServer,
                mfa: {Postgrex, :start_link, [advisory_locks_config]},
-               name: advisory_locks_postgrex_conn,
+               name: advisory_locks_postgrex_monitor,
                backoff_min: 30_000},
-              id: Module.concat([advisory_locks_postgrex_conn, MonitoredServer])
+              id: advisory_locks_postgrex_monitor
             ),
             {AdvisoryLocks,
              conn: advisory_locks_postgrex_conn,
+             monitor: advisory_locks_postgrex_monitor,
              query_timeout: query_timeout,
              schema: schema,
              name: advisory_locks_name}
diff --git a/lib/event_store/tasks/migrations.ex b/lib/event_store/tasks/migrations.ex
index 8b3dbc19..a8527611 100644
--- a/lib/event_store/tasks/migrations.ex
+++ b/lib/event_store/tasks/migrations.ex
@@ -19,7 +19,8 @@ defmodule EventStore.Tasks.Migrations do
     "1.1.0",
     "1.2.0",
     "1.3.0",
-    "1.3.2"
+    "1.3.2",
+    "1.5.0"
   ]
 
   @dialyzer {:no_return, exec: 2, handle_response: 1}
diff --git a/mix.exs b/mix.exs
index d390d60a..20a5ab06 100644
--- a/mix.exs
+++ b/mix.exs
@@ -50,7 +50,8 @@ defmodule EventStore.Mixfile do
       # Development and test tooling
       {:benchfella, "~> 0.3", only: :bench},
       {:dialyxir, "~> 1.4", only: :dev, runtime: false},
-      {:ex_doc, ">= 0.0.0", only: :dev}
+      {:ex_doc, ">= 0.0.0", only: :dev},
+      {:stream_data, "~> 1.1", only: [:test, :jsonb, :text_ids, :migration]}
     ]
   end
 
diff --git a/mix.lock b/mix.lock
index 85f2ddcb..12283e0d 100644
--- a/mix.lock
+++ b/mix.lock
@@ -15,5 +15,6 @@
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"},
   "poolboy": {:hex, :poolboy, "1.5.2", "392b007a1693a64540cead79830443abf5762f5d30cf50bc95cb2c1aaafa006b", [:rebar3], [], "hexpm", "dad79704ce5440f3d5a3681c8590b9dc25d1a561e8f5a9c995281012860901e3"},
   "postgrex": {:hex, :postgrex, "0.22.4", "d271f595dfd25230b6398354e19d17bb5e2d20130fd2d9bdca7e15f125d43552", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "4aae45a2d60e35b04eea2602440be152fae332901f1fc7a60fc7cb7f0f9a9c5a"},
+  "stream_data": {:hex, :stream_data, "1.4.0", "026f929db613aabea6208012ae9b8970d3fd5f88b3bdf26831bc536f98c42036", [:mix], [], "hexpm", "2b0ee3a340dcce1c8cf6302a763ee757d1e01c54d6e16d9069062509d68b1dc9"},
   "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"},
 }
diff --git a/priv/event_store/migrations/v1.5.0.sql b/priv/event_store/migrations/v1.5.0.sql
new file mode 100644
index 00000000..ebbddb42
--- /dev/null
+++ b/priv/event_store/migrations/v1.5.0.sql
@@ -0,0 +1,8 @@
+DO $migration$
+BEGIN
+  ALTER TABLE subscriptions ADD COLUMN lock_generation bigint NOT NULL DEFAULT 0;
+
+  -- Record schema migration
+  INSERT INTO schema_migrations (major_version, minor_version, patch_version) VALUES (1, 5, 0);
+END;
+$migration$ LANGUAGE plpgsql;
diff --git a/test/advisory_locks_test.exs b/test/advisory_locks_test.exs
index 7292fb50..f2e978bf 100644
--- a/test/advisory_locks_test.exs
+++ b/test/advisory_locks_test.exs
@@ -8,6 +8,7 @@ defmodule EventStore.AdvisoryLocksTest do
 
   @locks TestEventStore.EventStore.AdvisoryLocks
   @conn TestEventStore.EventStore.AdvisoryLocks.Postgrex
+  @monitor TestEventStore.EventStore.AdvisoryLocks.Postgrex.EventStore.MonitoredServer
 
   setup do
     postgrex_config = Config.parsed(TestEventStore, :eventstore) |> Config.default_postgrex_opts()
@@ -24,14 +25,22 @@ defmodule EventStore.AdvisoryLocksTest do
       assert is_reference(lock)
     end
 
-    test "should acquire lock when same process already has lock" do
+    test "should hand the owner the lock it already holds and free it with one release",
+         %{conn: conn, schema: schema} do
       assert {:ok, lock1} = AdvisoryLocks.try_advisory_lock(@locks, 1)
       assert {:ok, lock2} = AdvisoryLocks.try_advisory_lock(@locks, 1)
       assert {:ok, lock3} = AdvisoryLocks.try_advisory_lock(@locks, 1)
 
       assert is_reference(lock1)
-      assert is_reference(lock2)
-      assert is_reference(lock3)
+      assert lock2 == lock1
+      assert lock3 == lock1
+
+      assert :ok = AdvisoryLocks.release(@locks, lock1)
+
+      _state = :sys.get_state(@locks)
+
+      assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+      assert :ok = Storage.Lock.unlock(conn, 1, schema: schema)
     end
 
     test "should fail to acquire lock when already taken", %{conn: conn, schema: schema} do
@@ -41,6 +50,265 @@ defmodule EventStore.AdvisoryLocksTest do
     end
   end
 
+  describe "acquire subscription lock" do
+    test "should acquire the lock and record the lock owner", context do
+      %{subscription_id: subscription_id} = create_subscription(context)
+
+      assert {:ok, %AdvisoryLocks.Lease{} = lease} =
+               AdvisoryLocks.try_subscription_lock(@locks, subscription_id)
+
+      assert is_reference(lease.ref)
+      assert lease.token == EventStore.Subscriptions.FencingToken.new(subscription_id, 1)
+      assert lease.last_seen == nil
+    end
+
+    test "should record a new lock owner each time the lock is acquired", context do
+      %{subscription_id: subscription_id} = create_subscription(context)
+
+      assert {:ok, %{token: %{generation: 1}}} =
+               AdvisoryLocks.try_subscription_lock(@locks, subscription_id)
+
+      assert {:ok, %{token: %{generation: 2}}} =
+               AdvisoryLocks.try_subscription_lock(@locks, subscription_id)
+    end
+
+    test "should keep one hold when the owner claims the lock again", context do
+      %{conn: conn, schema: schema} = context
+      %{subscription_id: subscription_id} = create_subscription(context)
+
+      assert {:ok, %AdvisoryLocks.Lease{} = first} =
+               AdvisoryLocks.try_subscription_lock(@locks, subscription_id)
+
+      Postgrex.query!(
+        conn,
+        "UPDATE \"#{schema}\".subscriptions SET last_seen = 7 WHERE subscription_id = $1",
+        [subscription_id]
+      )
+
+      assert {:ok, %AdvisoryLocks.Lease{} = second} =
+               AdvisoryLocks.try_subscription_lock(@locks, subscription_id)
+
+      assert second.ref == first.ref
+      assert second.token.generation == first.token.generation + 1
+      assert second.last_seen == 7
+
+      assert :ok = AdvisoryLocks.release(@locks, first.ref)
+
+      _state = :sys.get_state(@locks)
+
+      assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, subscription_id, schema: schema)
+      assert :ok = Storage.Lock.unlock(conn, subscription_id, schema: schema)
+    end
+
+    test "should free the lock with one release after a claim outlived its caller", context do
+      %{conn: conn, schema: schema} = context
+      %{subscription_id: subscription_id} = create_subscription(context)
+
+      {:ok, busy_ref} = AdvisoryLocks.try_advisory_lock(@locks, 2)
+
+      holder = hold_only_connection()
+
+      try do
+        :ok = AdvisoryLocks.release(@locks, busy_ref)
+
+        wait_until_waiting_for_connection()
+
+        assert {:error, :timeout} =
+                 AdvisoryLocks.try_subscription_lock(@locks, subscription_id, 300)
+      after
+        send(holder, :release)
+      end
+
+      _state = :sys.get_state(@locks, 5_000)
+
+      assert {:ok, %AdvisoryLocks.Lease{ref: ref}} =
+               AdvisoryLocks.try_subscription_lock(@locks, subscription_id)
+
+      assert :ok = AdvisoryLocks.release(@locks, ref)
+
+      _state = :sys.get_state(@locks, 5_000)
+
+      assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, subscription_id, schema: schema)
+      assert :ok = Storage.Lock.unlock(conn, subscription_id, schema: schema)
+    end
+
+    test "should fail to acquire the lock when another session holds it", context do
+      %{conn: conn, schema: schema} = context
+      %{subscription_id: subscription_id} = create_subscription(context)
+
+      assert {:ok, _lease} =
+               Storage.Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+      assert {:error, :lock_already_taken} =
+               AdvisoryLocks.try_subscription_lock(@locks, subscription_id)
+    end
+
+    test "should fail to acquire the lock of a subscription that does not exist" do
+      assert {:error, :subscription_not_found} =
+               AdvisoryLocks.try_subscription_lock(@locks, 999_999)
+    end
+
+    test "should send `lock_released` message with the lease reference", context do
+      %{subscription_id: subscription_id} = create_subscription(context)
+
+      {:ok, %{ref: ref}} = AdvisoryLocks.try_subscription_lock(@locks, subscription_id)
+
+      shutdown_database_connection()
+
+      assert_receive({AdvisoryLocks, :lock_released, ^ref, :shutdown})
+    end
+
+    test "should release the lock when the owner terminates", context do
+      %{conn: conn, schema: schema} = context
+      %{subscription_id: subscription_id} = create_subscription(context)
+      reply_to = self()
+
+      pid =
+        spawn_link(fn ->
+          {:ok, _lease} = AdvisoryLocks.try_subscription_lock(@locks, subscription_id)
+
+          send(reply_to, :lock_acquired)
+
+          receive do
+            :shutdown -> :ok
+          end
+        end)
+
+      assert_receive :lock_acquired
+
+      assert {:error, :lock_already_taken} =
+               Storage.Lock.try_acquire_exclusive_lock(conn, subscription_id, schema: schema)
+
+      send(pid, :shutdown)
+
+      Wait.until(fn ->
+        assert :ok =
+                 Storage.Lock.try_acquire_exclusive_lock(conn, subscription_id, schema: schema)
+      end)
+    end
+  end
+
+  describe "release a lock by reference" do
+    test "should free the key for another session", %{conn: conn, schema: schema} do
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      assert {:error, :lock_already_taken} =
+               Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+
+      assert :ok = AdvisoryLocks.release(@locks, lock)
+
+      Wait.until(fn ->
+        assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+      end)
+    end
+
+    test "should free a subscription lock for another session", context do
+      %{conn: conn, schema: schema} = context
+      %{subscription_id: subscription_id} = create_subscription(context)
+
+      {:ok, %AdvisoryLocks.Lease{ref: ref}} =
+        AdvisoryLocks.try_subscription_lock(@locks, subscription_id)
+
+      assert {:error, :lock_already_taken} =
+               Storage.Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+      assert :ok = AdvisoryLocks.release(@locks, ref)
+
+      _ = :sys.get_state(@locks)
+
+      Wait.until(fn ->
+        assert {:ok, %{token: %{generation: 2}}} =
+                 Storage.Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+      end)
+    end
+
+    test "should keep the key held for another owner of the same key",
+         %{conn: conn, schema: schema} do
+      reply_to = self()
+
+      other_owner =
+        spawn_link(fn ->
+          {:ok, _lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+          send(reply_to, :lock_acquired)
+
+          receive do
+            :shutdown -> :ok
+          end
+        end)
+
+      assert_receive :lock_acquired
+
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      assert :ok = AdvisoryLocks.release(@locks, lock)
+
+      _state = :sys.get_state(@locks)
+
+      assert {:error, :lock_already_taken} =
+               Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+
+      send(other_owner, :shutdown)
+
+      Wait.until(fn ->
+        assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+      end)
+    end
+
+    test "should ignore a reference that is not held" do
+      assert :ok = AdvisoryLocks.release(@locks, make_ref())
+    end
+
+    test "should not notify the owner that the lock was released" do
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      assert :ok = AdvisoryLocks.release(@locks, lock)
+
+      refute_receive {AdvisoryLocks, :lock_released, ^lock, _reason}, 200
+    end
+
+    test "should not wait for the lock server to answer", %{conn: conn, schema: schema} do
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      :ok = :sys.suspend(@locks)
+
+      try do
+        {elapsed, :ok} = :timer.tc(fn -> AdvisoryLocks.release(@locks, lock) end, :millisecond)
+
+        assert elapsed < 1_000
+      after
+        :ok = :sys.resume(@locks)
+      end
+
+      _state = :sys.get_state(@locks)
+
+      assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+    end
+
+    test "should not crash a claim the busy lock server cannot answer in time", context do
+      %{subscription_id: held_id} = create_subscription(context)
+
+      {:ok, %AdvisoryLocks.Lease{ref: ref}} = AdvisoryLocks.try_subscription_lock(@locks, held_id)
+
+      other = create_subscription(context, "other_subscription")
+
+      holder = hold_only_connection()
+
+      try do
+        :ok = AdvisoryLocks.release(@locks, ref)
+
+        wait_until_waiting_for_connection()
+
+        assert {:error, :timeout} =
+                 AdvisoryLocks.try_subscription_lock(@locks, other.subscription_id, 300)
+      after
+        send(holder, :release)
+      end
+
+      _state = :sys.get_state(@locks, 5_000)
+    end
+  end
+
   describe "release lock" do
     test "should release lock when process terminates", %{conn: conn, schema: schema} do
       reply_to = self()
@@ -220,6 +488,50 @@ defmodule EventStore.AdvisoryLocksTest do
     end
   end
 
+  describe "lock pool down" do
+    test "should reply to a lock query in flight when the lock pool goes down" do
+      reply_to = self()
+
+      holder = hold_only_connection()
+
+      try do
+        spawn(fn ->
+          send(reply_to, {:lock, AdvisoryLocks.try_advisory_lock(@locks, 1, 15_000)})
+        end)
+
+        wait_until_waiting_for_connection()
+
+        Process.exit(lock_pool_pid(), :kill)
+
+        assert_receive {:lock, {:error, _error}}, 1_000
+      after
+        send(holder, :release)
+      end
+    end
+
+    test "should not wait for the lock pool to restart" do
+      Process.exit(lock_pool_pid(), :kill)
+
+      Wait.until(fn -> assert is_nil(lock_pool_pid()) end)
+
+      assert {:error, _error} = AdvisoryLocks.try_advisory_lock(@locks, 1, 1_000)
+    end
+
+    test "should keep running when the lock pool monitor does not answer" do
+      locks = Process.whereis(@locks)
+
+      :ok = :sys.suspend(@monitor)
+
+      try do
+        assert {:ok, _lock} = AdvisoryLocks.try_advisory_lock(@locks, 1, 10_000)
+      after
+        :ok = :sys.resume(@monitor)
+      end
+
+      assert Process.whereis(@locks) == locks
+    end
+  end
+
   describe "graceful shutdown" do
     test "should complete an in-flight lock query before terminating" do
       reply_to = self()
@@ -316,6 +628,7 @@ defmodule EventStore.AdvisoryLocksTest do
     test "should acquire lock and keep running" do
       locks_name = Module.concat([__MODULE__, SwitchedRole, AdvisoryLocks])
       conn_name = Module.concat([locks_name, Postgrex])
+      monitor_name = Module.concat([conn_name, MonitoredServer])
 
       config = Config.parsed(TestEventStore, :eventstore)
 
@@ -323,18 +636,20 @@ defmodule EventStore.AdvisoryLocksTest do
         config
         |> Config.advisory_locks_postgrex_opts()
         |> Config.add_connection_listener(locks_name)
+        |> Keyword.put(:name, conn_name)
         |> Keyword.put(:after_connect, {Postgrex, :query!, ["SET ROLE pg_signal_backend", []]})
 
       start_supervised!(
         {MonitoredServer,
-         mfa: {Postgrex, :start_link, [postgrex_config]}, name: conn_name, backoff_min: 30_000},
-        id: conn_name
+         mfa: {Postgrex, :start_link, [postgrex_config]}, name: monitor_name, backoff_min: 30_000},
+        id: monitor_name
       )
 
       locks =
         start_supervised!(
           {AdvisoryLocks,
            conn: conn_name,
+           monitor: monitor_name,
            query_timeout: Keyword.fetch!(config, :timeout),
            schema: "public",
            name: locks_name}
@@ -350,6 +665,7 @@ defmodule EventStore.AdvisoryLocksTest do
     test "should send `lock_released` message when the lock connection crashes" do
       locks_name = Module.concat([__MODULE__, ConnectedFirst, AdvisoryLocks])
       conn_name = Module.concat([locks_name, Postgrex])
+      monitor_name = Module.concat([conn_name, MonitoredServer])
 
       config = Config.parsed(TestEventStore, :eventstore)
 
@@ -357,11 +673,12 @@ defmodule EventStore.AdvisoryLocksTest do
         config
         |> Config.advisory_locks_postgrex_opts()
         |> Config.add_connection_listener(locks_name)
+        |> Keyword.put(:name, conn_name)
 
       start_supervised!(
         {MonitoredServer,
-         mfa: {Postgrex, :start_link, [postgrex_config]}, name: conn_name, backoff_min: 30_000},
-        id: conn_name
+         mfa: {Postgrex, :start_link, [postgrex_config]}, name: monitor_name, backoff_min: 30_000},
+        id: monitor_name
       )
 
       # The notice for this connection is sent to a lock server that does not exist yet.
@@ -370,6 +687,7 @@ defmodule EventStore.AdvisoryLocksTest do
       start_supervised!(
         {AdvisoryLocks,
          conn: conn_name,
+         monitor: monitor_name,
          query_timeout: Keyword.fetch!(config, :timeout),
          schema: "public",
          name: locks_name}
@@ -390,9 +708,111 @@ defmodule EventStore.AdvisoryLocksTest do
     end
   end
 
-  defp hold_only_connection do
-    %{pid: conn} = Process.whereis(@conn) |> :sys.get_state()
+  describe "lock pool down when the lock server starts" do
+    @tag capture_log: true
+    test "should start even when the lock pool is down" do
+      locks_name = Module.concat([__MODULE__, PoolDownAtStart, AdvisoryLocks])
+      conn_name = Module.concat([locks_name, Postgrex])
+      monitor_name = Module.concat([conn_name, MonitoredServer])
 
+      config = Config.parsed(TestEventStore, :eventstore)
+
+      postgrex_config =
+        config
+        |> Config.advisory_locks_postgrex_opts()
+        |> Config.add_connection_listener(locks_name)
+        |> Keyword.put(:name, conn_name)
+
+      start_supervised!(
+        {MonitoredServer,
+         mfa: {Postgrex, :start_link, [postgrex_config]}, name: monitor_name, backoff_min: 30_000},
+        id: monitor_name
+      )
+
+      Wait.until(fn -> assert is_pid(Process.whereis(conn_name)) end)
+
+      # The pool is down and its long backoff keeps the name unregistered while the
+      # lock server starts, so the startup disconnect has nothing to call.
+      Process.exit(Process.whereis(conn_name), :kill)
+      Wait.until(fn -> assert is_nil(Process.whereis(conn_name)) end)
+
+      assert {:ok, locks} =
+               start_supervised(
+                 {AdvisoryLocks,
+                  conn: conn_name,
+                  monitor: monitor_name,
+                  query_timeout: Keyword.fetch!(config, :timeout),
+                  schema: "public",
+                  name: locks_name}
+               )
+
+      assert Process.alive?(locks)
+    end
+  end
+
+  describe "release a lock while the lock connection cannot answer" do
+    @tag capture_log: true
+    test "should retry the unlock until the lock is released", %{conn: conn, schema: schema} do
+      alias EventStore.Storage.Session
+
+      locks_name = Module.concat([__MODULE__, ReleaseWithoutSession, AdvisoryLocks])
+      conn_name = Module.concat([locks_name, Postgrex])
+      monitor_name = Module.concat([conn_name, MonitoredServer])
+
+      config = Config.parsed(TestEventStore, :eventstore)
+
+      postgrex_config =
+        config
+        |> Config.advisory_locks_postgrex_opts()
+        |> Config.add_connection_listener(locks_name)
+        |> Keyword.put(:name, conn_name)
+        |> Keyword.merge(queue_target: 1, queue_interval: 1)
+
+      start_supervised!(
+        {MonitoredServer,
+         mfa: {Postgrex, :start_link, [postgrex_config]}, name: monitor_name, backoff_min: 30_000},
+        id: monitor_name
+      )
+
+      start_supervised!(
+        {AdvisoryLocks,
+         conn: conn_name,
+         monitor: monitor_name,
+         query_timeout: 100,
+         schema: schema,
+         name: locks_name}
+      )
+
+      # The startup replacement of the pre-registration connection settles once a
+      # session answers, so the lock rides a connection that survives.
+      Wait.until(5_000, fn ->
+        assert {:ok, _session} = Session.current(conn_name, timeout: 100)
+      end)
+
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(locks_name, 1)
+
+      assert {:error, :lock_already_taken} =
+               Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+
+      # The only pooled connection is busy, so the release cannot confirm its
+      # session and parks the unlock instead of completing it.
+      holder = hold_only_connection(conn_name)
+
+      try do
+        :ok = AdvisoryLocks.release(locks_name, lock)
+
+        Wait.until(fn -> refute Enum.empty?(:sys.get_state(locks_name).released) end)
+      after
+        send(holder, :release)
+      end
+
+      Wait.until(5_000, fn ->
+        assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+      end)
+    end
+  end
+
+  defp hold_only_connection(conn \\ lock_pool_pid()) do
     reply_to = self()
 
     holder =
@@ -485,8 +905,15 @@ defmodule EventStore.AdvisoryLocksTest do
   end
 
   defp shutdown_database_connection do
-    %{pid: pid} = Process.whereis(@conn) |> :sys.get_state()
+    EventStore.ProcessHelper.shutdown(lock_pool_pid())
+  end
+
+  defp lock_pool_pid, do: Process.whereis(@conn)
+
+  defp create_subscription(%{conn: conn, schema: schema}, name \\ "test_subscription") do
+    {:ok, %Storage.Subscription{} = subscription} =
+      Storage.subscribe_to_stream(conn, "$all", name, schema: schema)
 
-    EventStore.ProcessHelper.shutdown(pid)
+    subscription
   end
 end
diff --git a/test/list_event_store_migrations_test.exs b/test/list_event_store_migrations_test.exs
index b0ac0aee..f45bab7d 100644
--- a/test/list_event_store_migrations_test.exs
+++ b/test/list_event_store_migrations_test.exs
@@ -26,6 +26,7 @@ defmodule EventStore.ListEventStoreMigrationsTest do
         1.2.0     \tpending
         1.3.0     \tpending
         1.3.2     \tpending
+        1.5.0     \tpending
 
       """
 
diff --git a/test/migrate_event_store_test.exs b/test/migrate_event_store_test.exs
index 31588b0c..f958a109 100644
--- a/test/migrate_event_store_test.exs
+++ b/test/migrate_event_store_test.exs
@@ -88,6 +88,7 @@ defmodule EventStore.MigrateEventStoreTest do
       "1\t1\t0\tNOW" -> true
       "1\t2\t0\tNOW" -> true
       "1\t3\t0\tNOW" -> true
+      "1\t3\t2\tNOW" -> true
       _line -> false
     end)
   end
diff --git a/test/migrated_event_store_test.exs b/test/migrated_event_store_test.exs
index fa7c7a4e..45081c6f 100644
--- a/test/migrated_event_store_test.exs
+++ b/test/migrated_event_store_test.exs
@@ -124,3 +124,58 @@ defmodule EventStore.MigratedEventStoreTest do
     EventStore.Tasks.Migrate.exec(config, quiet: true)
   end
 end
+
+defmodule EventStore.MigratedSubscriptionLockTest do
+  use ExUnit.Case
+
+  alias EventStore.Config
+  alias EventStore.Storage
+  alias EventStore.Subscriptions.FencingToken
+
+  @moduletag :migration
+
+  setup_all do
+    config = TestEventStore.config()
+
+    EventStore.Storage.Database.drop(config)
+    :ok = EventStore.Storage.Database.create(config)
+    {_, 0} = EventStore.Storage.Database.restore(config, "test/fixture/eventstore_seed.dump")
+    EventStore.Tasks.Migrate.exec(config, quiet: true)
+
+    postgrex_config =
+      TestEventStore |> Config.parsed(:eventstore) |> Config.default_postgrex_opts()
+
+    [conn: start_supervised!({Postgrex, postgrex_config})]
+  end
+
+  test "claims the lock of a subscription that existed before the migration", %{conn: conn} do
+    assert [0, 0, 0, 0] == lock_generations(conn)
+
+    assert {:ok, lease} = Storage.Lock.claim_subscription_lock(conn, 2, schema: "public")
+
+    assert lease.last_seen == 100
+    assert lease.token == FencingToken.new(2, 1)
+    assert [0, 1, 0, 0] == lock_generations(conn)
+
+    assert :ok =
+             Storage.Subscription.ack_last_seen_event(conn, lease.token, 101, schema: "public")
+
+    assert %Postgrex.Result{rows: [[101]]} =
+             Postgrex.query!(
+               conn,
+               "SELECT last_seen FROM public.subscriptions WHERE subscription_id = 2",
+               []
+             )
+  end
+
+  defp lock_generations(conn) do
+    %Postgrex.Result{rows: rows} =
+      Postgrex.query!(
+        conn,
+        "SELECT lock_generation FROM public.subscriptions ORDER BY subscription_id",
+        []
+      )
+
+    List.flatten(rows)
+  end
+end
diff --git a/test/storage/subscription_fencing_property_test.exs b/test/storage/subscription_fencing_property_test.exs
new file mode 100644
index 00000000..36dbe94b
--- /dev/null
+++ b/test/storage/subscription_fencing_property_test.exs
@@ -0,0 +1,171 @@
+defmodule EventStore.Storage.SubscriptionFencingPropertyTest do
+  use EventStore.StorageCase
+  use ExUnitProperties
+
+  alias EventStore.{Storage, UUID, Wait}
+  alias EventStore.Storage.Lock
+
+  @sessions [:one, :two, :three]
+
+  defmodule Model do
+    defstruct owner: nil, generation: 0, last_seen: nil, tokens: %{}, connections: %{}
+  end
+
+  property "progress only moves forward and only for the latest lock owner", context do
+    check all(commands <- list_of(command(), max_length: 12), max_runs: 30) do
+      subscription_id = create_subscription(context)
+      model = start_sessions(%Model{}, context)
+
+      try do
+        Enum.reduce(commands, model, fn command, model ->
+          model = execute(command, model, subscription_id, context)
+
+          assert {model.generation, model.last_seen} == read(subscription_id, context)
+
+          model
+        end)
+      after
+        delete_subscription(subscription_id, context)
+        stop_sessions(model)
+      end
+    end
+  end
+
+  defp command do
+    session = member_of(@sessions)
+
+    one_of([
+      tuple({constant(:claim), session}),
+      tuple({constant(:ack), session, integer(0..50)}),
+      tuple({constant(:drop), session}),
+      tuple({constant(:take_over), session})
+    ])
+  end
+
+  defp execute({:take_over, session}, %Model{owner: owner} = model, subscription_id, context)
+       when owner not in [nil, session] do
+    model = execute({:drop, owner}, model, subscription_id, context)
+
+    execute({:claim, session}, model, subscription_id, context)
+  end
+
+  defp execute({:take_over, session}, %Model{} = model, subscription_id, context) do
+    execute({:claim, session}, model, subscription_id, context)
+  end
+
+  defp execute({:claim, session}, %Model{owner: owner} = model, subscription_id, context)
+       when owner not in [nil, session] do
+    assert {:error, :lock_already_taken} = claim(model, session, subscription_id, context)
+
+    model
+  end
+
+  defp execute({:claim, session}, %Model{} = model, subscription_id, context) do
+    assert {:ok, lease} = claim(model, session, subscription_id, context)
+
+    assert lease.last_seen == model.last_seen
+    assert lease.token.generation == model.generation + 1
+
+    %Model{
+      model
+      | owner: session,
+        generation: model.generation + 1,
+        tokens: Map.put(model.tokens, session, lease.token)
+    }
+  end
+
+  defp execute({:ack, session, position}, %Model{} = model, _subscription_id, context) do
+    case Map.get(model.tokens, session) do
+      nil ->
+        model
+
+      %{generation: generation} = token when generation == model.generation ->
+        assert :ok = ack(model, session, token, position, context)
+
+        %Model{model | last_seen: max(model.last_seen || position, position)}
+
+      token ->
+        assert {:error, :fenced} = ack(model, session, token, position, context)
+
+        model
+    end
+  end
+
+  defp execute({:drop, session}, %Model{} = model, _subscription_id, context) do
+    model = end_session(model, session, context)
+
+    %Model{model | owner: if(model.owner == session, do: nil, else: model.owner)}
+  end
+
+  defp claim(model, session, subscription_id, %{schema: schema}) do
+    Lock.claim_subscription_lock(connection(model, session), subscription_id, schema: schema)
+  end
+
+  defp ack(model, session, token, position, %{schema: schema}) do
+    Storage.ack_last_seen_event(connection(model, session), token, position, schema: schema)
+  end
+
+  defp stop_sessions(%Model{connections: connections}) do
+    Enum.each(connections, fn {_session, connection} -> GenServer.stop(connection) end)
+  end
+
+  defp connection(%Model{connections: connections}, session), do: Map.fetch!(connections, session)
+
+  defp start_sessions(%Model{} = model, %{postgrex_config: postgrex_config}) do
+    connections =
+      Map.new(@sessions, fn session ->
+        {:ok, connection} = Postgrex.start_link(postgrex_config)
+
+        {session, connection}
+      end)
+
+    %Model{model | connections: connections}
+  end
+
+  defp end_session(%Model{} = model, session, %{conn: conn}) do
+    connection = connection(model, session)
+
+    %Postgrex.Result{rows: [[backend_pid]]} =
+      Postgrex.query!(connection, "SELECT pg_backend_pid()", [])
+
+    %Postgrex.Result{rows: [[true]]} =
+      Postgrex.query!(conn, "SELECT pg_terminate_backend($1)", [backend_pid])
+
+    Wait.until(fn ->
+      assert %Postgrex.Result{rows: [[0]]} =
+               Postgrex.query!(
+                 conn,
+                 "SELECT count(*)::int FROM pg_stat_activity WHERE pid = $1",
+                 [backend_pid]
+               )
+    end)
+
+    Wait.until(fn -> assert {:ok, _result} = Postgrex.query(connection, "SELECT 1", []) end)
+
+    model
+  end
+
+  defp create_subscription(%{conn: conn, schema: schema}) do
+    {:ok, %Storage.Subscription{subscription_id: subscription_id}} =
+      Storage.subscribe_to_stream(conn, "$all", UUID.uuid4(), schema: schema)
+
+    subscription_id
+  end
+
+  defp delete_subscription(subscription_id, %{conn: conn}) do
+    Postgrex.query!(conn, "DELETE FROM public.subscriptions WHERE subscription_id = $1", [
+      subscription_id
+    ])
+  end
+
+  defp read(subscription_id, %{conn: conn}) do
+    %Postgrex.Result{rows: [[generation, last_seen]]} =
+      Postgrex.query!(
+        conn,
+        "SELECT lock_generation, last_seen FROM public.subscriptions WHERE subscription_id = $1",
+        [subscription_id]
+      )
+
+    {generation, last_seen}
+  end
+end
diff --git a/test/storage/subscription_lock_test.exs b/test/storage/subscription_lock_test.exs
new file mode 100644
index 00000000..36c78d74
--- /dev/null
+++ b/test/storage/subscription_lock_test.exs
@@ -0,0 +1,210 @@
+defmodule EventStore.Storage.SubscriptionLockTest do
+  use EventStore.StorageCase
+
+  alias EventStore.ProcessHelper
+  alias EventStore.Storage
+  alias EventStore.Storage.Lock
+  alias EventStore.Storage.Session
+  alias EventStore.Subscriptions.FencingToken
+
+  @all_stream "$all"
+  @subscription_name "test_subscription"
+
+  setup %{postgrex_config: postgrex_config} do
+    [conn: start_supervised!({Postgrex, postgrex_config})]
+  end
+
+  test "a new subscription has never had its lock claimed", context do
+    %{subscription_id: subscription_id} = create_subscription(context)
+
+    assert 0 == lock_generation(context, subscription_id)
+  end
+
+  test "claim the lock of a new subscription", context do
+    %{conn: conn, schema: schema} = context
+    %{subscription_id: subscription_id} = create_subscription(context)
+
+    assert {:ok, lease} =
+             Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+    assert is_struct(lease, Lock.Lease)
+    assert lease.token == FencingToken.new(subscription_id, 1)
+    assert lease.last_seen == nil
+    assert %Session{} = lease.session
+    assert 1 == lock_generation(context, subscription_id)
+  end
+
+  test "claim the lock resumes from the committed position", context do
+    %{conn: conn, schema: schema} = context
+    %{subscription_id: subscription_id} = create_subscription(context)
+
+    {:ok, %{token: token}} = Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+    :ok = Storage.ack_last_seen_event(conn, token, 3, schema: schema)
+
+    assert {:ok, %{last_seen: 3}} =
+             Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+  end
+
+  test "claim the lock again from the same connection", context do
+    %{conn: conn, schema: schema} = context
+    %{subscription_id: subscription_id} = create_subscription(context)
+
+    assert {:ok, %{token: %{generation: 1}}} =
+             Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+    assert {:ok, %{token: %{generation: 2}}} =
+             Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+  end
+
+  test "claim the lock while another connection holds it", context do
+    %{postgrex_config: postgrex_config, conn: conn, schema: schema} = context
+    %{subscription_id: subscription_id} = create_subscription(context)
+
+    {:ok, owner} = Postgrex.start_link(postgrex_config)
+
+    assert {:ok, _lease} =
+             Lock.claim_subscription_lock(owner, subscription_id, schema: schema)
+
+    assert {:error, :lock_already_taken} =
+             Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+    assert 1 == lock_generation(context, subscription_id)
+
+    ProcessHelper.shutdown(owner)
+  end
+
+  test "claim the lock of a subscription that no longer exists", context do
+    %{postgrex_config: postgrex_config, conn: conn, schema: schema} = context
+    %{subscription_id: subscription_id} = create_subscription(context)
+
+    :ok = Storage.delete_subscription(conn, @all_stream, @subscription_name, schema: schema)
+
+    assert {:error, :subscription_not_found} =
+             Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+    {:ok, other} = Postgrex.start_link(postgrex_config)
+
+    assert :ok = Lock.try_acquire_exclusive_lock(other, subscription_id, schema: schema)
+
+    ProcessHelper.shutdown(other)
+  end
+
+  test "a claim that fails after taking the advisory lock leaves the subscription claimable",
+       context do
+    %{postgrex_config: postgrex_config, conn: conn, schema: schema} = context
+    %{subscription_id: subscription_id} = create_subscription(context)
+
+    claiming_conn =
+      start_supervised!(
+        {Postgrex,
+         Keyword.merge(postgrex_config, parameters: [lock_timeout: "200ms"], pool_size: 1)},
+        id: :claiming_conn
+      )
+
+    test_pid = self()
+
+    holder =
+      Task.async(fn ->
+        Postgrex.transaction(conn, fn holder_conn ->
+          Postgrex.query!(
+            holder_conn,
+            "SELECT subscription_id FROM \"#{schema}\".subscriptions WHERE subscription_id = $1 FOR UPDATE",
+            [subscription_id]
+          )
+
+          send(test_pid, :row_locked)
+
+          receive do
+            :release -> :ok
+          end
+        end)
+      end)
+
+    assert_receive :row_locked
+
+    assert {:error, %Postgrex.Error{postgres: %{code: :lock_not_available}}} =
+             Lock.claim_subscription_lock(claiming_conn, subscription_id, schema: schema)
+
+    send(holder.pid, :release)
+    assert {:ok, :ok} = Task.await(holder)
+
+    assert {:ok, %Lock.Lease{}} =
+             Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+    assert :ok = Lock.unlock(conn, subscription_id, schema: schema)
+  end
+
+  describe "ack with a fencing token" do
+    test "a stale generation is fenced and leaves the position alone", context do
+      %{conn: conn, schema: schema} = context
+      %{subscription_id: subscription_id} = create_subscription(context)
+
+      {:ok, %{token: stale}} =
+        Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+      {:ok, %{token: current}} =
+        Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+      assert :ok = Storage.ack_last_seen_event(conn, current, 3, schema: schema)
+
+      assert {:error, :fenced} = Storage.ack_last_seen_event(conn, stale, 7, schema: schema)
+
+      assert 3 == last_seen(context, subscription_id)
+    end
+
+    test "the current generation is accepted", context do
+      %{conn: conn, schema: schema} = context
+      %{subscription_id: subscription_id} = create_subscription(context)
+
+      {:ok, %{token: token}} =
+        Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+      assert :ok = Storage.ack_last_seen_event(conn, token, 4, schema: schema)
+
+      assert 4 == last_seen(context, subscription_id)
+    end
+
+    test "a subscription that no longer exists is not found rather than fenced", context do
+      %{conn: conn, schema: schema} = context
+      %{subscription_id: subscription_id} = create_subscription(context)
+
+      {:ok, %{token: token}} =
+        Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+      :ok = Storage.delete_subscription(conn, @all_stream, @subscription_name, schema: schema)
+
+      assert {:error, :subscription_not_found} =
+               Storage.ack_last_seen_event(conn, token, 1, schema: schema)
+    end
+  end
+
+  defp create_subscription(%{conn: conn, schema: schema}) do
+    {:ok, %Storage.Subscription{} = subscription} =
+      Storage.subscribe_to_stream(conn, @all_stream, @subscription_name, schema: schema)
+
+    subscription
+  end
+
+  defp lock_generation(%{conn: conn, schema: schema}, subscription_id) do
+    %Postgrex.Result{rows: [[generation]]} =
+      Postgrex.query!(
+        conn,
+        "SELECT lock_generation FROM \"#{schema}\".subscriptions WHERE subscription_id = $1",
+        [subscription_id]
+      )
+
+    generation
+  end
+
+  defp last_seen(%{conn: conn, schema: schema}, subscription_id) do
+    %Postgrex.Result{rows: [[last_seen]]} =
+      Postgrex.query!(
+        conn,
+        "SELECT last_seen FROM \"#{schema}\".subscriptions WHERE subscription_id = $1",
+        [subscription_id]
+      )
+
+    last_seen
+  end
+end
diff --git a/test/storage/subscription_persistence_test.exs b/test/storage/subscription_persistence_test.exs
index fd7ce5f5..55774913 100644
--- a/test/storage/subscription_persistence_test.exs
+++ b/test/storage/subscription_persistence_test.exs
@@ -2,6 +2,8 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
   use EventStore.StorageCase
 
   alias EventStore.Storage
+  alias EventStore.Storage.Lock
+  alias EventStore.Subscriptions.FencingToken
 
   @all_stream "$all"
   @subscription_name "test_subscription"
@@ -74,10 +76,12 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
 
     {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context)
 
+    token = claim_token(context, subscription_id)
+
     :ok = delete_subscription(context)
 
     assert {:error, :subscription_not_found} =
-             Storage.ack_last_seen_event(conn, subscription_id, 1, schema: schema)
+             Storage.ack_last_seen_event(conn, token, 1, schema: schema)
   end
 
   test "ack last seen event only moves the subscription it names", context do
@@ -88,7 +92,7 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
     {:ok, %Storage.Subscription{subscription_id: untouched}} =
       Storage.subscribe_to_stream(conn, @all_stream, "another_subscription", schema: schema)
 
-    :ok = Storage.ack_last_seen_event(conn, acked, 1, schema: schema)
+    :ok = Storage.ack_last_seen_event(conn, claim_token(context, acked), 1, schema: schema)
 
     assert {:ok, %Storage.Subscription{subscription_id: ^acked, last_seen: 1}} =
              Storage.Subscription.subscription(conn, @all_stream, @subscription_name,
@@ -106,6 +110,8 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
 
     {:ok, %Storage.Subscription{subscription_id: deleted}} = subscribe_to_stream(context)
 
+    deleted_token = claim_token(context, deleted)
+
     :ok = delete_subscription(context)
 
     {:ok, %Storage.Subscription{subscription_id: recreated}} = subscribe_to_stream(context)
@@ -113,7 +119,7 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
     assert recreated != deleted
 
     assert {:error, :subscription_not_found} =
-             Storage.ack_last_seen_event(conn, deleted, 1, schema: schema)
+             Storage.ack_last_seen_event(conn, deleted_token, 1, schema: schema)
 
     # The stream and name are the same as the deleted subscription's, so anything keyed by those
     # would have credited this subscription with a position it never acknowledged.
@@ -130,7 +136,9 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
     {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context)
 
     assert {:error, %Postgrex.Error{}} =
-             Storage.ack_last_seen_event(conn, subscription_id, 1, schema: "no_such_schema")
+             Storage.ack_last_seen_event(conn, claim_token(context, subscription_id), 1,
+               schema: "no_such_schema"
+             )
   end
 
   test "ack last seen event for a subscription that never existed", context do
@@ -139,7 +147,9 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
     {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context)
 
     assert {:error, :subscription_not_found} =
-             Storage.ack_last_seen_event(conn, subscription_id + 1_000, 1, schema: schema)
+             Storage.ack_last_seen_event(conn, FencingToken.new(subscription_id + 1_000, 1), 1,
+               schema: schema
+             )
 
     assert {:ok, %Storage.Subscription{last_seen: nil}} = read_subscription(context)
   end
@@ -149,11 +159,13 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
 
     {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context)
 
-    assert :ok = Storage.ack_last_seen_event(conn, subscription_id, 3, schema: schema)
+    token = claim_token(context, subscription_id)
+
+    assert :ok = Storage.ack_last_seen_event(conn, token, 3, schema: schema)
 
     # Postgres counts the rows it matched rather than the rows it changed, so an acknowledgement
     # that moves nothing is still an acknowledgement its row accepted.
-    assert :ok = Storage.ack_last_seen_event(conn, subscription_id, 3, schema: schema)
+    assert :ok = Storage.ack_last_seen_event(conn, token, 3, schema: schema)
 
     assert {:ok, %Storage.Subscription{last_seen: 3}} = read_subscription(context)
   end
@@ -163,7 +175,9 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
 
     {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context)
 
-    assert :ok = Storage.ack_last_seen_event(conn, subscription_id, 0, schema: schema)
+    token = claim_token(context, subscription_id)
+
+    assert :ok = Storage.ack_last_seen_event(conn, token, 0, schema: schema)
 
     # Zero is a position that was acknowledged, where null is a subscription that never has.
     assert {:ok, %Storage.Subscription{last_seen: 0}} = read_subscription(context)
@@ -174,25 +188,26 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
 
     {:ok, %Storage.Subscription{} = before} = subscribe_to_stream(context)
 
-    :ok = Storage.ack_last_seen_event(conn, before.subscription_id, 3, schema: schema)
+    token = claim_token(context, before.subscription_id)
+
+    :ok = Storage.ack_last_seen_event(conn, token, 3, schema: schema)
 
     assert {:ok, %Storage.Subscription{} = after_ack} = read_subscription(context)
 
     assert after_ack == %Storage.Subscription{before | last_seen: 3}
   end
 
-  test "ack last seen event does not guard against a position going backwards", context do
+  test "ack last seen event late does not roll the position back", context do
     %{conn: conn, schema: schema} = context
 
     {:ok, %Storage.Subscription{subscription_id: subscription_id}} = subscribe_to_stream(context)
 
-    :ok = Storage.ack_last_seen_event(conn, subscription_id, 5, schema: schema)
-    :ok = Storage.ack_last_seen_event(conn, subscription_id, 2, schema: schema)
+    token = claim_token(context, subscription_id)
+
+    :ok = Storage.ack_last_seen_event(conn, token, 10, schema: schema)
+    :ok = Storage.ack_last_seen_event(conn, token, 5, schema: schema)
 
-    # Storage takes the position it is given. Nothing reaches here out of order, because the one
-    # process that owns a row acknowledges in order, and refusing the write would be indistinguish
-    # able from the row being gone.
-    assert {:ok, %Storage.Subscription{last_seen: 2}} = read_subscription(context)
+    assert {:ok, %Storage.Subscription{last_seen: 10}} = read_subscription(context)
   end
 
   defp read_subscription(context) do
@@ -207,7 +222,16 @@ defmodule EventStore.Storage.SubscriptionPersistenceTest do
     {:ok, %Storage.Subscription{subscription_id: subscription_id}} =
       Storage.Subscription.subscription(conn, @all_stream, @subscription_name, schema: schema)
 
-    Storage.ack_last_seen_event(conn, subscription_id, last_seen, schema: schema)
+    token = claim_token(context, subscription_id)
+
+    Storage.ack_last_seen_event(conn, token, last_seen, schema: schema)
+  end
+
+  defp claim_token(%{conn: conn, schema: schema}, subscription_id) do
+    {:ok, %Lock.Lease{token: token}} =
+      Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+    token
   end
 
   defp subscribe_to_stream(context) do
diff --git a/test/storage/subscription_previous_version_test.exs b/test/storage/subscription_previous_version_test.exs
new file mode 100644
index 00000000..57ad2a74
--- /dev/null
+++ b/test/storage/subscription_previous_version_test.exs
@@ -0,0 +1,91 @@
+defmodule EventStore.Storage.SubscriptionPreviousVersionTest do
+  use EventStore.StorageCase
+
+  alias EventStore.ProcessHelper
+  alias EventStore.Storage.Lock
+  alias EventStore.Subscriptions.FencingToken
+
+  @insert_subscription """
+  INSERT INTO "public".subscriptions
+    (
+      stream_uuid,
+      subscription_name,
+      last_seen
+    )
+  VALUES ($1, $2, $3)
+  RETURNING subscription_id, stream_uuid, subscription_name, last_seen, created_at;
+  """
+
+  @try_advisory_lock """
+  SELECT pg_try_advisory_lock(
+    '"public".subscriptions'::regclass::oid::int,
+    (CASE WHEN $1 > 2147483647 THEN mod($1, 2147483647) ELSE $1 END)::int
+  );
+  """
+
+  @subscription_ack """
+  UPDATE "public".subscriptions
+  SET last_seen = $2
+  WHERE subscription_id = $1;
+  """
+
+  @query_subscription """
+  SELECT
+    subscription_id,
+    stream_uuid,
+    subscription_name,
+    last_seen,
+    created_at
+  FROM "public".subscriptions
+  WHERE stream_uuid = $1 AND subscription_name = $2;
+  """
+
+  setup %{postgrex_config: postgrex_config} do
+    {:ok, previous_node} = Postgrex.start_link(postgrex_config)
+
+    on_exit(fn -> ProcessHelper.shutdown(previous_node) end)
+
+    [previous_node: previous_node]
+  end
+
+  test "a node on the previous version keeps working against the migrated schema", context do
+    %{conn: conn, previous_node: previous_node, schema: schema} = context
+
+    %Postgrex.Result{rows: [[subscription_id | _columns]]} =
+      Postgrex.query!(previous_node, @insert_subscription, ["$all", "previous_version", nil])
+
+    assert 0 == lock_generation(conn, subscription_id)
+
+    assert %Postgrex.Result{rows: [[true]]} =
+             Postgrex.query!(previous_node, @try_advisory_lock, [subscription_id])
+
+    Postgrex.query!(previous_node, @subscription_ack, [subscription_id, 7])
+
+    assert 0 == lock_generation(conn, subscription_id)
+
+    assert %Postgrex.Result{
+             rows: [[^subscription_id, "$all", "previous_version", 7, _created_at]]
+           } =
+             Postgrex.query!(previous_node, @query_subscription, ["$all", "previous_version"])
+
+    assert {:error, :lock_already_taken} =
+             Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+    :ok = Lock.unlock(previous_node, subscription_id, schema: schema)
+
+    assert {:ok, lease} = Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+    assert lease.token == FencingToken.new(subscription_id, 1)
+    assert lease.last_seen == 7
+  end
+
+  defp lock_generation(conn, subscription_id) do
+    %Postgrex.Result{rows: [[generation]]} =
+      Postgrex.query!(
+        conn,
+        "SELECT lock_generation FROM public.subscriptions WHERE subscription_id = $1",
+        [subscription_id]
+      )
+
+    generation
+  end
+end
diff --git a/test/subscriptions/subscription_churn_property_test.exs b/test/subscriptions/subscription_churn_property_test.exs
new file mode 100644
index 00000000..c45f81ba
--- /dev/null
+++ b/test/subscriptions/subscription_churn_property_test.exs
@@ -0,0 +1,332 @@
+defmodule EventStore.Subscriptions.SubscriptionChurnPropertyTest do
+  use EventStore.StorageCase
+  use ExUnitProperties
+
+  alias EventStore.{EventFactory, UUID, Wait}
+  alias EventStore.Subscriptions.Subscription
+
+  @moduletag :slow
+  @moduletag timeout: 600_000
+
+  @nodes [1, 2, 3]
+
+  defmodule Subscriber do
+    @moduledoc false
+
+    use GenServer
+
+    def start(), do: GenServer.start(__MODULE__, nil)
+
+    def seen(subscriber), do: GenServer.call(subscriber, :seen)
+    def subscription(subscriber), do: GenServer.call(subscriber, :subscription)
+
+    def ack(subscriber, event_number),
+      do: GenServer.call(subscriber, {:ack, event_number}, 20_000)
+
+    @impl GenServer
+    def init(_), do: {:ok, %{subscription: nil, seen: MapSet.new()}}
+
+    @impl GenServer
+    def handle_info({:subscribed, subscription}, state),
+      do: {:noreply, %{state | subscription: subscription}}
+
+    def handle_info({:events, events}, state) do
+      seen = Enum.reduce(events, state.seen, &MapSet.put(&2, &1.event_number))
+
+      {:noreply, %{state | seen: seen}}
+    end
+
+    @impl GenServer
+    def handle_call(:seen, _from, state),
+      do: {:reply, state.seen |> MapSet.to_list() |> Enum.sort(:desc), state}
+
+    def handle_call(:subscription, _from, state), do: {:reply, state.subscription, state}
+
+    def handle_call({:ack, _event_number}, _from, %{subscription: nil} = state),
+      do: {:reply, {:error, :not_subscribed}, state}
+
+    def handle_call({:ack, event_number}, _from, state),
+      do: {:reply, Subscription.ack(state.subscription, event_number, self()), state}
+  end
+
+  defmodule Model do
+    @moduledoc false
+
+    defstruct subscription_name: nil,
+              subscribers: %{},
+              generation: 0,
+              last_seen: 0,
+              legit_max: 0
+  end
+
+  property "progress only follows the owner of the current generation across node churn",
+           %{conn: conn, schema: schema} do
+    check all(
+            commands <- resize(list_of(command(), max_length: 10), 10),
+            max_runs: 10,
+            max_shrinking_steps: 20
+          ) do
+      model = %Model{subscription_name: UUID.uuid4()}
+
+      try do
+        model =
+          Enum.reduce(commands, model, fn command, model ->
+            model
+            |> apply_command(command, conn, schema)
+            |> check_invariants(model, conn, schema)
+          end)
+
+        finish(model, conn, schema)
+      after
+        for node <- @nodes, do: stop_node(%Model{}, node)
+        :timer.sleep(100)
+      end
+    end
+  end
+
+  defp command do
+    node = member_of(@nodes)
+
+    one_of([
+      tuple({constant(:start), node}),
+      tuple({constant(:stop), node}),
+      tuple({constant(:kill_lock_session), node}),
+      tuple({constant(:append), integer(1..5)}),
+      tuple({constant(:ack), node, integer(1..5)}),
+      constant(:settle)
+    ])
+  end
+
+  defp store_name(node), do: :"churn_node_#{node}"
+
+  defp apply_command(%Model{} = model, {:start, node}, _conn, schema) do
+    if Map.has_key?(model.subscribers, node), do: model, else: start_node(model, node, schema)
+  end
+
+  defp apply_command(%Model{} = model, {:stop, node}, _conn, _schema), do: stop_node(model, node)
+
+  defp apply_command(%Model{} = model, {:kill_lock_session, node}, conn, _schema) do
+    with true <- Map.has_key?(model.subscribers, node),
+         {:ok, backend_pid} <- lock_session_backend_pid(node) do
+      Postgrex.query!(conn, "SELECT pg_terminate_backend($1)", [backend_pid])
+    end
+
+    model
+  end
+
+  defp apply_command(%Model{} = model, {:append, count}, _conn, _schema) do
+    append(count)
+
+    model
+  end
+
+  defp apply_command(%Model{} = model, {:ack, node, k}, conn, schema) do
+    case Map.fetch(model.subscribers, node) do
+      {:ok, subscriber} ->
+        case Enum.at(Subscriber.seen(subscriber), k - 1) do
+          nil -> model
+          event_number -> ack(model, node, event_number, conn, schema)
+        end
+
+      :error ->
+        model
+    end
+  end
+
+  defp apply_command(%Model{} = model, :settle, conn, schema), do: settle(model, conn, schema)
+
+  # A node whose lock session was already killed has no lock connection to
+  # ask until its pool comes back, so a repeated kill has nothing to do.
+  defp lock_session_backend_pid(node) do
+    lock_pool = Module.concat([store_name(node), EventStore.AdvisoryLocks, Postgrex])
+
+    case Postgrex.query(lock_pool, "SELECT pg_backend_pid()", []) do
+      {:ok, %Postgrex.Result{rows: [[backend_pid]]}} -> {:ok, backend_pid}
+      {:error, _reason} -> :error
+    end
+  catch
+    :exit, _reason -> :error
+  end
+
+  defp start_node(%Model{} = model, node, schema) do
+    start_supervised!({TestEventStore, name: store_name(node), schema: schema},
+      id: store_name(node)
+    )
+
+    {:ok, subscriber} = Subscriber.start()
+
+    {:ok, _subscription} =
+      TestEventStore.subscribe_to_all_streams(model.subscription_name, subscriber,
+        name: store_name(node),
+        buffer_size: 10
+      )
+
+    %Model{model | subscribers: Map.put(model.subscribers, node, subscriber)}
+  end
+
+  defp stop_node(%Model{} = model, node) do
+    _ = stop_supervised(store_name(node))
+
+    case Map.pop(model.subscribers, node) do
+      {nil, _subscribers} ->
+        model
+
+      {subscriber, subscribers} ->
+        Process.exit(subscriber, :kill)
+        %Model{model | subscribers: subscribers}
+    end
+  end
+
+  defp append(count) do
+    :ok = TestEventStore.append_to_stream(UUID.uuid4(), 0, EventFactory.create_events(count))
+  end
+
+  # An ack counts as made by the owner when the acking node holds the
+  # generation the database holds around the time it acks. A takeover that
+  # completes while the ack is in flight is read as the owner's ack rather
+  # than failing the run.
+  defp ack(%Model{} = model, node, event_number, conn, schema) do
+    subscriber = Map.fetch!(model.subscribers, node)
+
+    legit_before? = holds_current_generation?(subscriber, model, conn, schema)
+
+    _ = Subscriber.ack(subscriber, event_number)
+
+    if legit_before? or holds_current_generation?(subscriber, model, conn, schema),
+      do: %Model{model | legit_max: max(model.legit_max, event_number)},
+      else: model
+  end
+
+  defp holds_current_generation?(subscriber, %Model{} = model, conn, schema) do
+    case subscription_row(conn, schema, model.subscription_name) do
+      {generation, _last_seen} -> token_generation(subscriber) == generation
+      nil -> false
+    end
+  end
+
+  defp token_generation(subscriber) do
+    with subscription when is_pid(subscription) <- Subscriber.subscription(subscriber),
+         %Subscription{subscription: %{data: %{fencing_token: token}}}
+         when token != nil <- :sys.get_state(subscription, 5_000) do
+      token.generation
+    else
+      _ -> nil
+    end
+  catch
+    :exit, _ -> nil
+  end
+
+  defp subscription_state(subscriber) do
+    with subscription when is_pid(subscription) <- Subscriber.subscription(subscriber),
+         %Subscription{subscription: %{state: state}} <- :sys.get_state(subscription, 5_000) do
+      state
+    else
+      _ -> nil
+    end
+  catch
+    :exit, _ -> nil
+  end
+
+  defp settle(%Model{subscribers: subscribers} = model, _conn, _schema)
+       when map_size(subscribers) == 0,
+       do: model
+
+  defp settle(%Model{} = model, conn, schema) do
+    max_event = max_event_number(conn)
+
+    Wait.until(15_000, fn ->
+      for {node, subscriber} <- model.subscribers, do: ack_all_seen(model, node, subscriber)
+
+      owners =
+        for {_node, subscriber} <- model.subscribers,
+            subscription_state(subscriber) == :subscribed,
+            do: subscriber
+
+      assert [_owner] = owners
+
+      {_generation, last_seen} = subscription_row(conn, schema, model.subscription_name)
+      assert last_seen >= max_event
+    end)
+
+    {generation, _last_seen} = subscription_row(conn, schema, model.subscription_name)
+
+    owner_seen =
+      for {_node, subscriber} <- model.subscribers,
+          token_generation(subscriber) == generation,
+          event_number <- Subscriber.seen(subscriber),
+          do: event_number
+
+    %Model{model | legit_max: Enum.max([model.legit_max | owner_seen])}
+  end
+
+  defp ack_all_seen(_model, _node, subscriber) do
+    for event_number <- subscriber |> Subscriber.seen() |> Enum.reverse() do
+      Subscriber.ack(subscriber, event_number)
+    end
+  end
+
+  defp check_invariants(%Model{} = model, previous, conn, schema) do
+    case subscription_row(conn, schema, model.subscription_name) do
+      nil ->
+        model
+
+      {generation, last_seen} ->
+        assert generation >= previous.generation,
+               "lock_generation went from #{previous.generation} to #{generation}"
+
+        assert last_seen >= previous.last_seen,
+               "last_seen went from #{previous.last_seen} to #{last_seen}"
+
+        assert last_seen <= model.legit_max,
+               "last_seen #{last_seen} exceeds the highest event acked by a holder of the " <>
+                 "current generation (#{model.legit_max})"
+
+        %Model{model | generation: generation, last_seen: last_seen}
+    end
+  end
+
+  defp finish(%Model{} = model, conn, schema) do
+    model = Enum.reduce(@nodes, model, &apply_command(&2, {:start, &1}, conn, schema))
+    model = settle(model, conn, schema)
+
+    append(3)
+    final = max_event_number(conn)
+
+    Wait.until(15_000, fn ->
+      for {node, subscriber} <- model.subscribers do
+        ack_all_seen(model, node, subscriber)
+      end
+
+      {_generation, last_seen} = subscription_row(conn, schema, model.subscription_name)
+      assert last_seen == final
+    end)
+
+    delivered =
+      for {node, subscriber} <- model.subscribers,
+          Enum.any?(Subscriber.seen(subscriber), &(&1 > final - 3)),
+          do: node
+
+    assert [_one_node] = delivered
+  end
+
+  defp max_event_number(conn) do
+    %Postgrex.Result{rows: [[max]]} =
+      Postgrex.query!(conn, "SELECT stream_version FROM streams WHERE stream_uuid = '$all'", [])
+
+    max
+  end
+
+  defp subscription_row(conn, schema, subscription_name) do
+    %Postgrex.Result{rows: rows} =
+      Postgrex.query!(
+        conn,
+        "SELECT lock_generation, COALESCE(last_seen, 0) FROM \"#{schema}\".subscriptions WHERE subscription_name = $1",
+        [subscription_name]
+      )
+
+    case rows do
+      [[generation, last_seen]] -> {generation, last_seen}
+      [] -> nil
+    end
+  end
+end
diff --git a/test/subscriptions/subscription_fencing_test.exs b/test/subscriptions/subscription_fencing_test.exs
new file mode 100644
index 00000000..68d5000c
--- /dev/null
+++ b/test/subscriptions/subscription_fencing_test.exs
@@ -0,0 +1,442 @@
+defmodule EventStore.Subscriptions.SubscriptionFencingTest do
+  use EventStore.StorageCase
+
+  alias EventStore.{EventFactory, TcpProxy, UUID, Wait}
+  alias EventStore.Subscriptions.{FencingToken, Subscription}
+
+  @store_a :fenced_store_a
+  @store_b :fenced_store_b
+
+  setup %{config: config, schema: schema} do
+    proxy =
+      start_supervised!(
+        {TcpProxy, {Keyword.fetch!(config, :hostname), Keyword.get(config, :port, 5432)}}
+      )
+
+    start_supervised!(
+      {TestEventStore,
+       name: @store_a,
+       schema: schema,
+       session_mode_pool: [port: TcpProxy.port(proxy), timeout: 60_000, idle_interval: 100]},
+      id: @store_a
+    )
+
+    start_supervised!({TestEventStore, name: @store_b, schema: schema}, id: @store_b)
+
+    [proxy: proxy, subscription_name: UUID.uuid4()]
+  end
+
+  test "a subscription that lost its lock does not overwrite the progress of the new owner",
+       %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do
+    {:ok, subscription_a} = subscribe_a(subscription_name, self())
+
+    fence_a(subscription_a, subscription_name, conn, schema, proxy)
+
+    :ok = TestEventStore.ack(subscription_a, 7)
+
+    assert 5 == last_seen(conn, schema, subscription_name)
+
+    Wait.until(fn -> assert :disconnected == subscription_state(subscription_a) end)
+
+    assert 5 == last_seen(conn, schema, subscription_name)
+
+    refute_receive {:events, _events}, 500
+  end
+
+  test "a fenced subscription resumes from the committed position when it takes the lock back",
+       %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do
+    {:ok, subscription_a} = subscribe_a(subscription_name, self())
+
+    fence_a(subscription_a, subscription_name, conn, schema, proxy)
+
+    :ok = TestEventStore.ack(subscription_a, 7)
+
+    Wait.until(fn -> assert :disconnected == subscription_state(subscription_a) end)
+
+    assert 5 == last_seen(conn, schema, subscription_name)
+
+    stop_supervised!(@store_b)
+    :ok = TcpProxy.forward(proxy)
+
+    assert_receive {:events, [%{event_number: 6} | _]}, 5_000
+
+    assert %FencingToken{generation: 3} = fencing_token(subscription_a)
+    assert 3 == lock_generation(conn, schema, subscription_name)
+
+    :ok = TestEventStore.ack(subscription_a, 7)
+
+    Wait.until(fn -> assert 7 == last_seen(conn, schema, subscription_name) end)
+  end
+
+  test "a fenced subscription that keeps retrying does not disturb the new owner",
+       %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do
+    {:ok, subscription_a} = subscribe_a(subscription_name, self())
+
+    {subscription_b, subscriber_b} =
+      fence_a(subscription_a, subscription_name, conn, schema, proxy)
+
+    :ok = TestEventStore.ack(subscription_a, 7)
+
+    Wait.until(fn -> assert :disconnected == subscription_state(subscription_a) end)
+
+    :timer.sleep(3_500)
+
+    # While its lock pool is unreachable the subscription may sit inside a
+    # bounded lock claim, so only the database is probed here.
+    assert 2 == lock_generation(conn, schema, subscription_name)
+    assert 5 == last_seen(conn, schema, subscription_name)
+    refute_received {:events, _events}
+
+    :ok = TcpProxy.forward(proxy)
+
+    :timer.sleep(3_500)
+
+    Wait.until(10_000, fn -> assert :disconnected == subscription_state(subscription_a) end)
+    assert 2 == lock_generation(conn, schema, subscription_name)
+    assert 5 == last_seen(conn, schema, subscription_name)
+    refute_received {:events, _events}
+
+    send(subscriber_b, {:ack, subscription_b, 6})
+
+    Wait.until(fn -> assert 6 == last_seen(conn, schema, subscription_name) end)
+
+    assert :subscribed == subscription_state(subscription_b)
+  end
+
+  test "a subscription fenced while catching up disconnects and the new owner resumes from the committed position",
+       %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do
+    for _ <- 1..4, do: append_to_a(50)
+
+    {:ok, subscription_a} = subscribe_a(subscription_name, self(), buffer_size: 1)
+
+    assert_receive {:subscribed, ^subscription_a}
+
+    for event_number <- 1..3 do
+      assert_receive {:events, [%{event_number: ^event_number}]}, 5_000
+
+      :ok = TestEventStore.ack(subscription_a, event_number)
+    end
+
+    assert_receive {:events, [%{event_number: 4}]}, 5_000
+
+    Wait.until(fn -> assert 3 == last_seen(conn, schema, subscription_name) end)
+
+    refute :subscribed == subscription_state(subscription_a)
+
+    monitor = Process.monitor(subscription_a)
+
+    {subscription_b, subscriber_b} = take_over(subscription_name, conn, schema, proxy)
+
+    :ok = TestEventStore.ack(subscription_a, 4)
+
+    Wait.until(fn -> assert :disconnected == subscription_state(subscription_a) end)
+
+    assert 5 == last_seen(conn, schema, subscription_name)
+
+    flush_events()
+
+    for event_number <- 6..12 do
+      send(subscriber_b, {:ack, subscription_b, event_number})
+    end
+
+    assert_receive_b_through(15)
+
+    refute_receive {:DOWN, ^monitor, :process, _pid, _reason}, 500
+    refute_receive {:events, _events}, 500
+
+    Wait.until(fn -> assert 12 == last_seen(conn, schema, subscription_name) end)
+  end
+
+  defp assert_receive_b_through(last_event_number) do
+    assert_receive {:b, :events, events}, 5_000
+
+    unless List.last(events).event_number >= last_event_number do
+      assert_receive_b_through(last_event_number)
+    end
+  end
+
+  defp subscribe_a(subscription_name, subscriber, opts \\ []) do
+    TestEventStore.subscribe_to_all_streams(
+      subscription_name,
+      subscriber,
+      Keyword.merge([name: @store_a, buffer_size: 10], opts)
+    )
+  end
+
+  defp fence_a(subscription_a, subscription_name, conn, schema, proxy) do
+    assert_receive {:subscribed, ^subscription_a}
+
+    append_to_a(7)
+
+    assert_receive_through(7)
+
+    :ok = TestEventStore.ack(subscription_a, 3)
+
+    Wait.until(fn -> assert 3 == last_seen(conn, schema, subscription_name) end)
+
+    take_over(subscription_name, conn, schema, proxy)
+  end
+
+  defp append_to_a(count) do
+    :ok =
+      TestEventStore.append_to_stream(UUID.uuid4(), 0, EventFactory.create_events(count),
+        name: @store_a
+      )
+  end
+
+  # Takes A's lock session away and has B own the subscription at event 5.
+  defp take_over(subscription_name, conn, schema, proxy) do
+    end_lock_session(conn, proxy)
+
+    subscriber_b = start_forwarding_subscriber(:b)
+
+    {:ok, subscription_b} =
+      TestEventStore.subscribe_to_all_streams(subscription_name, subscriber_b,
+        name: @store_b,
+        buffer_size: 10
+      )
+
+    assert_receive {:b, :subscribed, ^subscription_b}, 5_000
+    assert_receive {:b, :events, [%{event_number: 4} | _]}, 5_000
+
+    send(subscriber_b, {:ack, subscription_b, 5})
+
+    Wait.until(fn -> assert 5 == last_seen(conn, schema, subscription_name) end)
+
+    flush_events()
+
+    {subscription_b, subscriber_b}
+  end
+
+  defp end_lock_session(conn, proxy) do
+    lock_pool = Module.concat([@store_a, EventStore.AdvisoryLocks, Postgrex])
+
+    %Postgrex.Result{rows: [[backend_pid]]} =
+      Postgrex.query!(lock_pool, "SELECT pg_backend_pid()", [])
+
+    :ok = TcpProxy.blackhole(proxy)
+
+    %Postgrex.Result{rows: [[true]]} =
+      Postgrex.query!(conn, "SELECT pg_terminate_backend($1)", [backend_pid])
+
+    Wait.until(fn ->
+      assert %Postgrex.Result{rows: [[0]]} =
+               Postgrex.query!(
+                 conn,
+                 "SELECT count(*)::int FROM pg_stat_activity WHERE pid = $1",
+                 [backend_pid]
+               )
+    end)
+  end
+
+  defp take_over_all(tagged_names, conn, schema, proxy) do
+    end_lock_session(conn, proxy)
+
+    owners =
+      for {tag, subscription_name} <- tagged_names do
+        subscriber = start_forwarding_subscriber(tag)
+
+        {:ok, subscription} =
+          TestEventStore.subscribe_to_all_streams(subscription_name, subscriber,
+            name: @store_b,
+            buffer_size: 10
+          )
+
+        assert_receive {^tag, :subscribed, ^subscription}, 5_000
+        assert_receive {^tag, :events, [%{event_number: 4} | _]}, 5_000
+
+        send(subscriber, {:ack, subscription, 5})
+
+        Wait.until(fn -> assert 5 == last_seen(conn, schema, subscription_name) end)
+
+        {tag, subscription}
+      end
+
+    flush_tagged_events([:a1, :a2, :b1, :b2])
+
+    owners
+  end
+
+  defp flush_tagged_events(tags), do: Enum.each(tags, &flush_tagged/1)
+
+  defp flush_tagged(tag) do
+    receive do
+      {^tag, :events, _events} -> flush_tagged(tag)
+    after
+      0 -> :ok
+    end
+  end
+
+  test "every subscription on a node whose lock session ends is fenced and recovers",
+       %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do
+    other_name = UUID.uuid4()
+    subscriber_1 = start_forwarding_subscriber(:a1)
+    subscriber_2 = start_forwarding_subscriber(:a2)
+
+    {:ok, subscription_1} = subscribe_a(subscription_name, subscriber_1)
+    {:ok, subscription_2} = subscribe_a(other_name, subscriber_2)
+
+    assert_receive {:a1, :subscribed, ^subscription_1}
+    assert_receive {:a2, :subscribed, ^subscription_2}
+
+    append_to_a(7)
+
+    for tag <- [:a1, :a2], do: assert_receive_tagged_through(tag, 7)
+
+    send(subscriber_1, {:ack, subscription_1, 3})
+    send(subscriber_2, {:ack, subscription_2, 3})
+
+    for name <- [subscription_name, other_name] do
+      Wait.until(fn -> assert 3 == last_seen(conn, schema, name) end)
+    end
+
+    take_over_all([b1: subscription_name, b2: other_name], conn, schema, proxy)
+
+    send(subscriber_1, {:ack, subscription_1, 7})
+    send(subscriber_2, {:ack, subscription_2, 7})
+
+    for subscription <- [subscription_1, subscription_2] do
+      Wait.until(fn -> assert :disconnected == subscription_state(subscription) end)
+    end
+
+    for name <- [subscription_name, other_name] do
+      assert 5 == last_seen(conn, schema, name)
+    end
+
+    flush_tagged_events([:a1, :a2, :b1, :b2])
+
+    stop_supervised!(@store_b)
+    :ok = TcpProxy.forward(proxy)
+
+    assert_receive {:a1, :events, [%{event_number: 6} | _]}, 5_000
+    assert_receive {:a2, :events, [%{event_number: 6} | _]}, 5_000
+
+    for name <- [subscription_name, other_name] do
+      assert 3 == lock_generation(conn, schema, name)
+    end
+  end
+
+  defp assert_receive_tagged_through(tag, last_event_number) do
+    assert_receive {^tag, :events, events}
+
+    unless List.last(events).event_number >= last_event_number do
+      assert_receive_tagged_through(tag, last_event_number)
+    end
+  end
+
+  test "every subscriber of a fenced subscription stops receiving events",
+       %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do
+    subscriber_1 = start_forwarding_subscriber(:a1)
+    subscriber_2 = start_forwarding_subscriber(:a2)
+
+    {:ok, subscription_a} = subscribe_a(subscription_name, subscriber_1, concurrency_limit: 2)
+    {:ok, ^subscription_a} = subscribe_a(subscription_name, subscriber_2, concurrency_limit: 2)
+
+    assert_receive {:a1, :subscribed, ^subscription_a}
+    assert_receive {:a2, :subscribed, ^subscription_a}
+
+    append_to_a(7)
+
+    owners = receive_owners(%{a1: subscriber_1, a2: subscriber_2}, 7)
+
+    ack_in_order(owners, subscription_a, 1..3)
+
+    Wait.until(fn -> assert 3 == last_seen(conn, schema, subscription_name) end)
+
+    take_over(subscription_name, conn, schema, proxy)
+
+    ack_in_order(owners, subscription_a, 4..7)
+
+    Wait.until(fn -> assert :disconnected == subscription_state(subscription_a) end)
+
+    assert 5 == last_seen(conn, schema, subscription_name)
+
+    refute_receive {:a1, :events, _events}, 500
+    refute_receive {:a2, :events, _events}, 0
+  end
+
+  defp ack_in_order(owners, subscription, event_numbers) do
+    for event_number <- event_numbers do
+      send(Map.fetch!(owners, event_number), {:ack, subscription, event_number})
+    end
+  end
+
+  defp receive_owners(subscribers, last, acc \\ %{}) do
+    receive do
+      {tag, :events, events} when tag in [:a1, :a2] ->
+        acc =
+          Enum.reduce(events, acc, &Map.put(&2, &1.event_number, Map.fetch!(subscribers, tag)))
+
+        if map_size(acc) >= last, do: acc, else: receive_owners(subscribers, last, acc)
+    after
+      2_000 -> acc
+    end
+  end
+
+  defp assert_receive_through(last_event_number) do
+    assert_receive {:events, events}
+
+    unless List.last(events).event_number >= last_event_number do
+      assert_receive_through(last_event_number)
+    end
+  end
+
+  defp flush_events do
+    receive do
+      {:events, _events} -> flush_events()
+    after
+      0 -> :ok
+    end
+  end
+
+  defp start_forwarding_subscriber(tag) do
+    reply_to = self()
+
+    spawn_link(fn -> forward(tag, reply_to) end)
+  end
+
+  defp forward(tag, reply_to) do
+    receive do
+      {:subscribed, subscription} -> send(reply_to, {tag, :subscribed, subscription})
+      {:events, events} -> send(reply_to, {tag, :events, events})
+      {:ack, subscription, event_number} -> TestEventStore.ack(subscription, event_number)
+    end
+
+    forward(tag, reply_to)
+  end
+
+  defp subscription_state(subscription) do
+    %Subscription{subscription: %{state: state}} = :sys.get_state(subscription)
+
+    state
+  end
+
+  defp fencing_token(subscription) do
+    %Subscription{subscription: %{data: %{fencing_token: fencing_token}}} =
+      :sys.get_state(subscription)
+
+    fencing_token
+  end
+
+  defp lock_generation(conn, schema, subscription_name) do
+    %Postgrex.Result{rows: [[generation]]} =
+      Postgrex.query!(
+        conn,
+        "SELECT lock_generation FROM \"#{schema}\".subscriptions WHERE subscription_name = $1",
+        [subscription_name]
+      )
+
+    generation
+  end
+
+  defp last_seen(conn, schema, subscription_name) do
+    %Postgrex.Result{rows: [[last_seen]]} =
+      Postgrex.query!(
+        conn,
+        "SELECT last_seen FROM \"#{schema}\".subscriptions WHERE subscription_name = $1",
+        [subscription_name]
+      )
+
+    last_seen
+  end
+end
diff --git a/test/subscriptions/subscription_fsm_test.exs b/test/subscriptions/subscription_fsm_test.exs
index 4088c1a0..93066c43 100644
--- a/test/subscriptions/subscription_fsm_test.exs
+++ b/test/subscriptions/subscription_fsm_test.exs
@@ -1,7 +1,8 @@
 defmodule EventStore.Subscriptions.SubscriptionFsmTest do
   use EventStore.StorageCase
 
-  alias EventStore.{Storage, UUID}
+  alias EventStore.{EventFactory, Storage, UUID}
+  alias EventStore.Storage.Lock
   alias EventStore.Subscriptions.{SubscriptionFsm, SubscriptionState}
 
   # The states that own a stored position, and those that do not.
@@ -17,10 +18,15 @@ defmodule EventStore.Subscriptions.SubscriptionFsmTest do
         schema: context.schema
       )
 
+    fencing_token = claim_token(context, subscription_id)
+
+    on_exit(fn -> :ok = Lock.unlock(context.conn, subscription_id, schema: context.schema) end)
+
     [
       stream_uuid: stream_uuid,
       subscription_name: subscription_name,
-      subscription_id: subscription_id
+      subscription_id: subscription_id,
+      fencing_token: fencing_token
     ]
   end
 
@@ -155,6 +161,31 @@ defmodule EventStore.Subscriptions.SubscriptionFsmTest do
       assert_received {:checkpoint_failed, :subscription_not_found}
     end
 
+    test "tells the subscription it is fenced when the probe that classifies the rejection fails",
+         context do
+      %{conn: conn, schema: schema, subscription_id: subscription_id} = context
+
+      probe_conn = start_supervised!({Postgrex, context.postgrex_config}, id: :probe_conn)
+      lock_ref = make_ref()
+
+      fsm = fsm(context, checkpoints_pending: 1, last_ack: 3, conn: probe_conn)
+      fsm = %SubscriptionFsm{fsm | data: %SubscriptionState{fsm.data | lock_ref: lock_ref}}
+
+      {:ok, _result} =
+        Postgrex.query(
+          conn,
+          ~s|UPDATE "#{schema}".subscriptions SET lock_generation = lock_generation + 1 WHERE subscription_id = $1|,
+          [subscription_id]
+        )
+
+      drop_privileges_after_update(context)
+
+      assert %SubscriptionFsm{data: %SubscriptionState{checkpoints_pending: 0}} =
+               SubscriptionFsm.checkpoint(fsm)
+
+      assert_received {:fenced, ^lock_ref}
+    end
+
     test "does not write the position onto a subscription that reused the name", context do
       %{conn: conn, schema: schema, stream_uuid: stream_uuid} = context
       %{subscription_name: subscription_name, subscription_id: subscription_id} = context
@@ -305,26 +336,128 @@ defmodule EventStore.Subscriptions.SubscriptionFsmTest do
     end
   end
 
+  describe "subscribe/1 after another owner advanced the stored position" do
+    test "catches up to the stored position and subscribes instead of spinning on empty reads",
+         context do
+      reclaim_after_other_owner_acked(context, appended: 15, stored: 15)
+
+      fsm = fsm(context, state: :disconnected, last_received: 10, last_sent: 5, last_ack: 5)
+
+      assert %SubscriptionFsm{state: :request_catch_up, data: data} =
+               SubscriptionFsm.subscribe(fsm)
+
+      assert data.last_ack == 15
+      assert data.last_sent == 15
+
+      assert %SubscriptionFsm{state: :subscribed, data: data} =
+               SubscriptionFsm.catch_up(%SubscriptionFsm{state: :request_catch_up, data: data})
+
+      assert data.last_received == 15
+    end
+
+    test "does not move last_received backwards when events past the stored position were seen",
+         context do
+      reclaim_after_other_owner_acked(context, appended: 20, stored: 15)
+
+      fsm = fsm(context, state: :disconnected, last_received: 20, last_sent: 5, last_ack: 5)
+
+      assert %SubscriptionFsm{state: :request_catch_up, data: data} =
+               SubscriptionFsm.subscribe(fsm)
+
+      assert data.last_received == 20
+
+      assert %SubscriptionFsm{state: :catching_up, data: data} =
+               SubscriptionFsm.catch_up(%SubscriptionFsm{state: :request_catch_up, data: data})
+
+      assert data.queue_size == 5
+      assert data.last_received == 20
+    end
+  end
+
+  defp reclaim_after_other_owner_acked(context, appended: appended, stored: stored) do
+    %{conn: conn, schema: schema, subscription_id: subscription_id} = context
+
+    :ok =
+      TestEventStore.append_to_stream(
+        context.stream_uuid,
+        0,
+        EventFactory.create_events(appended)
+      )
+
+    {:ok, _result} =
+      Postgrex.query(
+        conn,
+        ~s|UPDATE "#{schema}".subscriptions SET last_seen = $2 WHERE subscription_id = $1|,
+        [subscription_id, stored]
+      )
+
+    :ok = Lock.unlock(conn, subscription_id, schema: schema)
+  end
+
+  defp claim_token(%{conn: conn, schema: schema}, subscription_id) do
+    {:ok, %Lock.Lease{token: token}} =
+      Lock.claim_subscription_lock(conn, subscription_id, schema: schema)
+
+    token
+  end
+
   defp fsm(context, overrides \\ []) do
     %{conn: conn, schema: schema} = context
     %{stream_uuid: stream_uuid, subscription_name: subscription_name} = context
     %{subscription_id: subscription_id} = context
 
     data = %SubscriptionState{
-      conn: conn,
+      conn: Keyword.get(overrides, :conn, conn),
       schema: Keyword.get(overrides, :schema, schema),
       stream_uuid: stream_uuid,
       subscription_name: subscription_name,
       subscription_id: subscription_id,
+      fencing_token: context.fencing_token,
       last_ack: Keyword.get(overrides, :last_ack, 0),
       checkpoints_pending: Keyword.get(overrides, :checkpoints_pending, 0),
       query_timeout: 15_000,
+      event_store: TestEventStore,
+      serializer: context.serializer,
+      correlation_id_type: context.correlation_id_type,
+      causation_id_type: context.causation_id_type,
+      max_size: 1_000,
+      last_received: Keyword.get(overrides, :last_received, 0),
+      last_sent: Keyword.get(overrides, :last_sent, 0),
       transient: Keyword.get(overrides, :transient, false)
     }
 
     %SubscriptionFsm{state: Keyword.get(overrides, :state, :subscribed), data: data}
   end
 
+  defp drop_privileges_after_update(%{conn: conn, schema: schema}) do
+    {:ok, _result} =
+      Postgrex.query(
+        conn,
+        """
+        CREATE FUNCTION fsm_test_drop_privileges() RETURNS trigger LANGUAGE plpgsql AS $$
+        BEGIN PERFORM set_config('role', 'pg_monitor', false); RETURN NULL; END $$
+        """,
+        []
+      )
+
+    {:ok, _result} =
+      Postgrex.query(
+        conn,
+        ~s|CREATE TRIGGER fsm_test_drop_privileges AFTER UPDATE ON "#{schema}".subscriptions FOR EACH STATEMENT EXECUTE FUNCTION fsm_test_drop_privileges()|,
+        []
+      )
+
+    on_exit(fn ->
+      Postgrex.query!(
+        conn,
+        ~s|DROP TRIGGER fsm_test_drop_privileges ON "#{schema}".subscriptions|,
+        []
+      )
+
+      Postgrex.query!(conn, "DROP FUNCTION fsm_test_drop_privileges()", [])
+    end)
+  end
+
   defp reset_position(context) do
     %{conn: conn, schema: schema, subscription_id: subscription_id} = context
 
diff --git a/test/subscriptions/subscription_locking_test.exs b/test/subscriptions/subscription_locking_test.exs
index fbf69d39..168aa69a 100644
--- a/test/subscriptions/subscription_locking_test.exs
+++ b/test/subscriptions/subscription_locking_test.exs
@@ -1,10 +1,13 @@
 defmodule EventStore.Subscriptions.SubscriptionLockingTest do
   use EventStore.StorageCase
 
-  alias EventStore.{Config, EventFactory, Storage, UUID}
-  alias EventStore.Subscriptions.Subscription
+  import ExUnit.CaptureLog
+
+  alias EventStore.{Config, EventFactory, Storage, UUID, Wait}
+  alias EventStore.Subscriptions.{FencingToken, Subscription}
 
   @conn TestEventStore.Postgrex
+  @advisory_locks TestEventStore.EventStore.AdvisoryLocks
   @advisory_locks_conn TestEventStore.EventStore.AdvisoryLocks.Postgrex
 
   setup do
@@ -81,7 +84,9 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do
       {:ok, %Storage.Subscription{subscription_id: subscription_id}} =
         Storage.Subscription.subscription(@conn, "$all", subscription_name, schema: schema)
 
-      :ok = Storage.Subscription.ack_last_seen_event(@conn, subscription_id, 2, schema: schema)
+      token = FencingToken.new(subscription_id, lock_generation(subscription_id, schema))
+
+      :ok = Storage.Subscription.ack_last_seen_event(@conn, token, 2, schema: schema)
 
       :ok = reconnect(subscription)
 
@@ -163,6 +168,173 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do
     end
   end
 
+  describe "late lock notice" do
+    setup [:create_subscription]
+
+    test "should keep running when the lock was already released", %{
+      event_store: event_store,
+      subscription: subscription
+    } do
+      assert_receive {:subscribed, ^subscription}
+
+      Process.flag(:trap_exit, true)
+      monitor_ref = Process.monitor(subscription)
+
+      lock_ref = lock_ref(subscription)
+
+      :ok = release_lock(subscription, lock_ref)
+      :sys.get_state(subscription)
+
+      :ok = release_lock(subscription, lock_ref)
+
+      refute_receive {:DOWN, ^monitor_ref, :process, ^subscription, _reason}, 200
+      assert Process.alive?(subscription)
+
+      :ok = reconnect(subscription)
+
+      append_events_to_stream(event_store, 3)
+
+      assert_receive_events([1, 2, 3])
+    end
+
+    test "should ignore a notice for a lock that was replaced by a new one", %{
+      event_store: event_store,
+      subscription: subscription
+    } do
+      assert_receive {:subscribed, ^subscription}
+
+      Process.flag(:trap_exit, true)
+      monitor_ref = Process.monitor(subscription)
+
+      old_lock_ref = lock_ref(subscription)
+
+      :ok = release_lock(subscription, old_lock_ref)
+      :ok = reconnect(subscription)
+
+      _new_lock_ref = wait_for_new_lock_ref(subscription, old_lock_ref)
+
+      :ok = release_lock(subscription, old_lock_ref)
+
+      refute_receive {:DOWN, ^monitor_ref, :process, ^subscription, _reason}, 200
+      assert Process.alive?(subscription)
+
+      append_events_to_stream(event_store, 3)
+
+      assert_receive_events([1, 2, 3])
+    end
+  end
+
+  describe "late fenced notice" do
+    setup [:create_subscription]
+
+    test "should keep running when the lock was already released", %{
+      event_store: event_store,
+      subscription: subscription
+    } do
+      assert_receive {:subscribed, ^subscription}
+
+      monitor_ref = Process.monitor(subscription)
+
+      lock_ref = lock_ref(subscription)
+
+      :ok = disconnect(subscription)
+      :sys.get_state(subscription)
+
+      send(subscription, {:fenced, lock_ref})
+
+      refute_receive {:DOWN, ^monitor_ref, :process, ^subscription, _reason}, 200
+      assert Process.alive?(subscription)
+
+      :ok = reconnect(subscription)
+
+      append_events_to_stream(event_store, 3)
+
+      assert_receive_events([1, 2, 3])
+    end
+
+    test "should ignore a notice for a lock that was replaced by a new one", %{
+      event_store: event_store,
+      subscription: subscription
+    } do
+      assert_receive {:subscribed, ^subscription}
+
+      monitor_ref = Process.monitor(subscription)
+
+      old_lock_ref = lock_ref(subscription)
+
+      :ok = disconnect(subscription)
+      :ok = reconnect(subscription)
+
+      new_lock_ref = wait_for_new_lock_ref(subscription, old_lock_ref)
+
+      Wait.until(fn -> assert :subscribed == subscription_state(subscription) end)
+
+      send(subscription, {:fenced, old_lock_ref})
+
+      refute_receive {:DOWN, ^monitor_ref, :process, ^subscription, _reason}, 200
+      assert Process.alive?(subscription)
+
+      assert :subscribed == subscription_state(subscription)
+      assert new_lock_ref == lock_ref(subscription)
+
+      append_events_to_stream(event_store, 3)
+
+      assert_receive_events([1, 2, 3])
+    end
+  end
+
+  describe "subscription stopped after another owner took over" do
+    test "should leave the progress of the new owner untouched", %{
+      conn: conn,
+      event_store: event_store,
+      schema: schema,
+      subscription_name: subscription_name
+    } do
+      {:ok, subscription} =
+        event_store.subscribe_to_all_streams(subscription_name, self(),
+          buffer_size: 3,
+          checkpoint_threshold: 100
+        )
+
+      assert_receive {:subscribed, ^subscription}
+
+      append_events_to_stream(event_store, 3)
+
+      assert_receive_events([1, 2, 3])
+
+      Postgrex.query!(
+        conn,
+        "UPDATE \"#{schema}\".subscriptions SET lock_generation = lock_generation + 1, last_seen = 2 WHERE subscription_name = $1",
+        [subscription_name]
+      )
+
+      :ok = Subscription.ack(subscription, 3)
+
+      monitor_ref = Process.monitor(subscription)
+
+      log =
+        capture_log(fn ->
+          assert :ok = GenServer.stop(subscription)
+        end)
+
+      assert_receive {:DOWN, ^monitor_ref, :process, ^subscription, :normal}
+      refute log =~ "terminating"
+
+      assert 2 == last_seen(conn, schema, subscription_name)
+    end
+  end
+
+  defp last_seen(conn, schema, subscription_name) do
+    %Postgrex.Result{rows: [[last_seen]]} =
+      Postgrex.query!(
+        conn,
+        "SELECT last_seen FROM \"#{schema}\".subscriptions WHERE subscription_name = $1",
+        [subscription_name]
+      )
+
+    last_seen
+  end
+
   defp lock_subscription(context) do
     %{config: config, schema: schema} = context
 
@@ -216,13 +388,51 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do
     :ok
   end
 
+  defp lock_generation(subscription_id, schema) do
+    %Postgrex.Result{rows: [[generation]]} =
+      Postgrex.query!(
+        @conn,
+        "SELECT lock_generation FROM \"#{schema}\".subscriptions WHERE subscription_id = $1",
+        [subscription_id]
+      )
+
+    generation
+  end
+
   defp reconnect(subscription) do
     Process.send(subscription, :subscribe_to_stream, [])
   end
 
   defp disconnect(subscription) do
+    release_lock(subscription, lock_ref(subscription))
+  end
+
+  defp lock_ref(subscription) do
     %Subscription{subscription: %{data: %{lock_ref: lock_ref}}} = :sys.get_state(subscription)
 
+    lock_ref
+  end
+
+  defp subscription_state(subscription) do
+    %Subscription{subscription: %{state: state}} = :sys.get_state(subscription)
+
+    state
+  end
+
+  defp wait_for_new_lock_ref(subscription, old_lock_ref) do
+    Wait.until(2_000, fn ->
+      lock_ref = lock_ref(subscription)
+
+      assert is_reference(lock_ref) and lock_ref != old_lock_ref
+
+      lock_ref
+    end)
+  end
+
+  defp release_lock(subscription, lock_ref) do
+    :ok = EventStore.AdvisoryLocks.release(@advisory_locks, lock_ref)
+    _state = :sys.get_state(@advisory_locks)
+
     Process.send(
       subscription,
       {EventStore.AdvisoryLocks, :lock_released, lock_ref, :shutdown},
diff --git a/test/subscriptions/subscription_unmigrated_schema_test.exs b/test/subscriptions/subscription_unmigrated_schema_test.exs
new file mode 100644
index 00000000..c66a85cc
--- /dev/null
+++ b/test/subscriptions/subscription_unmigrated_schema_test.exs
@@ -0,0 +1,55 @@
+defmodule EventStore.Subscriptions.SubscriptionUnmigratedSchemaTest do
+  use EventStore.StorageCase
+
+  alias EventStore.{EventFactory, UUID, Wait}
+  alias EventStore.Subscriptions.Subscription
+
+  @advisory_locks TestEventStore.EventStore.AdvisoryLocks
+  @advisory_locks_conn TestEventStore.EventStore.AdvisoryLocks.Postgrex
+
+  test "a database without the lock generation keeps a new node retrying instead of crashing",
+       %{conn: conn} do
+    on_exit(fn -> add_lock_generation(conn) end)
+
+    Postgrex.query!(conn, "ALTER TABLE subscriptions DROP COLUMN lock_generation", [])
+
+    advisory_locks = Process.whereis(@advisory_locks)
+    advisory_locks_conn = Process.whereis(@advisory_locks_conn)
+
+    {:ok, subscription} =
+      TestEventStore.subscribe_to_all_streams(UUID.uuid4(), self(), buffer_size: 10)
+
+    ref = Process.monitor(subscription)
+
+    refute_receive {:subscribed, _subscription}, 1_500
+    refute_receive {:DOWN, ^ref, :process, _pid, _reason}
+
+    assert :initial == subscription_state(subscription)
+    assert advisory_locks == Process.whereis(@advisory_locks)
+    assert advisory_locks_conn == Process.whereis(@advisory_locks_conn)
+
+    add_lock_generation(conn)
+
+    assert_receive {:subscribed, ^subscription}, 3_000
+
+    :ok = TestEventStore.append_to_stream(UUID.uuid4(), 0, EventFactory.create_events(1))
+
+    assert_receive {:events, [%{event_number: 1}]}
+
+    Wait.until(fn -> assert :subscribed == subscription_state(subscription) end)
+  end
+
+  defp add_lock_generation(conn) do
+    Postgrex.query!(
+      conn,
+      "ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS lock_generation bigint NOT NULL DEFAULT 0",
+      []
+    )
+  end
+
+  defp subscription_state(subscription) do
+    %Subscription{subscription: %{state: state}} = :sys.get_state(subscription)
+
+    state
+  end
+end
diff --git a/test/subscriptions/subscription_zombie_lock_test.exs b/test/subscriptions/subscription_zombie_lock_test.exs
new file mode 100644
index 00000000..9ccd8411
--- /dev/null
+++ b/test/subscriptions/subscription_zombie_lock_test.exs
@@ -0,0 +1,175 @@
+defmodule EventStore.Subscriptions.SubscriptionZombieLockTest do
+  use EventStore.StorageCase
+
+  alias EventStore.{EventFactory, TcpProxy, UUID, Wait}
+  alias EventStore.Subscriptions.Subscription
+
+  @store_a :zombie_store_a
+  @store_b :zombie_store_b
+
+  setup %{config: config, schema: schema} do
+    proxy =
+      start_supervised!(
+        {TcpProxy, {Keyword.fetch!(config, :hostname), Keyword.get(config, :port, 5432)}}
+      )
+
+    start_supervised!(
+      {TestEventStore,
+       name: @store_a,
+       schema: schema,
+       port: TcpProxy.port(proxy),
+       session_mode_pool: [timeout: 60_000]},
+      id: @store_a
+    )
+
+    start_supervised!({TestEventStore, name: @store_b, schema: schema}, id: @store_b)
+
+    [proxy: proxy, subscription_name: UUID.uuid4()]
+  end
+
+  test "a node that dies dirty cannot move the progress of the owner that replaced it",
+       %{conn: conn, schema: schema, proxy: proxy, subscription_name: subscription_name} do
+    {:ok, subscription_a} =
+      TestEventStore.subscribe_to_all_streams(subscription_name, self(),
+        name: @store_a,
+        buffer_size: 10
+      )
+
+    assert_receive {:subscribed, ^subscription_a}
+
+    :ok =
+      TestEventStore.append_to_stream(UUID.uuid4(), 0, EventFactory.create_events(7),
+        name: @store_a
+      )
+
+    assert_receive_through(7)
+
+    :ok = TestEventStore.ack(subscription_a, 3)
+
+    Wait.until(fn -> assert 3 == last_seen(conn, schema, subscription_name) end)
+    assert 1 == lock_generation(conn, schema, subscription_name)
+
+    :ok = TcpProxy.freeze(proxy)
+
+    test_pid = self()
+
+    pending_ack =
+      Task.async(fn ->
+        try do
+          Subscription.ack(subscription_a, 7, test_pid)
+        catch
+          :exit, reason -> {:exit, reason}
+        end
+      end)
+
+    :timer.sleep(300)
+
+    subscriber_b = start_forwarding_subscriber(:b)
+
+    {:ok, subscription_b} =
+      TestEventStore.subscribe_to_all_streams(subscription_name, subscriber_b,
+        name: @store_b,
+        buffer_size: 10
+      )
+
+    refute_receive {:b, :subscribed, _subscription}, 2_500
+    assert 1 == lock_generation(conn, schema, subscription_name)
+
+    terminate_lock_holder(conn)
+
+    assert_receive {:b, :subscribed, ^subscription_b}, 5_000
+    assert_receive {:b, :events, [%{event_number: 4} | _]}, 5_000
+
+    assert 2 == lock_generation(conn, schema, subscription_name)
+
+    send(subscriber_b, {:ack, subscription_b, 5})
+
+    Wait.until(fn -> assert 5 == last_seen(conn, schema, subscription_name) end)
+
+    :ok = TcpProxy.forward(proxy)
+
+    Task.await(pending_ack, 20_000)
+
+    assert 5 == last_seen(conn, schema, subscription_name)
+    assert 2 == lock_generation(conn, schema, subscription_name)
+
+    Wait.until(10_000, fn -> assert :disconnected == subscription_state(subscription_a) end)
+
+    assert 5 == last_seen(conn, schema, subscription_name)
+
+    flush_events()
+    refute_receive {:events, _events}, 1_500
+
+    send(subscriber_b, {:ack, subscription_b, 6})
+
+    Wait.until(fn -> assert 6 == last_seen(conn, schema, subscription_name) end)
+    assert 2 == lock_generation(conn, schema, subscription_name)
+  end
+
+  defp terminate_lock_holder(conn) do
+    %Postgrex.Result{rows: [[backend_pid]]} =
+      Postgrex.query!(
+        conn,
+        "SELECT pid FROM pg_locks WHERE locktype = 'advisory' AND granted",
+        []
+      )
+
+    %Postgrex.Result{rows: [[true]]} =
+      Postgrex.query!(conn, "SELECT pg_terminate_backend($1)", [backend_pid])
+  end
+
+  defp assert_receive_through(last_event_number) do
+    assert_receive {:events, events}
+
+    unless List.last(events).event_number >= last_event_number do
+      assert_receive_through(last_event_number)
+    end
+  end
+
+  defp flush_events do
+    receive do
+      {:events, _events} -> flush_events()
+    after
+      0 -> :ok
+    end
+  end
+
+  defp start_forwarding_subscriber(tag) do
+    reply_to = self()
+
+    spawn_link(fn -> forward(tag, reply_to) end)
+  end
+
+  defp forward(tag, reply_to) do
+    receive do
+      {:subscribed, subscription} -> send(reply_to, {tag, :subscribed, subscription})
+      {:events, events} -> send(reply_to, {tag, :events, events})
+      {:ack, subscription, event_number} -> TestEventStore.ack(subscription, event_number)
+    end
+
+    forward(tag, reply_to)
+  end
+
+  defp subscription_state(subscription) do
+    %Subscription{subscription: %{state: state}} = :sys.get_state(subscription)
+
+    state
+  end
+
+  defp lock_generation(conn, schema, subscription_name),
+    do: subscription_column(conn, schema, "lock_generation", subscription_name)
+
+  defp last_seen(conn, schema, subscription_name),
+    do: subscription_column(conn, schema, "last_seen", subscription_name)
+
+  defp subscription_column(conn, schema, column, subscription_name) do
+    %Postgrex.Result{rows: [[value]]} =
+      Postgrex.query!(
+        conn,
+        "SELECT #{column} FROM \"#{schema}\".subscriptions WHERE subscription_name = $1",
+        [subscription_name]
+      )
+
+    value
+  end
+end
diff --git a/test/support/tcp_proxy.ex b/test/support/tcp_proxy.ex
new file mode 100644
index 00000000..14392281
--- /dev/null
+++ b/test/support/tcp_proxy.ex
@@ -0,0 +1,181 @@
+defmodule EventStore.TcpProxy do
+  @moduledoc false
+
+  use GenServer
+
+  @grace 500
+
+  def start_link(upstream), do: GenServer.start_link(__MODULE__, upstream)
+
+  def port(proxy), do: GenServer.call(proxy, :port)
+
+  @doc """
+  Keeps the connections already piped and holds every new connection open
+  without ever answering it.
+  """
+  def blackhole(proxy), do: GenServer.call(proxy, :blackhole)
+
+  @doc """
+  Pipes new connections again and closes the ones held open by `blackhole/1`
+  so their clients fail fast and reconnect.
+  """
+  def forward(proxy), do: GenServer.call(proxy, :forward)
+
+  @doc """
+  Parks every byte and every close on the connections already piped without
+  closing any socket, and holds new connections like `blackhole/1`. Both ends
+  keep believing the connection is alive. `forward/1` delivers what was parked
+  and keeps relaying briefly before resetting those connections, the way late
+  packets still arrive when a partition heals.
+  """
+  def freeze(proxy), do: GenServer.call(proxy, :freeze)
+
+  def stop(proxy), do: GenServer.stop(proxy)
+
+  @impl GenServer
+  def init({host, port}) do
+    {:ok, listen} =
+      :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true, ip: {127, 0, 0, 1}])
+
+    {:ok, listen_port} = :inet.port(listen)
+    proxy = self()
+
+    acceptor =
+      spawn_link(fn ->
+        Process.put(:proxy, proxy)
+        accept(listen, proxy, to_charlist(host), port)
+      end)
+
+    {:ok,
+     %{
+       acceptor: acceptor,
+       listen: listen,
+       port: listen_port,
+       mode: :forward,
+       held: [],
+       relays: []
+     }}
+  end
+
+  @impl GenServer
+  def handle_call(:port, _from, state), do: {:reply, state.port, state}
+  def handle_call(:mode, _from, state), do: {:reply, state.mode, state}
+  def handle_call(:blackhole, _from, state), do: {:reply, :ok, %{state | mode: :blackhole}}
+
+  def handle_call({:hold, client}, _from, state),
+    do: {:reply, :ok, %{state | held: [client | state.held]}}
+
+  def handle_call(:freeze, _from, state), do: {:reply, :ok, %{state | mode: :frozen}}
+
+  def handle_call({:register, relay}, _from, state),
+    do: {:reply, :ok, %{state | relays: [relay | state.relays]}}
+
+  def handle_call(:forward, _from, state) do
+    Enum.each(state.held, &:gen_tcp.close/1)
+
+    if state.mode == :frozen, do: Enum.each(state.relays, &send(&1, :release))
+
+    {:reply, :ok, %{state | mode: :forward, held: [], relays: []}}
+  end
+
+  @impl GenServer
+  def terminate(_reason, %{acceptor: acceptor, listen: listen}) do
+    Process.unlink(acceptor)
+    Process.exit(acceptor, :kill)
+    :gen_tcp.close(listen)
+  end
+
+  defp accept(listen, proxy, host, port) do
+    case :gen_tcp.accept(listen) do
+      {:ok, client} ->
+        case GenServer.call(proxy, :mode) do
+          :forward -> pipe(client, host, port)
+          mode when mode in [:blackhole, :frozen] -> hold(client, proxy)
+        end
+
+        accept(listen, proxy, host, port)
+
+      {:error, _reason} ->
+        :ok
+    end
+  end
+
+  defp hold(client, proxy) do
+    :ok = :gen_tcp.controlling_process(client, proxy)
+    :ok = GenServer.call(proxy, {:hold, client})
+  end
+
+  defp pipe(client, host, port) do
+    {:ok, upstream} = :gen_tcp.connect(host, port, [:binary, active: false])
+
+    proxy = self_proxy()
+
+    for {from, to} <- [{client, upstream}, {upstream, client}] do
+      pid = spawn_link(fn -> receive do: (:go -> relay(from, to, proxy, :infinity)) end)
+      :ok = :gen_tcp.controlling_process(from, pid)
+      :ok = GenServer.call(proxy, {:register, pid})
+      send(pid, :go)
+    end
+  end
+
+  defp self_proxy, do: Process.get(:proxy)
+
+  defp relay(from, to, proxy, deadline) do
+    receive do
+      :release -> release(from, to, [])
+    after
+      0 -> relay_once(from, to, proxy, deadline)
+    end
+  end
+
+  defp relay_once(from, to, proxy, deadline) do
+    if expired?(deadline) do
+      close(from, to)
+    else
+      case :gen_tcp.recv(from, 0, 50) do
+        {:error, :timeout} ->
+          relay(from, to, proxy, deadline)
+
+        {:ok, data} ->
+          if deadline == :infinity and GenServer.call(proxy, :mode) == :frozen do
+            park(from, to, [data])
+          else
+            forward_data(from, to, proxy, deadline, data)
+          end
+
+        {:error, _reason} ->
+          if deadline == :infinity and GenServer.call(proxy, :mode) == :frozen do
+            park(from, to, [])
+          else
+            close(from, to)
+          end
+      end
+    end
+  end
+
+  defp expired?(:infinity), do: false
+  defp expired?(deadline), do: System.monotonic_time(:millisecond) > deadline
+
+  defp forward_data(from, to, proxy, deadline, data) do
+    case :gen_tcp.send(to, data) do
+      :ok -> relay(from, to, proxy, deadline)
+      {:error, _reason} -> close(from, to)
+    end
+  end
+
+  defp park(from, to, parked) do
+    receive do
+      :release -> release(from, to, parked)
+    end
+  end
+
+  defp release(from, to, parked) do
+    Enum.each(Enum.reverse(parked), &:gen_tcp.send(to, &1))
+    relay(from, to, nil, System.monotonic_time(:millisecond) + @grace)
+  end
+
+  defp close(from, to) do
+    :gen_tcp.close(from)
+    :gen_tcp.close(to)
+  end
+end