1 files changed, 69 insertions, 19 deletions

diff --git a/app/services/ci/process_pipeline_service.rb b/app/services/ci/process_pipeline_service.rb
index aaf56048b5c..99d4ff9ecd1 100644
--- a/app/services/ci/process_pipeline_service.rb
+++ b/app/services/ci/process_pipeline_service.rb
@@ -4,35 +4,69 @@ module Ci
   class ProcessPipelineService < BaseService
     attr_reader :pipeline
 
-    def execute(pipeline)
+    def execute(pipeline, trigger_build_ids = nil)
       @pipeline = pipeline
 
       update_retried
 
-      new_builds =
-        stage_indexes_of_created_processables.map do |index|
-          process_stage(index)
-        end
+      success = process_stages_without_needs
+
+      # we evaluate dependent needs,
+      # only when the another job has finished
+      success = process_builds_with_needs(trigger_build_ids) || success
 
       @pipeline.update_status
 
-      new_builds.flatten.any?
+      success
     end
 
     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
+
+    def process_builds_with_needs(trigger_build_ids)
+      return false unless trigger_build_ids.present?
+      return false unless Feature.enabled?(:ci_dag_support, project)
+
+      # rubocop: disable CodeReuse/ActiveRecord
+      trigger_build_names = pipeline.statuses
+        .where(id: trigger_build_ids)
+        .select(:name)
+      # rubocop: enable CodeReuse/ActiveRecord
+
+      created_processables
+        .with_needs(trigger_build_names)
+        .find_each
+        .map(&method(:process_build_with_needs))
+        .any?
+    end
+
+    def process_build_with_needs(build)
+      current_status = status_for_build_needs(build.needs.map(&:name))
+
+      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)
       end
     end
 
@@ -43,17 +77,33 @@ 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 status_for_build_needs(needs)
+      pipeline.builds.where(name: needs).latest.status || 'success'
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
     # rubocop: disable CodeReuse/ActiveRecord
-    def created_processables_in_stage(index)
-      created_processables.where(stage_idx: index)
+    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_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