Properly process `needs:` with `when:`

Currently, some of the jobs with `needs:`
would be processed in stages, it means
that `when:` for such jobs would not be
respected.

This changes the behavior to have a separate
execution paths for jobs with `needs:`.

app/models/commit_status.rb

@@ -49,6 +49,10 @@ class CommitStatus < ApplicationRecord
     where('EXISTS (?)', needs).preload(:needs)
   end
 
+  scope :without_needs, -> do
+    where('NOT EXISTS (?)', Ci::BuildNeed.scoped_build.select(1))
+  end
+
   # We use `CommitStatusEnums.failure_reasons` here so that EE can more easily
   # extend this `Hash` with new values.
   enum_with_nil failure_reason: ::CommitStatusEnums.failure_reasons

app/services/ci/process_pipeline_service.rb

@@ -9,10 +9,7 @@ module Ci
 
       update_retried
 
-      success =
-        stage_indexes_of_created_processables.flat_map do |index|
-          process_stage(index)
-        end.any?
+      success = process_stages_without_needs
 
       # we evaluate dependent needs,
       # only when the another job has finished
@@ -25,18 +22,19 @@ module Ci
 
     private
 
-    def process_stage(index)
+    def process_stages_without_needs
+      stage_indexes_of_created_processables_without_needs.flat_map do |index|
+        process_stage_without_needs(index)
+      end.any?
+    end
+
+    def process_stage_without_needs(index)
       current_status = status_for_prior_stages(index)
 
-      return if HasStatus::BLOCKED_STATUS.include?(current_status)
+      return unless HasStatus::COMPLETED_STATUSES.include?(current_status)
 
-      if HasStatus::COMPLETED_STATUSES.include?(current_status)
-        created_processables_in_stage(index).select do |build|
-          Gitlab::OptimisticLocking.retry_lock(build) do |subject|
-            Ci::ProcessBuildService.new(project, @user)
-              .execute(build, current_status)
-          end
-        end
+      created_processables_in_stage_without_needs(index).select do |build|
+        process_build(build, current_status)
       end
     end
 
@@ -56,6 +54,10 @@ module Ci
 
       return unless HasStatus::COMPLETED_STATUSES.include?(current_status)
 
+      process_build(build, current_status)
+    end
+
+    def process_build(build, current_status)
       Gitlab::OptimisticLocking.retry_lock(build) do |subject|
         Ci::ProcessBuildService.new(project, @user)
           .execute(subject, current_status)
@@ -75,17 +77,27 @@ module Ci
     # rubocop: enable CodeReuse/ActiveRecord
 
     # rubocop: disable CodeReuse/ActiveRecord
-    def stage_indexes_of_created_processables
-      created_processables.order(:stage_idx).pluck(Arel.sql('DISTINCT stage_idx'))
+    def stage_indexes_of_created_processables_without_needs
+      created_processables_without_needs.order(:stage_idx)
+        .pluck(Arel.sql('DISTINCT stage_idx'))
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
     # rubocop: disable CodeReuse/ActiveRecord
-    def created_processables_in_stage(index)
-      created_processables.where(stage_idx: index)
+    def created_processables_in_stage_without_needs(index)
+      created_processables_without_needs
+        .where(stage_idx: index)
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
+    def created_processables_without_needs
+      if Feature.enabled?(:ci_dag_support, project)
+        pipeline.processables.created.without_needs
+      else
+        pipeline.processables.created
+      end
+    end
+
     def created_processables
       pipeline.processables.created
     end

spec/models/ci/build_spec.rb

@@ -208,6 +208,22 @@ describe Ci::Build do
     end
   end
 
+  describe '.without_needs' do
+    let!(:build) { create(:ci_build) }
+
+    subject { described_class.without_needs }
+
+    context 'when no build_need is created' do
+      it { is_expected.to contain_exactly(build) }
+    end
+
+    context 'when a build_need is created' do
+      let!(:need_a) { create(:ci_build_need, build: build) }
+
+      it { is_expected.to be_empty }
+    end
+  end
+
   describe '#enqueue' do
     let(:build) { create(:ci_build, :created) }
 

spec/services/ci/process_pipeline_service_spec.rb

@@ -786,6 +786,50 @@ describe Ci::ProcessPipelineService, '#execute' do
         expect(builds.pending).to contain_exactly(deploy)
       end
     end
+
+    context 'when one of the jobs is run on a failure' do
+      let!(:linux_notify) { create_build('linux:notify', stage: 'deploy', stage_idx: 2, when: 'on_failure') }
+
+      let!(:linux_notify_on_build) { create(:ci_build_need, build: linux_notify, name: 'linux:build') }
+
+      context 'when another job in build phase fails first' do
+        context 'when ci_dag_support is enabled' do
+          it 'does skip linux:notify' do
+            expect(process_pipeline).to be_truthy
+
+            mac_build.reset.drop!
+            linux_build.reset.success!
+
+            expect(linux_notify.reset).to be_skipped
+          end
+        end
+
+        context 'when ci_dag_support is disabled' do
+          before do
+            stub_feature_flags(ci_dag_support: false)
+          end
+
+          it 'does run linux:notify' do
+            expect(process_pipeline).to be_truthy
+
+            mac_build.reset.drop!
+            linux_build.reset.success!
+
+            expect(linux_notify.reset).to be_pending
+          end
+        end
+      end
+
+      context 'when linux:build job fails first' do
+        it 'does run linux:notify' do
+          expect(process_pipeline).to be_truthy
+
+          linux_build.reset.drop!
+
+          expect(linux_notify.reset).to be_pending
+        end
+      end
+    end
   end
 
   def process_pipeline