Use an ExclusiveLease to prevent multiple instances of LogCursor to run

lib/gitlab/geo/log_cursor/daemon.rb

@@ -82,7 +82,7 @@ module Gitlab
         end
 
         def handle_repository_update(updated_event)
-          registry = ::Geo::ProjectRegistry.find_or_create_by(project_id: project_id)
+          registry = ::Geo::ProjectRegistry.find_or_initialize_by(project_id: project_id)
 
           case updated_event.source
           when 'repository'

lib/gitlab/geo/log_cursor/events.rb

@@ -5,19 +5,25 @@ module Gitlab
       class Events
         BATCH_SIZE = 50
         NAMESPACE = 'geo:gitlab'.freeze
+        LEASE_TIMEOUT = 5.minutes.freeze
+        LEASE_KEY = 'geo_log_cursor_processed'.freeze
 
         # fetches up to BATCH_SIZE next events and keep track of batches
         def self.fetch_in_batches
-          ::Geo::EventLog.where('id > ?', last_processed).find_in_batches(batch_size: BATCH_SIZE) do |batch|
-            yield batch
-            save_processed(batch.last.id)
+          try_obtain_lease do
+            ::Geo::EventLog.where('id > ?', last_processed).find_in_batches(batch_size: BATCH_SIZE) do |batch|
+              yield batch
+
+              save_processed(batch.last.id)
+              renew_lease!
+            end
           end
         end
 
         # saves last replicated event
         def self.save_processed(event_id)
-          ::Geo::EventLogState.create!(event_id: event_id)
-          ::Geo::EventLogState.where('event_id < ?', event_id).delete_all
+          event_state = ::Geo::EventLogState.last || ::Geo::EventLogState.new
+          event_state.update!(event_id: event_id)
         end
 
         # @return [Integer] id of last replicated event
@@ -27,6 +33,33 @@ module Gitlab
 
           ::Geo::EventLog.any? ? ::Geo::EventLog.last.id : -1
         end
+
+        # private methods
+
+        def self.try_obtain_lease
+          lease = exclusive_lease.try_obtain
+
+          unless lease
+            $stdout.puts 'Cannot obtain an exclusive lease. There must be another process already in execution.'
+            return
+          end
+
+          begin
+            yield lease
+          ensure
+            Gitlab::ExclusiveLease.cancel(LEASE_KEY, lease)
+          end
+        end
+
+        def self.renew_lease!
+          exclusive_lease.renew
+        end
+
+        def self.exclusive_lease
+          @lease ||= Gitlab::ExclusiveLease.new(LEASE_KEY, timeout: LEASE_TIMEOUT)
+        end
+
+        private_class_method :try_obtain_lease, :exclusive_lease
       end
     end
   end

spec/lib/gitlab/geo/log_cursor/events_spec.rb

@@ -17,6 +17,12 @@ describe Gitlab::Geo::LogCursor::Events, lib: true do
 
       described_class.fetch_in_batches { |batch| batch }
     end
+
+    it 'skips execution if cannot achieve a lease' do
+      expect_any_instance_of(Gitlab::ExclusiveLease).to receive(:try_obtain) { }
+
+      expect { |b| described_class.fetch_in_batches(&b) }.not_to yield_control
+    end
   end
 
   describe '.save_processed' do