Create the pipelines asynchronously when refreshing the merge requests

Issue: gitlab.com/gitlab-org/gitlab/-/issues/218410
MR: gitlab.com/gitlab-org/gitlab/-/merge_requests/58542

app/services/merge_requests/base_service.rb

@@ -143,8 +143,12 @@ module MergeRequests
         merge_request, merge_request.project, current_user, old_reviewers)
     end
 
-    def create_pipeline_for(merge_request, user)
-      MergeRequests::CreatePipelineService.new(project, user).execute(merge_request)
+    def create_pipeline_for(merge_request, user, async: false)
+      if async
+        MergeRequests::CreatePipelineWorker.perform_async(project.id, user.id, merge_request.id)
+      else
+        MergeRequests::CreatePipelineService.new(project, user).execute(merge_request)
+      end
     end
 
     def abort_auto_merge(merge_request, reason)

app/services/merge_requests/refresh_service.rb

@@ -162,9 +162,12 @@ module MergeRequests
     end
 
     def refresh_pipelines_on_merge_requests(merge_request)
-      create_pipeline_for(merge_request, current_user)
-
-      UpdateHeadPipelineForMergeRequestWorker.perform_async(merge_request.id)
+      if Feature.enabled?(:code_review_async_pipeline_creation, project, default_enabled: :yaml)
+        create_pipeline_for(merge_request, current_user, async: true)
+      else
+        create_pipeline_for(merge_request, current_user, async: false)
+        UpdateHeadPipelineForMergeRequestWorker.perform_async(merge_request.id)
+      end
     end
 
     def abort_auto_merges(merge_request)

app/workers/all_queues.yml

@@ -1227,6 +1227,14 @@
   :weight: 4
   :idempotent:
   :tags: []
+- :name: pipeline_creation:merge_requests_create_pipeline
+  :feature_category: :continuous_integration
+  :has_external_dependencies:
+  :urgency: :high
+  :resource_boundary: :cpu
+  :weight: 4
+  :idempotent: true
+  :tags: []
 - :name: pipeline_creation:run_pipeline_schedule
   :feature_category: :continuous_integration
   :has_external_dependencies:

app/workers/merge_requests/create_pipeline_worker.rb

+# frozen_string_literal: true
+
+module MergeRequests
+  class CreatePipelineWorker
+    include ApplicationWorker
+    include PipelineQueue
+
+    queue_namespace :pipeline_creation
+    feature_category :continuous_integration
+    urgency :high
+    worker_resource_boundary :cpu
+    idempotent!
+
+    def perform(project_id, user_id, merge_request_id)
+      project = Project.find_by_id(project_id)
+      return unless project
+
+      user = User.find_by_id(user_id)
+      return unless user
+
+      merge_request = MergeRequest.find_by_id(merge_request_id)
+      return unless merge_request
+
+      MergeRequests::CreatePipelineService.new(project, user).execute(merge_request)
+      merge_request.update_head_pipeline
+    end
+  end
+end

changelogs/unreleased/move_pipeline_creation_async.yml

+---
+title: Create the pipelines asynchronously when refreshing merge requests
+merge_request: 58542
+author:
+type: performance

config/feature_flags/development/code_review_async_pipeline_creation.yml

+---
+name: code_review_async_pipeline_creation
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/58542
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/327559
+milestone: '13.11'
+type: development
+group: group::code review
+default_enabled: false

ee/spec/services/ee/merge_requests/refresh_service_spec.rb

@@ -249,7 +249,7 @@ RSpec.describe MergeRequests::RefreshService do
       end
     end
 
-    describe 'Pipelines for merge requests' do
+    describe 'Pipelines for merge requests', :sidekiq_inline do
       let(:service) { described_class.new(project, current_user) }
       let(:current_user) { merge_request.author }
 

spec/services/merge_requests/refresh_service_spec.rb

@@ -198,7 +198,7 @@ RSpec.describe MergeRequests::RefreshService do
       end
     end
 
-    describe 'Pipelines for merge requests' do
+    shared_examples 'Pipelines for merge requests' do
       before do
         stub_ci_pipeline_yaml_file(config)
       end
@@ -256,7 +256,7 @@ RSpec.describe MergeRequests::RefreshService do
             stub_feature_flags(ci_disallow_to_create_merge_request_pipelines_in_target_project: false)
           end
 
-          it 'creates detached merge request pipeline for fork merge request', :sidekiq_inline do
+          it 'creates detached merge request pipeline for fork merge request' do
             expect { subject }
               .to change { @fork_merge_request.pipelines_for_merge_request.count }.by(1)
 
@@ -364,6 +364,18 @@ RSpec.describe MergeRequests::RefreshService do
       end
     end
 
+    context 'when the code_review_async_pipeline_creation feature flag is on', :sidekiq_inline do
+      it_behaves_like 'Pipelines for merge requests'
+    end
+
+    context 'when the code_review_async_pipeline_creation feature flag is off', :sidekiq_inline do
+      before do
+        stub_feature_flags(code_review_async_pipeline_creation: false)
+      end
+
+      it_behaves_like 'Pipelines for merge requests'
+    end
+
     context 'push to origin repo source branch' do
       let(:refresh_service) { service.new(@project, @user) }
       let(:notification_service) { spy('notification_service') }

spec/workers/merge_requests/create_pipeline_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe MergeRequests::CreatePipelineWorker do
+  subject(:worker) { described_class.new }
+
+  describe '#perform' do
+    let(:user) { create(:user) }
+    let(:project) { create(:project) }
+    let(:merge_request) { create(:merge_request) }
+
+    context 'when the objects exist' do
+      it 'calls the merge request create pipeline service and calls update head pipeline' do
+        aggregate_failures do
+          expect_next_instance_of(MergeRequests::CreatePipelineService, project, user) do |service|
+            expect(service).to receive(:execute).with(merge_request)
+          end
+
+          expect(MergeRequest).to receive(:find_by_id).with(merge_request.id).and_return(merge_request)
+          expect(merge_request).to receive(:update_head_pipeline)
+
+          subject.perform(project.id, user.id, merge_request.id)
+        end
+      end
+    end
+
+    shared_examples 'when object does not exist' do
+      it 'does not call the create pipeline service' do
+        expect(MergeRequests::CreatePipelineService).not_to receive(:new)
+
+        expect { subject.perform(project.id, user.id, merge_request.id) }
+          .not_to raise_exception
+      end
+    end
+
+    context 'when the project does not exist' do
+      before do
+        project.destroy!
+      end
+
+      it_behaves_like 'when object does not exist'
+    end
+
+    context 'when the user does not exist' do
+      before do
+        user.destroy!
+      end
+
+      it_behaves_like 'when object does not exist'
+    end
+
+    context 'when the merge request does not exist' do
+      before do
+        merge_request.destroy!
+      end
+
+      it_behaves_like 'when object does not exist'
+    end
+  end
+end