Move pipeline processing async

Moves the first pipeline processing step to an async worker

app/workers/all_queues.yml

@@ -1307,6 +1307,14 @@
   :weight: 5
   :idempotent:
   :tags: []
+- :name: pipeline_processing:ci_initial_pipeline_process
+  :feature_category: :continuous_integration
+  :has_external_dependencies:
+  :urgency: :high
+  :resource_boundary: :unknown
+  :weight: 5
+  :idempotent: true
+  :tags: []
 - :name: pipeline_processing:ci_resource_groups_assign_resource_from_resource_group
   :feature_category: :continuous_delivery
   :has_external_dependencies:

app/workers/ci/initial_pipeline_process_worker.rb

+# frozen_string_literal: true
+
+module Ci
+  class InitialPipelineProcessWorker
+    include ApplicationWorker
+    include PipelineQueue
+
+    queue_namespace :pipeline_processing
+    feature_category :continuous_integration
+    urgency :high
+    loggable_arguments 1
+    idempotent!
+
+    def perform(pipeline_id)
+      Ci::Pipeline.find_by_id(pipeline_id).try do |pipeline|
+        Ci::ProcessPipelineService
+          .new(pipeline)
+          .execute
+      end
+    end
+  end
+end

config/feature_flags/development/ci_async_initial_pipeline_processing.yml

+---
+name: ci_async_initial_pipeline_processing
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/57589
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/326217
+milestone: '13.11'
+type: development
+group: group::continuous integration
+default_enabled: false

ee/spec/services/ci/create_pipeline_service/needs_spec.rb

@@ -71,7 +71,9 @@ RSpec.describe Ci::CreatePipelineService do
     end
 
     it 'creates a pipeline with regular_job and bridge_dag_job pending' do
-      processables = execute.processables
+      pipeline = execute
+      processables = pipeline.processables
+      Ci::InitialPipelineProcessWorker.new.perform(pipeline.id)
 
       regular_job = processables.find { |processable| processable.name == 'regular_job' }
       bridge_dag_job = processables.find { |processable| processable.name == 'bridge_dag_job' }

ee/spec/services/ci/create_pipeline_service_spec.rb

@@ -32,7 +32,7 @@ RSpec.describe Ci::CreatePipelineService, '#execute' do
         pipeline = create_pipeline!
 
         expect(pipeline).to be_persisted
-        expect(pipeline).to be_pending
+        expect(pipeline).to be_created
       end
     end
 

ee/spec/services/ci/run_dast_scan_service_spec.rb

@@ -173,11 +173,6 @@ RSpec.describe Ci::RunDastScanService do
         expect(build.yaml_variables).to contain_exactly(*expected_variables)
       end
 
-      it 'enqueues a build' do
-        build = pipeline.builds.first
-        expect(build.queued_at).not_to be_nil
-      end
-
       context 'when the pipeline fails to save' do
         before do
           allow_any_instance_of(Ci::Pipeline).to receive(:created_successfully?).and_return(false)

lib/gitlab/ci/pipeline/chain/pipeline/process.rb

@@ -8,9 +8,11 @@ module Gitlab
           # After pipeline has been successfully created we can start processing it.
           class Process < Chain::Base
             def perform!
-              ::Ci::ProcessPipelineService
-                .new(@pipeline)
-                .execute
+              if ::Feature.enabled?(:ci_async_initial_pipeline_processing, project, default_enabled: :yaml)
+                ::Ci::InitialPipelineProcessWorker.perform_async(pipeline.id)
+              else
+                ::Ci::ProcessPipelineService.new(pipeline).execute
+              end
             end
 
             def break?

spec/controllers/projects/pipelines_controller_spec.rb

@@ -721,7 +721,7 @@ RSpec.describe Projects::PipelinesController do
 
           pipeline = project.ci_pipelines.last
           expected_redirect_path = Gitlab::Routing.url_helpers.project_pipeline_path(project, pipeline)
-          expect(pipeline).to be_pending
+          expect(pipeline).to be_created
           expect(response).to redirect_to(expected_redirect_path)
         end
       end

spec/features/merge_request/user_sees_merge_request_pipelines_spec.rb

@@ -62,7 +62,7 @@ RSpec.describe 'Merge request > User sees pipelines triggered by merge request',
 
     it 'sees branch pipelines and detached merge request pipelines in correct order' do
       page.within('.ci-table') do
