Optimise DAG processing

app/models/commit_status.rb

@@ -40,8 +40,11 @@ class CommitStatus < ApplicationRecord
   scope :ordered, -> { order(:name) }
   scope :latest_ordered, -> { latest.ordered.includes(project: :namespace) }
   scope :retried_ordered, -> { retried.ordered.includes(project: :namespace) }
+  scope :before_stage, -> (index) { where('stage_idx < ?', index) }
+  scope :for_stage, -> (index) { where(stage_idx: index) }
   scope :after_stage, -> (index) { where('stage_idx > ?', index) }
   scope :processables, -> { where(type: %w[Ci::Build Ci::Bridge]) }
+  scope :for_ids, -> (ids) { where(id: ids) }
 
   scope :with_needs, -> (names = nil) do
     needs = Ci::BuildNeed.scoped_build.select(1)
@@ -49,8 +52,10 @@ class CommitStatus < ApplicationRecord
     where('EXISTS (?)', needs).preload(:needs)
   end
 
-  scope :without_needs, -> do
-    where('NOT EXISTS (?)', Ci::BuildNeed.scoped_build.select(1))
+  scope :without_needs, -> (names = nil) do
+    needs = Ci::BuildNeed.scoped_build.select(1)
+    needs = needs.where(name: names) if names
+    where('NOT EXISTS (?)', needs)
   end
 
   # We use `CommitStatusEnums.failure_reasons` here so that EE can more easily
@@ -149,6 +154,18 @@ class CommitStatus < ApplicationRecord
     end
   end
 
+  def self.names
+    select(:name)
+  end
+
+  def self.status_for_prior_stages(index)
+    before_stage(index).latest.status || 'success'
+  end
+
+  def self.status_for_names(names)
+    where(name: names).latest.status || 'success'
+  end
+
   def locking_enabled?
     will_save_change_to_status?
   end

app/models/concerns/has_status.rb

@@ -106,10 +106,15 @@ module HasStatus
     scope :running_or_pending, -> { with_status(:running, :pending) }
     scope :finished, -> { with_status(:success, :failed, :canceled) }
     scope :failed_or_canceled, -> { with_status(:failed, :canceled) }
+    scope :incomplete, -> { without_statuses(completed_statuses) }
 
     scope :cancelable, -> do
       where(status: [:running, :preparing, :pending, :created, :scheduled])
     end
+
+    scope :without_statuses, -> (names) do
+      with_status(all_state_names - names.to_a)
+    end
   end
 
   def started?

app/services/ci/process_pipeline_service.rb

@@ -42,14 +42,19 @@ module Ci
       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
+      # 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?
@@ -70,17 +75,13 @@ module Ci
       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 status_for_build_needs(needs)
-      pipeline.builds.where(name: needs).latest.status || 'success'
+      pipeline.processables.status_for_names(needs)
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
     # rubocop: disable CodeReuse/ActiveRecord
     def stage_indexes_of_created_processables_without_needs
@@ -89,12 +90,10 @@ module Ci
     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)
+        .for_stage(index)
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
     def created_processables_without_needs
       if Feature.enabled?(:ci_dag_support, project)