diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex
index 8fe9481..97b284d 100644
--- a/lib/event_store/advisory_locks.ex
+++ b/lib/event_store/advisory_locks.ex
@@ -7,11 +7,24 @@ defmodule EventStore.AdvisoryLocks do
 
   use GenServer
 
+  require Logger
+
   alias EventStore.MonitoredServer
 
+  @verify_retry 1_000
+
   defmodule State do
     @moduledoc false
-    defstruct [:conn, :query_timeout, :ref, :schema, locks: %{}]
+    defstruct [
+      :conn,
+      :connection,
+      :query_timeout,
+      :ref,
+      :schema,
+      :verify_timer,
+      locks: %{},
+      released: []
+    ]
 
     def new(opts) do
       conn = Keyword.fetch!(opts, :conn)
@@ -25,22 +38,47 @@ defmodule EventStore.AdvisoryLocks do
   defmodule Lock do
     @moduledoc false
 
+    alias EventStore.Storage.Session
+
     @type t :: %Lock{
             key: non_neg_integer(),
             owner: pid(),
-            ref: reference()
+            ref: reference(),
+            session: Session.t()
           }
-    defstruct [:key, :owner, :ref]
+    defstruct [:key, :owner, :ref, :session]
 
-    def new(key, owner) do
-      %Lock{key: key, owner: owner, ref: make_ref()}
+    def new(key, owner, %Session{} = session) do
+      %Lock{key: key, owner: owner, ref: make_ref(), session: session}
     end
 
     def ref(%Lock{ref: ref}), do: ref
+
+    def held_by?(%Lock{session: session}, %Session{} = current), do: session == current
+  end
+
+  defmodule Connection do
+    @moduledoc false
+
+    @type t :: %Connection{pid: pid(), monitor_ref: reference()}
+    defstruct [:pid, :monitor_ref]
+
+    def monitor(pid), do: %Connection{pid: pid, monitor_ref: Process.monitor(pid)}
+
+    def demonitor(nil), do: :ok
+
+    def demonitor(%Connection{monitor_ref: monitor_ref}),
+      do: Process.demonitor(monitor_ref, [:flush])
+
+    # Without a known connection any notice may be about the one holding the locks.
+    def current?(nil, _pid), do: true
+    def current?(%Connection{pid: pid}, pid), do: true
+    def current?(%Connection{}, _pid), do: false
   end
 
-  alias EventStore.AdvisoryLocks.{Lock, State}
+  alias EventStore.AdvisoryLocks.{Connection, Lock, State}
   alias EventStore.Storage
+  alias EventStore.Storage.Session
 
   def start_link(opts) do
     {start_opts, advisory_locks_opts} =
@@ -58,6 +96,10 @@ defmodule EventStore.AdvisoryLocks do
 
     {:ok, ref} = MonitoredServer.monitor(conn)
 
+    # 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)
+
     {:ok, %State{state | ref: ref}}
   end
 
@@ -101,9 +143,12 @@ defmodule EventStore.AdvisoryLocks do
   defp try_acquire_exclusive_lock(key, owner, timeout, %State{} = state) do
     %State{conn: conn, schema: schema} = state
 
-    with :ok <-
-           Storage.Lock.try_acquire_exclusive_lock(conn, key, schema: schema, timeout: timeout) do
-      {:ok, Lock.new(key, owner)}
+    with {:ok, %Session{} = session} <-
+           Storage.Lock.try_acquire_exclusive_session_lock(conn, key,
+             schema: schema,
+             timeout: timeout
+           ) do
+      {:ok, Lock.new(key, owner, session)}
     end
   end
 
@@ -121,9 +166,34 @@ defmodule EventStore.AdvisoryLocks do
 
     notify_lost_locks(locks, reason)
 
-    {:noreply, %State{state | locks: %{}}}
+    {:noreply, %State{state | connection: nil, locks: %{}, released: [], verify_timer: nil}}
   end
 