-        expect(page).to have_selector('.ci-pending', count: 2)
+        expect(page).to have_selector('.ci-created', count: 2)
         expect(first('[data-testid="pipeline-url-link"]')).to have_content("##{detached_merge_request_pipeline.id}")
       end
     end
@@ -154,7 +154,7 @@ RSpec.describe 'Merge request > User sees pipelines triggered by merge request',
 
       context 'when detached merge request pipeline succeeds' do
         before do
-          detached_merge_request_pipeline.succeed!
+          detached_merge_request_pipeline.reload.succeed!
 
           wait_for_requests
         end
@@ -168,7 +168,7 @@ RSpec.describe 'Merge request > User sees pipelines triggered by merge request',
       context 'when branch pipeline succeeds' do
         before do
           click_link 'Overview'
-          push_pipeline.succeed!
+          push_pipeline.reload.succeed!
 
           wait_for_requests
         end
@@ -197,7 +197,7 @@ RSpec.describe 'Merge request > User sees pipelines triggered by merge request',
 
       it 'sees a branch pipeline in pipeline tab' do
         page.within('.ci-table') do
-          expect(page).to have_selector('.ci-pending', count: 1)
+          expect(page).to have_selector('.ci-created', count: 1)
           expect(first('[data-testid="pipeline-url-link"]')).to have_content("##{push_pipeline.id}")
         end
       end
@@ -341,7 +341,7 @@ RSpec.describe 'Merge request > User sees pipelines triggered by merge request',
 
       context 'when the previous pipeline failed in the fork project' do
         before do
-          detached_merge_request_pipeline.drop!
+          detached_merge_request_pipeline.reload.drop!
         end
 
         context 'when the parent project enables pipeline must succeed' do
@@ -376,7 +376,7 @@ RSpec.describe 'Merge request > User sees pipelines triggered by merge request',
 
       context 'when detached merge request pipeline succeeds' do
         before do
-          detached_merge_request_pipeline.succeed!
+          detached_merge_request_pipeline.reload.succeed!
 
           wait_for_requests
         end
@@ -389,7 +389,7 @@ RSpec.describe 'Merge request > User sees pipelines triggered by merge request',
 
       context 'when branch pipeline succeeds' do
         before do
-          push_pipeline.succeed!
+          push_pipeline.reload.succeed!
 
           wait_for_requests
         end

spec/lib/gitlab/ci/pipeline/chain/pipeline/process_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Ci::Pipeline::Chain::Pipeline::Process do
+  let_it_be(:project) { build(:project) }
+  let_it_be(:user) { build(:user) }
+  let_it_be(:pipeline) { build(:ci_pipeline, project: project, id: 42) }
+
+  let_it_be(:command) do
+    Gitlab::Ci::Pipeline::Chain::Command.new(project: project, current_user: user)
+  end
+
+  let(:step) { described_class.new(pipeline, command) }
+
+  describe '#perform!' do
+    subject(:perform) { step.perform! }
+
+    it 'schedules a job to process the pipeline' do
+      expect(Ci::InitialPipelineProcessWorker)
+        .to receive(:perform_async)
+        .with(42)
+
+      perform
+    end
+
+    context 'with async processing disabled' do
+      before do
+        stub_feature_flags(ci_async_initial_pipeline_processing: false)
+      end
+
+      it 'processes pipeline inline' do
+        expect(::Ci::ProcessPipelineService)
+          .to receive(:new)
+          .with(pipeline)
+          .and_call_original
+
+        perform
+      end
+    end
+  end
+
+  describe '#break?' do
+    it { expect(step.break?).to be_falsey }
+  end
+end

spec/requests/api/triggers_spec.rb

@@ -49,8 +49,6 @@ RSpec.describe API::Triggers do
 
         expect(response).to have_gitlab_http_status(:created)
         expect(json_response).to include('id' => pipeline.id)
-        pipeline.builds.reload
-        expect(pipeline.builds.pending.size).to eq(2)
         expect(pipeline.builds.size).to eq(5)
       end
 

spec/services/ci/create_downstream_pipeline_service_spec.rb

