Lock migrations to support multiple concurrent migrators

* Locks migrations table in a transaction
* Wraps each migration in a Task, which may have its own transaction
* Propogates errors up, while ensure transactions are committed

Signed-off-by: José Valim <jose.valim@plataformatec.com.br>

integration_test/cases/migrator.exs

@@ -111,7 +111,7 @@ defmodule Ecto.Integration.MigratorTest do
   end
 
   defp count_entries() do
-    length Process.get(:migrations)
+    PoolRepo.aggregate(SchemaMigration, :count, :version)
   end
 
   defp create_migration(num) do

integration_test/sql/alter.exs

@@ -58,26 +58,8 @@ defmodule Ecto.Integration.AlterTest do
       result ->
         assert [%Decimal{}] = result
     end
-
-    PoolRepo.transaction(fn() ->
-      assert [%Decimal{}] = PoolRepo.all(values)
-
-      assert :ok == down(PoolRepo, 20161112130000, AlterMigrationTwo, log: false)
-
-      # optionally fail once with database error when already prepared on
-      # connection (and clear cache)
-      try do
-        PoolRepo.all(values, [mode: :savepoint])
-      catch
-        :error, _ ->
-          assert PoolRepo.all(values) == [1]
-      else
-        result ->
-          assert result == [1]
-      end
-    end)
-
   after
+    assert :ok == down(PoolRepo, 20161112130000, AlterMigrationTwo, log: false)
     assert :ok == down(PoolRepo, 20161112120000, AlterMigrationOne, log: false)
   end
 
@@ -101,16 +83,8 @@ defmodule Ecto.Integration.AlterTest do
       result ->
         assert result == {1, nil}
     end
-
-    PoolRepo.transaction(fn() ->
-      assert PoolRepo.update_all(values, [set: [value: Decimal.new(5)]]) == {1, nil}
-
-      assert :ok == down(PoolRepo, 20161112130000, AlterMigrationTwo, log: false)
-
-      assert PoolRepo.update_all(values, [set: [value: 6]]) == {1, nil}
-    end)
-
   after
+    assert :ok == down(PoolRepo, 20161112130000, AlterMigrationTwo, log: false)
     assert :ok == down(PoolRepo, 20161112120000, AlterMigrationOne, log: false)
   end
 end

lib/ecto/adapter/migration.ex

@@ -50,4 +50,14 @@ defmodule Ecto.Adapter.Migration  do
     * `:log` - When false, does not log begin/commit/rollback queries
   """
   @callback execute_ddl(repo :: Ecto.Repo.t, command, options :: Keyword.t) :: :ok | no_return
+
+  @doc """
+  The callback responsible for locking the migrations table
+  and emitting the locked versions for callback execution.
+
+  It returns the result of calling the given function with a
+  list of versions.
+  """
+  @callback lock_for_migrations(repo :: Ecto.Repo.t, query :: Ecto.Query.t,
+                                options :: Keyword.t, ([version :: integer] -> result)) :: result when result: var
 end

lib/ecto/adapters/sql.ex

@@ -136,8 +136,14 @@ defmodule Ecto.Adapters.SQL do
         :ok
       end
 
+      @doc false
+      def lock_for_migrations(repo, query, opts, fun) do
+        Ecto.Adapters.SQL.lock_for_migrations(repo, query, opts, fun)
+      end
+
       defoverridable [prepare: 2, execute: 6, insert: 6, update: 6, delete: 4, insert_all: 7,
-                      execute_ddl: 3, loaders: 2, dumpers: 2, autogenerate: 1, ensure_all_started: 2]
+                      execute_ddl: 3, loaders: 2, dumpers: 2, autogenerate: 1, ensure_all_started: 2,
+                      lock_for_migrations: 4]
     end
   end
 
@@ -596,6 +602,21 @@ defmodule Ecto.Adapters.SQL do
     end
   end
 
+  ## Migrations
+
+  @doc false
+  def lock_for_migrations(repo, query, opts, fun) do
+    {:ok, result} =
+      transaction(repo, opts ++ [log: false, timeout: :infinity], fn ->
+        query
+        |> Map.put(:lock, "FOR UPDATE")
+        |> repo.all()
+        |> fun.()
+      end)
+
+    result
+  end
+
   ## Log
 
   defp with_log(repo, params, opts) do

lib/ecto/migration/schema_migration.ex

@@ -18,10 +18,9 @@ defmodule Ecto.Migration.SchemaMigration do
     create_migrations_table(adapter, repo, prefix)
   end
 
-  def migrated_versions(repo, prefix) do
-    from(p in {get_source(repo), __MODULE__}, select: p.version)
+  def versions(repo, prefix) do
+    from(p in get_source(repo), select: type(p.version, :integer))
     |> Map.put(:prefix, prefix)
-    |> repo.all(@opts)
   end
 
   def up(repo, version, prefix) do
@@ -31,7 +30,7 @@ defmodule Ecto.Migration.SchemaMigration do
   end
 
   def down(repo, version, prefix) do
-    from(p in {get_source(repo), __MODULE__}, where: p.version == ^version)
+    from(p in get_source(repo), where: p.version == type(^version, :integer))
     |> Map.put(:prefix, prefix)
     |> repo.delete_all(@opts)
   end