diff --git a/guides/Cluster.md b/guides/Cluster.md index b72e28de..8de77d76 100644 --- a/guides/Cluster.md +++ b/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 diff --git a/lib/event_store/advisory_locks.ex b/lib/event_store/advisory_locks.ex index bce4da40..59e086d6 100644 --- a/lib/event_store/advisory_locks.ex +++ b/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) diff --git a/lib/event_store/advisory_locks/protocol.ex b/lib/event_store/advisory_locks/protocol.ex index b34bb5b2..e5df0e76 100644 --- a/lib/event_store/advisory_locks/protocol.ex +++ b/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 diff --git a/test/advisory_locks_test.exs b/test/advisory_locks_test.exs index 9554fb4b..65066215 100644 --- a/test/advisory_locks_test.exs +++ b/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)