@@ -112,7 +112,7 @@ RSpec.describe Ci::CreateDownstreamPipelineService, '#execute' do
     it 'updates bridge status when downstream pipeline gets processed' do
       pipeline = service.execute(bridge)
 
-      expect(pipeline.reload).to be_pending
+      expect(pipeline.reload).to be_created
       expect(bridge.reload).to be_success
     end
 
@@ -227,7 +227,7 @@ RSpec.describe Ci::CreateDownstreamPipelineService, '#execute' do
           it 'updates bridge status when downstream pipeline gets processed' do
             pipeline = service.execute(bridge)
 
-            expect(pipeline.reload).to be_pending
+            expect(pipeline.reload).to be_created
             expect(bridge.reload).to be_success
           end
 

spec/services/ci/create_pipeline_service/cross_project_pipeline_spec.rb

@@ -40,6 +40,7 @@ RSpec.describe Ci::CreatePipelineService, '#execute' do
 
     it 'creates bridge job with resource group' do
       pipeline = create_pipeline!
+      Ci::InitialPipelineProcessWorker.new.perform(pipeline.id)
 
       test = pipeline.statuses.find_by(name: 'instrumentation_test')
       expect(pipeline).to be_created_successfully

spec/services/ci/create_pipeline_service/needs_spec.rb

@@ -202,7 +202,7 @@ RSpec.describe Ci::CreatePipelineService do
         YAML
       end
 
-      it 'creates a pipeline with build_a and test_b pending; deploy_b manual' do
+      it 'creates a pipeline with build_a and test_b pending; deploy_b manual', :sidekiq_inline do
         processables = pipeline.processables
 
         build_a = processables.find { |processable| processable.name == 'build_a' }

spec/services/ci/create_pipeline_service/parent_child_pipeline_spec.rb

@@ -91,6 +91,7 @@ RSpec.describe Ci::CreatePipelineService, '#execute' do
 
       it 'creates bridge job with resource group', :aggregate_failures do
         pipeline = create_pipeline!
+        Ci::InitialPipelineProcessWorker.new.perform(pipeline.id)
 
         test = pipeline.statuses.find_by(name: 'instrumentation_test')
         expect(pipeline).to be_created_successfully

spec/services/ci/create_pipeline_service/rules_spec.rb

@@ -230,8 +230,8 @@ RSpec.describe Ci::CreatePipelineService do
       end
 
       context 'matching the first rule in the list' do
-        it 'saves a pending pipeline' do
-          expect(pipeline).to be_pending
+        it 'saves a created pipeline' do
+          expect(pipeline).to be_created
           expect(pipeline).to be_persisted
         end
       end
@@ -239,8 +239,8 @@ RSpec.describe Ci::CreatePipelineService do
       context 'matching the last rule in the list' do
         let(:ref) { 'refs/heads/feature' }
 
-        it 'saves a pending pipeline' do
-          expect(pipeline).to be_pending
+        it 'saves a created pipeline' do
+          expect(pipeline).to be_created
           expect(pipeline).to be_persisted
         end
       end
@@ -280,8 +280,8 @@ RSpec.describe Ci::CreatePipelineService do
       end
 
       context 'matching the first rule in the list' do
-        it 'saves a pending pipeline' do
-          expect(pipeline).to be_pending
+        it 'saves a created pipeline' do
+          expect(pipeline).to be_created
           expect(pipeline).to be_persisted
         end
       end
@@ -305,8 +305,8 @@ RSpec.describe Ci::CreatePipelineService do
       context 'with partial match' do
         let(:ref) { 'refs/heads/feature' }
 
-        it 'saves a pending pipeline' do
-          expect(pipeline).to be_pending
+        it 'saves a created pipeline' do
+          expect(pipeline).to be_created
           expect(pipeline).to be_persisted
         end
       end
@@ -349,8 +349,8 @@ RSpec.describe Ci::CreatePipelineService do
       context 'where workflow passes and the job passes' do
         let(:ref) { 'refs/heads/feature' }
 
-        it 'saves a pending pipeline' do
-          expect(pipeline).to be_pending
+        it 'saves a created pipeline' do
+          expect(pipeline).to be_created
           expect(pipeline).to be_persisted
         end
       end

spec/services/ci/create_pipeline_service_spec.rb