+  # A session that could not be verified never reported a disconnect, so the
+  # pending unlocks are reconciled again once it can answer.
+  def handle_info(:verify_sessions, %State{} = state),
+    do: {:noreply, verify_sessions(%State{state | verify_timer: nil})}
+
+  # The connection pool outlives its connections, so a reconnect replaces the
+  # session that held the locks without the pool going down.
+  def handle_info({:connected, conn_pid}, %State{} = state) when is_pid(conn_pid),
+    do: {:noreply, connected(conn_pid, state)}
+
+  def handle_info({:connected, conn_pid, _tag}, %State{} = state) when is_pid(conn_pid),
+    do: {:noreply, connected(conn_pid, state)}
+
+  def handle_info({:disconnected, conn_pid}, %State{} = state) when is_pid(conn_pid),
+    do: {:noreply, disconnected(conn_pid, :session_ended, state)}
+
+  def handle_info({:disconnected, conn_pid, _tag}, %State{} = state) when is_pid(conn_pid),
+    do: {:noreply, disconnected(conn_pid, :session_ended, state)}
+
+  def handle_info(
+        {:DOWN, monitor_ref, :process, conn_pid, reason},
+        %State{connection: %Connection{monitor_ref: monitor_ref}} = state
+      ),
+      do: {:noreply, disconnected(conn_pid, reason, state)}
+
   # Release lock when the lock owner process terminates.
   def handle_info({:DOWN, ref, :process, _pid, _reason}, %State{} = state) do
     %State{locks: locks} = state
@@ -150,10 +220,80 @@ defmodule EventStore.AdvisoryLocks do
     {:stop, reason, state}
   end
 
+  defp connected(conn_pid, %State{connection: connection} = state) do
+    Connection.demonitor(connection)
+
+    verify_sessions(%State{state | connection: Connection.monitor(conn_pid)})
+  end
+
+  # A lock taken through the next connection can be recorded before this notice
+  # is handled, so it is released here too and only unlocked once verified.
+  defp disconnected(conn_pid, reason, %State{connection: connection} = state) do
+    if Connection.current?(connection, conn_pid) do
+      %State{locks: locks, released: released} = state
+
+      Connection.demonitor(connection)
+
+      notify_lost_locks(locks, reason)
+
+      %State{state | connection: nil, locks: %{}, released: Map.values(locks) ++ released}
+    else
+      state
+    end
+  end
+
+  defp verify_sessions(%State{locks: locks, released: []} = state) when map_size(locks) == 0,
+    do: state
+
+  defp verify_sessions(%State{} = state) do
+    %State{conn: conn, locks: locks, query_timeout: query_timeout, released: released} = state
+
+    case Session.current(conn, timeout: query_timeout) do
+      {:ok, %Session{} = session} ->
+        {held, lost} = Map.split_with(locks, fn {_ref, lock} -> Lock.held_by?(lock, session) end)
+
+        notify_lost_locks(lost, :session_ended)
+
+        # A lock whose session has ended died with it, but one still held by this
+        # session is only cleared once its unlock is acknowledged.
+        pending =
+          for %Lock{key: key} = lock <- released,
+              Lock.held_by?(lock, session),
+              release_lock(key, state) != :ok,
+              do: lock
+
+        # A live session confirms the locks still held, so polling stops once
+        # the pending unlocks are drained.
+        maybe_schedule_verify_retry(%State{state | locks: held, released: pending})
+
+      # A failed query does not prove the session ended, so verification is
+      # retried until a live session can confirm the locks and release them.
+      {:error, _error} ->
+        schedule_verify_retry(state)
+    end
+  end
+
+  defp maybe_schedule_verify_retry(%State{released: []} = state), do: state
+  defp maybe_schedule_verify_retry(%State{} = state), do: schedule_verify_retry(state)
+
+  defp schedule_verify_retry(%State{verify_timer: ref} = state) when is_reference(ref), do: state
+
+  defp schedule_verify_retry(%State{} = state),
+    do: %State{state | verify_timer: Process.send_after(self(), :verify_sessions, @verify_retry)}
+
+  defp notify_lost_locks(locks, _reason) when map_size(locks) == 0, do: :ok
+
   defp notify_lost_locks(locks, reason) do
