fix(subscriptions): tell lock owners their session is gone when the call that hit it replies before the connection says so

guides/Cluster.md

@@ -23,7 +23,7 @@ Subscriptions hold their locks on a dedicated connection. When the database stop
 - While a lock query is running, the node learns of the loss once the query reaches its `:timeout` (15,000ms by default), plus up to two seconds spent trying to cancel it. With `timeout: :infinity` the query never gives up and `ping_timeout` does not apply to it, so the node is not told until the socket itself errors, and until then its subscriptions keep running on a session that has ended. Keep `:timeout` finite on any node that runs subscriptions.
 - While the connection is idle, the node learns of the loss when the connection next pings the database: within one to two times `idle_interval` (1,000ms by default), plus however long the ping waits for a reply. That wait is `ping_timeout`, which defaults to `:timeout`. With `timeout: :infinity` the ping waits forever, so set `ping_timeout` in `session_mode_pool` to bound it, for example `session_mode_pool: [ping_timeout: 15_000]`.
 
-PostgreSQL only releases the locks of a session once it notices the client is gone, which can take longer still unless `tcp_keepalives_idle` or `client_connection_check_interval` is configured on the server. Until then no other node can take the lock, so a slow release delays the hand-off but never lets two nodes record progress at the same time.
+PostgreSQL only releases the locks of a session once it notices the client is gone. When a node's host dies or the network drops without closing the connection, the server only notices through TCP keepalives, which with the Linux defaults takes over two hours. Set `tcp_keepalives_idle`, `tcp_keepalives_interval` and `tcp_keepalives_count` together on the server to shorten it, for example to 10, 5 and 3 for a release within about 25 seconds; `tcp_keepalives_idle` alone still leaves the probes to the default interval and count, and `client_connection_check_interval` does not help, since it only notices a connection the operating system already knows is closed. Until then no other node can take the lock, so a slow release delays the hand-off but never lets two nodes record progress at the same time.
 
 ## Automatic cluster formation
 

lib/event_store/advisory_locks.ex

@@ -12,6 +12,7 @@ defmodule EventStore.AdvisoryLocks do
   alias EventStore.MonitoredServer
 
   @verify_retry 1_000
+  @session_lost_wait 1_000
 
   defmodule State do
     @moduledoc false
@@ -393,6 +394,9 @@ defmodule EventStore.AdvisoryLocks do
   def handle_info({__MODULE__, :session_lost, conn_pid}, %State{} = state) when is_pid(conn_pid),
     do: {:noreply, disconnected(conn_pid, :session_ended, state)}
 
+  def handle_info({__MODULE__, :query_disconnected}, %State{} = state),
+    do: {:noreply, receive_session_loss(state, @session_lost_wait)}
+
   def handle_info({:disconnected, conn_pid}, %State{} = state) when is_pid(conn_pid),
     do: {:noreply, disconnected(conn_pid, :session_ended, state)}
 
@@ -432,16 +436,26 @@ defmodule EventStore.AdvisoryLocks do
 
   # A query held up by a connection that stopped answering leaves the notice of
   # its lost session queued behind the requests that arrived meanwhile, so it is
-  # handled before them instead of after each one has waited on the pool.
-  defp handle_pending_session_loss(%State{connection: %Connection{pid: conn_pid}} = state) do
+  # handled before them instead of after each one has waited on the pool. A query
+  # that failed on that connection replies before the notice is sent, so it is
+  # awaited briefly then.
+  defp handle_pending_session_loss(%State{} = state) do
+    receive do
+      {__MODULE__, :query_disconnected} -> receive_session_loss(state, @session_lost_wait)
+    after
+      0 -> receive_session_loss(state, 0)
+    end
+  end
+
+  defp receive_session_loss(%State{connection: %Connection{pid: conn_pid}} = state, timeout) do
     receive do
       {__MODULE__, :session_lost, ^conn_pid} -> disconnected(conn_pid, :session_ended, state)
     after
-      0 -> state
+      timeout -> state
     end
   end
 
-  defp handle_pending_session_loss(%State{} = state), do: state
+  defp receive_session_loss(%State{} = state, _timeout), do: state
 
   defp connected(conn_pid, %State{connection: connection} = state) do
     Connection.demonitor(connection)

lib/event_store/advisory_locks/protocol.ex

@@ -47,7 +47,16 @@ defmodule EventStore.AdvisoryLocks.Protocol do
 
   # A retry runs on the next connection before the lock server has handled the
   # loss of this session, so the lock owners would hear of it only afterwards.
-  defp without_retry({:disconnect_and_retry, error, state}), do: {:disconnect, error, state}
+  defp without_retry({:disconnect_and_retry, error, state}),
+    do: without_retry({:disconnect, error, state})
+
+  # The query runs in the lock server, which replies before this connection gets
+  # to announce the loss of its session, so it is told to wait for that notice.
+  defp without_retry({:disconnect, _error, _state} = result) do
+    send(self(), {EventStore.AdvisoryLocks, :query_disconnected})
+    result
+  end
+
   defp without_retry(result), do: result
 
   @impl DBConnection

test/advisory_locks_test.exs

@@ -1174,6 +1174,45 @@ defmodule EventStore.AdvisoryLocksTest do
     end
   end
 
+  describe "lock session ends while lock calls are queued and new connections hang" do
+    @tag capture_log: true
+    test "should send `lock_released` message before the next queued call waits on the pool",
+         %{conn: conn, schema: schema} do
+      {locks, proxy} =
+        start_proxied_lock_store(:failover_lock_store, schema, idle_interval: 60_000)
+
+      lock_pool = Module.concat([locks, Postgrex])
+
+      try do
+        {:ok, lock} = AdvisoryLocks.try_advisory_lock(locks, 1)
+
+        lock_connection = lock_connection_pid(lock_pool)
+        backend_pid = backend_pid(lock_pool)
+
+        :ok = EventStore.TcpProxy.blackhole(proxy)
+
+        %Postgrex.Result{rows: [[true]]} =
+          Postgrex.query!(conn, "SELECT pg_terminate_backend($1)", [backend_pid])
+
+        # Holds back the connection's notice until the failed call has replied.
+        :ok = :sys.suspend(lock_connection)
+
+        for key <- 2..5, do: spawn(fn -> AdvisoryLocks.try_advisory_lock(locks, key) end)
+
+        Wait.until(fn ->
+          {:messages, messages} = Process.info(lock_connection, :messages)
+          assert messages != []
+        end)
+
+        :ok = :sys.resume(lock_connection)
+
+        assert_receive {AdvisoryLocks, :lock_released, ^lock, _reason}, 1_000
+      after
+        stop_supervised!(EventStore.TcpProxy)
+      end
+    end
+  end
+
   # An event store whose lock connection goes through a proxy that can stop answering.
   defp start_proxied_lock_store(name, schema, session_mode_pool_opts) do
     config = Config.parsed(TestEventStore, :eventstore)