Use advisory locks to prevent concurrent migrations

- Addresses issue #22092
- Works on Postgres and MySQL
- Uses advisory locks because of two important properties:
  1. The can be obtained outside of the context of a transaction
  2. They are automatically released when the session ends, so if a
  migration process crashed for whatever reason the lock is not left
  open perpetually
- Adds get_advisory_lock and release_advisory_lock methods to database
  adapters
- Attempting to run a migration while another one is in process will
  raise a ConcurrentMigrationError instead of attempting to run in
  parallel with undefined behavior. This could be rescued and
  the migration could exit cleanly instead. Perhaps as a configuration
  option?

Technical Notes
==============

The Migrator uses generate_migrator_advisory_lock_key to build the key
for the lock. In order to be compatible across multiple adapters there
are some constraints on this key.
- Postgres limits us to 64 bit signed integers
- MySQL advisory locks are server-wide so we have to scope to the
  database
- To fulfil these requirements we use a Migrator salt (a randomly
  chosen signed integer with max length of 31 bits) that identifies
  the Rails migration process as the owner of the lock. We multiply
  this salt with a CRC32 unsigned integer hash of the database name to
  get a signed 64 bit integer that can also be converted to a string
  to act as a lock key in MySQL databases.
- It is important for subsequent versions of the Migrator to use the
  same salt, otherwise different versions of the Migrator will not see
  each other's locks.

4 files changed, 199 insertions, 0 deletions

diff --git a/activerecord/test/cases/adapters/mysql/connection_test.rb b/activerecord/test/cases/adapters/mysql/connection_test.rb
index decac9e83b..75653ee9af 100644
--- a/activerecord/test/cases/adapters/mysql/connection_test.rb
+++ b/activerecord/test/cases/adapters/mysql/connection_test.rb
@@ -170,6 +170,34 @@ class MysqlConnectionTest < ActiveRecord::MysqlTestCase
     end
   end
 
+  def test_get_and_release_advisory_lock
+    key = "test_key"
+
+    got_lock = @connection.get_advisory_lock(key)
+    assert got_lock, "get_advisory_lock should have returned true but it didn't"
+
+    assert_equal test_lock_free(key), false,
+      "expected the test advisory lock to be held but it wasn't"
+
+    released_lock = @connection.release_advisory_lock(key)
+    assert released_lock, "expected release_advisory_lock to return true but it didn't"
+
+    assert test_lock_free(key), 'expected the test key to be available after releasing'
+  end
+
+  def test_release_non_existent_advisory_lock
+    fake_key = "fake_key"
+    released_non_existent_lock = @connection.release_advisory_lock(fake_key)
+    assert_equal released_non_existent_lock, false,
+      'expected release_advisory_lock to return false when there was no lock to release'
+  end
+
+  protected
+
+  def test_lock_free(key)
+    @connection.select_value("SELECT IS_FREE_LOCK('#{key}');") == '1'
+  end
+
   private
 
   def with_example_table(&block)
diff --git a/activerecord/test/cases/adapters/mysql2/connection_test.rb b/activerecord/test/cases/adapters/mysql2/connection_test.rb
index 000bcadebe..71c4028675 100644
--- a/activerecord/test/cases/adapters/mysql2/connection_test.rb
+++ b/activerecord/test/cases/adapters/mysql2/connection_test.rb
@@ -131,4 +131,32 @@ class Mysql2ConnectionTest < ActiveRecord::Mysql2TestCase
   ensure
     @connection.execute "DROP TABLE `bar_baz`"
   end
+
+  def test_get_and_release_advisory_lock
+    key = "test_key"
+
+    got_lock = @connection.get_advisory_lock(key)
+    assert got_lock, "get_advisory_lock should have returned true but it didn't"
+
+    assert_equal test_lock_free(key), false,
+      "expected the test advisory lock to be held but it wasn't"
+
+    released_lock = @connection.release_advisory_lock(key)
+    assert released_lock, "expected release_advisory_lock to return true but it didn't"
+
+    assert test_lock_free(key), 'expected the test key to be available after releasing'
+  end
+
+  def test_release_non_existent_advisory_lock
+    fake_key = "fake_key"
+    released_non_existent_lock = @connection.release_advisory_lock(fake_key)
+    assert_equal released_non_existent_lock, false,
+      'expected release_advisory_lock to return false when there was no lock to release'
+  end
+
+  protected
+
+  def test_lock_free(key)
+    @connection.select_value("SELECT IS_FREE_LOCK('#{key}');") == 1
+  end
 end
diff --git a/activerecord/test/cases/adapters/postgresql/connection_test.rb b/activerecord/test/cases/adapters/postgresql/connection_test.rb
index 722e2377c1..b12beb91de 100644
--- a/activerecord/test/cases/adapters/postgresql/connection_test.rb
+++ b/activerecord/test/cases/adapters/postgresql/connection_test.rb
@@ -209,5 +209,47 @@ module ActiveRecord
         ActiveRecord::Base.establish_connection(orig_connection.deep_merge({:variables => {:debug_print_plan => :default}}))
       end
     end
