Packages
electric
1.2.1
1.7.8
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.10
1.6.9
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.16
1.4.16-beta-1
1.4.15
1.4.14
1.4.13
1.4.12
1.4.11
1.4.10
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.4
1.3.3
1.3.2
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.14
1.1.13
1.1.12
1.1.11
1.1.10
1.1.9
1.1.8
1.1.7
1.1.6
retired
1.1.5
retired
1.1.4
retired
1.1.3
retired
1.1.2
1.1.1
1.1.0
1.0.24
1.0.23
1.0.22
1.0.21
1.0.20
1.0.19
1.0.18
1.0.17
1.0.15
1.0.13
1.0.12
1.0.11
1.0.10
1.0.9
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
1.0.0-beta.23
1.0.0-beta.22
1.0.0-beta.20
1.0.0-beta.19
1.0.0-beta.18
1.0.0-beta.17
1.0.0-beta.16
1.0.0-beta.15
1.0.0-beta.14
1.0.0-beta.13
1.0.0-beta.12
1.0.0-beta.11
1.0.0-beta.10
1.0.0-beta.9
1.0.0-beta.8
1.0.0-beta.7
1.0.0-beta.6
1.0.0-beta.5
1.0.0-beta.4
1.0.0-beta.3
1.0.0-beta.2
1.0.0-beta.1
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.3
0.6.2
0.6.1
0.5.2
0.4.4
Postgres sync engine. Sync little subsets of your Postgres data into local apps and services.
Current section
Files
Jump to
Current section
Files
lib/electric/postgres/lock_breaker_connection.ex
defmodule Electric.Postgres.LockBreakerConnection do
@moduledoc """
A Postgres connection that is used to break an abandoned lock.
Electric takes out a session-level advisory lock on a separate connection to better manage the
ownership of the replication slot. Unfortunately, we have seen instances (especially on Neon),
where the Electric disconnects, but the lock is not auto-released.
For these cases, this breaker exists - it'll connect to the database, and check that for
a given lock name, if that lock is taken, there also exists an active replication slot with the
same name. If not, it'll terminate the backend that is holding the lock, under the assumption
that it's one of the abandoned locks.
"""
require Logger
@behaviour Postgrex.SimpleConnection
@type option ::
{:connection_opts, Keyword.t()}
| {:stack_id, String.t()}
@type options :: [option]
@spec start(options()) :: {:ok, pid()} | {:error, Postgrex.Error.t() | term()}
def start(opts) do
{connection_opts, init_opts} = Keyword.pop(opts, :connection_opts)
connection_opts = Electric.Utils.deobfuscate_password(connection_opts)
with {:ok, pid} <-
Postgrex.SimpleConnection.start_link(
__MODULE__,
init_opts |> Keyword.put(:database, connection_opts[:database]),
[
auto_reconnect: false,
sync_connect: true
] ++
connection_opts
) do
# unlink the lock breaker so that if it crashes it does not affect the caller,
# since it is a one shot fix attempt anyway
Process.unlink(pid)
{:ok, pid}
end
end
def stop_backends_and_close(server, lock_name, lock_connection_backend_pid \\ nil) do
send(server, {:stop_backends_and_close, lock_name, lock_connection_backend_pid})
end
@impl true
def init(opts) do
opts = Map.new(opts)
Process.set_label({:lock_breaker_connection, opts.stack_id})
metadata = [
is_connection_process?: true,
stack_id: opts.stack_id
]
Logger.metadata(metadata)
Electric.Telemetry.Sentry.set_tags_context(metadata)
{:ok, opts}
end
@impl true
def handle_connect(state) do
{:noreply, state}
end
@impl true
def handle_info(
{:stop_backends_and_close, lock_name, lock_connection_backend_pid},
state
) do
{:query, lock_breaker_query(lock_name, lock_connection_backend_pid, state.database),
Map.put(state, :lock_name, lock_name)}
end
@impl true
def handle_result([%Postgrex.Result{columns: ["pg_terminate_backend"]} = result], state) do
if result.num_rows == 0 do
Logger.debug("No stuck backends found")
else
Logger.notice(
"Terminated a stuck backend to free the lock #{state.lock_name} because slot with same name was inactive"
)
end
exit(:shutdown)
end
def handle_result(%Postgrex.Error{} = error, _) do
raise error
end
@impl true
def notify(_, _, _), do: :ok
defp lock_breaker_query(lock_name, lock_connection_backend_pid, database) do
# We're using a `WITH` clause to execute all this in one statement
# - See if there are existing but inactive replication slots with the given name
# - Find all backends that are holding locks with the same name
# - Terminate those backends
#
# It's generally impossible for this to return more than one row
"""
WITH inactive_slots AS (
select slot_name
from pg_replication_slots
where active = false and database = '#{database}' and slot_name = '#{lock_name}'
),
stuck_backends AS (
select pid
from pg_locks, inactive_slots
where
hashtext(slot_name) = (classid::bigint << 32) | objid::bigint
and locktype = 'advisory'
and objsubid = 1
and database = (select oid from pg_database where datname = '#{database}')
and granted
and pid != #{lock_connection_backend_pid || 0}
)
SELECT pg_terminate_backend(pid) FROM stuck_backends;
"""
end
end