Move external PR pipeline creation to Sidekiq

app/services/ci/external_pull_requests/create_pipeline_service.rb

@@ -16,9 +16,15 @@ module Ci
       private
 
       def create_pipeline_for(pull_request)
+        if ::Feature.enabled?(:ci_create_external_pr_pipeline_async, project, default_enabled: :yaml)
+          Ci::ExternalPullRequests::CreatePipelineWorker.perform_async(
+            project.id, current_user.id, pull_request.id
+          )
+        else
           Ci::CreatePipelineService.new(project, current_user, create_params(pull_request))
             .execute(:external_pull_request_event, external_pull_request: pull_request)
         end
+      end
 
       def create_params(pull_request)
         {

app/workers/all_queues.yml

@@ -1470,6 +1470,15 @@
   :weight: 3
   :idempotent:
   :tags: []
+- :name: pipeline_creation:ci_external_pull_requests_create_pipeline
+  :worker_name: Ci::ExternalPullRequests::CreatePipelineWorker
+  :feature_category: :pipeline_authoring
+  :has_external_dependencies:
+  :urgency: :high
+  :resource_boundary: :cpu
+  :weight: 4
+  :idempotent:
+  :tags: []
 - :name: pipeline_creation:create_pipeline
   :worker_name: CreatePipelineWorker
   :feature_category: :continuous_integration

app/workers/ci/external_pull_requests/create_pipeline_worker.rb

+# frozen_string_literal: true
+
+module Ci
+  module ExternalPullRequests
+    class CreatePipelineWorker # rubocop:disable Scalability/IdempotentWorker
+      include ApplicationWorker
+
+      data_consistency :always
+      queue_namespace :pipeline_creation
+      feature_category :pipeline_authoring
+      urgency :high
+      worker_resource_boundary :cpu
+
+      def perform(project_id, user_id, external_pull_request_id)
+        user = User.find_by_id(user_id)
+        return unless user
+
+        project = Project.find_by_id(project_id)
+        return unless project
+
+        external_pull_request = project.external_pull_requests.find_by_id(external_pull_request_id)
+        return unless external_pull_request
+
+        ::Ci::CreatePipelineService
+          .new(project, user, execute_params(external_pull_request))
+          .execute(:external_pull_request_event, external_pull_request: external_pull_request)
+      end
+
+      private
+
+      def execute_params(pull_request)
+        {
+          ref: pull_request.source_ref,
+          source_sha: pull_request.source_sha,
+          target_sha: pull_request.target_sha
+        }
+      end
+    end
+  end
+end

config/feature_flags/development/ci_create_external_pr_pipeline_async.yml

+---
+name: ci_create_external_pr_pipeline_async
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/68567
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/338908
+milestone: '14.3'
+type: development
+group: group::pipeline authoring
+default_enabled: false

ee/spec/requests/api/project_mirror_spec.rb

@@ -48,6 +48,7 @@ RSpec.describe API::ProjectMirror do
           let(:source_branch) { branch.name }
           let(:source_sha) { branch.target }
           let(:action) { 'opened' }
+          let(:user) { project_mirrored.mirror_user }
 
           let(:params) do
             {
@@ -80,26 +81,55 @@ RSpec.describe API::ProjectMirror do
             stub_licensed_features(ci_cd_projects: true, github_project_service_integration: true)
           end
 
+          subject(:send_request) { do_post(params: params) }
+
+          shared_examples_for 'triggering pipeline creation' do
+            context 'when the FF ci_create_external_pr_pipeline_async is disabled' do
+              before do
+                stub_feature_flags(ci_create_external_pr_pipeline_async: false)
+              end
+
+              let(:create_pipeline_service) { instance_double(Ci::CreatePipelineService) }
+
               it 'triggers a pipeline for pull request' do
                 expect(Ci::CreatePipelineService)
                   .to receive(:new)
-              .with(project_mirrored, project_mirrored.mirror_user, pipeline_params)
+                  .with(project_mirrored, user, pipeline_params)
                   .and_return(create_pipeline_service)
 
                 expect(create_pipeline_service)
                   .to receive(:execute)
                   .with(:external_pull_request_event, any_args)
 
-            do_post(params: params)
+                send_request
+
+                expect(response).to have_gitlab_http_status(:ok)
+              end
+            end
+
+            it 'enqueues Ci::ExternalPullRequests::CreatePipelineWorker' do
+              expect { send_request }
+                .to change { ExternalPullRequest.count }.by(1)
+                .and change { ::Ci::ExternalPullRequests::CreatePipelineWorker.jobs.count }.by(1)
 
               expect(response).to have_gitlab_http_status(:ok)
+
+              args = ::Ci::ExternalPullRequests::CreatePipelineWorker.jobs.last['args']
+              pull_request = ExternalPullRequest.last
+
+              expect(args[0]).to eq(project_mirrored.id)
+              expect(args[1]).to eq(user.id)
+              expect(args[2]).to eq(pull_request.id)
             end
+          end
+
+          it_behaves_like 'triggering pipeline creation'
 
           context 'when any param is missing' do
             let(:source_sha) { nil }
 
             it 'returns the error message' do
-              do_post(params: params)
+              send_request
 
               expect(response).to have_gitlab_http_status(:bad_request)
             end
@@ -111,31 +141,22 @@ RSpec.describe API::ProjectMirror do
             it 'ignores it and return success status' do
               expect(Ci::CreatePipelineService).not_to receive(:new)
 
-              do_post(params: params)
+              send_request
 
               expect(response).to have_gitlab_http_status(:unprocessable_entity)
             end
           end
 
           context 'when authenticated as user' do
-            let(:user) { create(:user) }
+            let_it_be(:user) { create(:user) }
 
-            it 'triggers a pipeline for pull request' do
+            before do
               project_member(:maintainer, user)
+            end
 
-              expect(Ci::CreatePipelineService)
-                .to receive(:new)
-                .with(project_mirrored, user, pipeline_params)
-                .and_return(create_pipeline_service)
-
-              expect(create_pipeline_service)
-                .to receive(:execute)
-                .with(:external_pull_request_event, any_args)
-
-              do_post(params: params, user: user, headers: {})
+            subject(:send_request) { do_post(params: params, user: user, headers: {}) }
 
-              expect(response).to have_gitlab_http_status(:ok)
-            end
+            it_behaves_like 'triggering pipeline creation'
           end
 
           context 'when ci_cd_projects is not available' do
@@ -144,7 +165,7 @@ RSpec.describe API::ProjectMirror do
             end
 
             it 'returns the error message' do
-              do_post(params: params)
+              send_request
 
               expect(response).to have_gitlab_http_status(:unprocessable_entity)
             end
@@ -156,7 +177,7 @@ RSpec.describe API::ProjectMirror do
             end
 
             it 'returns the error message' do
-              do_post(params: params)
+              send_request
 
               expect(response).to have_gitlab_http_status(:unprocessable_entity)
             end

ee/spec/services/ci/external_pull_requests/process_github_event_service_spec.rb

@@ -53,6 +53,11 @@ RSpec.describe Ci::ExternalPullRequests::ProcessGithubEventService do
           let(:source_branch) { branch.name }
           let(:source_sha) { branch.target }
 
+          context 'when the FF ci_create_external_pr_pipeline_async is disabled' do
+            before do
+              stub_feature_flags(ci_create_external_pr_pipeline_async: false)
+            end
+
             let(:create_pipeline_service) { instance_double(Ci::CreatePipelineService) }
 
             it 'creates a pipeline and the external pull request' do
@@ -71,6 +76,20 @@ RSpec.describe Ci::ExternalPullRequests::ProcessGithubEventService do
             end
           end
 
+          it 'enqueues Ci::ExternalPullRequests::CreatePipelineWorker' do
+            expect { subject.execute(params) }
+             .to change { ExternalPullRequest.count }.by(1)
+             .and change { ::Ci::ExternalPullRequests::CreatePipelineWorker.jobs.count }.by(1)
+
+            args = ::Ci::ExternalPullRequests::CreatePipelineWorker.jobs.last['args']
+            pull_request = ExternalPullRequest.last
+
+            expect(args[0]).to eq(project.id)
+            expect(args[1]).to eq(user.id)
+            expect(args[2]).to eq(pull_request.id)
+          end
+        end
+
         context 'when the pull request webhook occurs before mirror update' do
           let(:source_branch) { 'a-new-branch' }
           let(:source_sha) { 'non-existent-sha' }

spec/services/ci/external_pull_requests/create_pipeline_service_spec.rb

@@ -12,7 +12,7 @@ RSpec.describe Ci::ExternalPullRequests::CreatePipelineService do
       project.add_maintainer(user)
     end
 
-    subject(:response) { described_class.new(project, user).execute(pull_request) }
+    subject(:execute) { described_class.new(project, user).execute(pull_request) }
 
     context 'when pull request is open' do
       before do
@@ -21,16 +21,20 @@ RSpec.describe Ci::ExternalPullRequests::CreatePipelineService do
 
       context 'when source sha is the head of the source branch' do
         let(:source_branch) { project.repository.branches.last }
-        let(:create_pipeline_service) { instance_double(Ci::CreatePipelineService) }
 
         before do
           pull_request.update!(source_branch: source_branch.name, source_sha: source_branch.target)
         end
 
+        context 'when the FF ci_create_external_pr_pipeline_async is disabled' do
+          before do
+            stub_feature_flags(ci_create_external_pr_pipeline_async: false)
+          end
+
           it 'creates a pipeline for external pull request', :aggregate_failures do
-          pipeline = response.payload
+            pipeline = execute.payload
 
-          expect(response).to be_success
+            expect(execute).to be_success
             expect(pipeline).to be_valid
             expect(pipeline).to be_persisted
             expect(pipeline).to be_external_pull_request_event
@@ -44,6 +48,19 @@ RSpec.describe Ci::ExternalPullRequests::CreatePipelineService do
           end
         end
 
+        it 'enqueues Ci::ExternalPullRequests::CreatePipelineWorker' do
+          expect { execute }
+            .to change { ::Ci::ExternalPullRequests::CreatePipelineWorker.jobs.count }
+            .by(1)
+
+          args = ::Ci::ExternalPullRequests::CreatePipelineWorker.jobs.last['args']
+
+          expect(args[0]).to eq(project.id)
+          expect(args[1]).to eq(user.id)
+          expect(args[2]).to eq(pull_request.id)
+        end
+      end
+
       context 'when source sha is not the head of the source branch (force push upon rebase)' do
         let(:source_branch) { project.repository.branches.first }
         let(:commit) { project.repository.commits(source_branch.name, limit: 2).last }
@@ -53,11 +70,12 @@ RSpec.describe Ci::ExternalPullRequests::CreatePipelineService do
         end
 
         it 'does nothing', :aggregate_failures do
-          expect(Ci::CreatePipelineService).not_to receive(:new)
+          expect { execute }
+            .not_to change { ::Ci::ExternalPullRequests::CreatePipelineWorker.jobs.count }
 
-          expect(response).to be_error
-          expect(response.message).to eq('The source sha is not the head of the source branch')
-          expect(response.payload).to be_nil
+          expect(execute).to be_error
+          expect(execute.message).to eq('The source sha is not the head of the source branch')
+          expect(execute.payload).to be_nil
         end
       end
     end
@@ -68,11 +86,12 @@ RSpec.describe Ci::ExternalPullRequests::CreatePipelineService do
       end
 
       it 'does nothing', :aggregate_failures do
-        expect(Ci::CreatePipelineService).not_to receive(:new)
+        expect { execute }
+          .not_to change { ::Ci::ExternalPullRequests::CreatePipelineWorker.jobs.count }
 
-        expect(response).to be_error
-        expect(response.message).to eq('The pull request is not opened')
-        expect(response.payload).to be_nil
+        expect(execute).to be_error
+        expect(execute.message).to eq('The pull request is not opened')
+        expect(execute.payload).to be_nil
       end
     end
   end

spec/workers/ci/external_pull_requests/create_pipeline_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::ExternalPullRequests::CreatePipelineWorker do
+  let_it_be(:project) { create(:project, :auto_devops, :repository) }
+  let_it_be(:user) { project.owner }
+  let_it_be(:external_pull_request) do
+    branch = project.repository.branches.last
+    create(:external_pull_request, project: project, source_branch: branch.name, source_sha: branch.target)
+  end
+
+  let(:worker) { described_class.new }
+
+  describe '#perform' do
+    let(:project_id) { project.id }
+    let(:user_id) { user.id }
+    let(:external_pull_request_id) { external_pull_request.id }
+
+    subject(:perform) { worker.perform(project_id, user_id, external_pull_request_id) }
+
+    it 'creates the pipeline' do
+      pipeline = perform.payload
+
+      expect(pipeline).to be_valid
+      expect(pipeline).to be_persisted
+      expect(pipeline).to be_external_pull_request_event
+      expect(pipeline.project).to eq(project)
+      expect(pipeline.user).to eq(user)
+      expect(pipeline.external_pull_request).to eq(external_pull_request)
+      expect(pipeline.status).to eq('created')
+      expect(pipeline.ref).to eq(external_pull_request.source_branch)
+      expect(pipeline.sha).to eq(external_pull_request.source_sha)
+      expect(pipeline.source_sha).to eq(external_pull_request.source_sha)
+      expect(pipeline.target_sha).to eq(external_pull_request.target_sha)
+    end
+
+    shared_examples_for 'not calling service' do
+      it 'does not call the service' do
+        expect(Ci::CreatePipelineService).not_to receive(:new)
+        perform
+      end
+    end
+
+    context 'when the project not found' do
+      let(:project_id) { non_existing_record_id }
+
+      it_behaves_like 'not calling service'
+    end
+
+    context 'when the user not found' do
+      let(:user_id) { non_existing_record_id }
+
+      it_behaves_like 'not calling service'
+    end
+
+    context 'when the pull request not found' do
+      let(:external_pull_request_id) { non_existing_record_id }
+
+      it_behaves_like 'not calling service'
+    end
+
+    context 'when the pull request does not belong to the project' do
+      let(:external_pull_request_id) { create(:external_pull_request).id }
+
+      it_behaves_like 'not calling service'
+    end
+  end
+end