+
+    def test_get_and_release_advisory_lock
+      key = 5295901941911233559
+      list_advisory_locks = <<-SQL
+        SELECT locktype,
+              (classid::bigint << 32) | objid::bigint AS lock_key
+        FROM pg_locks
+        WHERE locktype = 'advisory'
+      SQL
+
+      got_lock = @connection.get_advisory_lock(key)
+      assert got_lock, "get_advisory_lock should have returned true but it didn't"
+
+      advisory_lock = @connection.query(list_advisory_locks).find {|l| l[1] == key}
+      assert advisory_lock,
+        "expected to find an advisory lock with key #{key} but there wasn't one"
+
+      released_lock = @connection.release_advisory_lock(key)
+      assert released_lock, "expected release_advisory_lock to return true but it didn't"
+
+      advisory_locks = @connection.query(list_advisory_locks).select {|l| l[1] == key}
+      assert_empty advisory_locks,
+        "expected to have released advisory lock with key #{key} but it was still held"
+    end
+
+    def test_release_non_existent_advisory_lock
+      fake_key = 2940075057017742022
+      with_warning_suppression do
+        released_non_existent_lock = @connection.release_advisory_lock(fake_key)
+        assert_equal released_non_existent_lock, false,
+          'expected release_advisory_lock to return false when there was no lock to release'
+      end
+    end
+
+    protected
+
+    def with_warning_suppression
+      log_level = @connection.client_min_messages
+      @connection.client_min_messages = 'error'
+      yield
+      @connection.client_min_messages = log_level
+    end
   end
 end
diff --git a/activerecord/test/cases/migration_test.rb b/activerecord/test/cases/migration_test.rb
index 10f1c7216f..9c4fa0df4f 100644
--- a/activerecord/test/cases/migration_test.rb
+++ b/activerecord/test/cases/migration_test.rb
@@ -522,6 +522,79 @@ class MigrationTest < ActiveRecord::TestCase
     end
   end
 
+  if ActiveRecord::Base.connection.supports_advisory_locks?
+    def test_migrator_generates_valid_lock_key
+      migration = Class.new(ActiveRecord::Migration).new
+      migrator = ActiveRecord::Migrator.new(:up, [migration], 100)
+
+      lock_key = migrator.send(:generate_migrator_advisory_lock_key)
+
+      assert ActiveRecord::Base.connection.get_advisory_lock(lock_key),
+        "the Migrator should have generated a valid lock key, but it didn't"
+      assert ActiveRecord::Base.connection.release_advisory_lock(lock_key),
+        "the Migrator should have generated a valid lock key, but it didn't"
+    end
+
+    def test_generate_migrator_advisory_lock_key
+      # It is important we are consistent with how we generate this so that
+      # exclusive locking works across migrator versions
+      migration = Class.new(ActiveRecord::Migration).new
+      migrator = ActiveRecord::Migrator.new(:up, [migration], 100)
+
+      lock_key = migrator.send(:generate_migrator_advisory_lock_key)
+
+      current_database = ActiveRecord::Base.connection.current_database
+      salt = ActiveRecord::Migrator::MIGRATOR_SALT
+      expected_key = Zlib.crc32(current_database) * salt
+
+      assert lock_key == expected_key, "expected lock key generated by the migrator to be #{expected_key}, but it was #{lock_key} instead"
+      assert lock_key.is_a?(Fixnum), "expected lock key to be a Fixnum, but it wasn't"
+      assert lock_key.bit_length <= 63, "lock key must be a signed integer of max 63 bits magnitude"
+    end
+
+    def test_migrator_one_up_with_unavailable_lock
+      assert_no_column Person, :last_name
+
+      migration = Class.new(ActiveRecord::Migration) {
+        def version; 100 end
+        def migrate(x)
+          add_column "people", "last_name", :string
+        end
+      }.new
+
+      migrator = ActiveRecord::Migrator.new(:up, [migration], 100)
+      lock_key = migrator.send(:generate_migrator_advisory_lock_key)
+
+      with_another_process_holding_lock(lock_key) do
+        assert_raise(ActiveRecord::ConcurrentMigrationError) { migrator.migrate }
+      end
+
+      assert_no_column Person, :last_name,
+        "without an advisory lock, the Migrator should not make any changes, but it did."
+    end
+
+    def test_migrator_one_up_with_unavailable_lock_using_run
+      assert_no_column Person, :last_name
+
+      migration = Class.new(ActiveRecord::Migration) {
+        def version; 100 end
+        def migrate(x)
+          add_column "people", "last_name", :string
+        end
+      }.new
+
+      migrator = ActiveRecord::Migrator.new(:up, [migration], 100)
+      lock_key = migrator.send(:generate_migrator_advisory_lock_key)
+
+      with_another_process_holding_lock(lock_key) do
+        assert_raise(ActiveRecord::ConcurrentMigrationError) { migrator.run }
+      end
+
+      assert_no_column Person, :last_name,
+        "without an advisory lock, the Migrator should not make any changes, but it did."
+    end
+  end
+
   protected
     # This is needed to isolate class_attribute assignments like `table_name_prefix`
     # for each test case.
@@ -531,6 +604,34 @@ class MigrationTest < ActiveRecord::TestCase
         def self.base_class; self; end
       }
     end
+
+    def with_another_process_holding_lock(lock_key)
+      other_process_has_lock = false
+      test_terminated = false
+
+      other_process = Thread.new do
+        begin
+          conn = ActiveRecord::Base.connection_pool.checkout
+          conn.get_advisory_lock(lock_key)
+          other_process_has_lock = true
+          while !test_terminated do # hold the lock open until we tested everything
+            sleep(0.01)
+          end
+        ensure
+          conn.release_advisory_lock(lock_key)
+          ActiveRecord::Base.connection_pool.checkin(conn)
+        end
+      end
+
+      while !other_process_has_lock # wait until the 'other process' has the lock
+        sleep(0.01)
+      end
+
+      yield
+
+      test_terminated = true
+      other_process.join
+    end
 end
 
 class ReservedWordsMigrationTest < ActiveRecord::TestCase