Merge branch 'id-make-merge-service-idempotent' into 'master'

Make MergeService idempotent

See merge request gitlab-org/gitlab!55368

app/services/merge_requests/merge_service.rb

@@ -8,7 +8,10 @@ module MergeRequests
   # Executed when you do merge via GitLab UI
   #
   class MergeService < MergeRequests::MergeBaseService
+    include Gitlab::Utils::StrongMemoize
+
     GENERIC_ERROR_MESSAGE = 'An error occurred while merging'
+    LEASE_TIMEOUT = 15.minutes.to_i
 
     delegate :merge_jid, :state, to: :@merge_request
 
@@ -18,6 +21,9 @@ module MergeRequests
         return
       end
 
+      return if merge_request.merged?
+      return unless exclusive_lease(merge_request.id).try_obtain
+
       @merge_request = merge_request
       @options = options
 
@@ -34,6 +40,8 @@ module MergeRequests
       log_info("Merge process finished on JID #{merge_jid} with state #{state}")
     rescue MergeError => e
       handle_merge_error(log_message: e.message, save_message_on_model: true)
+    ensure
+      exclusive_lease(merge_request.id).cancel
     end
 
     private
@@ -146,5 +154,13 @@ module MergeRequests
       # loaded from the database they're strings
       params.with_indifferent_access[:sha] == merge_request.diff_head_sha
     end
+
+    def exclusive_lease(merge_request_id)
+      strong_memoize(:"exclusive_lease_#{merge_request_id}") do
+        lease_key = ['merge_requests_merge_service', merge_request_id].join(':')
+
+        Gitlab::ExclusiveLease.new(lease_key, timeout: LEASE_TIMEOUT)
+      end
+    end
   end
 end

app/services/merge_requests/post_merge_service.rb

@@ -12,20 +12,28 @@ module MergeRequests
     MAX_RETARGET_MERGE_REQUESTS = 4
 
     def execute(merge_request)
+      return if merge_request.merged?
+
+      # Mark the merge request as merged, everything that happens afterwards is
+      # executed once
       merge_request.mark_as_merged
-      close_issues(merge_request)
-      todo_service.merge_merge_request(merge_request, current_user)
+
       create_event(merge_request)
-      create_note(merge_request)
+      todo_service.merge_merge_request(merge_request, current_user)
+
       merge_request_activity_counter.track_merge_mr_action(user: current_user)
+
+      create_note(merge_request)
+      close_issues(merge_request)
       notification_service.merge_mr(merge_request, current_user)
-      execute_hooks(merge_request, 'merge')
       invalidate_cache_counts(merge_request, users: merge_request.assignees | merge_request.reviewers)
       merge_request.update_project_counter_caches
       delete_non_latest_diffs(merge_request)
       cancel_review_app_jobs!(merge_request)
       cleanup_environments(merge_request)
       cleanup_refs(merge_request)
+
+      execute_hooks(merge_request, 'merge')
     end
 
     private

app/workers/all_queues.yml

@@ -1890,7 +1890,7 @@
   :urgency: :high
   :resource_boundary: :unknown
   :weight: 5
-  :idempotent:
+  :idempotent: true
   :tags: []
 - :name: merge_request_cleanup_refs
   :feature_category: :code_review

app/workers/merge_worker.rb

@@ -7,11 +7,19 @@ class MergeWorker # rubocop:disable Scalability/IdempotentWorker
   urgency :high
   weight 5
   loggable_arguments 2
+  idempotent!
+
+  deduplicate :until_executed, including_scheduled: true
 
   def perform(merge_request_id, current_user_id, params)
     params = params.with_indifferent_access
-    current_user = User.find(current_user_id)
-    merge_request = MergeRequest.find(merge_request_id)
+
+    begin
+      current_user = User.find(current_user_id)
+      merge_request = MergeRequest.find(merge_request_id)
+    rescue ActiveRecord::RecordNotFound
+      return
+    end
 
     MergeRequests::MergeService.new(merge_request.target_project, current_user, params)
       .execute(merge_request)

