Serialize migrations across pods with an advisory lock

Both bob pods run Bob.ReleaseTasks.migrate() before starting the app, and
the Deployment's maxUnavailable: 1 creates both new pods at once, so two
pods race to migrate on every deploy. Ecto's own migration lock cannot
serialize them when a migration uses CREATE INDEX CONCURRENTLY, which is
why those migrations set @disable_migration_lock true.

On 2026-08-19 the deploy of 2c287e9 hit this: both pods ran the same
CREATE INDEX CONCURRENTLY statements from 20260815120000 and deadlocked
three times, each time leaving the index in progress invalid. Because
IF NOT EXISTS treats an invalid index as present, every restart skipped
the builds and failed the validity check, so the rollout crash looped.

Wrap migrate and rollback in a session level advisory lock held on its
own Postgrex connection, polled with pg_try_advisory_lock so the waiting
pod holds no snapshot that the holder's CREATE INDEX CONCURRENTLY would
wait for. The lock is released explicitly when the function returns and
by Postgres when the connection ends, including when a pod dies.

Have the migration drop invalid leftovers of its own indexes before
creating them, so a retry after an interrupted build can succeed instead
of failing the validity check forever.

Author: ericmj

lib/bob/release_tasks.ex

@@ -4,19 +4,85 @@ defmodule Bob.ReleaseTasks do
 
   Run on deploy with: `bin/bob eval "Bob.ReleaseTasks.migrate()"`.
   """
+  require Logger
+
   @app :bob
+  @migration_lock_key 4_771_003
+  @migration_lock_poll_interval 1_000
 
   def migrate() do
     load_app()
 
     for repo <- repos() do
-      {:ok, _, _} = Ecto.Migrator.with_repo(repo, &Ecto.Migrator.run(&1, :up, all: true))
+      {:ok, _, _} =
+        Ecto.Migrator.with_repo(repo, fn repo ->
+          with_migration_lock(repo, fn -> Ecto.Migrator.run(repo, :up, all: true) end)
+        end)
     end
   end
 
   def rollback(repo, version) do
     load_app()
-    {:ok, _, _} = Ecto.Migrator.with_repo(repo, &Ecto.Migrator.run(&1, :down, to: version))
+
+    {:ok, _, _} =
+      Ecto.Migrator.with_repo(repo, fn repo ->
+        with_migration_lock(repo, fn -> Ecto.Migrator.run(repo, :down, to: version) end)
+      end)
+  end
+
+  @doc false
+  # Every pod runs migrate() at boot, so they race. Ecto's own migration lock
+  # cannot serialize them here: it holds a transaction for the duration, and
+  # CREATE INDEX CONCURRENTLY waits for concurrent transactions to finish, so
+  # the two deadlock. A session level advisory lock on its own connection holds
+  # no transaction once acquired, and the lock is released when that connection
+  # ends, including when the pod dies mid-migration.
+  #
+  # Waiting for the lock must not hold a transaction either. A blocking
+  # pg_advisory_lock() call is a running statement with a snapshot for as long
+  # as it waits, and the holder's CREATE INDEX CONCURRENTLY waits for every
+  # older snapshot to go away, so the two would deadlock through the app.
+  # Polling pg_try_advisory_lock() leaves the connection idle between attempts.
+  def with_migration_lock(repo, fun) do
+    opts =
+      Keyword.take(repo.config(), [
+        :hostname,
+        :port,
+        :username,
+        :password,
+        :database,
+        :socket_options,
+        :ssl
+      ])
+
+    {:ok, conn} = Postgrex.start_link(opts)
+
+    try do
+      Logger.info("Waiting for migration lock")
+      acquire_migration_lock(conn)
+      Logger.info("Acquired migration lock")
+
+      try do
+        fun.()
+      after
+        Postgrex.query!(conn, "SELECT pg_advisory_unlock($1)", [@migration_lock_key])
+      end
+    after
+      GenServer.stop(conn)
+    end
+  end
+
+  @doc false
+  def migration_lock_key(), do: @migration_lock_key
+
+  defp acquire_migration_lock(conn) do
+    %{rows: [[locked?]]} =
+      Postgrex.query!(conn, "SELECT pg_try_advisory_lock($1)", [@migration_lock_key])
+
+    unless locked? do
+      Process.sleep(@migration_lock_poll_interval)
+      acquire_migration_lock(conn)
+    end
   end
 
   defp repos() do

priv/repo/migrations/20260815120000_index_docker_tags_by_version.exs

@@ -4,6 +4,13 @@ defmodule Bob.Repo.Migrations.IndexDockerTagsByVersion do
   @disable_ddl_transaction true
   @disable_migration_lock true
 
+  @indexes [
+    "docker_tags_elixir_version_desc_index",
+    "docker_tags_erlang_version_desc_index",
+    "docker_tags_os_version_desc_index",
+    "docker_tags_os_desc_index"
+  ]
+
   def up do
     execute("""
     CREATE OR REPLACE FUNCTION docker_tag_natural_sort_key(value text)
