Fix concurrent index migrations with schema-driven multitenancy #610

lib/multitenancy.ex

@@ -5,7 +5,6 @@
 defmodule AshPostgres.MultiTenancy do
   @moduledoc false
 
-  @dialyzer {:nowarn_function, load_migration!: 1}
   require Logger
 
   # sobelow_skip ["SQL.Query"]
@@ -44,21 +43,32 @@ defmodule AshPostgres.MultiTenancy do
     end)
     |> Enum.map(&extract_migration_info/1)
     |> Enum.filter(& &1)
-    |> Enum.map(&load_migration!/1)
-    |> Enum.each(fn {version, mod} ->
-      Ecto.Migration.Runner.run(
-        repo,
-        [],
-        version,
-        mod,
-        :forward,
-        :up,
-        :up,
-        all: true,
-        prefix: tenant_name
-      )
-
-      Ecto.Migration.SchemaMigration.up(repo, repo.config(), version, prefix: tenant_name)
+    |> Enum.map(&load_migration_with_file!/1)
+    |> Enum.each(fn {version, mod, file} ->
+      requires_no_transaction? = migration_requires_no_transaction?(mod)
+
+      if requires_no_transaction? do
+        # For migrations that require no transaction (e.g., concurrent indexes),
+        # we need to ensure they run outside of any transaction.
+        # Ecto.Migration.Runner.run will handle @disable_ddl_transaction correctly
+        # if we're not already in a transaction, so we use a separate connection
+        # or ensure we're not in a transaction.
+        run_migration_without_transaction(repo, version, mod, tenant_name)
+      else
+        Ecto.Migration.Runner.run(
+          repo,
+          [],
+          version,
+          mod,
+          :forward,
+          :up,
+          :up,
+          all: true,
+          prefix: tenant_name
+        )
+
+        Ecto.Migration.SchemaMigration.up(repo, repo.config(), version, prefix: tenant_name)
+      end
     end)
   end
 
@@ -74,19 +84,18 @@ defmodule AshPostgres.MultiTenancy do
     :ok
   end
 
-  defp load_migration!({version, _, file}) when is_binary(file) do
+  defp load_migration_with_file!({version, _, file}) when is_binary(file) do
     loaded_modules = file |> compile_file() |> Enum.map(&elem(&1, 0))
 
-    case Enum.find(loaded_modules, &migration?/1) do
-      nil ->
-        raise Ecto.MigrationError,
-              "file #{Path.relative_to_cwd(file)} does not define an Ecto.Migration"
-
-      mod ->
-        {version, mod}
+    if mod = Enum.find(loaded_modules, &migration?/1) do
+      {version, mod, file}
+    else
+      raise Ecto.MigrationError,
+            "file #{Path.relative_to_cwd(file)} does not define an Ecto.Migration"
     end
   end
 
+
   defp compile_file(file) do
     AshPostgres.MigrationCompileCache.start_link()
     AshPostgres.MigrationCompileCache.compile_file(file)
@@ -123,4 +132,45 @@ defmodule AshPostgres.MultiTenancy do
   defp tenant_name_regex do
     ~r/^[a-zA-Z0-9_-]+$/
   end
+
+  # Check if a migration requires no transaction by examining the compiled module's
+  # migration metadata. The module is already compiled at this point, so we ask
+  # the module directly rather than reading the file. This also catches cases
+  # where the attribute is set programmatically via Module.put_attribute/3.
+  defp migration_requires_no_transaction?(mod) do
+    if function_exported?(mod, :__migration__, 0) do
+      migration_info = mod.__migration__()
+      Map.get(migration_info, :disable_ddl_transaction, false)
+    else
+      false
+    end
+  end
+
+  # Run a migration that requires no transaction outside of any transaction context
+  defp run_migration_without_transaction(repo, version, mod, tenant_name) do
+    # For migrations that require no transaction (e.g., concurrent indexes),
+    # we need to ensure they run outside of any transaction.
+    # Ecto.Migration.Runner.run respects @disable_ddl_transaction, but if we're
+    # already in a transaction, PostgreSQL will still error.
+    #
+    # We use Ecto.Adapters.SQL.checkout/3 to get a fresh connection from the pool
+    # that's not part of any transaction, ensuring the migration runs correctly.
+    config = repo.config()
+
+    Ecto.Adapters.SQL.checkout(repo, config, fn ->
+      Ecto.Migration.Runner.run(
+        repo,
+        [],
+        version,
+        mod,
+        :forward,
+        :up,
+        :up,
+        all: true,
+        prefix: tenant_name
+      )
+
+      Ecto.Migration.SchemaMigration.up(repo, config, version, prefix: tenant_name)
+    end)
+  end
 end