-    for {_ref, %Lock{} = lock} <- locks do
+    Logger.warning(
+      "Lost #{map_size(locks)} advisory lock(s) held by this event store due to: " <>
+        inspect(reason)
+    )
+
+    for {owner_ref, %Lock{} = lock} <- locks do
       %Lock{owner: owner, ref: ref} = lock
 
+      Process.demonitor(owner_ref, [:flush])
+
       :ok = Process.send(owner, {__MODULE__, :lock_released, ref, reason}, [])
     end
 
diff --git a/lib/event_store/config.ex b/lib/event_store/config.ex
index b8b28c6..97cb4f1 100644
--- a/lib/event_store/config.ex
+++ b/lib/event_store/config.ex
@@ -196,6 +196,23 @@ defmodule EventStore.Config do
     |> Keyword.put(:pool_size, 1)
   end
 
+  @doc """
+  Add a connection listener while keeping any listeners already configured.
+
+  A listener given as a registered name is wrapped in a `{name, node}` tuple, so
+  notifications sent before the name is registered are dropped instead of
+  crashing the connection.
+  """
+  def add_connection_listener(opts, name) when is_atom(name) do
+    listener = {name, node()}
+
+    case Keyword.get(opts, :connection_listeners) do
+      nil -> Keyword.put(opts, :connection_listeners, [listener])
+      {listeners, tag} -> Keyword.put(opts, :connection_listeners, {[listener | listeners], tag})
+      listeners -> Keyword.put(opts, :connection_listeners, [listener | listeners])
+    end
+  end
+
   # Get the optional session mode pool to be used for persistent Postgres
   # connections (advisory locks and notifications). Uses the default Postgres
   # configuration if not specified.
diff --git a/lib/event_store/sql/statements.ex b/lib/event_store/sql/statements.ex
index c980173..6d53011 100644
--- a/lib/event_store/sql/statements.ex
+++ b/lib/event_store/sql/statements.ex
@@ -19,11 +19,13 @@ defmodule EventStore.Sql.Statements do
         {:insert_subscription, [:schema]},
         {:delete_subscription, [:schema]},
         {:try_advisory_lock, [:schema]},
+        {:try_advisory_session_lock, [:schema]},
         {:advisory_unlock, [:schema]},
         {:subscription_ack, [:schema]},
         {:insert_snapshot, [:schema]},
         {:delete_snapshot, [:schema]},
         {:query_all_subscriptions, [:schema]},
+        {:query_session, []},
         {:query_snapshot, [:schema]},
         {:query_stream_info, [:schema]},
         {:query_stream_events_backward, [:schema]},
diff --git a/lib/event_store/sql/statements/query_session.sql.eex b/lib/event_store/sql/statements/query_session.sql.eex
new file mode 100644
index 0000000..8881143
--- /dev/null
+++ b/lib/event_store/sql/statements/query_session.sql.eex
@@ -0,0 +1,4 @@
+SELECT pg_backend_pid(), COALESCE(
+  NULLIF(current_setting('eventstore.session_id', true), ''),
+  set_config('eventstore.session_id', md5(random()::text || clock_timestamp()::text), false)
+);
diff --git a/lib/event_store/sql/statements/try_advisory_session_lock.sql.eex b/lib/event_store/sql/statements/try_advisory_session_lock.sql.eex
new file mode 100644
index 0000000..2634350
--- /dev/null
+++ b/lib/event_store/sql/statements/try_advisory_session_lock.sql.eex
@@ -0,0 +1,7 @@
+SELECT pg_try_advisory_lock(
+  '"<%= schema %>".subscriptions'::regclass::oid::int,
+  (CASE WHEN $1 > 2147483647 THEN mod($1, 2147483647) ELSE $1 END)::int
+), pg_backend_pid(), COALESCE(
+  NULLIF(current_setting('eventstore.session_id', true), ''),
+  set_config('eventstore.session_id', md5(random()::text || clock_timestamp()::text), false)
+);
diff --git a/lib/event_store/storage/lock.ex b/lib/event_store/storage/lock.ex
index 06e19d0..9fd609b 100644
--- a/lib/event_store/storage/lock.ex
+++ b/lib/event_store/storage/lock.ex
@@ -2,6 +2,7 @@ defmodule EventStore.Storage.Lock do
   @moduledoc false
 
   alias EventStore.Sql.Statements
