fix: Handle replication slot conflicts

.changeset/poor-candles-fly.md

@@ -0,0 +1,7 @@
+---
+"@core/sync-service": patch
+---
+
+- Wait for advisory lock on replication slot to enable rolling deploys.
+- Configurable replication slot and publication name using `REPLICATION_STREAM_ID` environment variable.
+- Add `HealthCheckPlug` API endopint at `v1/health` that returns `waiting`, `starting`, `active`, and `stopping` statuses.

integration-tests/tests/invalidated-replication-slot.lux

@@ -7,7 +7,7 @@
 [my invalidated_slot_error=
   """
   [error] GenServer Electric.ConnectionManager terminating
-  ** (Postgrex.Error) ERROR 55000 (object_not_in_prerequisite_state) cannot read from logical replication slot "electric_slot"
+  ** (Postgrex.Error) ERROR 55000 (object_not_in_prerequisite_state) cannot read from logical replication slot "electric_slot_default"
 
   This slot has been invalidated because it exceeded the maximum reserved size.
   """]
@@ -34,7 +34,7 @@
 
 ## Confirm slot invalidation in Postgres.
 [shell pg]
-  ?invalidating slot "electric_slot" because its restart_lsn \d+/\d+ exceeds max_slot_wal_keep_size
+  ?invalidating slot "electric_slot_default" because its restart_lsn [\d\w]+/[\d\w]+ exceeds max_slot_wal_keep_size
 
 ## Observe the fatal connection error.
 [shell electric]

integration-tests/tests/macros.luxinc

@@ -68,10 +68,14 @@
 [endmacro]
 
 [macro setup_electric]
-  [shell electric]
+  [invoke setup_electric_shell "electric" "3000"]
+[endmacro]
+
+[macro setup_electric_shell shell_name port]
+  [shell $shell_name]
     -$fail_pattern
 
-    !DATABASE_URL=$database_url ../electric_dev.sh
+    !DATABASE_URL=$database_url PORT=$port ../electric_dev.sh
 [endmacro]
 
 [macro teardown]

integration-tests/tests/postgres-disconnection.lux

@@ -22,7 +22,7 @@
 
 ## Observe the connection error.
 [shell electric]
-  ??[warning] Database connection in replication mode failed
+  ??[warning] Database connection in lock_connection mode failed
   ??[warning] Reconnecting in 
 
 ## Start the Postgres container back up.

integration-tests/tests/rolling-deploy.lux

@@ -0,0 +1,62 @@
+[doc Verify handling of an Electric rolling deploy]
+
+[include macros.luxinc]
+
+[global pg_container_name=rolling-deploy__pg]
+
+###
+
+## Start a new Postgres cluster
+[invoke setup_pg "" ""]
+
+## Start the first sync service.
+[invoke setup_electric_shell "electric_1" "3000"]
+
+[shell electric_1]
+  ??[info] Acquiring lock from postgres with name electric_slot_default
+  ??[info] Lock acquired from postgres with name electric_slot_default
+  ??[info] Starting replication from postgres
+
+# First service should be health and active
+[shell orchestator]
+  !curl -X GET http://localhost:3000/v1/health
+  ??{"status":"active"}
+
+## Start the second sync service.
+[invoke setup_electric_shell "electric_2" "3001"]
+
+## Assert that the lock is not acquired and replication does not start
+## in the second electric
+[shell electric_2]
+  -Lock acquired from postgres|Starting replication from postgres|$fail_pattern
+  ??[info] Acquiring lock from postgres with name electric_slot_default
+  [sleep 2]
+
+
+# Second service should be in waiting state, ready to take over
+[shell orchestator]
+  !curl -X GET http://localhost:3000/v1/health
+  ??{"status":"active"}
+  !curl -X GET http://localhost:3001/v1/health
+  ??{"status":"waiting"}
+
+## Terminate first electric
+[shell electric_1]
+  !System.halt()
+
+  # Confirm Electric process exit.
+  ??$PS1
+
+## Lock should now be acquired and replication starting
+[shell electric_2]
+  -$fail_pattern
+  ??[info] Lock acquired from postgres with name electric_slot_default
+  ??[info] Starting replication from postgres
+
+# Second service is now healthy and active
+[shell orchestator]
+  !curl -X GET http://localhost:3001/v1/health
+  ??{"status":"active"}
+
+[cleanup]
+  [invoke teardown]

packages/sync-service/config/runtime.exs

@@ -161,6 +161,8 @@ config :electric,
   instance_id: instance_id,
   telemetry_statsd_host: statsd_host,
   db_pool_size: env!("DB_POOL_SIZE", :integer, 50),
+  replication_stream_id: env!("REPLICATION_STREAM_ID", :string, "default"),
+  service_port: env!("PORT", :integer, 3000),
   prometheus_port: prometheus_port,
   storage: storage,
   persistent_kv: persistent_kv

