5 files changed, 106 insertions, 33 deletions

diff --git a/app/services/ci/archive_trace_service.rb b/app/services/ci/archive_trace_service.rb
index a1dd00721b5..700d78361a4 100644
--- a/app/services/ci/archive_trace_service.rb
+++ b/app/services/ci/archive_trace_service.rb
@@ -2,8 +2,25 @@
 
 module Ci
   class ArchiveTraceService
-    def execute(job)
+    def execute(job, worker_name:)
+      # TODO: Remove this logging once we confirmed new live trace architecture is functional.
+      # See https://gitlab.com/gitlab-com/gl-infra/infrastructure/issues/4667.
+      unless job.has_live_trace?
+        Sidekiq.logger.warn(class: worker_name,
+                            message: 'The job does not have live trace but going to be archived.',
+                            job_id: job.id)
+        return
+      end
+
       job.trace.archive!
+
+      # TODO: Remove this logging once we confirmed new live trace architecture is functional.
+      # See https://gitlab.com/gitlab-com/gl-infra/infrastructure/issues/4667.
+      unless job.has_archived_trace?
+        Sidekiq.logger.warn(class: worker_name,
+                            message: 'The job does not have archived trace after archiving.',
+                            job_id: job.id)
+      end
     rescue ::Gitlab::Ci::Trace::AlreadyArchivedError
       # It's already archived, thus we can safely ignore this exception.
     rescue => e
@@ -11,7 +28,7 @@ module Ci
       # If `archive!` keeps failing for over a week, that could incur data loss.
       # (See more https://docs.gitlab.com/ee/administration/job_traces.html#new-live-trace-architecture)
       # In order to avoid interrupting the system, we do not raise an exception here.
-      archive_error(e, job)
+      archive_error(e, job, worker_name)
     end
 
     private
@@ -22,13 +39,16 @@ module Ci
                                 "Counter of failed attempts of trace archiving")
     end
 
-    def archive_error(error, job)
+    def archive_error(error, job, worker_name)
       failed_archive_counter.increment
-      Rails.logger.error "Failed to archive trace. id: #{job.id} message: #{error.message}"
+
+      Sidekiq.logger.warn(class: worker_name,
+        message: "Failed to archive trace. message: #{error.message}.",
+        job_id: job.id)
 
       Gitlab::Sentry
         .track_exception(error,
-                         issue_url: 'https://gitlab.com/gitlab-org/gitlab-ce/issues/51502',
+                          issue_url: 'https://gitlab.com/gitlab-org/gitlab-ce/issues/51502',
                          extra: { job_id: job.id })
     end
   end
diff --git a/app/services/ci/play_build_service.rb b/app/services/ci/play_build_service.rb
index eb0b070657d..9f922ffde81 100644
--- a/app/services/ci/play_build_service.rb
+++ b/app/services/ci/play_build_service.rb
@@ -2,7 +2,7 @@
 
 module Ci
   class PlayBuildService < ::BaseService
-    def execute(build)
+    def execute(build, job_variables_attributes = nil)
       unless can?(current_user, :update_build, build)
         raise Gitlab::Access::AccessDeniedError
       end
@@ -10,7 +10,7 @@ module Ci
       # Try to enqueue the build, otherwise create a duplicate.
       #
       if build.enqueue
-        build.tap { |action| action.update(user: current_user) }
+        build.tap { |action| action.update(user: current_user, job_variables_attributes: job_variables_attributes || []) }
       else
         Ci::Build.retry(build, current_user)
       end
diff --git a/app/services/ci/process_pipeline_service.rb b/app/services/ci/process_pipeline_service.rb
index 4a7ce00b8e2..3b145a65d79 100644
--- a/app/services/ci/process_pipeline_service.rb
+++ b/app/services/ci/process_pipeline_service.rb
@@ -4,56 +4,105 @@ 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, default_enabled: true)
+
+      # we find processables that are dependent:
+      # 1. because of current dependency,
+      trigger_build_names = pipeline.processables.latest
+        .for_ids(trigger_build_ids).names
+
+      # 2. does not have builds that not yet complete
+      incomplete_build_names = pipeline.processables.latest
+        .incomplete.names
+
+      # Each found processable is guaranteed here to have completed status
+      created_processables
+        .with_needs(trigger_build_names)
+        .without_needs(incomplete_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
 
-    # rubocop: disable CodeReuse/ActiveRecord
     def status_for_prior_stages(index)
-      pipeline.builds.where('stage_idx < ?', index).latest.status || 'success'
+      pipeline.processables.status_for_prior_stages(index)
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
-    # rubocop: disable CodeReuse/ActiveRecord
-    def stage_indexes_of_created_processables
-      created_processables.order(:stage_idx).pluck('distinct stage_idx')
+    def status_for_build_needs(needs)
+      pipeline.processables.status_for_names(needs)
     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
 
+    def created_processables_in_stage_without_needs(index)
+      created_processables_without_needs
+        .for_stage(index)
+    end
+
+    def created_processables_without_needs
+      if Feature.enabled?(:ci_dag_support, project, default_enabled: true)
+        pipeline.processables.created.without_needs
+      else
+        pipeline.processables.created
+      end
+    end
+
     def created_processables
       pipeline.processables.created
     end
@@ -68,7 +117,7 @@ module Ci
       latest_statuses = pipeline.statuses.latest
         .group(:name)
         .having('count(*) > 1')
-        .pluck('max(id)', 'name')
+        .pluck(Arel.sql('MAX(id)'), 'name')
 
       # mark builds that are retried
       pipeline.statuses.latest
diff --git a/app/services/ci/retry_build_service.rb b/app/services/ci/retry_build_service.rb
index fab8a179843..338495ba030 100644
--- a/app/services/ci/retry_build_service.rb
+++ b/app/services/ci/retry_build_service.rb
@@ -5,7 +5,7 @@ module Ci
     CLONE_ACCESSORS = %i[pipeline project ref tag options name
                          allow_failure stage stage_id stage_idx trigger_request
                          yaml_variables when environment coverage_regex
-                         description tag_list protected].freeze
+                         description tag_list protected needs].freeze
 
     def execute(build)
       reprocess!(build).tap do |new_build|
diff --git a/app/services/ci/update_build_queue_service.rb b/app/services/ci/update_build_queue_service.rb
index 9c589d910eb..31c7178c9e7 100644
--- a/app/services/ci/update_build_queue_service.rb
+++ b/app/services/ci/update_build_queue_service.rb
@@ -9,6 +9,10 @@ module Ci
     private
 
     def tick_for(build, runners)
+      if Feature.enabled?(:ci_update_queues_for_online_runners, build.project, default_enabled: true)
+        runners = runners.with_recent_runner_queue
+      end
+
       runners.each do |runner|
         runner.pick_build!(build)
       end