Add exclusive lease to mergeability check process

Concurrent calls to UserMergeToRef RPC updating a single ref
can lead to an opaque fail that is being rescued at Gitaly.

So this commit adds an exclusive lease to the mergeability
check process with the key as the current MR ID.

app/models/merge_request.rb

@@ -754,7 +754,7 @@ class MergeRequest < ApplicationRecord
   end
 
   def check_mergeability
-    MergeRequests::MergeabilityCheckService.new(self).execute
+    MergeRequests::MergeabilityCheckService.new(self).execute(retry_lease: false)
   end
   # rubocop: enable CodeReuse/ServiceClass
 

app/services/merge_requests/mergeability_check_service.rb

@@ -3,6 +3,7 @@
 module MergeRequests
   class MergeabilityCheckService < ::BaseService
     include Gitlab::Utils::StrongMemoize
+    include Gitlab::ExclusiveLeaseHelpers
 
     delegate :project, to: :@merge_request
     delegate :repository, to: :project
@@ -21,13 +22,35 @@ module MergeRequests
     # where we need the current state of the merge ref in repository, the `recheck`
     # argument is required.
     #
+    # retry_lease - Concurrent calls wait for at least 10 seconds until the
+    # lease is granted (other process finishes running). Returns an error
+    # ServiceResponse if the lease is not granted during this time.
+    #
     # Returns a ServiceResponse indicating merge_status is/became can_be_merged
     # and the merge-ref is synced. Success in case of being/becoming mergeable,
     # error otherwise.
-    def execute(recheck: false)
+    def execute(recheck: false, retry_lease: true)
       return ServiceResponse.error(message: 'Invalid argument') unless merge_request
       return ServiceResponse.error(message: 'Unsupported operation') if Gitlab::Database.read_only?
+      return check_mergeability(recheck) unless merge_ref_auto_sync_lock_enabled?
+
+      in_write_lock(retry_lease: retry_lease) do |retried|
+        # When multiple calls are waiting for the same lock (retry_lease),
+        # it's possible that when granted, the MR status was already updated for
+        # that object, therefore we reset if there was a lease retry.
+        merge_request.reset if retried
+
+        check_mergeability(recheck)
+      end
+    rescue FailedToObtainLockError => error
+      ServiceResponse.error(message: error.message)
+    end
+
+    private
+
+    attr_reader :merge_request
 
+    def check_mergeability(recheck)
       recheck! if recheck
       update_merge_status
 
@@ -46,9 +69,21 @@ module MergeRequests
       ServiceResponse.success(payload: payload)
     end
 
-    private
+    # It's possible for this service to send concurrent requests to Gitaly in order
+    # to "git update-ref" the same ref. Therefore we handle a light exclusive
+    # lease here.
+    #
+    def in_write_lock(retry_lease:, &block)
+      lease_key = "mergeability_check:#{merge_request.id}"
 
-    attr_reader :merge_request
+      lease_opts = {
+        ttl:       1.minute,
+        retries:   retry_lease ? 10 : 0,
+        sleep_sec: retry_lease ? 1.second : 0
+      }
+
+      in_lock(lease_key, lease_opts, &block)
+    end
 
     def payload
       strong_memoize(:payload) do
@@ -116,5 +151,9 @@ module MergeRequests
     def merge_ref_auto_sync_enabled?
       Feature.enabled?(:merge_ref_auto_sync, project, default_enabled: true)
     end
+
+    def merge_ref_auto_sync_lock_enabled?
+      Feature.enabled?(:merge_ref_auto_sync_lock, project, default_enabled: true)
+    end
   end
 end

changelogs/unreleased/osw-avoid-errors-due-to-concurrent-calls.yml

+---
+title: Add exclusive lease to mergeability check process
+merge_request: 31082
+author:
+type: fixed

ee/spec/services/ee/merge_requests/create_pipeline_service_spec.rb

@@ -2,7 +2,7 @@
 
 require 'spec_helper'
 
-describe MergeRequests::CreatePipelineService do
+describe MergeRequests::CreatePipelineService, :clean_gitlab_redis_shared_state do
   include ProjectForksHelper
 
   describe '#execute' do

lib/gitlab/exclusive_lease_helpers.rb

@@ -15,17 +15,18 @@ module Gitlab
       raise ArgumentError, 'Key needs to be specified' unless key
 
       lease = Gitlab::ExclusiveLease.new(key, timeout: ttl)
+      retried = false
 
       until uuid = lease.try_obtain
         # Keep trying until we obtain the lease. To prevent hammering Redis too
         # much we'll wait for a bit.
         sleep(sleep_sec)
-        break if (retries -= 1) < 0
+        (retries -= 1) < 0 ? break : retried ||= true
       end
 
       raise FailedToObtainLockError, 'Failed to obtain a lock' unless uuid
 
-      yield
+      yield(retried)
     ensure
       Gitlab::ExclusiveLease.cancel(key, uuid)
     end

spec/lib/gitlab/exclusive_lease_helpers_spec.rb

@@ -25,13 +25,13 @@ describe Gitlab::ExclusiveLeaseHelpers, :clean_gitlab_redis_shared_state do
       end
 
       it 'calls the given block' do
-        expect { |b| class_instance.in_lock(unique_key, &b) }.to yield_control.once
+        expect { |b| class_instance.in_lock(unique_key, &b) }.to yield_with_args(false)
       end
 
       it 'calls the given block continuously' do
