4 files changed, 41 insertions, 9 deletions
diff --git a/app/models/ci/build.rb b/app/models/ci/build.rb
index bb3762c26f6..369a793f3d5 100644
--- a/app/models/ci/build.rb
+++ b/app/models/ci/build.rb
@@ -447,10 +447,6 @@ module Ci
         options_retry_when.include?('always')
     end
 
-    def latest?
-      !retried?
-    end
-
     def any_unmet_prerequisites?
       prerequisites.present?
     end
diff --git a/app/models/ci/pipeline.rb b/app/models/ci/pipeline.rb
index 3943d991c87..7e3ba98d86c 100644
--- a/app/models/ci/pipeline.rb
+++ b/app/models/ci/pipeline.rb
@@ -515,7 +515,9 @@ module Ci
     # rubocop: enable CodeReuse/ServiceClass
 
     def mark_as_processable_after_stage(stage_idx)
-      builds.skipped.after_stage(stage_idx).find_each(&:process)
+      builds.skipped.after_stage(stage_idx).find_each do |build|
+        Gitlab::OptimisticLocking.retry_lock(build, &:process)
+      end
     end
 
     def latest?
@@ -554,6 +556,13 @@ module Ci
       end
     end
 
+    def needs_processing?
+      statuses
+        .where(processed: [false, nil])
+        .latest
+        .exists?
+    end
+
     # TODO: this logic is duplicate with Pipeline::Chain::Config::Content
     # we should persist this is `ci_pipelines.config_path`
     def config_path
@@ -583,9 +592,8 @@ module Ci
       project.notes.for_commit_id(sha)
     end
 
-    def update_status
+    def set_status(new_status)
       retry_optimistic_lock(self) do
-        new_status = latest_builds_status.to_s
         case new_status
         when 'created' then nil
         when 'waiting_for_resource' then request_resource
@@ -605,6 +613,10 @@ module Ci
       end
     end
 
+    def update_legacy_status
+      set_status(latest_builds_status.to_s)
+    end
+
     def protected_ref?
       strong_memoize(:protected_ref) { project.protected_for?(git_ref) }
     end
diff --git a/app/models/ci/processable.rb b/app/models/ci/processable.rb
index 9c56aa67e20..6c4b271cd2c 100644
--- a/app/models/ci/processable.rb
+++ b/app/models/ci/processable.rb
@@ -8,8 +8,26 @@ module Ci
 
     scope :preload_needs, -> { preload(:needs) }
 
+    def self.select_with_aggregated_needs(project)
+      return all unless Feature.enabled?(:ci_dag_support, project, default_enabled: true)
+
+      aggregated_needs_names = Ci::BuildNeed
+        .scoped_build
+        .select("ARRAY_AGG(name)")
+        .to_sql
+
+      all.select(
+        '*',
+        "(#{aggregated_needs_names}) as aggregated_needs_names"
+      )
+    end
+
     validates :type, presence: true
 
+    def aggregated_needs_names
+      read_attribute(:aggregated_needs_names)
+    end
+
     def schedulable?
       raise NotImplementedError
     end
diff --git a/app/models/ci/stage.rb b/app/models/ci/stage.rb
index 96041e02337..75f73429c2a 100644
--- a/app/models/ci/stage.rb
+++ b/app/models/ci/stage.rb
@@ -13,9 +13,12 @@ module Ci
     belongs_to :pipeline
 
     has_many :statuses, class_name: 'CommitStatus', foreign_key: :stage_id
+    has_many :processables, class_name: 'Ci::Processable', foreign_key: :stage_id
     has_many :builds, foreign_key: :stage_id
     has_many :bridges, foreign_key: :stage_id
 
+    scope :ordered, -> { order(position: :asc) }
+
     with_options unless: :importing? do
       validates :project, presence: true
       validates :pipeline, presence: true
@@ -80,9 +83,8 @@ module Ci
       end
     end
 
-    def update_status
+    def set_status(new_status)
       retry_optimistic_lock(self) do
-        new_status = latest_stage_status.to_s
         case new_status
         when 'created' then nil
         when 'waiting_for_resource' then request_resource
@@ -102,6 +104,10 @@ module Ci
       end
     end
 
+    def update_legacy_status
+      set_status(latest_stage_status.to_s)
+    end
+
     def groups
       @groups ||= Ci::Group.fabricate(self)
     end