Move GeoRepositoryFetchWorker logic to service class and use ExclusiveLease

app/services/geo/repository_update_service.rb

+module Geo
+  class RepositoryUpdateService
+    attr_reader :project, :clone_url
+
+    LEASE_TIMEOUT = 1.hour.freeze
+    LEASE_KEY_PREFIX = 'geo_repository_fetch'.freeze
+
+    def initialize(project, clone_url, logger = nil)
+      @project = project
+      @clone_url = clone_url
+      @logger = logger
+    end
+
+    def execute
+      try_obtain_lease do
+        project.create_repository unless project.repository_exists?
+        project.repository.after_create if project.empty_repo?
+        project.repository.fetch_geo_mirror(clone_url)
+        project.repository.expire_all_method_caches
+        project.repository.expire_branch_cache
+        project.repository.expire_content_cache
+      end
+    rescue Gitlab::Shell::Error => e
+      logger.error "Error fetching repository for project #{project.path_with_namespace}: #{e}"
+    end
+
+    private
+
+    def try_obtain_lease
+      log('Trying to obtain lease to sync repository')
+
+      repository_lease = Gitlab::ExclusiveLease.new(lease_key, timeout: LEASE_TIMEOUT).try_obtain
+      unless repository_lease.present?
+        log('Could not obtain lease to sync repository')
+
+        return
+      end
+
+      begin
+        yield
+      ensure
+        if repository_lease.present?
+          log('Releasing leases to sync repository')
+          Gitlab::ExclusiveLease.cancel(lease_key, repository_lease)
+        end
+      end
+    end
+
+    def lease_key
+      @lease_key ||= "#{LEASE_KEY_PREFIX}:#{project.id}"
+    end
+
+    def log(message)
+      logger.info("#{self.class.name}: #{message} for project #{project.path_with_namespace} (#{project.id})")
+    end
+
+    def logger
+      @logger || Rails.logger
+    end
+  end
+end

app/workers/geo_repository_fetch_worker.rb

@@ -8,14 +8,6 @@ class GeoRepositoryFetchWorker
 
   def perform(project_id, clone_url)
     project = Project.find(project_id)
-
-    project.create_repository unless project.repository_exists?
-    project.repository.after_create if project.empty_repo?
-    project.repository.fetch_geo_mirror(clone_url)
-    project.repository.expire_all_method_caches
-    project.repository.expire_branch_cache
-    project.repository.expire_content_cache
-  rescue Gitlab::Shell::Error => e
-    logger.error "Error fetching repository for project #{project.path_with_namespace}: #{e}"
+    Geo::RepositoryUpdateService.new(project, clone_url, logger).execute
   end
 end

spec/services/geo/repository_update_service_spec.rb

+require 'spec_helper'
+
+describe Geo::RepositoryUpdateService, services: true do
+  let(:project) { create(:empty_project) }
+  let(:clone_url) { project.ssh_url_to_repo }
+
+  subject { described_class.new(project, clone_url) }
+
+  describe '#execute' do
+    before do
+      allow_any_instance_of(Gitlab::Geo).to receive_messages(secondary?: true)
+      allow(project.repository).to receive(:fetch_geo_mirror).and_return(true)
+      allow(project).to receive(:repository_exists?) { false }
+      allow(project).to receive(:empty_repo?) { true }
+
+      allow(project.repository).to receive(:expire_all_method_caches)
+      allow(project.repository).to receive(:expire_branch_cache)
+      allow(project.repository).to receive(:expire_content_cache)
+    end
+
+    it 'releases the lease' do
+      expect(Gitlab::ExclusiveLease).to receive(:cancel).once.and_call_original
+
+      subject.execute
+    end
+
+    it 'creates a new repository' do
+      expect(project).to receive(:create_repository)
+
+      subject.execute
+    end
+
+    it 'executes after_create hook' do
+      expect(project.repository).to receive(:after_create).at_least(:once)
+
+      subject.execute
+    end
+
+    it 'fetches the Geo mirror' do
+      expect(project.repository).to receive(:fetch_geo_mirror)
+
+      subject.execute
+    end
+
+    it 'expires repository caches' do
+      expect(project.repository).to receive(:expire_all_method_caches)
+      expect(project.repository).to receive(:expire_branch_cache)
+      expect(project.repository).to receive(:expire_content_cache)
+
+      subject.execute
+    end
+
+    it 'does not raise exception when git failures occurs' do
+      expect(project.repository).to receive(:fetch_geo_mirror).and_raise(Gitlab::Shell::Error)
+
+      expect { subject.execute }.not_to raise_error
+    end
+  end
+end

spec/workers/geo_repository_fetch_worker_spec.rb

@@ -4,48 +4,11 @@ describe GeoRepositoryFetchWorker do
   describe '#perform' do
     let(:project) { create(:empty_project) }
 
-    before do
-      allow_any_instance_of(Gitlab::Geo).to receive_messages(secondary?: true)
-      allow_any_instance_of(Repository).to receive(:fetch_geo_mirror).and_return(true)
-      allow_any_instance_of(Project).to receive(:repository_exists?) { false }
-      allow_any_instance_of(Project).to receive(:empty_repo?) { true }
-
-      allow_any_instance_of(Repository).to receive(:expire_all_method_caches)
-      allow_any_instance_of(Repository).to receive(:expire_branch_cache)
-      allow_any_instance_of(Repository).to receive(:expire_content_cache)
-    end
-
-    it 'creates a new repository' do
-      expect_any_instance_of(Project).to receive(:create_repository)
-
-      perform
-    end
-
-    it 'executes after_create hook' do
-      expect_any_instance_of(Repository).to receive(:after_create).at_least(:once)
-
-      perform
-    end
-
-    it 'fetches the Geo mirror' do
-      expect_any_instance_of(Repository).to receive(:fetch_geo_mirror)
+    it 'delegates to Geo::RepositoryUpdateService' do
+      expect_any_instance_of(Geo::RepositoryUpdateService).to receive(:execute)
 
       perform
     end
-
-    it 'expires repository caches' do
-      expect_any_instance_of(Repository).to receive(:expire_all_method_caches)
-      expect_any_instance_of(Repository).to receive(:expire_branch_cache)
-      expect_any_instance_of(Repository).to receive(:expire_content_cache)
-
-      perform
-    end
-
-    it 'does not raise exception when git failures occurs' do
-      expect_any_instance_of(Repository).to receive(:fetch_geo_mirror).and_raise(Gitlab::Shell::Error)
-
-      expect { perform }.not_to raise_error
-    end
   end
 
   def perform