-        expect { |b| class_instance.in_lock(unique_key, &b) }.to yield_control.once
-        expect { |b| class_instance.in_lock(unique_key, &b) }.to yield_control.once
-        expect { |b| class_instance.in_lock(unique_key, &b) }.to yield_control.once
+        expect { |b| class_instance.in_lock(unique_key, &b) }.to yield_with_args(false)
+        expect { |b| class_instance.in_lock(unique_key, &b) }.to yield_with_args(false)
+        expect { |b| class_instance.in_lock(unique_key, &b) }.to yield_with_args(false)
       end
 
       it 'cancels the exclusive lease after the block' do
@@ -68,6 +68,15 @@ describe Gitlab::ExclusiveLeaseHelpers, :clean_gitlab_redis_shared_state do
 
           expect { subject }.to raise_error('Failed to obtain a lock')
         end
+
+        context 'when lease is granted after retry' do
+          it 'yields block with true' do
+            expect(lease).to receive(:try_obtain).exactly(3).times { nil }
+            expect(lease).to receive(:try_obtain).once { unique_key }
+
+            expect { |b| class_instance.in_lock(unique_key, &b) }.to yield_with_args(true)
+          end
+        end
       end
 
       context 'when sleep second is specified' do

spec/requests/api/merge_requests_spec.rb

@@ -1571,7 +1571,7 @@ describe API::MergeRequests do
     end
   end
 
-  describe "GET /projects/:id/merge_requests/:merge_request_iid/merge_ref" do
+  describe "GET /projects/:id/merge_requests/:merge_request_iid/merge_ref", :clean_gitlab_redis_shared_state do
     before do
       merge_request.mark_as_unchecked!
     end

spec/services/merge_requests/mergeability_check_service_spec.rb

@@ -2,7 +2,7 @@
 
 require 'spec_helper'
 
-describe MergeRequests::MergeabilityCheckService do
+describe MergeRequests::MergeabilityCheckService, :clean_gitlab_redis_shared_state do
   shared_examples_for 'unmergeable merge request' do
     it 'updates or keeps merge status as cannot_be_merged' do
       subject
@@ -60,24 +60,69 @@ describe MergeRequests::MergeabilityCheckService do
 
     subject { described_class.new(merge_request).execute }
 
+    def execute_within_threads(amount:, retry_lease: true)
+      threads = []
+
+      amount.times do
+        # Let's use a different object for each thread to get closer
+        # to a real world scenario.
+        mr = MergeRequest.find(merge_request.id)
+
+        threads << Thread.new do
+          described_class.new(mr).execute(retry_lease: retry_lease)
+        end
+      end
+
+      threads.each(&:join)
+      threads
+    end
+
     before do
       project.add_developer(merge_request.author)
     end
 
     it_behaves_like 'mergeable merge request'
 
-    context 'when multiple calls to the service' do
-      it 'returns success' do
-        subject
-        result = subject
+    context 'when lock is disabled' do
+      before do
+        stub_feature_flags(merge_ref_auto_sync_lock: false)
+      end
 
-        expect(result).to be_a(ServiceResponse)
-        expect(result.success?).to be(true)
+      it_behaves_like 'mergeable merge request'
+    end
+
+    context 'when concurrent calls' do
+      it 'waits first lock and returns "cached" result in subsequent calls' do
+        threads = execute_within_threads(amount: 3)
+        results = threads.map { |t| t.value.status }
+
+        expect(results).to contain_exactly(:success, :success, :success)
+      end
+
+      it 'writes the merge-ref once' do
+        service = instance_double(MergeRequests::MergeToRefService)
+
+        expect(MergeRequests::MergeToRefService).to receive(:new).once { service }
+        expect(service).to receive(:execute).once.and_return(success: true)
+
+        execute_within_threads(amount: 3)
       end
 
-      it 'second call does not change the merge-ref' do
-        expect { subject }.to change(merge_request, :merge_ref_head).from(nil)
-        expect { subject }.not_to change(merge_request.merge_ref_head, :id)
+      it 'resets one merge request upon execution' do
+        expect_any_instance_of(MergeRequest).to receive(:reset).once
+
+        execute_within_threads(amount: 2)
+      end
+
+      context 'when retry_lease flag is false' do
+        it 'the first call succeeds, subsequent concurrent calls get a lock error response' do
+          threads = execute_within_threads(amount: 3, retry_lease: false)
+          results = threads.map { |t| [t.value.status, t.value.message] }
+
+          expect(results).to contain_exactly([:error, 'Failed to obtain a lock'],
+                                             [:error, 'Failed to obtain a lock'],
+                                             [:success, nil])
+        end
       end
     end
 
@@ -102,8 +147,7 @@ describe MergeRequests::MergeabilityCheckService do
 
     context 'when broken' do
       before do
-        allow(merge_request).to receive(:broken?) { true }
-        allow(project.repository).to receive(:can_be_merged?) { false }
+        expect(merge_request).to receive(:broken?) { true }
       end
 
       it_behaves_like 'unmergeable merge request'
@@ -117,10 +161,13 @@ describe MergeRequests::MergeabilityCheckService do
       end
     end
 
-    context 'when it has conflicts' do
-      before do
-        allow(merge_request).to receive(:broken?) { false }
-        allow(project.repository).to receive(:can_be_merged?) { false }
+    context 'when it cannot be merged on git' do
+      let(:merge_request) do
+        create(:merge_request,
+               merge_status: :unchecked,
+               source_branch: 'conflict-resolvable',
+               source_project: project,
+               target_branch: 'conflict-start')
       end
 
       it_behaves_like 'unmergeable merge request'