Permalink
Browse files

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.
  • Loading branch information...
Sam Davies
Sam Davies committed Oct 29, 2015
1 parent 0174837 commit 2c2a8755460ec3d32ece91c9766dbd0304ece028
@@ -214,6 +214,11 @@ def supports_savepoints?
false
end
# Does this adapter support application-enforced advisory locking?
def supports_advisory_locks?
false
end
# Should primary key values be selected from their corresponding
# sequence before the insert statement? If true, next_sequence_value
# is called before each insert to set the record's primary key.
@@ -280,6 +285,20 @@ def disable_extension(name)
def enable_extension(name)
end
# This is meant to be implemented by the adapters that support advisory
# locks
#
# Return true if we got the lock, otherwise false
def get_advisory_lock(key) # :nodoc:
end
# This is meant to be implemented by the adapters that support advisory
# locks.
#
# Return true if we released the lock, otherwise false
def release_advisory_lock(key) # :nodoc:
end
# A list of extensions, to be filled in by adapters that support them.
def extensions
[]
@@ -220,6 +220,20 @@ def supports_datetime_with_precision?
version >= '5.6.4'
end
# 5.0.0 definitely supports it, possibly supported by earlier versions but
# not sure
def supports_advisory_locks?
version >= '5.0.0'
end
def get_advisory_lock(key, timeout = 0) # :nodoc:
select_value("SELECT GET_LOCK('#{key}', #{timeout});").to_s == '1'
end
def release_advisory_lock(key) # :nodoc:
select_value("SELECT RELEASE_LOCK('#{key}')").to_s == '1'
end
def native_database_types
NATIVE_DATABASE_TYPES
end
@@ -293,6 +293,10 @@ def supports_ddl_transactions?
true
end
def supports_advisory_locks?
true
end
def supports_explain?
true
end
@@ -311,6 +315,20 @@ def supports_materialized_views?
postgresql_version >= 90300
end
def get_advisory_lock(key) # :nodoc:
unless key.is_a?(Integer) && key.bit_length <= 63
raise(ArgumentError, "Postgres requires advisory lock keys to be a signed 64 bit integer")
end
select_value("SELECT pg_try_advisory_lock(#{key});")
end
def release_advisory_lock(key) # :nodoc:
unless key.is_a?(Integer) && key.bit_length <= 63
raise(ArgumentError, "Postgres requires advisory lock keys to be a signed 64 bit integer")
end
select_value("SELECT pg_advisory_unlock(#{key})")
end
def enable_extension(name)
exec_query("CREATE EXTENSION IF NOT EXISTS \"#{name}\"").tap {
reload_type_map
@@ -135,6 +135,14 @@ def initialize(message = nil)
end
end
class ConcurrentMigrationError < MigrationError #:nodoc:
DEFAULT_MESSAGE = "Cannot run migrations because another migration process is currently running.".freeze
def initialize(message = DEFAULT_MESSAGE)
super
end
end
# = Active Record Migrations
#
# Migrations can manage the evolution of a schema used by several physical
@@ -1042,32 +1050,18 @@ def current_migration
alias :current :current_migration
def run
migration = migrations.detect { |m| m.version == @target_version }
raise UnknownMigrationVersionError.new(@target_version) if migration.nil?
unless (up? && migrated.include?(migration.version.to_i)) || (down? && !migrated.include?(migration.version.to_i))
begin
execute_migration_in_transaction(migration, @direction)
rescue => e
canceled_msg = use_transaction?(migration) ? ", this migration was canceled" : ""
raise StandardError, "An error has occurred#{canceled_msg}:\n\n#{e}", e.backtrace
end
if use_advisory_lock?
with_advisory_lock { run_without_lock }
else
run_without_lock
end
end
def migrate
if !target && @target_version && @target_version > 0
raise UnknownMigrationVersionError.new(@target_version)
end
runnable.each do |migration|
Base.logger.info "Migrating to #{migration.name} (#{migration.version})" if Base.logger
begin
execute_migration_in_transaction(migration, @direction)
rescue => e
canceled_msg = use_transaction?(migration) ? "this and " : ""
raise StandardError, "An error has occurred, #{canceled_msg}all later migrations canceled:\n\n#{e}", e.backtrace
end
if use_advisory_lock?
with_advisory_lock { migrate_without_lock }
else
migrate_without_lock
end
end
@@ -1092,10 +1086,45 @@ def pending_migrations
end
def migrated
@migrated_versions ||= Set.new(self.class.get_all_versions)
@migrated_versions || load_migrated
end
def load_migrated
@migrated_versions = Set.new(self.class.get_all_versions)
end
private
def run_without_lock
migration = migrations.detect { |m| m.version == @target_version }
raise UnknownMigrationVersionError.new(@target_version) if migration.nil?
unless (up? && migrated.include?(migration.version.to_i)) || (down? && !migrated.include?(migration.version.to_i))
begin
execute_migration_in_transaction(migration, @direction)
rescue => e
canceled_msg = use_transaction?(migration) ? ", this migration was canceled" : ""
raise StandardError, "An error has occurred#{canceled_msg}:\n\n#{e}", e.backtrace
end
end
end
def migrate_without_lock
if !target && @target_version && @target_version > 0
raise UnknownMigrationVersionError.new(@target_version)
end
runnable.each do |migration|
Base.logger.info "Migrating to #{migration.name} (#{migration.version})" if Base.logger
begin
execute_migration_in_transaction(migration, @direction)
rescue => e
canceled_msg = use_transaction?(migration) ? "this and " : ""
raise StandardError, "An error has occurred, #{canceled_msg}all later migrations canceled:\n\n#{e}", e.backtrace
end
end
end
def ran?(migration)
migrated.include?(migration.version.to_i)
end
@@ -1157,5 +1186,25 @@ def ddl_transaction(migration)
def use_transaction?(migration)
!migration.disable_ddl_transaction && Base.connection.supports_ddl_transactions?
end
def use_advisory_lock?
Base.connection.supports_advisory_locks?
end
def with_advisory_lock
key = generate_migrator_advisory_lock_key
got_lock = Base.connection.get_advisory_lock(key)
raise ConcurrentMigrationError unless got_lock
load_migrated # reload schema_migrations to be sure it wasn't changed by another process before we got the lock
yield
ensure
Base.connection.release_advisory_lock(key) if got_lock
end
MIGRATOR_SALT = 2053462845
def generate_migrator_advisory_lock_key
db_name_hash = Zlib.crc32(Base.connection.current_database)
MIGRATOR_SALT * db_name_hash
end
end
end
@@ -170,6 +170,34 @@ def test_mysql_set_session_variable_to_default
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)
@@ -131,4 +131,32 @@ def test_logs_name_rename_column_sql
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
@@ -209,5 +209,47 @@ def test_set_session_variable_default
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
Oops, something went wrong.

0 comments on commit 2c2a875

Please sign in to comment.