Add latest changes from gitlab-org/gitlab@13-10-stable-eev13.10.0-rc40

23 files changed, 1203 insertions, 528 deletions

diff --git a/spec/services/ci/build_report_result_service_spec.rb b/spec/services/ci/build_report_result_service_spec.rb
index 7c2702af086..c5238b7f5e0 100644
--- a/spec/services/ci/build_report_result_service_spec.rb
+++ b/spec/services/ci/build_report_result_service_spec.rb
@@ -10,13 +10,17 @@ RSpec.describe Ci::BuildReportResultService do
       let(:build) { create(:ci_build, :success, :test_reports) }
 
       it 'creates a build report result entry', :aggregate_failures do
+        expect { build_report_result }.to change { Ci::BuildReportResult.count }.by(1)
         expect(build_report_result.tests_name).to eq("test")
         expect(build_report_result.tests_success).to eq(2)
         expect(build_report_result.tests_failed).to eq(2)
         expect(build_report_result.tests_errored).to eq(0)
         expect(build_report_result.tests_skipped).to eq(0)
         expect(build_report_result.tests_duration).to eq(0.010284)
-        expect(Ci::BuildReportResult.count).to eq(1)
+      end
+
+      it 'tracks unique test cases parsed' do
+        build_report_result
 
         unique_test_cases_parsed = Gitlab::UsageDataCounters::HLLRedisCounter.unique_events(
           event_names: described_class::EVENT_NAME,
@@ -26,6 +30,32 @@ RSpec.describe Ci::BuildReportResultService do
         expect(unique_test_cases_parsed).to eq(4)
       end
 
+      context 'and build has test report parsing errors' do
+        let(:build) { create(:ci_build, :success, :broken_test_reports) }
+
+        it 'creates a build report result entry with suite error', :aggregate_failures do
+          expect { build_report_result }.to change { Ci::BuildReportResult.count }.by(1)
+          expect(build_report_result.tests_name).to eq("test")
+          expect(build_report_result.tests_success).to eq(0)
+          expect(build_report_result.tests_failed).to eq(0)
+          expect(build_report_result.tests_errored).to eq(0)
+          expect(build_report_result.tests_skipped).to eq(0)
+          expect(build_report_result.tests_duration).to eq(0)
+          expect(build_report_result.suite_error).to be_present
+        end
+
+        it 'does not track unique test cases parsed' do
+          build_report_result
+
+          unique_test_cases_parsed = Gitlab::UsageDataCounters::HLLRedisCounter.unique_events(
+            event_names: described_class::EVENT_NAME,
+            start_date: 2.weeks.ago,
+            end_date: 2.weeks.from_now
+          )
+          expect(unique_test_cases_parsed).to eq(0)
+        end
+      end
+
       context 'when data has already been persisted' do
         it 'raises an error and do not persist the same data twice' do
           expect { 2.times { described_class.new.execute(build) } }.to raise_error(ActiveRecord::RecordNotUnique)
diff --git a/spec/services/ci/create_pipeline_service/environment_spec.rb b/spec/services/ci/create_pipeline_service/environment_spec.rb
new file mode 100644
index 00000000000..0ed63012325
--- /dev/null
+++ b/spec/services/ci/create_pipeline_service/environment_spec.rb
@@ -0,0 +1,48 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::CreatePipelineService do
+  let_it_be(:project) { create(:project, :repository) }
+  let_it_be(:developer) { create(:user) }
+  let(:service) { described_class.new(project, user, ref: 'master') }
+  let(:user) { developer }
+
+  before_all do
+    project.add_developer(developer)
+  end
+
+  describe '#execute' do
+    subject { service.execute(:push) }
+
+    context 'with deployment tier' do
+      before do
+        config = YAML.dump(
+          deploy: {
+            script: 'ls',
+            environment: { name: "review/$CI_COMMIT_REF_NAME", deployment_tier: tier }
+          })
+
+        stub_ci_pipeline_yaml_file(config)
+      end
+
+      let(:tier) { 'development' }
+
+      it 'creates the environment with the expected tier' do
+        is_expected.to be_created_successfully
+
+        expect(Environment.find_by_name("review/master")).to be_development
+      end
+
+      context 'when tier is testing' do
+        let(:tier) { 'testing' }
+
+        it 'creates the environment with the expected tier' do
+          is_expected.to be_created_successfully
+
+          expect(Environment.find_by_name("review/master")).to be_testing
+        end
+      end
+    end
+  end
+end
diff --git a/spec/services/ci/create_pipeline_service/needs_spec.rb b/spec/services/ci/create_pipeline_service/needs_spec.rb
index 512091035a2..a6b0a9662c9 100644
--- a/spec/services/ci/create_pipeline_service/needs_spec.rb
+++ b/spec/services/ci/create_pipeline_service/needs_spec.rb
@@ -238,5 +238,51 @@ RSpec.describe Ci::CreatePipelineService do
           .to eq('jobs:invalid_dag_job:needs config can not be an empty hash')
       end
     end
+
+    context 'when the needed job has rules' do
+      let(:config) do
+        <<~YAML
+          build:
+            stage: build
+            script: exit 0
+            rules:
+              - if: $CI_COMMIT_REF_NAME == "invalid"
+
+          test:
+            stage: test
+            script: exit 0
+            needs: [build]
+        YAML
+      end
+
+      it 'returns error' do
+        expect(pipeline.yaml_errors)
+          .to eq("'test' job needs 'build' job, but it was not added to the pipeline")
+      end
+
+      context 'when need is optional' do
+        let(:config) do
+          <<~YAML
+            build:
+              stage: build
+              script: exit 0
+              rules:
+                - if: $CI_COMMIT_REF_NAME == "invalid"
+
+            test:
+              stage: test
+              script: exit 0
+              needs:
+                - job: build
+                  optional: true
+          YAML
+        end
+
+        it 'creates the pipeline without an error' do
+          expect(pipeline).to be_persisted
+          expect(pipeline.builds.pluck(:name)).to contain_exactly('test')
+        end
+      end
+    end
   end
 end
diff --git a/spec/services/ci/create_pipeline_service/parallel_spec.rb b/spec/services/ci/create_pipeline_service/parallel_spec.rb
new file mode 100644
index 00000000000..5e34a67d376
--- /dev/null
+++ b/spec/services/ci/create_pipeline_service/parallel_spec.rb
@@ -0,0 +1,118 @@
+# frozen_string_literal: true
+require 'spec_helper'
+
+RSpec.describe Ci::CreatePipelineService do
+  let_it_be(:project) { create(:project, :repository) }
+  let_it_be(:user)    { project.owner }
+
+  let(:service)  { described_class.new(project, user, { ref: 'master' }) }
+  let(:pipeline) { service.execute(:push) }
+
+  before do
+    stub_ci_pipeline_yaml_file(config)
+  end
+
+  context 'job:parallel' do
+    context 'numeric' do
+      let(:config) do
+        <<-EOY
+        job:
+          script: "echo job"
+          parallel: 3
+        EOY
+      end
+
+      it 'creates the pipeline' do
+        expect(pipeline).to be_created_successfully
+      end
+
+      it 'creates 3 jobs' do
+        expect(pipeline.processables.pluck(:name)).to contain_exactly(
+          'job 1/3', 'job 2/3', 'job 3/3'
+        )
+      end
+    end
+
+    context 'matrix' do
+      let(:config) do
+        <<-EOY
+        job:
+          script: "echo job"
+          parallel:
+            matrix:
+              - PROVIDER: ovh
+                STACK: [monitoring, app]
+              - PROVIDER: [gcp, vultr]
+                STACK: [data]
+        EOY
+      end
+
+      it 'creates the pipeline' do
+        expect(pipeline).to be_created_successfully
+      end
+
+      it 'creates 4 builds with the corresponding matrix variables' do
+        expect(pipeline.processables.pluck(:name)).to contain_exactly(
+          'job: [gcp, data]', 'job: [ovh, app]', 'job: [ovh, monitoring]', 'job: [vultr, data]'
+        )
+
+        job1 = find_job('job: [gcp, data]')
+        job2 = find_job('job: [ovh, app]')
+        job3 = find_job('job: [ovh, monitoring]')
+        job4 = find_job('job: [vultr, data]')
+
+        expect(job1.scoped_variables.to_hash).to include('PROVIDER' => 'gcp', 'STACK' => 'data')
+        expect(job2.scoped_variables.to_hash).to include('PROVIDER' => 'ovh', 'STACK' => 'app')
+        expect(job3.scoped_variables.to_hash).to include('PROVIDER' => 'ovh', 'STACK' => 'monitoring')
+        expect(job4.scoped_variables.to_hash).to include('PROVIDER' => 'vultr', 'STACK' => 'data')
+      end
+
+      context 'when a bridge is using parallel:matrix' do
+        let(:config) do
+          <<-EOY
+          job:
+            stage: test
+            script: "echo job"
+
+          deploy:
+            stage: deploy
+            trigger:
+              include: child.yml
+            parallel:
+              matrix:
+                - PROVIDER: ovh
+                  STACK: [monitoring, app]
+                - PROVIDER: [gcp, vultr]
+                  STACK: [data]
+          EOY
+        end
+
+        it 'creates the pipeline' do
+          expect(pipeline).to be_created_successfully
+        end
+
+        it 'creates 1 build and 4 bridges with the corresponding matrix variables' do
+          expect(pipeline.processables.pluck(:name)).to contain_exactly(
+            'job', 'deploy: [gcp, data]', 'deploy: [ovh, app]', 'deploy: [ovh, monitoring]', 'deploy: [vultr, data]'
+          )
+
+          bridge1 = find_job('deploy: [gcp, data]')
+          bridge2 = find_job('deploy: [ovh, app]')
+          bridge3 = find_job('deploy: [ovh, monitoring]')
+          bridge4 = find_job('deploy: [vultr, data]')
+
+          expect(bridge1.scoped_variables.to_hash).to include('PROVIDER' => 'gcp', 'STACK' => 'data')
+          expect(bridge2.scoped_variables.to_hash).to include('PROVIDER' => 'ovh', 'STACK' => 'app')
+          expect(bridge3.scoped_variables.to_hash).to include('PROVIDER' => 'ovh', 'STACK' => 'monitoring')
+          expect(bridge4.scoped_variables.to_hash).to include('PROVIDER' => 'vultr', 'STACK' => 'data')
+        end
+      end
+    end
+  end
+
+  private
+
+  def find_job(name)
+    pipeline.processables.find { |job| job.name == name }
+  end
+end
diff --git a/spec/services/ci/create_pipeline_service/rules_spec.rb b/spec/services/ci/create_pipeline_service/rules_spec.rb
index 04ecac6a85a..e97e74c1515 100644
--- a/spec/services/ci/create_pipeline_service/rules_spec.rb
+++ b/spec/services/ci/create_pipeline_service/rules_spec.rb
@@ -174,33 +174,19 @@ RSpec.describe Ci::CreatePipelineService do
           let(:ref) { 'refs/heads/master' }
 
           it 'overrides VAR1' do
-            variables = job.scoped_variables_hash
+            variables = job.scoped_variables.to_hash
 
             expect(variables['VAR1']).to eq('overridden var 1')
             expect(variables['VAR2']).to eq('my var 2')
             expect(variables['VAR3']).to be_nil
           end
-
-          context 'when FF ci_rules_variables is disabled' do
-            before do
-              stub_feature_flags(ci_rules_variables: false)
-            end
-
-            it 'does not affect variables' do
-              variables = job.scoped_variables_hash
-
-              expect(variables['VAR1']).to eq('my var 1')
-              expect(variables['VAR2']).to eq('my var 2')
-              expect(variables['VAR3']).to be_nil
-            end
-          end
         end
 
         context 'when matching to the second rule' do
           let(:ref) { 'refs/heads/feature' }
 
           it 'overrides VAR2 and adds VAR3' do
-            variables = job.scoped_variables_hash
+            variables = job.scoped_variables.to_hash
 
             expect(variables['VAR1']).to eq('my var 1')
             expect(variables['VAR2']).to eq('overridden var 2')
@@ -212,7 +198,7 @@ RSpec.describe Ci::CreatePipelineService do
           let(:ref) { 'refs/heads/wip' }
 
           it 'does not affect vars' do
-            variables = job.scoped_variables_hash
+            variables = job.scoped_variables.to_hash
 
             expect(variables['VAR1']).to eq('my var 1')
             expect(variables['VAR2']).to eq('my var 2')
diff --git a/spec/services/ci/create_pipeline_service_spec.rb b/spec/services/ci/create_pipeline_service_spec.rb
index 1005985b3e4..9fafc57a770 100644
--- a/spec/services/ci/create_pipeline_service_spec.rb
+++ b/spec/services/ci/create_pipeline_service_spec.rb
@@ -101,7 +101,7 @@ RSpec.describe Ci::CreatePipelineService do
 
       describe 'recording a conversion event' do
         it 'schedules a record conversion event worker' do
-          expect(Experiments::RecordConversionEventWorker).to receive(:perform_async).with(:ci_syntax_templates, user.id)
+          expect(Experiments::RecordConversionEventWorker).to receive(:perform_async).with(:ci_syntax_templates_b, user.id)
 
           pipeline
         end
diff --git a/spec/services/ci/destroy_expired_job_artifacts_service_spec.rb b/spec/services/ci/destroy_expired_job_artifacts_service_spec.rb
index 1edcef2977b..d315dd35632 100644
--- a/spec/services/ci/destroy_expired_job_artifacts_service_spec.rb
+++ b/spec/services/ci/destroy_expired_job_artifacts_service_spec.rb
@@ -77,14 +77,6 @@ RSpec.describe Ci::DestroyExpiredJobArtifactsService, :clean_gitlab_redis_shared
           it 'does not remove the files' do
             expect { subject }.not_to change { artifact.file.exists? }
           end
-
-          it 'reports metrics for destroyed artifacts' do
-            counter = service.send(:destroyed_artifacts_counter)
-
-            expect(counter).to receive(:increment).with({}, 1).and_call_original
-
-            subject
-          end
         end
       end
 
@@ -244,5 +236,17 @@ RSpec.describe Ci::DestroyExpiredJobArtifactsService, :clean_gitlab_redis_shared
         expect { subject }.to change { Ci::JobArtifact.count }.by(-1)
       end
     end
+
+    context 'when all artifacts are locked' do
+      before do
+        pipeline = create(:ci_pipeline, locked: :artifacts_locked)
+        job = create(:ci_build, pipeline: pipeline)
+        artifact.update!(job: job)
+      end
+
+      it 'destroys no artifacts' do
+        expect { subject }.to change { Ci::JobArtifact.count }.by(0)
+      end
+    end
   end
 end
diff --git a/spec/services/ci/expire_pipeline_cache_service_spec.rb b/spec/services/ci/expire_pipeline_cache_service_spec.rb
index 8df5d0bc159..3dbf2dbb8f1 100644
--- a/spec/services/ci/expire_pipeline_cache_service_spec.rb
+++ b/spec/services/ci/expire_pipeline_cache_service_spec.rb
@@ -13,10 +13,14 @@ RSpec.describe Ci::ExpirePipelineCacheService do
       pipelines_path = "/#{project.full_path}/-/pipelines.json"
       new_mr_pipelines_path = "/#{project.full_path}/-/merge_requests/new.json"
       pipeline_path = "/#{project.full_path}/-/pipelines/#{pipeline.id}.json"
+      graphql_pipeline_path = "/api/graphql:pipelines/id/#{pipeline.id}"
 
-      expect_any_instance_of(Gitlab::EtagCaching::Store).to receive(:touch).with(pipelines_path)
-      expect_any_instance_of(Gitlab::EtagCaching::Store).to receive(:touch).with(new_mr_pipelines_path)
-      expect_any_instance_of(Gitlab::EtagCaching::Store).to receive(:touch).with(pipeline_path)
+      expect_next_instance_of(Gitlab::EtagCaching::Store) do |store|
+        expect(store).to receive(:touch).with(pipelines_path)
+        expect(store).to receive(:touch).with(new_mr_pipelines_path)
+        expect(store).to receive(:touch).with(pipeline_path)
+        expect(store).to receive(:touch).with(graphql_pipeline_path)
+      end
 
       subject.execute(pipeline)
     end
@@ -59,5 +63,36 @@ RSpec.describe Ci::ExpirePipelineCacheService do
         expect(Project.find(project_with_repo.id).pipeline_status.has_status?).to be_falsey
       end
     end
+
+    context 'when the pipeline is triggered by another pipeline' do
+      let(:source) { create(:ci_sources_pipeline, pipeline: pipeline) }
+
+      it 'updates the cache of dependent pipeline' do
+        dependent_pipeline_path = "/#{source.source_project.full_path}/-/pipelines/#{source.source_pipeline.id}.json"
+
+        expect_next_instance_of(Gitlab::EtagCaching::Store) do |store|
+          allow(store).to receive(:touch)
+          expect(store).to receive(:touch).with(dependent_pipeline_path)
+        end
+
+        subject.execute(pipeline)
+      end
+    end
+
+    context 'when the pipeline triggered another pipeline' do
+      let(:build) { create(:ci_build, pipeline: pipeline) }
+      let(:source) { create(:ci_sources_pipeline, source_job: build) }
+
+      it 'updates the cache of dependent pipeline' do
+        dependent_pipeline_path = "/#{source.project.full_path}/-/pipelines/#{source.pipeline.id}.json"
+
+        expect_next_instance_of(Gitlab::EtagCaching::Store) do |store|
+          allow(store).to receive(:touch)
+          expect(store).to receive(:touch).with(dependent_pipeline_path)
+        end
+
+        subject.execute(pipeline)
+      end
+    end
   end
 end
diff --git a/spec/services/ci/job_artifacts_destroy_batch_service_spec.rb b/spec/services/ci/job_artifacts_destroy_batch_service_spec.rb
new file mode 100644
index 00000000000..74fbbf28ef1
--- /dev/null
+++ b/spec/services/ci/job_artifacts_destroy_batch_service_spec.rb
@@ -0,0 +1,81 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::JobArtifactsDestroyBatchService do
+  include ExclusiveLeaseHelpers
+
+  let(:artifacts) { Ci::JobArtifact.all }
+  let(:service) { described_class.new(artifacts, pick_up_at: Time.current) }
+
+  describe '.execute' do
+    subject(:execute) { service.execute }
+
+    let_it_be(:artifact, refind: true) do
+      create(:ci_job_artifact)
+    end
+
+    context 'when the artifact has a file attached to it' do
+      before do
+        artifact.file = fixture_file_upload(Rails.root.join('spec/fixtures/ci_build_artifacts.zip'), 'application/zip')
+        artifact.save!
+      end
+
+      it 'creates a deleted object' do
+        expect { subject }.to change { Ci::DeletedObject.count }.by(1)
+      end
+
+      it 'resets project statistics' do
+        expect(ProjectStatistics).to receive(:increment_statistic).once
+          .with(artifact.project, :build_artifacts_size, -artifact.file.size)
+          .and_call_original
+
+        execute
+      end
+
+      it 'does not remove the files' do
+        expect { execute }.not_to change { artifact.file.exists? }
+      end
+
+      it 'reports metrics for destroyed artifacts' do
+        expect_next_instance_of(Gitlab::Ci::Artifacts::Metrics) do |metrics|
+          expect(metrics).to receive(:increment_destroyed_artifacts).with(1).and_call_original
+        end
+
+        execute
+      end
+    end
+
+    context 'when failed to destroy artifact' do
+      context 'when the import fails' do
+        before do
+          expect(Ci::DeletedObject)
+            .to receive(:bulk_import)
+            .once
+            .and_raise(ActiveRecord::RecordNotDestroyed)
+        end
+
+        it 'raises an exception and stop destroying' do
+          expect { execute }.to raise_error(ActiveRecord::RecordNotDestroyed)
+                            .and not_change { Ci::JobArtifact.count }.from(1)
+        end
+      end
+    end
+
+    context 'when there are no artifacts' do
+      let(:artifacts) { Ci::JobArtifact.none }
+
+      before do
+        artifact.destroy!
+      end
+
+      it 'does not raise error' do
+        expect { execute }.not_to raise_error
+      end
+
+      it 'reports the number of destroyed artifacts' do
+        is_expected.to eq(destroyed_artifacts_count: 0, status: :success)
+      end
+    end
+  end
+end
diff --git a/spec/services/ci/pipeline_processing/shared_processing_service.rb b/spec/services/ci/pipeline_processing/shared_processing_service.rb
index bbd7422b435..13c924a3089 100644
--- a/spec/services/ci/pipeline_processing/shared_processing_service.rb
+++ b/spec/services/ci/pipeline_processing/shared_processing_service.rb
@@ -1,21 +1,13 @@
 # frozen_string_literal: true
 
 RSpec.shared_examples 'Pipeline Processing Service' do
-  let(:user) { create(:user) }
-  let(:project) { create(:project) }
+  let(:project) { create(:project, :repository) }
+  let(:user)    { project.owner }
 
   let(:pipeline) do
     create(:ci_empty_pipeline, ref: 'master', project: project)
   end
 
-  before do
-    stub_ci_pipeline_to_return_yaml_file
-
-    stub_not_protect_default_branch
-
-    project.add_developer(user)
-  end
-
   context 'when simple pipeline is defined' do
     before do
       create_build('linux', stage_idx: 0)
@@ -843,19 +835,97 @@ RSpec.shared_examples 'Pipeline Processing Service' do
       create(:ci_build_need, build: deploy, name: 'linux:build')
     end
 
-    it 'makes deploy DAG to be waiting for optional manual to finish' do
+    it 'makes deploy DAG to be skipped' do
       expect(process_pipeline).to be_truthy
 
-      expect(stages).to eq(%w(skipped created))
+      expect(stages).to eq(%w(skipped skipped))
       expect(all_builds.manual).to contain_exactly(linux_build)
-      expect(all_builds.created).to contain_exactly(deploy)
+      expect(all_builds.skipped).to contain_exactly(deploy)
+    end
+
+    context 'when FF ci_fix_pipeline_status_for_dag_needs_manual is disabled' do
+      before do
+        stub_feature_flags(ci_fix_pipeline_status_for_dag_needs_manual: false)
+      end
+
+      it 'makes deploy DAG to be waiting for optional manual to finish' do
+        expect(process_pipeline).to be_truthy
+
+        expect(stages).to eq(%w(skipped created))
+        expect(all_builds.manual).to contain_exactly(linux_build)
+        expect(all_builds.created).to contain_exactly(deploy)
+      end
+    end
+  end
+
+  context 'when a bridge job has parallel:matrix config', :sidekiq_inline do
+    let(:parent_config) do
+      <<-EOY
+      test:
+        stage: test
+        script: echo test
+
+      deploy:
+        stage: deploy
+        trigger:
+          include: .child.yml
+        parallel:
+          matrix:
+            - PROVIDER: ovh
+              STACK: [monitoring, app]
+      EOY
+    end
+
+    let(:child_config) do
+      <<-EOY
+      test:
+        stage: test
+        script: echo test
+      EOY
+    end
+
+    let(:pipeline) do
+      Ci::CreatePipelineService.new(project, user, { ref: 'master' }).execute(:push)
+    end
+
+    before do
+      allow_next_instance_of(Repository) do |repository|
+        allow(repository)
+          .to receive(:blob_data_at)
+          .with(an_instance_of(String), '.gitlab-ci.yml')
+          .and_return(parent_config)
+
+        allow(repository)
+          .to receive(:blob_data_at)
+          .with(an_instance_of(String), '.child.yml')
+          .and_return(child_config)
+      end
+    end
+
+    it 'creates pipeline with bridges, then passes the matrix variables to downstream jobs' do
+      expect(all_builds_names).to contain_exactly('test', 'deploy: [ovh, monitoring]', 'deploy: [ovh, app]')
+      expect(all_builds_statuses).to contain_exactly('pending', 'created', 'created')
+
+      succeed_pending
+
+      # bridge jobs directly transition to success
+      expect(all_builds_statuses).to contain_exactly('success', 'success', 'success')
+
+      bridge1 = all_builds.find_by(name: 'deploy: [ovh, monitoring]')
+      bridge2 = all_builds.find_by(name: 'deploy: [ovh, app]')
+
+      downstream_job1 = bridge1.downstream_pipeline.processables.first
+      downstream_job2 = bridge2.downstream_pipeline.processables.first
+
+      expect(downstream_job1.scoped_variables.to_hash).to include('PROVIDER' => 'ovh', 'STACK' => 'monitoring')
+      expect(downstream_job2.scoped_variables.to_hash).to include('PROVIDER' => 'ovh', 'STACK' => 'app')
     end
   end
 
   private
 
   def all_builds
-    pipeline.builds.order(:stage_idx, :id)
+    pipeline.processables.order(:stage_idx, :id)
   end
 
   def builds
diff --git a/spec/services/ci/pipeline_processing/shared_processing_service_tests_with_yaml.rb b/spec/services/ci/pipeline_processing/shared_processing_service_tests_with_yaml.rb
index 2936d6fae4d..a9f9db8c689 100644
--- a/spec/services/ci/pipeline_processing/shared_processing_service_tests_with_yaml.rb
+++ b/spec/services/ci/pipeline_processing/shared_processing_service_tests_with_yaml.rb
@@ -1,21 +1,19 @@
 # frozen_string_literal: true
 
 RSpec.shared_context 'Pipeline Processing Service Tests With Yaml' do
+  let_it_be(:project) { create(:project, :repository) }
+  let_it_be(:user)    { project.owner }
+
   where(:test_file_path) do
     Dir.glob(Rails.root.join('spec/services/ci/pipeline_processing/test_cases/*.yml'))
   end
 
   with_them do
     let(:test_file) { YAML.load_file(test_file_path) }
-
-    let(:user)     { create(:user) }
-    let(:project)  { create(:project, :repository) }
     let(:pipeline) { Ci::CreatePipelineService.new(project, user, ref: 'master').execute(:pipeline) }
 
     before do
       stub_ci_pipeline_yaml_file(YAML.dump(test_file['config']))
-      stub_not_protect_default_branch
-      project.add_developer(user)
     end
 
     it 'follows transitions' do
diff --git a/spec/services/ci/pipeline_processing/test_cases/dag_build2_build1_rules_out_test_needs_build1_with_optional.yml b/spec/services/ci/pipeline_processing/test_cases/dag_build2_build1_rules_out_test_needs_build1_with_optional.yml
new file mode 100644
index 00000000000..170e1b589bb
--- /dev/null
+++ b/spec/services/ci/pipeline_processing/test_cases/dag_build2_build1_rules_out_test_needs_build1_with_optional.yml
@@ -0,0 +1,50 @@
+config:
+  build1:
+    stage: build
+    script: exit 0
+    rules:
+      - if: $CI_COMMIT_REF_NAME == "invalid"
+
+  build2:
+    stage: build
+    script: exit 0
+
+  test:
+    stage: test
+    script: exit 0
+    needs:
+      - job: build1
+        optional: true
+
+init:
+  expect:
+    pipeline: pending
+    stages:
+      build: pending
+      test: pending
+    jobs:
+      build2: pending
+      test: pending
+
+transitions:
+  - event: success
+    jobs: [test]
+    expect:
+      pipeline: running
+      stages:
+        build: pending
+        test: success
+      jobs:
+        build2: pending
+        test: success
+
+  - event: success
+    jobs: [build2]
+    expect:
+      pipeline: success
+      stages:
+        build: success
+        test: success
+      jobs:
+        build2: success
+        test: success
diff --git a/spec/services/ci/pipeline_processing/test_cases/dag_build_rules_out_test_needs_build_with_optional.yml b/spec/services/ci/pipeline_processing/test_cases/dag_build_rules_out_test_needs_build_with_optional.yml
new file mode 100644
index 00000000000..85e7aa04a24
--- /dev/null
+++ b/spec/services/ci/pipeline_processing/test_cases/dag_build_rules_out_test_needs_build_with_optional.yml
@@ -0,0 +1,31 @@
+config:
+  build:
+    stage: build
+    script: exit 0
+    rules:
+      - if: $CI_COMMIT_REF_NAME == "invalid"
+
+  test:
+    stage: test
+    script: exit 0
+    needs:
+      - job: build
+        optional: true
+
+init:
+  expect:
+    pipeline: pending
+    stages:
+      test: pending
+    jobs:
+      test: pending
+
+transitions:
+  - event: success
+    jobs: [test]
+    expect:
+      pipeline: success
+      stages:
+        test: success
+      jobs:
+        test: success
diff --git a/spec/services/ci/pipeline_processing/test_cases/dag_build_succeds_test_manual_allow_failure_true_deploy_needs_both.yml b/spec/services/ci/pipeline_processing/test_cases/dag_build_succeds_test_manual_allow_failure_true_deploy_needs_both.yml
index 60f803bc3d0..96377b00c85 100644
--- a/spec/services/ci/pipeline_processing/test_cases/dag_build_succeds_test_manual_allow_failure_true_deploy_needs_both.yml
+++ b/spec/services/ci/pipeline_processing/test_cases/dag_build_succeds_test_manual_allow_failure_true_deploy_needs_both.yml
@@ -30,12 +30,12 @@ transitions:
   - event: success
     jobs: [build]
     expect:
-      pipeline: running
+      pipeline: success
       stages:
         build: success
         test: skipped
-        deploy: created
+        deploy: skipped
       jobs:
         build: success
         test: manual
-        deploy: created
+        deploy: skipped
diff --git a/spec/services/ci/pipeline_processing/test_cases/dag_build_succeds_test_manual_allow_failure_true_deploy_needs_test.yml b/spec/services/ci/pipeline_processing/test_cases/dag_build_succeds_test_manual_allow_failure_true_deploy_needs_test.yml
index 4e4b2f22224..69640630ef4 100644
--- a/spec/services/ci/pipeline_processing/test_cases/dag_build_succeds_test_manual_allow_failure_true_deploy_needs_test.yml
+++ b/spec/services/ci/pipeline_processing/test_cases/dag_build_succeds_test_manual_allow_failure_true_deploy_needs_test.yml
@@ -30,12 +30,12 @@ transitions:
   - event: success
     jobs: [build]
     expect:
-      pipeline: running
+      pipeline: success
       stages:
         build: success
         test: skipped
-        deploy: created
+        deploy: skipped
       jobs:
         build: success
         test: manual
-        deploy: created
+        deploy: skipped
diff --git a/spec/services/ci/pipeline_processing/test_cases/dag_build_test_manual_review_deploy.yml b/spec/services/ci/pipeline_processing/test_cases/dag_build_test_manual_review_deploy.yml
index fef28dcfbbe..8de484d6793 100644
--- a/spec/services/ci/pipeline_processing/test_cases/dag_build_test_manual_review_deploy.yml
+++ b/spec/services/ci/pipeline_processing/test_cases/dag_build_test_manual_review_deploy.yml
@@ -54,29 +54,29 @@ transitions:
       stages:
         build: success
         test: pending
-        review: created
-        deploy: created
+        review: skipped
+        deploy: skipped
       jobs:
         build: success
         test: pending
         release_test: manual
-        review: created
-        staging: created
-        production: created
+        review: skipped
+        staging: skipped
+        production: skipped
 
   - event: success
     jobs: [test]
     expect:
-      pipeline: running
+      pipeline: success
       stages:
         build: success
         test: success
-        review: created
-        deploy: created
+        review: skipped
+        deploy: skipped
       jobs:
         build: success
         test: success
         release_test: manual
-        review: created
-        staging: created
-        production: created
+        review: skipped
+        staging: skipped
+        production: skipped
diff --git a/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true.yml b/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true.yml
index d8ca563b141..b8fcdd1566a 100644
--- a/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true.yml
+++ b/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true.yml
@@ -12,13 +12,13 @@ config:
 
 init:
   expect:
-    pipeline: created
+    pipeline: skipped
     stages:
       test: skipped
-      deploy: created
+      deploy: skipped
     jobs:
       test: manual
-      deploy: created
+      deploy: skipped
 
 transitions:
   - event: enqueue
@@ -27,10 +27,10 @@ transitions:
       pipeline: pending
       stages:
         test: pending
-        deploy: created
+        deploy: skipped
       jobs:
         test: pending
-        deploy: created
+        deploy: skipped
 
   - event: run
     jobs: [test]
@@ -38,21 +38,18 @@ transitions:
       pipeline: running
       stages:
         test: running
-        deploy: created
+        deploy: skipped
       jobs:
         test: running
-        deploy: created
+        deploy: skipped
 
   - event: drop
     jobs: [test]
     expect:
-      pipeline: running
+      pipeline: success
       stages:
         test: success
-        deploy: pending
+        deploy: skipped
       jobs:
         test: failed
-        deploy: pending
-
-# TOOD: should we run deploy?
-# Further discussions: https://gitlab.com/gitlab-org/gitlab/-/issues/213080
+        deploy: skipped
diff --git a/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_deploy_always.yml b/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_deploy_always.yml
index ba0a20f49a7..a4a98bf4629 100644
--- a/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_deploy_always.yml
+++ b/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_deploy_always.yml
@@ -13,15 +13,12 @@ config:
 
 init:
   expect:
-    pipeline: created
+    pipeline: pending
     stages:
       test: skipped
-      deploy: created
+      deploy: pending
     jobs:
       test: manual
-      deploy: created
+      deploy: pending
 
 transitions: []
-
-# TODO: should we run `deploy`?
-# Further discussions: https://gitlab.com/gitlab-org/gitlab/-/issues/213080
diff --git a/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_deploy_on_failure.yml b/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_deploy_on_failure.yml
index d375c6a49e0..81aad4940b6 100644
--- a/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_deploy_on_failure.yml
+++ b/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_deploy_on_failure.yml
@@ -13,13 +13,13 @@ config:
 
 init:
   expect:
-    pipeline: created
+    pipeline: skipped
     stages:
       test: skipped
-      deploy: created
+      deploy: skipped
     jobs:
       test: manual
-      deploy: created
+      deploy: skipped
 
 transitions:
   - event: enqueue
@@ -28,10 +28,10 @@ transitions:
       pipeline: pending
       stages:
         test: pending
-        deploy: created
+        deploy: skipped
       jobs:
         test: pending
-        deploy: created
+        deploy: skipped
 
   - event: drop
     jobs: [test]
@@ -43,6 +43,3 @@ transitions:
       jobs:
         test: failed
         deploy: skipped
-
-# TODO: should we run `deploy`?
-# Further discussions: https://gitlab.com/gitlab-org/gitlab/-/issues/213080
diff --git a/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_other_test_succeeds_deploy_needs_both.yml b/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_other_test_succeeds_deploy_needs_both.yml
index 34073b92ccc..a5bb103d1a5 100644
--- a/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_other_test_succeeds_deploy_needs_both.yml
+++ b/spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_other_test_succeeds_deploy_needs_both.yml
@@ -19,24 +19,21 @@ init:
     pipeline: pending
     stages:
       test: pending
-      deploy: created
+      deploy: skipped
     jobs:
       test1: pending
       test2: manual
-      deploy: created
+      deploy: skipped
 
 transitions:
   - event: success
     jobs: [test1]
     expect:
-      pipeline: running
+      pipeline: success
       stages:
         test: success
-        deploy: created
+        deploy: skipped
       jobs:
         test1: success
         test2: manual
-        deploy: created
-
-# TODO: should deploy run?
-# Further discussions: https://gitlab.com/gitlab-org/gitlab/-/issues/213080
+        deploy: skipped
diff --git a/spec/services/ci/process_pipeline_service_spec.rb b/spec/services/ci/process_pipeline_service_spec.rb
index d316c9a262b..e02536fd07f 100644
--- a/spec/services/ci/process_pipeline_service_spec.rb
+++ b/spec/services/ci/process_pipeline_service_spec.rb
@@ -43,42 +43,59 @@ RSpec.describe Ci::ProcessPipelineService do
     let!(:build) { create_build('build') }
     let!(:test) { create_build('test') }
 
-    it 'returns unique statuses' do
-      subject.execute
+    context 'when FF ci_remove_update_retried_from_process_pipeline is enabled' do
+      it 'does not update older builds as retried' do
+        subject.execute
 
-      expect(all_builds.latest).to contain_exactly(build, test)
-      expect(all_builds.retried).to contain_exactly(build_retried)
+        expect(all_builds.latest).to contain_exactly(build, build_retried, test)
+        expect(all_builds.retried).to be_empty
+      end
     end
 
-    context 'counter ci_legacy_update_jobs_as_retried_total' do
-      let(:counter) { double(increment: true) }
-
+    context 'when FF ci_remove_update_retried_from_process_pipeline is disabled' do
       before do
-        allow(Gitlab::Metrics).to receive(:counter).and_call_original
-        allow(Gitlab::Metrics).to receive(:counter)
-                                    .with(:ci_legacy_update_jobs_as_retried_total, anything)
-                                    .and_return(counter)
+        stub_feature_flags(ci_remove_update_retried_from_process_pipeline: false)
       end
 
-      it 'increments the counter' do
-        expect(counter).to receive(:increment)
-
+      it 'returns unique statuses' do
         subject.execute
+
+        expect(all_builds.latest).to contain_exactly(build, test)
+        expect(all_builds.retried).to contain_exactly(build_retried)
       end
 
-      context 'when the previous build has already retried column true' do
+      context 'counter ci_legacy_update_jobs_as_retried_total' do
+        let(:counter) { double(increment: true) }
+
         before do
-          build_retried.update_columns(retried: true)
+          allow(Gitlab::Metrics).to receive(:counter).and_call_original
+          allow(Gitlab::Metrics).to receive(:counter)
+                                      .with(:ci_legacy_update_jobs_as_retried_total, anything)
+                                      .and_return(counter)
         end
 
-        it 'does not increment the counter' do
-          expect(counter).not_to receive(:increment)
+        it 'increments the counter' do
+          expect(counter).to receive(:increment)
 
           subject.execute
         end
+
+        context 'when the previous build has already retried column true' do
+          before do
+            build_retried.update_columns(retried: true)
+          end
+
+          it 'does not increment the counter' do
+            expect(counter).not_to receive(:increment)
+
+            subject.execute
+          end
+        end
       end
     end
 
+    private
+
     def create_build(name, **opts)
       create(:ci_build, :created, pipeline: pipeline, name: name, **opts)
     end
diff --git a/spec/services/ci/register_job_service_spec.rb b/spec/services/ci/register_job_service_spec.rb
index 88770c8095b..9187dd4f300 100644
--- a/spec/services/ci/register_job_service_spec.rb
+++ b/spec/services/ci/register_job_service_spec.rb
@@ -13,573 +13,656 @@ module Ci
     let!(:pending_job) { create(:ci_build, pipeline: pipeline) }
 
     describe '#execute' do
-      context 'runner follow tag list' do
-        it "picks build with the same tag" do
-          pending_job.update!(tag_list: ["linux"])
-          specific_runner.update!(tag_list: ["linux"])
-          expect(execute(specific_runner)).to eq(pending_job)
-        end
-
-        it "does not pick build with different tag" do
-          pending_job.update!(tag_list: ["linux"])
-          specific_runner.update!(tag_list: ["win32"])
-          expect(execute(specific_runner)).to be_falsey
-        end
+      shared_examples 'handles runner assignment' do
+        context 'runner follow tag list' do
+          it "picks build with the same tag" do
+            pending_job.update!(tag_list: ["linux"])
+            specific_runner.update!(tag_list: ["linux"])
+            expect(execute(specific_runner)).to eq(pending_job)
+          end
 
-        it "picks build without tag" do
-          expect(execute(specific_runner)).to eq(pending_job)
-        end
+          it "does not pick build with different tag" do
+            pending_job.update!(tag_list: ["linux"])
+            specific_runner.update!(tag_list: ["win32"])
+            expect(execute(specific_runner)).to be_falsey
+          end
 
-        it "does not pick build with tag" do
-          pending_job.update!(tag_list: ["linux"])
-          expect(execute(specific_runner)).to be_falsey
-        end
+          it "picks build without tag" do
+            expect(execute(specific_runner)).to eq(pending_job)
+          end
 
-        it "pick build without tag" do
-          specific_runner.update!(tag_list: ["win32"])
-          expect(execute(specific_runner)).to eq(pending_job)
-        end
-      end
+          it "does not pick build with tag" do
+            pending_job.update!(tag_list: ["linux"])
+            expect(execute(specific_runner)).to be_falsey
+          end
 
-      context 'deleted projects' do
-        before do
-          project.update!(pending_delete: true)
+          it "pick build without tag" do
+            specific_runner.update!(tag_list: ["win32"])
+            expect(execute(specific_runner)).to eq(pending_job)
+          end
         end
 
-        context 'for shared runners' do
+        context 'deleted projects' do
           before do
-            project.update!(shared_runners_enabled: true)
+            project.update!(pending_delete: true)
           end
 
-          it 'does not pick a build' do
-            expect(execute(shared_runner)).to be_nil
+          context 'for shared runners' do
+            before do
+              project.update!(shared_runners_enabled: true)
+            end
+
+            it 'does not pick a build' do
+              expect(execute(shared_runner)).to be_nil
+            end
           end
-        end
 
-        context 'for specific runner' do
-          it 'does not pick a build' do
-            expect(execute(specific_runner)).to be_nil
+          context 'for specific runner' do
+            it 'does not pick a build' do
+              expect(execute(specific_runner)).to be_nil
+            end
           end
         end
-      end
 
-      context 'allow shared runners' do
-        before do
-          project.update!(shared_runners_enabled: true)
-        end
+        context 'allow shared runners' do
+          before do
+            project.update!(shared_runners_enabled: true)
+          end
+
+          context 'for multiple builds' do
+            let!(:project2) { create :project, shared_runners_enabled: true }
+            let!(:pipeline2) { create :ci_pipeline, project: project2 }
+            let!(:project3) { create :project, shared_runners_enabled: true }
+            let!(:pipeline3) { create :ci_pipeline, project: project3 }
+            let!(:build1_project1) { pending_job }
+            let!(:build2_project1) { FactoryBot.create :ci_build, pipeline: pipeline }
+            let!(:build3_project1) { FactoryBot.create :ci_build, pipeline: pipeline }
+            let!(:build1_project2) { FactoryBot.create :ci_build, pipeline: pipeline2 }
+            let!(:build2_project2) { FactoryBot.create :ci_build, pipeline: pipeline2 }
+            let!(:build1_project3) { FactoryBot.create :ci_build, pipeline: pipeline3 }
+
+            it 'prefers projects without builds first' do
+              # it gets for one build from each of the projects
+              expect(execute(shared_runner)).to eq(build1_project1)
+              expect(execute(shared_runner)).to eq(build1_project2)
+              expect(execute(shared_runner)).to eq(build1_project3)
+
+              # then it gets a second build from each of the projects
+              expect(execute(shared_runner)).to eq(build2_project1)
+              expect(execute(shared_runner)).to eq(build2_project2)
+
+              # in the end the third build
+              expect(execute(shared_runner)).to eq(build3_project1)
+            end
 
-        context 'for multiple builds' do
-          let!(:project2) { create :project, shared_runners_enabled: true }
-          let!(:pipeline2) { create :ci_pipeline, project: project2 }
-          let!(:project3) { create :project, shared_runners_enabled: true }
-          let!(:pipeline3) { create :ci_pipeline, project: project3 }
-          let!(:build1_project1) { pending_job }
-          let!(:build2_project1) { FactoryBot.create :ci_build, pipeline: pipeline }
-          let!(:build3_project1) { FactoryBot.create :ci_build, pipeline: pipeline }
-          let!(:build1_project2) { FactoryBot.create :ci_build, pipeline: pipeline2 }
-          let!(:build2_project2) { FactoryBot.create :ci_build, pipeline: pipeline2 }
-          let!(:build1_project3) { FactoryBot.create :ci_build, pipeline: pipeline3 }
-
-          it 'prefers projects without builds first' do
-            # it gets for one build from each of the projects
-            expect(execute(shared_runner)).to eq(build1_project1)
-            expect(execute(shared_runner)).to eq(build1_project2)
-            expect(execute(shared_runner)).to eq(build1_project3)
-
-            # then it gets a second build from each of the projects
-            expect(execute(shared_runner)).to eq(build2_project1)
-            expect(execute(shared_runner)).to eq(build2_project2)
-
-            # in the end the third build
-            expect(execute(shared_runner)).to eq(build3_project1)
-          end
-
-          it 'equalises number of running builds' do
-            # after finishing the first build for project 1, get a second build from the same project
-            expect(execute(shared_runner)).to eq(build1_project1)
-            build1_project1.reload.success
-            expect(execute(shared_runner)).to eq(build2_project1)
-
-            expect(execute(shared_runner)).to eq(build1_project2)
-            build1_project2.reload.success
-            expect(execute(shared_runner)).to eq(build2_project2)
-            expect(execute(shared_runner)).to eq(build1_project3)
-            expect(execute(shared_runner)).to eq(build3_project1)
+            it 'equalises number of running builds' do
+              # after finishing the first build for project 1, get a second build from the same project
+              expect(execute(shared_runner)).to eq(build1_project1)
+              build1_project1.reload.success
+              expect(execute(shared_runner)).to eq(build2_project1)
+
+              expect(execute(shared_runner)).to eq(build1_project2)
+              build1_project2.reload.success
+              expect(execute(shared_runner)).to eq(build2_project2)
+              expect(execute(shared_runner)).to eq(build1_project3)
+              expect(execute(shared_runner)).to eq(build3_project1)
+            end
           end
-        end
 
-        context 'shared runner' do
-          let(:response) { described_class.new(shared_runner).execute }
-          let(:build) { response.build }
+          context 'shared runner' do
+            let(:response) { described_class.new(shared_runner).execute }
+            let(:build) { response.build }
 
-          it { expect(build).to be_kind_of(Build) }
-          it { expect(build).to be_valid }
-          it { expect(build).to be_running }
-          it { expect(build.runner).to eq(shared_runner) }
-          it { expect(Gitlab::Json.parse(response.build_json)['id']).to eq(build.id) }
-        end
+            it { expect(build).to be_kind_of(Build) }
+            it { expect(build).to be_valid }
+            it { expect(build).to be_running }
+            it { expect(build.runner).to eq(shared_runner) }
+            it { expect(Gitlab::Json.parse(response.build_json)['id']).to eq(build.id) }
+          end
 
-        context 'specific runner' do
-          let(:build) { execute(specific_runner) }
+          context 'specific runner' do
+            let(:build) { execute(specific_runner) }
 
-          it { expect(build).to be_kind_of(Build) }
-          it { expect(build).to be_valid }
-          it { expect(build).to be_running }
-          it { expect(build.runner).to eq(specific_runner) }
+            it { expect(build).to be_kind_of(Build) }
+            it { expect(build).to be_valid }
+            it { expect(build).to be_running }
+            it { expect(build.runner).to eq(specific_runner) }
+          end
         end
-      end
 
-      context 'disallow shared runners' do
-        before do
-          project.update!(shared_runners_enabled: false)
-        end
+        context 'disallow shared runners' do
+          before do
+            project.update!(shared_runners_enabled: false)
+          end
 
-        context 'shared runner' do
-          let(:build) { execute(shared_runner) }
+          context 'shared runner' do
+            let(:build) { execute(shared_runner) }
 
-          it { expect(build).to be_nil }
-        end
+            it { expect(build).to be_nil }
+          end
 
-        context 'specific runner' do
-          let(:build) { execute(specific_runner) }
+          context 'specific runner' do
+            let(:build) { execute(specific_runner) }
 
-          it { expect(build).to be_kind_of(Build) }
-          it { expect(build).to be_valid }
-          it { expect(build).to be_running }
-          it { expect(build.runner).to eq(specific_runner) }
+            it { expect(build).to be_kind_of(Build) }
+            it { expect(build).to be_valid }
+            it { expect(build).to be_running }
+            it { expect(build.runner).to eq(specific_runner) }
+          end
         end
-      end
 
-      context 'disallow when builds are disabled' do
-        before do
-          project.update!(shared_runners_enabled: true, group_runners_enabled: true)
-          project.project_feature.update_attribute(:builds_access_level, ProjectFeature::DISABLED)
-        end
+        context 'disallow when builds are disabled' do
+          before do
+            project.update!(shared_runners_enabled: true, group_runners_enabled: true)
+            project.project_feature.update_attribute(:builds_access_level, ProjectFeature::DISABLED)
+          end
 
-        context 'and uses shared runner' do
-          let(:build) { execute(shared_runner) }
+          context 'and uses shared runner' do
+            let(:build) { execute(shared_runner) }
 
-          it { expect(build).to be_nil }
-        end
+            it { expect(build).to be_nil }
+          end
 
-        context 'and uses group runner' do
-          let(:build) { execute(group_runner) }
+          context 'and uses group runner' do
+            let(:build) { execute(group_runner) }
 
-          it { expect(build).to be_nil }
-        end
+            it { expect(build).to be_nil }
+          end
 
-        context 'and uses project runner' do
-          let(:build) { execute(specific_runner) }
+          context 'and uses project runner' do
+            let(:build) { execute(specific_runner) }
 
-          it { expect(build).to be_nil }
+            it { expect(build).to be_nil }
+          end
         end
-      end
 
-      context 'allow group runners' do
-        before do
-          project.update!(group_runners_enabled: true)
-        end
+        context 'allow group runners' do
+          before do
+            project.update!(group_runners_enabled: true)
+          end
 
-        context 'for multiple builds' do
-          let!(:project2) { create(:project, group_runners_enabled: true, group: group) }
-          let!(:pipeline2) { create(:ci_pipeline, project: project2) }
-          let!(:project3) { create(:project, group_runners_enabled: true, group: group) }
-          let!(:pipeline3) { create(:ci_pipeline, project: project3) }
+          context 'for multiple builds' do
+            let!(:project2) { create(:project, group_runners_enabled: true, group: group) }
+            let!(:pipeline2) { create(:ci_pipeline, project: project2) }
+            let!(:project3) { create(:project, group_runners_enabled: true, group: group) }
+            let!(:pipeline3) { create(:ci_pipeline, project: project3) }
 
-          let!(:build1_project1) { pending_job }
-          let!(:build2_project1) { create(:ci_build, pipeline: pipeline) }
-          let!(:build3_project1) { create(:ci_build, pipeline: pipeline) }
-          let!(:build1_project2) { create(:ci_build, pipeline: pipeline2) }
-          let!(:build2_project2) { create(:ci_build, pipeline: pipeline2) }
-          let!(:build1_project3) { create(:ci_build, pipeline: pipeline3) }
+            let!(:build1_project1) { pending_job }
+            let!(:build2_project1) { create(:ci_build, pipeline: pipeline) }
+            let!(:build3_project1) { create(:ci_build, pipeline: pipeline) }
+            let!(:build1_project2) { create(:ci_build, pipeline: pipeline2) }
+            let!(:build2_project2) { create(:ci_build, pipeline: pipeline2) }
+            let!(:build1_project3) { create(:ci_build, pipeline: pipeline3) }
 
-          # these shouldn't influence the scheduling
-          let!(:unrelated_group) { create(:group) }
-          let!(:unrelated_project) { create(:project, group_runners_enabled: true, group: unrelated_group) }
-          let!(:unrelated_pipeline) { create(:ci_pipeline, project: unrelated_project) }
-          let!(:build1_unrelated_project) { create(:ci_build, pipeline: unrelated_pipeline) }
-          let!(:unrelated_group_runner) { create(:ci_runner, :group, groups: [unrelated_group]) }
+            # these shouldn't influence the scheduling
+            let!(:unrelated_group) { create(:group) }
+            let!(:unrelated_project) { create(:project, group_runners_enabled: true, group: unrelated_group) }
+            let!(:unrelated_pipeline) { create(:ci_pipeline, project: unrelated_project) }
+            let!(:build1_unrelated_project) { create(:ci_build, pipeline: unrelated_pipeline) }
+            let!(:unrelated_group_runner) { create(:ci_runner, :group, groups: [unrelated_group]) }
 
-          it 'does not consider builds from other group runners' do
-            expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 6
-            execute(group_runner)
+            it 'does not consider builds from other group runners' do
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 6
+              execute(group_runner)
 
-            expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 5
-            execute(group_runner)
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 5
+              execute(group_runner)
 
-            expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 4
-            execute(group_runner)
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 4
+              execute(group_runner)
 
-            expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 3
-            execute(group_runner)
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 3
+              execute(group_runner)
 
-            expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 2
-            execute(group_runner)
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 2
+              execute(group_runner)
 
-            expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 1
-            execute(group_runner)
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 1
+              execute(group_runner)
 
-            expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 0
-            expect(execute(group_runner)).to be_nil
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 0
+              expect(execute(group_runner)).to be_nil
+            end
           end
-        end
 
-        context 'group runner' do
-          let(:build) { execute(group_runner) }
+          context 'group runner' do
+            let(:build) { execute(group_runner) }
 
-          it { expect(build).to be_kind_of(Build) }
-          it { expect(build).to be_valid }
-          it { expect(build).to be_running }
-          it { expect(build.runner).to eq(group_runner) }
+            it { expect(build).to be_kind_of(Build) }
+            it { expect(build).to be_valid }
+            it { expect(build).to be_running }
+            it { expect(build.runner).to eq(group_runner) }
+          end
         end
-      end
 
-      context 'disallow group runners' do
-        before do
-          project.update!(group_runners_enabled: false)
-        end
+        context 'disallow group runners' do
+          before do
+            project.update!(group_runners_enabled: false)
+          end
 
-        context 'group runner' do
-          let(:build) { execute(group_runner) }
+          context 'group runner' do
+            let(:build) { execute(group_runner) }
 
-          it { expect(build).to be_nil }
+            it { expect(build).to be_nil }
+          end
         end
-      end
 
-      context 'when first build is stalled' do
-        before do
-          allow_any_instance_of(Ci::RegisterJobService).to receive(:assign_runner!).and_call_original
-          allow_any_instance_of(Ci::RegisterJobService).to receive(:assign_runner!)
-            .with(pending_job, anything).and_raise(ActiveRecord::StaleObjectError)
-        end
+        context 'when first build is stalled' do
+          before do
+            allow_any_instance_of(Ci::RegisterJobService).to receive(:assign_runner!).and_call_original
+            allow_any_instance_of(Ci::RegisterJobService).to receive(:assign_runner!)
+              .with(pending_job, anything).and_raise(ActiveRecord::StaleObjectError)
+          end
 
-        subject { described_class.new(specific_runner).execute }
+          subject { described_class.new(specific_runner).execute }
 
-        context 'with multiple builds are in queue' do
-          let!(:other_build) { create :ci_build, pipeline: pipeline }
+          context 'with multiple builds are in queue' do
+            let!(:other_build) { create :ci_build, pipeline: pipeline }
 
-          before do
-            allow_any_instance_of(Ci::RegisterJobService).to receive(:builds_for_project_runner)
-              .and_return(Ci::Build.where(id: [pending_job, other_build]))
-          end
+            before do
+              allow_any_instance_of(Ci::RegisterJobService).to receive(:builds_for_project_runner)
+                .and_return(Ci::Build.where(id: [pending_job, other_build]))
+            end
 
-          it "receives second build from the queue" do
-            expect(subject).to be_valid
-            expect(subject.build).to eq(other_build)
+            it "receives second build from the queue" do
+              expect(subject).to be_valid
+              expect(subject.build).to eq(other_build)
+            end
           end
-        end
 
-        context 'when single build is in queue' do
-          before do
-            allow_any_instance_of(Ci::RegisterJobService).to receive(:builds_for_project_runner)
-              .and_return(Ci::Build.where(id: pending_job))
-          end
+          context 'when single build is in queue' do
+            before do
+              allow_any_instance_of(Ci::RegisterJobService).to receive(:builds_for_project_runner)
+                .and_return(Ci::Build.where(id: pending_job))
+            end
 
-          it "does not receive any valid result" do
-            expect(subject).not_to be_valid
+            it "does not receive any valid result" do
+              expect(subject).not_to be_valid
+            end
           end
-        end
 
-        context 'when there is no build in queue' do
-          before do
-            allow_any_instance_of(Ci::RegisterJobService).to receive(:builds_for_project_runner)
-              .and_return(Ci::Build.none)
-          end
+          context 'when there is no build in queue' do
+            before do
+              allow_any_instance_of(Ci::RegisterJobService).to receive(:builds_for_project_runner)
+                .and_return(Ci::Build.none)
+            end
 
-          it "does not receive builds but result is valid" do
-            expect(subject).to be_valid
-            expect(subject.build).to be_nil
+            it "does not receive builds but result is valid" do
+              expect(subject).to be_valid
+              expect(subject.build).to be_nil
+            end
           end
         end
-      end
 
-      context 'when access_level of runner is not_protected' do
-        let!(:specific_runner) { create(:ci_runner, :project, projects: [project]) }
+        context 'when access_level of runner is not_protected' do
+          let!(:specific_runner) { create(:ci_runner, :project, projects: [project]) }
 
-        context 'when a job is protected' do
-          let!(:pending_job) { create(:ci_build, :protected, pipeline: pipeline) }
+          context 'when a job is protected' do
+            let!(:pending_job) { create(:ci_build, :protected, pipeline: pipeline) }
 
-          it 'picks the job' do
-            expect(execute(specific_runner)).to eq(pending_job)
+            it 'picks the job' do
+              expect(execute(specific_runner)).to eq(pending_job)
+            end
           end
-        end
 
-        context 'when a job is unprotected' do
-          let!(:pending_job) { create(:ci_build, pipeline: pipeline) }
+          context 'when a job is unprotected' do
+            let!(:pending_job) { create(:ci_build, pipeline: pipeline) }
 
-          it 'picks the job' do
-            expect(execute(specific_runner)).to eq(pending_job)
+            it 'picks the job' do
+              expect(execute(specific_runner)).to eq(pending_job)
+            end
           end
-        end
 
-        context 'when protected attribute of a job is nil' do
-          let!(:pending_job) { create(:ci_build, pipeline: pipeline) }
+          context 'when protected attribute of a job is nil' do
+            let!(:pending_job) { create(:ci_build, pipeline: pipeline) }
 
-          before do
-            pending_job.update_attribute(:protected, nil)
-          end
+            before do
+              pending_job.update_attribute(:protected, nil)
+            end
 
-          it 'picks the job' do
-            expect(execute(specific_runner)).to eq(pending_job)
+            it 'picks the job' do
+              expect(execute(specific_runner)).to eq(pending_job)
+            end
           end
         end
-      end
 
-      context 'when access_level of runner is ref_protected' do
-        let!(:specific_runner) { create(:ci_runner, :project, :ref_protected, projects: [project]) }
+        context 'when access_level of runner is ref_protected' do
+          let!(:specific_runner) { create(:ci_runner, :project, :ref_protected, projects: [project]) }
 
-        context 'when a job is protected' do
-          let!(:pending_job) { create(:ci_build, :protected, pipeline: pipeline) }
+          context 'when a job is protected' do
+            let!(:pending_job) { create(:ci_build, :protected, pipeline: pipeline) }
 
-          it 'picks the job' do
-            expect(execute(specific_runner)).to eq(pending_job)
+            it 'picks the job' do
+              expect(execute(specific_runner)).to eq(pending_job)
+            end
           end
-        end
 
-        context 'when a job is unprotected' do
-          let!(:pending_job) { create(:ci_build, pipeline: pipeline) }
+          context 'when a job is unprotected' do
+            let!(:pending_job) { create(:ci_build, pipeline: pipeline) }
 
-          it 'does not pick the job' do
-            expect(execute(specific_runner)).to be_nil
+            it 'does not pick the job' do
+              expect(execute(specific_runner)).to be_nil
+            end
           end
-        end
 
-        context 'when protected attribute of a job is nil' do
-          let!(:pending_job) { create(:ci_build, pipeline: pipeline) }
+          context 'when protected attribute of a job is nil' do
+            let!(:pending_job) { create(:ci_build, pipeline: pipeline) }
 
-          before do
-            pending_job.update_attribute(:protected, nil)
-          end
+            before do
+              pending_job.update_attribute(:protected, nil)
+            end
 
-          it 'does not pick the job' do
-            expect(execute(specific_runner)).to be_nil
+            it 'does not pick the job' do
+              expect(execute(specific_runner)).to be_nil
+            end
           end
         end
-      end
 
-      context 'runner feature set is verified' do
-        let(:options) { { artifacts: { reports: { junit: "junit.xml" } } } }
-        let!(:pending_job) { create(:ci_build, :pending, pipeline: pipeline, options: options) }
+        context 'runner feature set is verified' do
+          let(:options) { { artifacts: { reports: { junit: "junit.xml" } } } }
+          let!(:pending_job) { create(:ci_build, :pending, pipeline: pipeline, options: options) }
 
-        subject { execute(specific_runner, params) }
+          subject { execute(specific_runner, params) }
 
-        context 'when feature is missing by runner' do
-          let(:params) { {} }
+          context 'when feature is missing by runner' do
+            let(:params) { {} }
 
-          it 'does not pick the build and drops the build' do
-            expect(subject).to be_nil
-            expect(pending_job.reload).to be_failed
-            expect(pending_job).to be_runner_unsupported
+            it 'does not pick the build and drops the build' do
+              expect(subject).to be_nil
+              expect(pending_job.reload).to be_failed
+              expect(pending_job).to be_runner_unsupported
+            end
           end
-        end
 
-        context 'when feature is supported by runner' do
-          let(:params) do
-            { info: { features: { upload_multiple_artifacts: true } } }
-          end
+          context 'when feature is supported by runner' do
+            let(:params) do
+              { info: { features: { upload_multiple_artifacts: true } } }
+            end
 
-          it 'does pick job' do
-            expect(subject).not_to be_nil
+            it 'does pick job' do
+              expect(subject).not_to be_nil
+            end
           end
         end
-      end
 
-      context 'when "dependencies" keyword is specified' do
-        shared_examples 'not pick' do
-          it 'does not pick the build and drops the build' do
-            expect(subject).to be_nil
-            expect(pending_job.reload).to be_failed
-            expect(pending_job).to be_missing_dependency_failure
+        context 'when "dependencies" keyword is specified' do
+          shared_examples 'not pick' do
+            it 'does not pick the build and drops the build' do
+              expect(subject).to be_nil
+              expect(pending_job.reload).to be_failed
+              expect(pending_job).to be_missing_dependency_failure
+            end
           end
-        end
 
-        shared_examples 'validation is active' do
-          context 'when depended job has not been completed yet' do
-            let!(:pre_stage_job) { create(:ci_build, :manual, pipeline: pipeline, name: 'test', stage_idx: 0) }
+          shared_examples 'validation is active' do
+            context 'when depended job has not been completed yet' do
+              let!(:pre_stage_job) { create(:ci_build, :manual, pipeline: pipeline, name: 'test', stage_idx: 0) }
 
-            it { expect(subject).to eq(pending_job) }
-          end
+              it { expect(subject).to eq(pending_job) }
+            end
 
-          context 'when artifacts of depended job has been expired' do
-            let!(:pre_stage_job) { create(:ci_build, :success, :expired, pipeline: pipeline, name: 'test', stage_idx: 0) }
+            context 'when artifacts of depended job has been expired' do
+              let!(:pre_stage_job) { create(:ci_build, :success, :expired, pipeline: pipeline, name: 'test', stage_idx: 0) }
 
-            it_behaves_like 'not pick'
-          end
+              it_behaves_like 'not pick'
+            end
 
-          context 'when artifacts of depended job has been erased' do
-            let!(:pre_stage_job) { create(:ci_build, :success, pipeline: pipeline, name: 'test', stage_idx: 0, erased_at: 1.minute.ago) }
+            context 'when artifacts of depended job has been erased' do
+              let!(:pre_stage_job) { create(:ci_build, :success, pipeline: pipeline, name: 'test', stage_idx: 0, erased_at: 1.minute.ago) }
 
-            before do
-              pre_stage_job.erase
+              before do
+                pre_stage_job.erase
+              end
+
+              it_behaves_like 'not pick'
             end
 
-            it_behaves_like 'not pick'
+            context 'when job object is staled' do
+              let!(:pre_stage_job) { create(:ci_build, :success, :expired, pipeline: pipeline, name: 'test', stage_idx: 0) }
+
+              before do
+                allow_any_instance_of(Ci::Build).to receive(:drop!)
+                  .and_raise(ActiveRecord::StaleObjectError.new(pending_job, :drop!))
+              end
+
+              it 'does not drop nor pick' do
+                expect(subject).to be_nil
+              end
+            end
           end
 
-          context 'when job object is staled' do
-            let!(:pre_stage_job) { create(:ci_build, :success, :expired, pipeline: pipeline, name: 'test', stage_idx: 0) }
+          shared_examples 'validation is not active' do
+            context 'when depended job has not been completed yet' do
+              let!(:pre_stage_job) { create(:ci_build, :manual, pipeline: pipeline, name: 'test', stage_idx: 0) }
 
-            before do
-              allow_any_instance_of(Ci::Build).to receive(:drop!)
-                .and_raise(ActiveRecord::StaleObjectError.new(pending_job, :drop!))
+              it { expect(subject).to eq(pending_job) }
             end
 
-            it 'does not drop nor pick' do
-              expect(subject).to be_nil
+            context 'when artifacts of depended job has been expired' do
+              let!(:pre_stage_job) { create(:ci_build, :success, :expired, pipeline: pipeline, name: 'test', stage_idx: 0) }
+
+              it { expect(subject).to eq(pending_job) }
             end
-          end
-        end
 
-        shared_examples 'validation is not active' do
-          context 'when depended job has not been completed yet' do
-            let!(:pre_stage_job) { create(:ci_build, :manual, pipeline: pipeline, name: 'test', stage_idx: 0) }
+            context 'when artifacts of depended job has been erased' do
+              let!(:pre_stage_job) { create(:ci_build, :success, pipeline: pipeline, name: 'test', stage_idx: 0, erased_at: 1.minute.ago) }
 
-            it { expect(subject).to eq(pending_job) }
+              before do
+                pre_stage_job.erase
+              end
+
+              it { expect(subject).to eq(pending_job) }
+            end
           end
 
-          context 'when artifacts of depended job has been expired' do
-            let!(:pre_stage_job) { create(:ci_build, :success, :expired, pipeline: pipeline, name: 'test', stage_idx: 0) }
+          before do
+            stub_feature_flags(ci_validate_build_dependencies_override: false)
+          end
+
+          let!(:pre_stage_job) { create(:ci_build, :success, pipeline: pipeline, name: 'test', stage_idx: 0) }
 
-            it { expect(subject).to eq(pending_job) }
+          let!(:pending_job) do
+            create(:ci_build, :pending,
+              pipeline: pipeline, stage_idx: 1,
+              options: { script: ["bash"], dependencies: ['test'] })
           end
 
-          context 'when artifacts of depended job has been erased' do
-            let!(:pre_stage_job) { create(:ci_build, :success, pipeline: pipeline, name: 'test', stage_idx: 0, erased_at: 1.minute.ago) }
+          subject { execute(specific_runner) }
 
+          context 'when validates for dependencies is enabled' do
             before do
-              pre_stage_job.erase
+              stub_feature_flags(ci_validate_build_dependencies_override: false)
             end
 
-            it { expect(subject).to eq(pending_job) }
+            it_behaves_like 'validation is active'
+
+            context 'when the main feature flag is enabled for a specific project' do
+              before do
+                stub_feature_flags(ci_validate_build_dependencies: pipeline.project)
+              end
+
+              it_behaves_like 'validation is active'
+            end
+
+            context 'when the main feature flag is enabled for a different project' do
+              before do
+                stub_feature_flags(ci_validate_build_dependencies: create(:project))
+              end
+
+              it_behaves_like 'validation is not active'
+            end
           end
-        end
 
-        before do
-          stub_feature_flags(ci_validate_build_dependencies_override: false)
+          context 'when validates for dependencies is disabled' do
+            before do
+              stub_feature_flags(ci_validate_build_dependencies_override: true)
+            end
+
+            it_behaves_like 'validation is not active'
+          end
         end
 
-        let!(:pre_stage_job) { create(:ci_build, :success, pipeline: pipeline, name: 'test', stage_idx: 0) }
+        context 'when build is degenerated' do
+          let!(:pending_job) { create(:ci_build, :pending, :degenerated, pipeline: pipeline) }
+
+          subject { execute(specific_runner, {}) }
+
+          it 'does not pick the build and drops the build' do
+            expect(subject).to be_nil
 
-        let!(:pending_job) do
-          create(:ci_build, :pending,
-            pipeline: pipeline, stage_idx: 1,
-            options: { script: ["bash"], dependencies: ['test'] })
+            pending_job.reload
+            expect(pending_job).to be_failed
+            expect(pending_job).to be_archived_failure
+          end
         end
 
-        subject { execute(specific_runner) }
+        context 'when build has data integrity problem' do
+          let!(:pending_job) do
+            create(:ci_build, :pending, pipeline: pipeline)
+          end
 
-        context 'when validates for dependencies is enabled' do
           before do
-            stub_feature_flags(ci_validate_build_dependencies_override: false)
+            pending_job.update_columns(options: "string")
           end
 
-          it_behaves_like 'validation is active'
+          subject { execute(specific_runner, {}) }
 
-          context 'when the main feature flag is enabled for a specific project' do
-            before do
-              stub_feature_flags(ci_validate_build_dependencies: pipeline.project)
-            end
+          it 'does drop the build and logs both failures' do
+            expect(Gitlab::ErrorTracking).to receive(:track_exception)
+              .with(anything, a_hash_including(build_id: pending_job.id))
+              .twice
+              .and_call_original
 
-            it_behaves_like 'validation is active'
-          end
-
-          context 'when the main feature flag is enabled for a different project' do
-            before do
-              stub_feature_flags(ci_validate_build_dependencies: create(:project))
-            end
+            expect(subject).to be_nil
 
-            it_behaves_like 'validation is not active'
+            pending_job.reload
+            expect(pending_job).to be_failed
+            expect(pending_job).to be_data_integrity_failure
           end
         end
 
-        context 'when validates for dependencies is disabled' do
+        context 'when build fails to be run!' do
+          let!(:pending_job) do
+            create(:ci_build, :pending, pipeline: pipeline)
+          end
+
           before do
-            stub_feature_flags(ci_validate_build_dependencies_override: true)
+            expect_any_instance_of(Ci::Build).to receive(:run!)
+              .and_raise(RuntimeError, 'scheduler error')
           end
 
-          it_behaves_like 'validation is not active'
+          subject { execute(specific_runner, {}) }
+
+          it 'does drop the build and logs failure' do
+            expect(Gitlab::ErrorTracking).to receive(:track_exception)
+              .with(anything, a_hash_including(build_id: pending_job.id))
+              .once
+              .and_call_original
+
+            expect(subject).to be_nil
+
+            pending_job.reload
+            expect(pending_job).to be_failed
+            expect(pending_job).to be_scheduler_failure
+          end
         end
-      end
 
-      context 'when build is degenerated' do
-        let!(:pending_job) { create(:ci_build, :pending, :degenerated, pipeline: pipeline) }
+        context 'when an exception is raised during a persistent ref creation' do
+          before do
+            allow_any_instance_of(Ci::PersistentRef).to receive(:exist?) { false }
+            allow_any_instance_of(Ci::PersistentRef).to receive(:create_ref) { raise ArgumentError }
+          end
 
-        subject { execute(specific_runner, {}) }
+          subject { execute(specific_runner, {}) }
 
-        it 'does not pick the build and drops the build' do
-          expect(subject).to be_nil
+          it 'picks the build' do
+            expect(subject).to eq(pending_job)
 
-          pending_job.reload
-          expect(pending_job).to be_failed
-          expect(pending_job).to be_archived_failure
+            pending_job.reload
+            expect(pending_job).to be_running
+          end
         end
-      end
 
-      context 'when build has data integrity problem' do
-        let!(:pending_job) do
-          create(:ci_build, :pending, pipeline: pipeline)
-        end
+        context 'when only some builds can be matched by runner' do
+          let!(:specific_runner) { create(:ci_runner, :project, projects: [project], tag_list: %w[matching]) }
+          let!(:pending_job) { create(:ci_build, pipeline: pipeline, tag_list: %w[matching]) }
 
-        before do
-          pending_job.update_columns(options: "string")
+          before do
+            # create additional matching and non-matching jobs
+            create_list(:ci_build, 2, pipeline: pipeline, tag_list: %w[matching])
+            create(:ci_build, pipeline: pipeline, tag_list: %w[non-matching])
+          end
+
+          it "observes queue size of only matching jobs" do
+            # pending_job + 2 x matching ones
+            expect(Gitlab::Ci::Queue::Metrics.queue_size_total).to receive(:observe).with({}, 3)
+
+            expect(execute(specific_runner)).to eq(pending_job)
+          end
         end
 
-        subject { execute(specific_runner, {}) }
+        context 'when ci_register_job_temporary_lock is enabled' do
+          before do
+            stub_feature_flags(ci_register_job_temporary_lock: true)
 
-        it 'does drop the build and logs both failures' do
-          expect(Gitlab::ErrorTracking).to receive(:track_exception)
-            .with(anything, a_hash_including(build_id: pending_job.id))
-            .twice
-            .and_call_original
+            allow(Gitlab::Ci::Queue::Metrics.queue_operations_total).to receive(:increment)
+          end
 
-          expect(subject).to be_nil
+          context 'when a build is temporarily locked' do
+            let(:service) { described_class.new(specific_runner) }
 
-          pending_job.reload
-          expect(pending_job).to be_failed
-          expect(pending_job).to be_data_integrity_failure
-        end
-      end
+            before do
+              service.send(:acquire_temporary_lock, pending_job.id)
+            end
 
-      context 'when build fails to be run!' do
-        let!(:pending_job) do
-          create(:ci_build, :pending, pipeline: pipeline)
-        end
+            it 'skips this build and marks queue as invalid' do
+              expect(Gitlab::Ci::Queue::Metrics.queue_operations_total).to receive(:increment)
+                .with(operation: :queue_iteration)
+              expect(Gitlab::Ci::Queue::Metrics.queue_operations_total).to receive(:increment)
+                .with(operation: :build_temporary_locked)
 
-        before do
-          expect_any_instance_of(Ci::Build).to receive(:run!)
-            .and_raise(RuntimeError, 'scheduler error')
-        end
+              expect(service.execute).not_to be_valid
+            end
 
-        subject { execute(specific_runner, {}) }
+            context 'when there is another build in queue' do
+              let!(:next_pending_job) { create(:ci_build, pipeline: pipeline) }
 
-        it 'does drop the build and logs failure' do
-          expect(Gitlab::ErrorTracking).to receive(:track_exception)
-            .with(anything, a_hash_including(build_id: pending_job.id))
-            .once
-            .and_call_original
+              it 'skips this build and picks another build' do
+                expect(Gitlab::Ci::Queue::Metrics.queue_operations_total).to receive(:increment)
+                  .with(operation: :queue_iteration).twice
+                expect(Gitlab::Ci::Queue::Metrics.queue_operations_total).to receive(:increment)
+                  .with(operation: :build_temporary_locked)
 
-          expect(subject).to be_nil
+                result = service.execute
 
-          pending_job.reload
-          expect(pending_job).to be_failed
-          expect(pending_job).to be_scheduler_failure
+                expect(result.build).to eq(next_pending_job)
+                expect(result).to be_valid
+              end
+            end
+          end
         end
       end
 
-      context 'when an exception is raised during a persistent ref creation' do
+      context 'when ci_register_job_service_one_by_one is enabled' do
         before do
-          allow_any_instance_of(Ci::PersistentRef).to receive(:exist?) { false }
-          allow_any_instance_of(Ci::PersistentRef).to receive(:create_ref) { raise ArgumentError }
+          stub_feature_flags(ci_register_job_service_one_by_one: true)
         end
 
-        subject { execute(specific_runner, {}) }
+        it 'picks builds one-by-one' do
+          expect(Ci::Build).to receive(:find).with(pending_job.id).and_call_original
 
-        it 'picks the build' do
-          expect(subject).to eq(pending_job)
+          expect(execute(specific_runner)).to eq(pending_job)
+        end
+
+        include_examples 'handles runner assignment'
+      end
 
-          pending_job.reload
-          expect(pending_job).to be_running
+      context 'when ci_register_job_service_one_by_one is disabled' do
+        before do
+          stub_feature_flags(ci_register_job_service_one_by_one: false)
         end
+
+        include_examples 'handles runner assignment'
       end
     end
 
@@ -590,22 +673,14 @@ module Ci
 
       before do
         allow(Time).to receive(:now).and_return(current_time)
-
-        # Stub defaults for any metrics other than the ones we're testing
-        allow(Gitlab::Metrics).to receive(:counter)
-                                    .with(any_args)
-                                    .and_return(Gitlab::Metrics::NullMetric.instance)
-        allow(Gitlab::Metrics).to receive(:histogram)
-                                    .with(any_args)
-                                    .and_return(Gitlab::Metrics::NullMetric.instance)
-
         # Stub tested metrics
-        allow(Gitlab::Metrics).to receive(:counter)
-                                    .with(:job_register_attempts_total, anything)
-                                    .and_return(attempt_counter)
-        allow(Gitlab::Metrics).to receive(:histogram)
-                                    .with(:job_queue_duration_seconds, anything, anything, anything)
-                                    .and_return(job_queue_duration_seconds)
+        allow(Gitlab::Ci::Queue::Metrics)
+          .to receive(:attempt_counter)
+          .and_return(attempt_counter)
+
+        allow(Gitlab::Ci::Queue::Metrics)
+          .to receive(:job_queue_duration_seconds)
+          .and_return(job_queue_duration_seconds)
 
         project.update!(shared_runners_enabled: true)
         pending_job.update!(created_at: current_time - 3600, queued_at: current_time - 1800)
@@ -655,7 +730,7 @@ module Ci
       context 'when shared runner is used' do
         let(:runner) { create(:ci_runner, :instance, tag_list: %w(tag1 tag2)) }
         let(:expected_shared_runner) { true }
-        let(:expected_shard) { Ci::RegisterJobService::DEFAULT_METRICS_SHARD }
+        let(:expected_shard) { ::Gitlab::Ci::Queue::Metrics::DEFAULT_METRICS_SHARD }
         let(:expected_jobs_running_for_project_first_job) { 0 }
         let(:expected_jobs_running_for_project_third_job) { 2 }
 
@@ -694,7 +769,7 @@ module Ci
       context 'when specific runner is used' do
         let(:runner) { create(:ci_runner, :project, projects: [project], tag_list: %w(tag1 metrics_shard::shard_tag tag2)) }
         let(:expected_shared_runner) { false }
-        let(:expected_shard) { Ci::RegisterJobService::DEFAULT_METRICS_SHARD }
+        let(:expected_shard) { ::Gitlab::Ci::Queue::Metrics::DEFAULT_METRICS_SHARD }
         let(:expected_jobs_running_for_project_first_job) { '+Inf' }
         let(:expected_jobs_running_for_project_third_job) { '+Inf' }
 
@@ -715,6 +790,46 @@ module Ci
       end
     end
 
+    context 'when max queue depth is reached' do
+      let!(:pending_job) { create(:ci_build, :pending, :degenerated, pipeline: pipeline) }
+      let!(:pending_job_2) { create(:ci_build, :pending, :degenerated, pipeline: pipeline) }
+      let!(:pending_job_3) { create(:ci_build, :pending, pipeline: pipeline) }
+
+      before do
+        stub_const("#{described_class}::MAX_QUEUE_DEPTH", 2)
+      end
+
+      context 'when feature is enabled' do
+        before do
+          stub_feature_flags(gitlab_ci_builds_queue_limit: true)
+        end
+
+        it 'returns 409 conflict' do
+          expect(Ci::Build.pending.unstarted.count).to eq 3
+
+          result = described_class.new(specific_runner).execute
+
+          expect(result).not_to be_valid
+          expect(result.build).to be_nil
+        end
+      end
+
+      context 'when feature is disabled' do
+        before do
+          stub_feature_flags(gitlab_ci_builds_queue_limit: false)
+        end
+
+        it 'returns a valid result' do
+          expect(Ci::Build.pending.unstarted.count).to eq 3
+
+          result = described_class.new(specific_runner).execute
+
+          expect(result).to be_valid
+          expect(result.build).to eq pending_job_3
+        end
+      end
+    end
+
     def execute(runner, params = {})
       described_class.new(runner).execute(params).build
     end
diff --git a/spec/services/ci/update_build_queue_service_spec.rb b/spec/services/ci/update_build_queue_service_spec.rb
index ebccfdc5140..2d9f80a249d 100644
--- a/spec/services/ci/update_build_queue_service_spec.rb
+++ b/spec/services/ci/update_build_queue_service_spec.rb
@@ -26,6 +26,25 @@ RSpec.describe Ci::UpdateBuildQueueService do
       end
 
       it_behaves_like 'refreshes runner'
+
+      it 'avoids running redundant queries' do
+        expect(Ci::Runner).not_to receive(:owned_or_instance_wide)
+
+        subject.execute(build)
+      end
+
+      context 'when feature flag ci_reduce_queries_when_ticking_runner_queue is disabled' do
+        before do
+          stub_feature_flags(ci_reduce_queries_when_ticking_runner_queue: false)
+          stub_feature_flags(ci_runners_short_circuit_assignable_for: false)
+        end
+
+        it 'runs redundant queries using `owned_or_instance_wide` scope' do
+          expect(Ci::Runner).to receive(:owned_or_instance_wide).and_call_original
+
+          subject.execute(build)
+        end
+      end
     end
   end
 
@@ -97,4 +116,43 @@ RSpec.describe Ci::UpdateBuildQueueService do
       it_behaves_like 'does not refresh runner'
     end
   end
+
+  context 'avoids N+1 queries', :request_store do
+    let!(:build) { create(:ci_build, pipeline: pipeline, tag_list: %w[a b]) }
+    let!(:project_runner) { create(:ci_runner, :project, :online, projects: [project], tag_list: %w[a b c]) }
+
+    context 'when ci_preload_runner_tags and ci_reduce_queries_when_ticking_runner_queue are enabled' do
+      before do
+        stub_feature_flags(
+          ci_reduce_queries_when_ticking_runner_queue: true,
+          ci_preload_runner_tags: true
+        )
+      end
+
+      it 'does execute the same amount of queries regardless of number of runners' do
+        control_count = ActiveRecord::QueryRecorder.new { subject.execute(build) }.count
+
+        create_list(:ci_runner, 10, :project, :online, projects: [project], tag_list: %w[b c d])
+
+        expect { subject.execute(build) }.not_to exceed_all_query_limit(control_count)
+      end
+    end
+
+    context 'when ci_preload_runner_tags and ci_reduce_queries_when_ticking_runner_queue are disabled' do
+      before do
+        stub_feature_flags(
+          ci_reduce_queries_when_ticking_runner_queue: false,
+          ci_preload_runner_tags: false
+        )
+      end
+
+      it 'does execute more queries for more runners' do
+        control_count = ActiveRecord::QueryRecorder.new { subject.execute(build) }.count
+
+        create_list(:ci_runner, 10, :project, :online, projects: [project], tag_list: %w[b c d])
+
+        expect { subject.execute(build) }.to exceed_all_query_limit(control_count)
+      end
+    end
+  end
 end