@@ -63,7 +63,7 @@ RSpec.describe Ci::CreatePipelineService do
         expect(pipeline).to be_push
         expect(pipeline).to eq(project.ci_pipelines.last)
         expect(pipeline).to have_attributes(user: user)
-        expect(pipeline).to have_attributes(status: 'pending')
+        expect(pipeline).to have_attributes(status: 'created')
         expect(pipeline.iid).not_to be_nil
         expect(pipeline.repository_source?).to be true
         expect(pipeline.builds.first).to be_kind_of(Ci::Build)
@@ -253,7 +253,7 @@ RSpec.describe Ci::CreatePipelineService do
           pipeline
           pipeline_on_previous_commit
 
-          expect(pipeline.reload).to have_attributes(status: 'pending', auto_canceled_by_id: nil)
+          expect(pipeline.reload).to have_attributes(status: 'created', auto_canceled_by_id: nil)
         end
 
         it 'auto cancel pending non-HEAD pipelines', :sidekiq_might_not_need_inline do
@@ -263,8 +263,8 @@ RSpec.describe Ci::CreatePipelineService do
           expect(pipeline_on_previous_commit.reload).to have_attributes(status: 'canceled', auto_canceled_by_id: pipeline.id)
         end
 
-        it 'cancels running outdated pipelines', :sidekiq_might_not_need_inline do
-          pipeline_on_previous_commit.run
+        it 'cancels running outdated pipelines', :sidekiq_inline do
+          pipeline_on_previous_commit.reload.run
           head_pipeline = execute_service
 
           expect(pipeline_on_previous_commit.reload).to have_attributes(status: 'canceled', auto_canceled_by_id: head_pipeline.id)
@@ -278,13 +278,13 @@ RSpec.describe Ci::CreatePipelineService do
         end
 
         it 'does not cancel pipelines from the other branches' do
-          pending_pipeline = execute_service(
+          new_pipeline = execute_service(
             ref: 'refs/heads/feature',
             after: previous_commit_sha_from_ref('feature')
           )
           pipeline
 
-          expect(pending_pipeline.reload).to have_attributes(status: 'pending', auto_canceled_by_id: nil)
+          expect(new_pipeline.reload).to have_attributes(status: 'created', auto_canceled_by_id: nil)
         end
 
         context 'when the interruptible attribute is' do
@@ -465,12 +465,12 @@ RSpec.describe Ci::CreatePipelineService do
           project.update!(auto_cancel_pending_pipelines: 'disabled')
         end
 
-        it 'does not auto cancel pending non-HEAD pipelines' do
+        it 'does not auto cancel created non-HEAD pipelines' do
           pipeline_on_previous_commit
           pipeline
 
           expect(pipeline_on_previous_commit.reload)
-            .to have_attributes(status: 'pending', auto_canceled_by_id: nil)
+            .to have_attributes(status: 'created', auto_canceled_by_id: nil)
         end
       end
 
@@ -770,7 +770,7 @@ RSpec.describe Ci::CreatePipelineService do
         stub_ci_pipeline_yaml_file(config)
       end
 
-      it 'does not create a new pipeline' do
+      it 'does not create a new pipeline', :sidekiq_inline do
         result = execute_service
 
         expect(result).to be_persisted

spec/services/ci/external_pull_requests/create_pipeline_service_spec.rb

@@ -34,7 +34,7 @@ RSpec.describe Ci::ExternalPullRequests::CreatePipelineService do
           expect(subject).to eq(project.ci_pipelines.last)
           expect(subject.external_pull_request).to eq(pull_request)
           expect(subject.user).to eq(user)
-          expect(subject.status).to eq('pending')
+          expect(subject.status).to eq('created')
           expect(subject.ref).to eq(pull_request.source_branch)
           expect(subject.sha).to eq(pull_request.source_sha)
           expect(subject.source_sha).to eq(pull_request.source_sha)

spec/workers/ci/initial_pipeline_process_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::InitialPipelineProcessWorker do
+  describe '#perform' do
+    let_it_be(:pipeline) { create(:ci_pipeline, :with_job, status: :created) }
+
+    include_examples 'an idempotent worker' do
+      let(:job_args) { pipeline.id }
+
+      it 'marks the pipeline as pending' do
+        expect(pipeline).to be_created
+
+        subject
+
+        expect(pipeline.reload).to be_pending
+      end
+    end
+  end
+end