@@ -39,6 +46,10 @@ defmodule Bob.Repo.Migrations.IndexDockerTagsByVersion do
     $$
     """)
 
+    for index <- invalid_indexes() do
+      execute("DROP INDEX CONCURRENTLY IF EXISTS #{index}")
+    end
+
     execute("""
     CREATE INDEX CONCURRENTLY IF NOT EXISTS docker_tags_elixir_version_desc_index
     ON docker_tags (
@@ -101,10 +112,29 @@ defmodule Bob.Repo.Migrations.IndexDockerTagsByVersion do
   end
 
   def down do
-    execute("DROP INDEX CONCURRENTLY IF EXISTS docker_tags_os_desc_index")
-    execute("DROP INDEX CONCURRENTLY IF EXISTS docker_tags_os_version_desc_index")
-    execute("DROP INDEX CONCURRENTLY IF EXISTS docker_tags_erlang_version_desc_index")
-    execute("DROP INDEX CONCURRENTLY IF EXISTS docker_tags_elixir_version_desc_index")
+    for index <- Enum.reverse(@indexes) do
+      execute("DROP INDEX CONCURRENTLY IF EXISTS #{index}")
+    end
+
     execute("DROP FUNCTION IF EXISTS docker_tag_natural_sort_key(text)")
   end
+
+  defp invalid_indexes() do
+    %{rows: rows} =
+      repo().query!(
+        """
+        SELECT index_class.relname
+        FROM pg_index AS index_metadata
+        JOIN pg_class AS index_class ON index_class.oid = index_metadata.indexrelid
+        JOIN pg_namespace AS namespace ON namespace.oid = index_class.relnamespace
+        WHERE namespace.nspname = current_schema()
+          AND index_class.relname = ANY($1)
+          AND NOT index_metadata.indisvalid
+        ORDER BY index_class.relname
+        """,
+        [@indexes]
+      )
+
+    List.flatten(rows)
+  end
 end

test/bob/release_tasks_test.exs

@@ -0,0 +1,145 @@
+defmodule Bob.ReleaseTasksTest do
+  use Bob.DataCase, async: false
+
+  alias Bob.ReleaseTasks
+
+  defp locks_held() do
+    %{rows: [[count]]} =
+      Repo.query!(
+        "SELECT count(*) FROM pg_locks WHERE locktype = 'advisory' AND classid = 0 AND objid = $1",
+        [ReleaseTasks.migration_lock_key()]
+      )
+
+    count
+  end
+
+  defp lock_connections() do
+    Repo.query!("SELECT pg_stat_clear_snapshot()", [])
+
+    %{rows: [[count]]} =
+      Repo.query!(
+        """
+        SELECT count(*) FROM pg_stat_activity
+        WHERE datname = current_database()
+          AND pid <> pg_backend_pid()
+          AND query LIKE '%advisory_lock($1)'
+        """,
+        []
+      )
+
+    count
+  end
+
+  defp wait_until(fun, attempts \\ 100) do
+    cond do
+      fun.() ->
+        :ok
+
+      attempts == 0 ->
+        flunk("condition not met")
+
+      true ->
+        Process.sleep(50)
+        wait_until(fun, attempts - 1)
+    end
+  end
+
+  defp create_index_concurrently() do
+    opts = Keyword.take(Repo.config(), [:hostname, :port, :username, :password, :database])
+    {:ok, conn} = Postgrex.start_link(opts)
+
+    try do
+      Postgrex.query!(conn, "DROP TABLE IF EXISTS migration_lock_test", [])
+      Postgrex.query!(conn, "CREATE TABLE migration_lock_test (id integer)", [])
+
+      Postgrex.query!(
+        conn,
+        "CREATE INDEX CONCURRENTLY migration_lock_test_id_index ON migration_lock_test (id)",
+        []
+      )
+
+      Postgrex.query!(conn, "DROP TABLE migration_lock_test", [])
+    after
+      GenServer.stop(conn)
+    end
+  end
+
+  describe "with_migration_lock/2" do
+    test "holds the lock while the function runs" do
+      assert locks_held() == 0
+
+      ReleaseTasks.with_migration_lock(Repo, fn ->
+        assert locks_held() == 1
+      end)
+
+      assert locks_held() == 0
+    end
+
+    test "releases the lock when the function raises" do
+      assert_raise RuntimeError, "boom", fn ->
+        ReleaseTasks.with_migration_lock(Repo, fn -> raise "boom" end)
+      end
+
+      assert locks_held() == 0
+    end
+
+    test "returns what the function returned" do
+      assert ReleaseTasks.with_migration_lock(Repo, fn -> :migrated end) == :migrated
+    end
+
+    test "a second caller waits for the first to finish" do
+      test = self()
+
+      holder =
+        Task.async(fn ->
+          ReleaseTasks.with_migration_lock(Repo, fn ->
+            send(test, :holding)
+            receive do: (:release -> :ok)
+          end)
+        end)
+
+      assert_receive :holding, 10_000
+
+      waiter =
+        Task.async(fn ->
+          ReleaseTasks.with_migration_lock(Repo, fn -> send(test, :second_ran) end)
+        end)
+
+      refute_receive :second_ran, 200
+
+      send(holder.pid, :release)
+      Task.await(holder, 10_000)
+
+      assert_receive :second_ran, 10_000
+      Task.await(waiter, 10_000)
+      assert locks_held() == 0
+    end
+
+    test "a waiting caller does not block CREATE INDEX CONCURRENTLY in the holder" do
+      test = self()
+
+      holder =
+        Task.async(fn ->
+          ReleaseTasks.with_migration_lock(Repo, fn ->
+            send(test, :holding)
+            receive do: (:create_index -> :ok)
+            create_index_concurrently()
+          end)
+        end)
+
+      assert_receive :holding, 10_000
+
+      waiter =
+        Task.async(fn ->
+          ReleaseTasks.with_migration_lock(Repo, fn -> :ok end)
+        end)
+
+      wait_until(fn -> lock_connections() == 2 end)
+
+      send(holder.pid, :create_index)
+      Task.await(holder, 10_000)
+      Task.await(waiter, 10_000)
+      assert locks_held() == 0
+    end
+  end
+end