fix: Fix publication update deadlocks (#3285)

We observed deadlocks during publication updates in high traffic usage.

The protection against concurrent transactions causing deadlocks was
removed here: #3218

The idea was that since we linearise everything behind the publication
manager, this protection is redundant, but it seems this can still occur
(and DDL operations on Postgres have much weaker guarantees).

I've added a slow test (tagged slow to only run on CI) that managed to
reproduce the deadlocks, although not always but that's the problem with
concurrency bugs, and restoring the logic of updating replica identities
only after touching the publication seems to have done away with the
error completely.

Author: msfstef

.changeset/pink-bikes-bathe.md

@@ -0,0 +1,5 @@
+---
+'@core/sync-service': patch
+---
+
+Fix deadlock appearing during high concurrency publication updates.

packages/sync-service/lib/electric/postgres/configuration.ex

@@ -168,11 +168,36 @@ defmodule Electric.Postgres.Configuration do
       )
     end
 
-    configuration_result =
+    # Notes on avoiding deadlocks
+    # - `ALTER TABLE` should be after the publication altering, because it takes out an exclusive lock over this table,
+    #   but the publication altering takes out a shared lock on all mentioned tables, so a concurrent transaction will
+    #   deadlock if the order is reversed, and we've seen this happen even within the context of a single process perhaps
+    #   across multiple calls from separate deployments or timing issues.
+    # - It is important for all table operations to also occur in the same order to avoid deadlocks due to
+    #   lock ordering issues, so despite splitting drop and add operations we sort them and process them together
+    #   in a sorted single pass
+    results =
       Enum.concat(
-        Enum.map(to_drop, &{&1, drop_table_from_publication(conn, publication_name, &1)}),
-        Enum.map(to_add, &{&1, add_table_to_publication(conn, publication_name, &1)})
+        Enum.map(to_drop, &{&1, :drop}),
+        Enum.map(to_add, &{&1, :add})
       )
+      |> Enum.sort(&(elem(&1, 0) <= elem(&2, 0)))
+      |> Enum.map(fn
+        {rel, :drop} -> {rel, drop_table_from_publication(conn, publication_name, rel)}
+        {rel, :add} -> {rel, add_table_to_publication(conn, publication_name, rel)}
+      end)
+      |> Enum.map(fn
+        {rel, {:ok, :added}} ->
+          with {:ok, :configured} <- set_table_replica_identity_full(conn, rel) do
+            {rel, {:ok, :added}}
+          end
+
+        res ->
+          res
+      end)
+
+    configuration_result =
+      results
       |> Enum.concat(Enum.map(to_preserve, &{&1, {:ok, :validated}}))
       |> Enum.concat(Enum.map(to_invalidate, &{&1, {:error, :schema_changed}}))
       |> Map.new()
@@ -186,17 +211,46 @@ defmodule Electric.Postgres.Configuration do
     {_oid, relation} = oid_relation
     table = Utils.relation_to_sql(relation)
 
-    Logger.debug(
-      "Adding #{table} to publication #{publication_name} and " <>
-        "setting its replica identity to FULL"
-    )
+    Logger.debug("Adding #{table} to publication #{publication_name}")
 
-    with :ok <- exec_alter_publication_for_table(conn, publication_name, :add, table),
-         :ok <- exec_set_replica_identity_full(conn, table) do
+    with :ok <- exec_alter_publication_for_table(conn, publication_name, :add, table) do
       {:ok, :added}
     end
   end
 
+  @spec set_table_replica_identity_full(Postgrex.conn(), Electric.oid_relation()) ::
+          {:ok, :configured} | {:error, term()}
+  defp set_table_replica_identity_full(conn, oid_relation) do
+    {_oid, relation} = oid_relation
+    {schema, name} = relation
+    table = Utils.relation_to_sql(relation)
+
+    case Postgrex.query(
+           conn,
+           """
+           SELECT c.relreplident
+           FROM pg_class c
+           JOIN pg_namespace n ON n.oid = c.relnamespace
+           WHERE n.nspname = $1 AND c.relname = $2
+           """,
+           [schema, name]
+         ) do
+      {:ok, %Postgrex.Result{rows: [[<<"f">>]]}} ->
+        Logger.debug("Replica identity already FULL for #{table}, skipping")
+        {:ok, :configured}
+
+      {:ok, _} ->
+        Logger.debug("Setting #{table} replica identity to FULL")
+
+        with :ok <- exec_set_replica_identity_full(conn, table) do
+          {:ok, :configured}
+        end
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
   @spec drop_table_from_publication(Postgrex.conn(), String.t(), Electric.oid_relation()) ::
           {:ok, :dropped} | {:error, term()}
   defp drop_table_from_publication(conn, publication_name, oid_relation) do
@@ -211,9 +265,9 @@ defmodule Electric.Postgres.Configuration do
 
   @spec exec_alter_publication_for_table(Postgrex.conn(), String.t(), :add | :drop, String.t()) ::
           :ok | {:error, term()}
-  defp exec_alter_publication_for_table(conn, publication_name, op, table) do
+  defp exec_alter_publication_for_table(conn, publication_name, op_atom, table) do
     op =
-      case op do
+      case op_atom do
         :add -> "ADD"
         :drop -> "DROP"
       end
@@ -233,9 +287,10 @@ defmodule Electric.Postgres.Configuration do
         Postgrex.query!(conn, "RELEASE SAVEPOINT before_publication", [])
 
         case reason do
-          # Duplicate object error is raised if we're trying to add a table
-          # to the publication when it's already there.
-          %{postgres: %{code: :undefined_table}} -> :ok
+          # undefined table can happen when removing a table that was already removed
+          %{postgres: %{code: :undefined_object}} when op_atom == :drop -> :ok
+          # duplicate object can happen when adding a table that was already added
+          %{postgres: %{code: :duplicate_object}} when op_atom == :add -> :ok
           _ -> {:error, reason}
         end
     end

packages/sync-service/test/electric/postgres/configuration_test.exs

@@ -416,6 +416,56 @@ defmodule Electric.Postgres.ConfigurationTest do
     end
   end
 
+  describe "concurrent publication updates" do
+    @tag slow: true
+    @tag connection_opt_overrides: [pool_size: 10, queue_target: 1_000, queue_interval: 20_000]
+    test "should not cause deadlocks", %{
+      pool: conn,
+      publication_name: publication
+    } do
+      num_relations = 10
+      num_to_pick = div(num_relations, 3)
+
+      deadlock_oid_rels =
+        for i <- 1..num_relations do
+          table_name = "deadlock_table_#{i}"
+
+          Postgrex.query!(
+            conn,
+            """
+            CREATE TABLE #{table_name} (
+              id UUID PRIMARY KEY,
+              value TEXT NOT NULL
+            )
+            """,
+            []
+          )
+
+          table_oid = get_table_oid(conn, {"public", table_name})
+          {table_oid, {"public", table_name}}
+        end
+        |> MapSet.new()
+
+      tasks =
+        for _i <- 1..30 do
+          Task.async(fn ->
+            Configuration.configure_publication!(
+              conn,
+              publication,
+              Enum.take_random(deadlock_oid_rels, num_to_pick) |> MapSet.new()
+            )
+          end)
+        end
+
+      error_results =
+        Task.await_many(tasks, 60_000)
+        |> Enum.flat_map(& &1)
+        |> Enum.filter(&match?({_, {:error, _}}, &1))
+
+      assert error_results == []
+    end
+  end
+
   defp get_table_identity(conn, {schema, table}) do
     %{rows: [[ident]]} =
       Postgrex.query!(