changelogs/unreleased/make-mergeservice-idempotent.yml

+---
+title: Make MergeService idempotent
+merge_request: 55368
+author:
+type: performance

spec/services/merge_requests/merge_service_spec.rb

@@ -3,6 +3,8 @@
 require 'spec_helper'
 
 RSpec.describe MergeRequests::MergeService do
+  include ExclusiveLeaseHelpers
+
   let_it_be(:user) { create(:user) }
   let_it_be(:user2) { create(:user) }
 
@@ -20,6 +22,9 @@ RSpec.describe MergeRequests::MergeService do
       { commit_message: 'Awesome message', sha: merge_request.diff_head_sha }
     end
 
+    let(:lease_key) { "merge_requests_merge_service:#{merge_request.id}" }
+    let!(:lease) { stub_exclusive_lease(lease_key) }
+
     context 'valid params' do
       before do
         allow(service).to receive(:execute_hooks)
@@ -90,6 +95,20 @@ RSpec.describe MergeRequests::MergeService do
       end
     end
 
+    context 'running the service multiple time' do
+      it 'is idempotent' do
+        2.times { service.execute(merge_request) }
+
+        expect(merge_request.merge_error).to be_falsey
+        expect(merge_request).to be_valid
+        expect(merge_request).to be_merged
+
+        commit_messages = project.repository.commits('master', limit: 2).map(&:message)
+        expect(commit_messages.uniq.size).to eq(2)
+        expect(merge_request.in_progress_merge_commit_sha).to be_nil
+      end
+    end
+
     context 'when an invalid sha is passed' do
       let(:merge_request) do
         create(:merge_request, :simple,
@@ -306,6 +325,8 @@ RSpec.describe MergeRequests::MergeService do
       end
 
       it 'logs and saves error if user is not authorized' do
+        stub_exclusive_lease
+
         unauthorized_user = create(:user)
         project.add_reporter(unauthorized_user)
 
@@ -423,6 +444,7 @@ RSpec.describe MergeRequests::MergeService do
               merge_request.project.update!(merge_method: merge_method)
               error_message = 'Only fast-forward merge is allowed for your project. Please update your source branch'
               allow(service).to receive(:execute_hooks)
+              expect(lease).to receive(:cancel)
 
               service.execute(merge_request)
 
@@ -473,5 +495,17 @@ RSpec.describe MergeRequests::MergeService do
         end
       end
     end
+
+    context 'when the other sidekiq worker has already been running' do
+      before do
+        stub_exclusive_lease_taken(lease_key)
+      end
+
+      it 'does not execute service' do
+        expect(service).not_to receive(:commit)
+
+        service.execute(merge_request)
+      end
+    end
   end
 end

spec/services/merge_requests/post_merge_service_spec.rb

@@ -22,7 +22,6 @@ RSpec.describe MergeRequests::PostMergeService do
     it 'refreshes the number of open merge requests for a valid MR', :use_clean_rails_memory_store_caching do
       # Cache the counter before the MR changed state.
       project.open_merge_requests_count
-      merge_request.update!(state: 'merged')
 
       expect { subject }.to change { project.open_merge_requests_count }.from(1).to(0)
     end

spec/workers/merge_worker_spec.rb

@@ -29,5 +29,23 @@ RSpec.describe MergeWorker do
       source_project.repository.expire_branches_cache
       expect(source_project.repository.branch_names).not_to include('markdown')
     end
+
+    it_behaves_like 'an idempotent worker' do
+      let(:job_args) do
+        [
+          merge_request.id,
+          merge_request.author_id,
+          commit_message: 'wow such merge',
+          sha: merge_request.diff_head_sha
+        ]
+      end
+
+      it 'the merge request is still shown as merged' do
+        subject
+
+        merge_request.reload
+        expect(merge_request).to be_merged
+      end
+    end
   end
 end