packages/sync-service/lib/electric/application.ex

@@ -22,8 +22,9 @@ defmodule Electric.Application do
 
     persistent_kv = apply(kv_module, kv_fun, [kv_params])
 
-    publication_name = "electric_publication"
-    slot_name = "electric_slot"
+    replication_stream_id = Application.fetch_env!(:electric, :replication_stream_id)
+    publication_name = "electric_publication_#{replication_stream_id}"
+    slot_name = "electric_slot_#{replication_stream_id}"
 
     with {:ok, storage_opts} <- storage_module.shared_opts(storage_opts) do
       storage = {storage_module, storage_opts}
@@ -32,6 +33,14 @@ defmodule Electric.Application do
         Electric.ConnectionManager.get_pg_version(Electric.ConnectionManager)
       end
 
+      get_service_status = fn ->
+        Electric.ServiceStatus.check(
+          get_connection_status: fn ->
+            Electric.ConnectionManager.get_status(Electric.ConnectionManager)
+          end
+        )
+      end
+
       prepare_tables_fn =
         {Electric.Postgres.Configuration, :configure_tables_for_replication!,
          [get_pg_version, publication_name]}
@@ -103,12 +112,13 @@ defmodule Electric.Application do
                 storage: storage,
                 registry: Registry.ShapeChanges,
                 shape_cache: {Electric.ShapeCache, []},
+                get_service_status: get_service_status,
                 inspector: inspector,
                 long_poll_timeout: 20_000,
                 max_age: Application.fetch_env!(:electric, :cache_max_age),
                 stale_age: Application.fetch_env!(:electric, :cache_stale_age),
                 allow_shape_deletion: Application.get_env(:electric, :allow_shape_deletion, false)},
-             port: 3000,
+             port: Application.fetch_env!(:electric, :service_port),
              thousand_island_options: http_listener_options()}
           ]
           |> add_prometheus_router(Application.fetch_env!(:electric, :prometheus_port))

packages/sync-service/lib/electric/connection_manager.ex

@@ -43,10 +43,14 @@ defmodule Electric.ConnectionManager do
       :shape_cache,
       # PID of the replication client.
       :replication_client_pid,
+      # PID of the Postgres connection lock.
+      :lock_connection_pid,
       # PID of the database connection pool (a `Postgrex` process).
       :pool_pid,
       # Backoff term used for reconnection with exponential back-off.
       :backoff,
+      # Flag indicating whether the lock on the replication has been acquired.
+      :pg_lock_acquired,
       # PostgreSQL server version
       :pg_version,
       :electric_instance_id
@@ -57,6 +61,8 @@ defmodule Electric.ConnectionManager do
 
   require Logger
 
+  @type status :: :waiting | :starting | :active
+
   @type option ::
           {:connection_opts, Keyword.t()}
           | {:replication_opts, Keyword.t()}
@@ -69,6 +75,8 @@ defmodule Electric.ConnectionManager do
 
   @name __MODULE__
 
+  @lock_status_logging_interval 10_000
+
   @doc """
   Returns the version of the PostgreSQL server.
   """
@@ -77,6 +85,14 @@ defmodule Electric.ConnectionManager do
     GenServer.call(server, :get_pg_version)
   end
 
+  @doc """
+  Returns the status of the connection manager.
+  """
+  @spec get_status(GenServer.server()) :: status()
+  def get_status(server) do
+    GenServer.call(server, :get_status)
+  end
+
   @spec start_link(options) :: GenServer.on_start()
   def start_link(opts) do
     GenServer.start_link(__MODULE__, opts, name: @name)
@@ -112,21 +128,54 @@ defmodule Electric.ConnectionManager do
         timeline_opts: timeline_opts,
         log_collector: Keyword.fetch!(opts, :log_collector),
         shape_cache: Keyword.fetch!(opts, :shape_cache),
+        pg_lock_acquired: false,
         backoff: {:backoff.init(1000, 10_000), nil},
         electric_instance_id: Keyword.fetch!(opts, :electric_instance_id)
       }
 
-    # We try to start the replication connection first because it requires additional
-    # priveleges compared to regular "pooled" connections, so failure to open a replication
-    # connection should be reported ASAP.
-    {:ok, state, {:continue, :start_replication_client}}
+    # Try to acquire the connection lock on the replication slot
+    # before starting shape and replication processes, to ensure
+    # a single active sync service is connected to Postgres per slot.
+    {:ok, state, {:continue, :start_lock_connection}}
   end
 
   @impl true
   def handle_call(:get_pg_version, _from, %{pg_version: pg_version} = state) do
     {:reply, pg_version, state}
   end
 