+  alias EventStore.Storage.Session
 
   @doc """
   Use two 32-bit key values for advisory locks where the first key acts as the
@@ -27,6 +28,27 @@ defmodule EventStore.Storage.Lock do
     end
   end
 
+  @doc """
+  Like `try_acquire_exclusive_lock/3`, but also returns the session holding the
+  lock, which releases it when it ends.
+  """
+  def try_acquire_exclusive_session_lock(conn, key, opts) do
+    {schema, opts} = Keyword.pop(opts, :schema)
+
+    query = Statements.try_advisory_session_lock(schema)
+
+    case Postgrex.query(conn, query, [key], opts) do
+      {:ok, %Postgrex.Result{rows: [[true, backend_pid, session_id]]}} ->
+        {:ok, Session.new(backend_pid, session_id)}
+
+      {:ok, %Postgrex.Result{rows: [[false, _backend_pid, _session_id]]}} ->
+        {:error, :lock_already_taken}
+
+      {:error, _error} = reply ->
+        reply
+    end
+  end
+
   def unlock(conn, key, opts) do
     {schema, opts} = Keyword.pop(opts, :schema)
 
diff --git a/lib/event_store/storage/session.ex b/lib/event_store/storage/session.ex
new file mode 100644
index 0000000..ece87dc
--- /dev/null
+++ b/lib/event_store/storage/session.ex
@@ -0,0 +1,26 @@
+defmodule EventStore.Storage.Session do
+  @moduledoc false
+
+  # The id is a session level setting the session sets the first time it is read,
+  # because pg_stat_activity hides backend_start from a role switched to with SET ROLE.
+
+  alias EventStore.Sql.Statements
+
+  @type t :: %__MODULE__{backend_pid: pos_integer(), id: String.t()}
+
+  defstruct [:backend_pid, :id]
+
+  def new(backend_pid, id) when is_integer(backend_pid) and is_binary(id) and id != "" do
+    %__MODULE__{backend_pid: backend_pid, id: id}
+  end
+
+  def current(conn, opts) do
+    case Postgrex.query(conn, Statements.query_session(), [], opts) do
+      {:ok, %Postgrex.Result{rows: [[backend_pid, id]]}} ->
+        {:ok, new(backend_pid, id)}
+
+      {:error, _error} = reply ->
+        reply
+    end
+  end
+end
diff --git a/lib/event_store/supervisor.ex b/lib/event_store/supervisor.ex
index 7812ee3..d2da6da 100644
--- a/lib/event_store/supervisor.ex
+++ b/lib/event_store/supervisor.ex
@@ -58,7 +58,11 @@ defmodule EventStore.Supervisor do
 
         advisory_locks_name = Module.concat([name, AdvisoryLocks])
         advisory_locks_postgrex_conn = Module.concat([advisory_locks_name, Postgrex])
-        advisory_locks_config = Config.advisory_locks_postgrex_opts(config)
+
+        advisory_locks_config =
+          config
+          |> Config.advisory_locks_postgrex_opts()
+          |> Config.add_connection_listener(advisory_locks_name)
 
         subscriptions_name = Module.concat([name, Subscriptions.Supervisor])
         subscriptions_registry_name = Module.concat([name, Subscriptions.Registry])
diff --git a/test/advisory_locks_test.exs b/test/advisory_locks_test.exs
index f51ce4c..7292fb5 100644
--- a/test/advisory_locks_test.exs
+++ b/test/advisory_locks_test.exs
@@ -3,7 +3,7 @@ defmodule EventStore.AdvisoryLocksTest do
 
   import ExUnit.CaptureLog
 
-  alias EventStore.{AdvisoryLocks, Config, Wait}
+  alias EventStore.{AdvisoryLocks, Config, MonitoredServer, Wait}
   alias EventStore.Storage
 
   @locks TestEventStore.EventStore.AdvisoryLocks
@@ -79,6 +79,145 @@ defmodule EventStore.AdvisoryLocksTest do
 
       assert_receive({AdvisoryLocks, :lock_released, ^lock, :shutdown})
     end
+
+    test "should send `lock_released` message when the lock session ends and the connection reconnects",
+         %{conn: conn, schema: schema} do
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      terminate_lock_session(conn)
+
+      assert_receive {AdvisoryLocks, :lock_released, ^lock, _reason}, 5_000
+
+      Wait.until(fn ->
+        assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+      end)
+    end
+
+    test "should keep a lock acquired after the connection reconnects", %{conn: conn} do
+      {:ok, lost_lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      terminate_lock_session(conn)
+
+      assert_receive {AdvisoryLocks, :lock_released, ^lost_lock, _reason}, 5_000
+
+      kept_lock =
+        Wait.until(fn ->
+          assert {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 2)
+
+          lock
+        end)
+
+      refute_receive {AdvisoryLocks, :lock_released, ^kept_lock, _reason}, 1_000
+    end
+
+    test "should send `lock_released` message without waiting for the connection to reconnect",
+         %{conn: conn} do
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      without_reconnects(fn _lock_connection ->
+        terminate_lock_session(conn)
+
+        # Using the ended session makes the connection notice it now, not at its next idle ping.
+        # The query bypasses the lock server so it stays free to handle the notice, and
+        # `queue: false` stops a retry from waiting for a connection that cannot start.
+        assert {:error, _error} = Postgrex.query(@conn, "SELECT 1", [], queue: false)
+
+        assert_receive {AdvisoryLocks, :lock_released, ^lock, _reason}, 500
+      end)
+    end
+
+    test "should keep locks when a notice for a previous connection arrives late" do
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      send(@locks, {:connected, lock_connection_pid()})
+
+      previous_connection = spawn(fn -> :ok end)
+
+      send(@locks, {:disconnected, previous_connection})
+
+      refute_receive {AdvisoryLocks, :lock_released, ^lock, _reason}, 500
+    end
+
+    test "should unlock a lock reported lost while its session is still alive",
+         %{conn: conn, schema: schema} do
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      lock_connection = lock_connection_pid()
+
+      send(@locks, {:connected, lock_connection})
+      send(@locks, {:disconnected, lock_connection})
+
+      assert_receive {AdvisoryLocks, :lock_released, ^lock, :session_ended}
+
+      send(@locks, {:connected, lock_connection})
+
+      Wait.until(fn ->
+        assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+      end)
+    end
+
+    test "should unlock a lock reported lost once a session that could not be verified answers",
+         %{conn: conn, schema: schema} do
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      lock_connection = lock_connection_pid()
+
+      send(@locks, {:connected, lock_connection})
+      send(@locks, {:disconnected, lock_connection})
+
+      assert_receive {AdvisoryLocks, :lock_released, ^lock, :session_ended}
+
+      # The session still holds the lock but cannot answer the verification query,
+      # and it never disconnects, so nothing else would retry the pending unlock.
+      Postgrex.query!(@conn, "BEGIN", [])
+      assert {:error, _error} = Postgrex.query(@conn, "SELECT 1 / 0", [])
+
+      send(@locks, {:connected, lock_connection})
+      :sys.get_state(@locks)
+
+      Postgrex.query!(@conn, "ROLLBACK", [])
+
+      Wait.until(5_000, fn ->
+        assert :ok = Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+      end)
+    end
+
+    test "should retry verification for an active lock when a reconnect query fails",
+         %{conn: conn, schema: schema} do
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      lock_connection = lock_connection_pid()
+
+      # A reconnect notice makes verification, not the dropped monitor, responsible
+      # for the lock. The verification query cannot answer while the session is in a
+      # failed transaction, so with only an active lock held it must still schedule a
+      # retry rather than leave the owner holding a lock nothing will reconcile.
+      Postgrex.query!(@conn, "BEGIN", [])
+      assert {:error, _error} = Postgrex.query(@conn, "SELECT 1 / 0", [])
+
+      send(@locks, {:connected, lock_connection})
+
+      assert %{verify_timer: timer} = :sys.get_state(@locks)
+      assert is_reference(timer)
+
+      Postgrex.query!(@conn, "ROLLBACK", [])
+
+      # The session was alive all along, so the retry confirms the lock and keeps it.
+      refute_receive {AdvisoryLocks, :lock_released, ^lock, _reason}, 1_500
+
+      assert {:error, :lock_already_taken} =
+               Storage.Lock.try_acquire_exclusive_lock(conn, 1, schema: schema)
+    end
+
+    test "should send `lock_released` message when the lock connection crashes" do
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(@locks, 1)
+
+      without_reconnects(fn lock_connection ->
+        Process.exit(lock_connection, :kill)
+
+        assert_receive {AdvisoryLocks, :lock_released, ^lock, _reason}, 500
+      end)
+    end
   end
 
   describe "graceful shutdown" do
@@ -173,6 +312,84 @@ defmodule EventStore.AdvisoryLocksTest do
     end
   end
 
+  describe "lock connection that switches role after connecting" do
+    test "should acquire lock and keep running" do
+      locks_name = Module.concat([__MODULE__, SwitchedRole, AdvisoryLocks])
+      conn_name = Module.concat([locks_name, Postgrex])
+
+      config = Config.parsed(TestEventStore, :eventstore)
+
+      postgrex_config =
+        config
+        |> Config.advisory_locks_postgrex_opts()
+        |> Config.add_connection_listener(locks_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
+      )
+
+      locks =
+        start_supervised!(
+          {AdvisoryLocks,
+           conn: conn_name,
+           query_timeout: Keyword.fetch!(config, :timeout),
+           schema: "public",
+           name: locks_name}
+        )
+
+      assert {:ok, lock} = AdvisoryLocks.try_advisory_lock(locks_name, 1)
+      assert is_reference(lock)
+      assert Process.alive?(locks)
+    end
+  end
+
+  describe "lock connection that connected before the lock server started" 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])
+
+      config = Config.parsed(TestEventStore, :eventstore)
+
+      postgrex_config =
+        config
+        |> Config.advisory_locks_postgrex_opts()
+        |> Config.add_connection_listener(locks_name)
+
+      start_supervised!(
+        {MonitoredServer,
+         mfa: {Postgrex, :start_link, [postgrex_config]}, name: conn_name, backoff_min: 30_000},
+        id: conn_name
+      )
+
+      # The notice for this connection is sent to a lock server that does not exist yet.
+      first_session = backend_pid(conn_name)
+
+      start_supervised!(
+        {AdvisoryLocks,
+         conn: conn_name,
+         query_timeout: Keyword.fetch!(config, :timeout),
+         schema: "public",
+         name: locks_name}
+      )
+
+      Wait.until(5_000, fn ->
+        refute backend_pid(conn_name) == first_session,
+               "the connection made before the lock server started was not replaced"
+      end)
+
+      {:ok, lock} = AdvisoryLocks.try_advisory_lock(locks_name, 1)
+
+      without_reconnects(conn_name, fn lock_connection ->
+        Process.exit(lock_connection, :kill)
+
+        assert_receive {AdvisoryLocks, :lock_released, ^lock, _reason}, 500
+      end)
+    end
+  end
+
   defp hold_only_connection do
     %{pid: conn} = Process.whereis(@conn) |> :sys.get_state()
 
@@ -206,6 +423,67 @@ defmodule EventStore.AdvisoryLocksTest do
     end)
   end
 
+  defp terminate_lock_session(conn) do
+    %Postgrex.Result{rows: [[backend_pid]]} =
+      Postgrex.query!(@conn, "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)
+  end
+
+  defp without_reconnects(conn \\ @conn, fun) do
+    lock_connection = lock_connection_pid(conn)
+
+    {:dictionary, dictionary} = Process.info(lock_connection, :dictionary)
+    [connection_supervisor | _ancestors] = dictionary[:"$ancestors"]
+
+    :ok = :sys.suspend(connection_supervisor)
+
+    try do
+      fun.(lock_connection)
+    after
+      :ok = :sys.resume(connection_supervisor)
+    end
+  end
+
+  defp lock_connection_pid(conn \\ @conn) do
+    backend_pid = backend_pid(conn)
+
+    Enum.find(Process.list(), fn pid ->
+      with {:dictionary, dictionary} <- Process.info(pid, :dictionary),
+           {DBConnection.Connection, _function, _arity} <- dictionary[:"$initial_call"],
+           {_state, %{state: %Postgrex.Protocol{connection_id: ^backend_pid}}} <-
+             connection_state(pid) do
+        true
+      else
+        _ -> false
+      end
+    end)
+  end
+
+  defp backend_pid(conn) do
+    %Postgrex.Result{rows: [[backend_pid]]} = Postgrex.query!(conn, "SELECT pg_backend_pid()", [])
+
+    backend_pid
+  end
+
+  defp connection_state(pid) do
+    :sys.get_state(pid, 100)
+  catch
+    :exit, _reason -> nil
+  end
+
   defp shutdown_database_connection do
     %{pid: pid} = Process.whereis(@conn) |> :sys.get_state()
 
diff --git a/test/config_test.exs b/test/config_test.exs
index 3aa8322..f91fb4c 100644
--- a/test/config_test.exs
+++ b/test/config_test.exs
@@ -224,6 +224,27 @@ defmodule EventStore.ConfigTest do
     end
   end
 
+  describe "add connection listener" do
+    test "adds the listener as a `{name, node}` tuple" do
+      opts = Config.add_connection_listener([], :listener)
+
+      assert opts[:connection_listeners] == [{:listener, node()}]
+    end
+
+    test "keeps configured listeners" do
+      opts = Config.add_connection_listener([connection_listeners: [:configured]], :listener)
+
+      assert opts[:connection_listeners] == [{:listener, node()}, :configured]
+    end
+
+    test "keeps the tag of configured tagged listeners" do
+      opts =
+        Config.add_connection_listener([connection_listeners: {[:configured], :tag}], :listener)
+
+      assert opts[:connection_listeners] == {[{:listener, node()}, :configured], :tag}
+    end
+  end
+
   test "parse url with query parameters" do
     config = [
       url: "postgres://username:password@localhost/database?ssl=true&pool_size=5&timeout=120000"
diff --git a/test/storage/lock_test.exs b/test/storage/lock_test.exs
index b446016..4fc2dce 100644
--- a/test/storage/lock_test.exs
+++ b/test/storage/lock_test.exs
@@ -2,7 +2,7 @@ defmodule EventStore.Storage.LockTest do
   use EventStore.StorageCase
 
   alias EventStore.ProcessHelper
-  alias EventStore.Storage.Lock
+  alias EventStore.Storage.{Lock, Session}
   alias EventStore.Wait
 
   test "acquire exclusive subscription lock", %{postgrex_config: postgrex_config, schema: schema} do
@@ -41,5 +41,44 @@ defmodule EventStore.Storage.LockTest do
     ProcessHelper.shutdown(conn2)
   end
 
+  describe "on a connection that switches role after connecting" do
+    test "acquire exclusive lock", %{postgrex_config: postgrex_config, schema: schema} do
+      conn1 = start_switched_role_connection(postgrex_config, :conn1)
+      conn2 = start_switched_role_connection(postgrex_config, :conn2)
+
+      key = random_key()
+
+      assert :ok = Lock.try_acquire_exclusive_lock(conn1, key, schema: schema)
+
+      assert {:error, :lock_already_taken} =
+               Lock.try_acquire_exclusive_lock(conn2, key, schema: schema)
+    end
+
+    test "acquire exclusive session lock", %{postgrex_config: postgrex_config, schema: schema} do
+      conn1 = start_switched_role_connection(postgrex_config, :conn1)
+      conn2 = start_switched_role_connection(postgrex_config, :conn2)
+
+      assert {:ok, %Session{} = session} =
+               Lock.try_acquire_exclusive_session_lock(conn1, random_key(), schema: schema)
+
+      assert {:ok, ^session} = Session.current(conn1, [])
+      assert {:ok, ^session} = Session.current(conn1, [])
+
+      assert {:ok, %Session{} = other_session} = Session.current(conn2, [])
+      refute other_session.id == session.id
+    end
+  end
+
+  defp start_switched_role_connection(postgrex_config, id) do
+    postgrex_config =
+      Keyword.put(
+        postgrex_config,
+        :after_connect,
+        {Postgrex, :query!, ["SET ROLE pg_signal_backend", []]}
+      )
+
+    start_supervised!({Postgrex, postgrex_config}, id: id)
+  end
+
   def random_key, do: :rand.uniform(2_147_483_647)
 end
diff --git a/test/subscriptions/subscription_locking_test.exs b/test/subscriptions/subscription_locking_test.exs
index 1930073..fbf69d3 100644
--- a/test/subscriptions/subscription_locking_test.exs
+++ b/test/subscriptions/subscription_locking_test.exs
@@ -5,6 +5,7 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do
   alias EventStore.Subscriptions.Subscription
 
   @conn TestEventStore.Postgrex
+  @advisory_locks_conn TestEventStore.EventStore.AdvisoryLocks.Postgrex
 
   setup do
     subscription_name = UUID.uuid4()
@@ -105,6 +106,26 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do
     end
   end
 
+  describe "subscription lock session ended" do
+    setup [:create_subscription]
+
+    test "should resend in-flight events after the lock connection reconnects", %{
+      conn: conn,
+      event_store: event_store,
+      subscription: subscription
+    } do
+      assert_receive {:subscribed, ^subscription}
+
+      append_events_to_stream(event_store, 3)
+
+      assert_receive_events([1, 2, 3])
+
+      :ok = terminate_lock_session(conn)
+
+      assert_receive_events([1, 2, 3], 5_000)
+    end
+  end
+
   describe "duplicate subscriptions" do
     setup [:lock_subscription, :create_subscription]
 
@@ -185,6 +206,16 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do
     [subscription: subscription]
   end
 
+  defp terminate_lock_session(conn) do
+    %Postgrex.Result{rows: [[backend_pid]]} =
+      Postgrex.query!(@advisory_locks_conn, "SELECT pg_backend_pid()", [])
+
+    %Postgrex.Result{rows: [[true]]} =
+      Postgrex.query!(conn, "SELECT pg_terminate_backend($1)", [backend_pid])
+
+    :ok
+  end
+
   defp reconnect(subscription) do
     Process.send(subscription, :subscribe_to_stream, [])
   end
@@ -208,8 +239,8 @@ defmodule EventStore.Subscriptions.SubscriptionLockingTest do
     stream_uuid
   end
 
-  defp assert_receive_events(expected_event_numbers) do
-    assert_receive {:events, received_events}
+  defp assert_receive_events(expected_event_numbers, timeout \\ 2_000) do
+    assert_receive {:events, received_events}, timeout
     assert pluck(received_events, :event_number) == expected_event_numbers
   end