Prevent multiple instances of schedulers from being running simultane…

…ously

It was possible to get into a situation, where multiple instances were running
simultaneously, because of cron schedules and previous instances had not finished
their work. That situation could lead to database overload and resource exhaustion.

CHANGELOG.md

@@ -1,5 +1,7 @@
 ## master (unreleased)
 
+- Prevent multiple instances of schedulers from being running simultaneously
+
 - Reduce default batch sizes for background data migrations
 
     `batch_size` was 20_000, now 1_000; `sub_batch_size` was 1_000, now 100

lib/online_migrations.rb

@@ -23,6 +23,7 @@ class UnsafeMigration < Error; end
 
   extend ActiveSupport::Autoload
 
+  autoload :AdvisoryLock
   autoload :ApplicationRecord
   autoload :BatchIterator
   autoload :VerboseSqlLogs

lib/online_migrations/advisory_lock.rb

@@ -0,0 +1,60 @@
+# frozen_string_literal: true
+
+require "zlib"
+
+module OnlineMigrations
+  # @private
+  class AdvisoryLock
+    attr_reader :name, :connection
+
+    def initialize(name:, connection: ApplicationRecord.connection)
+      @name = name
+      @connection = connection
+    end
+
+    def try_lock
+      locked = connection.select_value("SELECT pg_try_advisory_lock(#{lock_key})")
+      Utils.to_bool(locked)
+    end
+
+    def unlock
+      connection.select_value("SELECT pg_advisory_unlock(#{lock_key})")
+    end
+
+    # Runs the given block if an advisory lock is able to be acquired.
+    def try_with_lock
+      if try_lock
+        begin
+          yield
+        ensure
+          unlock
+        end
+      end
+    end
+
+    def active?
+      objid = lock_key & 0xffffffff
+      classid = lock_key >> 32
+
+      active = connection.select_value(<<~SQL)
+        SELECT granted
+        FROM pg_locks
+        WHERE locktype = 'advisory'
+          AND pid = pg_backend_pid()
+          AND mode = 'ExclusiveLock'
+          AND classid = #{classid}
+          AND objid = #{objid}
+      SQL
+
+      Utils.to_bool(active)
+    end
+
+    private
+      SALT = 936723412
+
+      def lock_key
+        name_hash = Zlib.crc32(name)
+        SALT * name_hash
+      end
+  end
+end

lib/online_migrations/background_migrations/scheduler.rb

@@ -27,9 +27,18 @@ def run
 
         if runnable_migration
           runner = MigrationRunner.new(runnable_migration)
-          runner.run_migration_job
+
+          try_with_lock do
+            runner.run_migration_job
+          end
         end
       end
+
+      private
+        def try_with_lock(&block)
+          lock = AdvisoryLock.new(name: "online_migrations_data_scheduler")
+          lock.try_with_lock(&block)
+        end
     end
   end
 end

lib/online_migrations/background_schema_migrations/scheduler.rb

@@ -23,7 +23,10 @@ def run
         migration = find_migration
         if migration
           runner = MigrationRunner.new(migration)
-          runner.run
+
+          try_with_lock do
+            runner.run
+          end
         end
       end
 
@@ -40,6 +43,11 @@ def find_migration
             end
           end
         end
+
+        def try_with_lock(&block)
+          lock = AdvisoryLock.new(name: "online_migrations_schema_scheduler")
+          lock.try_with_lock(&block)
+        end
     end
   end
 end

test/advisory_lock_test.rb

@@ -0,0 +1,36 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class AdvisoryLockTest < Minitest::Test
+  def setup
+    @lock = OnlineMigrations::AdvisoryLock.new(name: "somename")
+  end
+
+  def teardown
+    @lock.unlock if @lock.active?
+  end
+
+  def test_try_lock
+    locked = @lock.try_lock
+    assert locked
+    assert @lock.active?
+  end
+
+  def test_unlock
+    @lock.try_lock
+    assert @lock.active?
+    @lock.unlock
+    assert_not @lock.active?
+  end
+
+  def test_try_with_lock
+    assert_not @lock.active?
+
+    @lock.try_with_lock do
+      assert @lock.active?
+    end
+
+    assert_not @lock.active?
+  end
+end