+  def handle_call(:get_status, _from, %{pg_lock_acquired: pg_lock_acquired} = state) do
+    status =
+      cond do
+        not pg_lock_acquired ->
+          :waiting
+
+        is_nil(state.replication_client_pid) || is_nil(state.pool_pid) ||
+            not Process.alive?(state.pool_pid) ->
+          :starting
+
+        true ->
+          :active
+      end
+
+    {:reply, status, state}
+  end
+
+  def handle_continue(:start_lock_connection, state) do
+    case Electric.Postgres.LockConnection.start_link(
+           connection_opts: state.connection_opts,
+           connection_manager: self(),
+           lock_name: Keyword.fetch!(state.replication_opts, :slot_name)
+         ) do
+      {:ok, lock_connection_pid} ->
+        Process.send_after(self(), :log_lock_connection_status, @lock_status_logging_interval)
+        {:noreply, %{state | lock_connection_pid: lock_connection_pid}}
+
+      {:error, reason} ->
+        handle_connection_error(reason, state, "lock_connection")
+    end
+  end
+
   @impl true
   def handle_continue(:start_replication_client, state) do
     case start_replication_client(state) do
@@ -188,6 +237,7 @@ defmodule Electric.ConnectionManager do
 
     tag =
       cond do
+        pid == state.lock_connection_pid -> :lock_connection
         pid == state.replication_client_pid -> :replication_connection
         pid == state.pool_pid -> :database_pool
       end
@@ -203,6 +253,17 @@ defmodule Electric.ConnectionManager do
     {:noreply, %{state | replication_client_pid: nil}}
   end
 
+  # Periodically log the status of the lock connection until it is acquired for
+  # easier debugging and diagnostics.
+  def handle_info(:log_lock_connection_status, state) do
+    if not state.pg_lock_acquired do
+      Logger.warning(fn -> "Waiting for postgres lock to be acquired..." end)
+      Process.send_after(self(), :log_lock_connection_status, @lock_status_logging_interval)
+    end
+
+    {:noreply, state}
+  end
+
   @impl true
   def handle_cast({:connection_opts, pid, connection_opts}, state) do
     Process.monitor(pid)
@@ -220,6 +281,13 @@ defmodule Electric.ConnectionManager do
     end
   end
 
+  def handle_cast(:lock_connection_acquired, %{pg_lock_acquired: false} = state) do
+    # As soon as we acquire the connection lock, we try to start the replication connection
+    # first because it requires additional privileges compared to regular "pooled" connections,
+    # so failure to open a replication connection should be reported ASAP.
+    {:noreply, %{state | pg_lock_acquired: true}, {:continue, :start_replication_client}}
+  end
+
   defp start_replication_client(state) do
     Electric.Shapes.Supervisor.start_link(
       electric_instance_id: state.electric_instance_id,
@@ -273,6 +341,7 @@ defmodule Electric.ConnectionManager do
 
     step =
       cond do
+        is_nil(state.lock_connection_pid) -> :start_lock_connection
         is_nil(state.replication_client_pid) -> :start_replication_client
         is_nil(state.pool_pid) -> :start_connection_pool
       end

packages/sync-service/lib/electric/plug/health_check_plug.ex

@@ -0,0 +1,38 @@
+defmodule Electric.Plug.HealthCheckPlug do
+  alias Plug.Conn
+  require Logger
+  use Plug.Builder
+
+  plug :check_service_status
+  plug :put_relevant_headers
+  plug :send_response
+
+  # Match service status to a status code and status message,
+  # keeping the message name decoupled from the internal representation
+  # of the status to ensure the API is stable
+  defp check_service_status(conn, _) do
+    get_service_status = Access.fetch!(conn.assigns.config, :get_service_status)
+
+    {status_code, status_text} =
+      case get_service_status.() do
+        :waiting -> {200, "waiting"}
+        :starting -> {200, "starting"}
+        :active -> {200, "active"}
+        :stopping -> {503, "stopping"}
+      end
+
+    conn |> assign(:status_text, status_text) |> assign(:status_code, status_code)
+  end
+
+  defp put_relevant_headers(conn, _),
+    do:
+      conn
+      |> put_resp_header("content-type", "application/json")
+      |> put_resp_header("cache-control", "no-cache, no-store, must-revalidate")
+
+  defp send_response(
+         %Conn{assigns: %{status_text: status_text, status_code: status_code}} = conn,
+         _
+       ),
+       do: send_resp(conn, status_code, Jason.encode!(%{status: status_text}))
+end

packages/sync-service/lib/electric/plug/router.ex

@@ -18,6 +18,8 @@ defmodule Electric.Plug.Router do
   delete "/v1/shape/:root_table", to: Electric.Plug.DeleteShapePlug
   match "/v1/shape/:root_table", via: :options, to: Electric.Plug.OptionsShapePlug
 
+  get "/v1/health", to: Electric.Plug.HealthCheckPlug
+
   match _ do
     send_resp(conn, 404, "Not found")
   end