Merge branch 'improve-pipeline-processing' into 'master'

Implement AtomicProcessing service

See merge request gitlab-org/gitlab!20229

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

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

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

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

app/models/commit_status.rb

@@ -40,6 +40,7 @@ class CommitStatus < ApplicationRecord
   scope :latest, -> { where(retried: [false, nil]) }
   scope :retried, -> { where(retried: true) }
   scope :ordered, -> { order(:name) }
+  scope :ordered_by_stage, -> { order(stage_idx: :asc) }
   scope :latest_ordered, -> { latest.ordered.includes(project: :namespace) }
   scope :retried_ordered, -> { retried.ordered.includes(project: :namespace) }
   scope :before_stage, -> (index) { where('stage_idx < ?', index) }
@@ -57,6 +58,10 @@ class CommitStatus < ApplicationRecord
     preload(:project, :user)
   end
 
+  scope :with_project_preload, -> do
+    preload(project: :namespace)
+  end
+
   scope :with_needs, -> (names = nil) do
     needs = Ci::BuildNeed.scoped_build.select(1)
     needs = needs.where(name: names) if names
@@ -69,6 +74,15 @@ class CommitStatus < ApplicationRecord
     where('NOT EXISTS (?)', needs)
   end
 
+  scope :match_id_and_lock_version, -> (slice) do
+    # it expects that items are an array of attributes to match
+    # each hash needs to have `id` and `lock_version`
+    slice.inject(self) do |relation, item|
+      match = CommitStatus.where(item.slice(:id, :lock_version))
+      relation.or(match)
+    end
+  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
@@ -86,6 +100,16 @@ class CommitStatus < ApplicationRecord
     # rubocop: enable CodeReuse/ServiceClass
   end
 
+  before_save if: :status_changed?, unless: :importing? do
+    if Feature.disabled?(:ci_atomic_processing, project)
+      self.processed = nil
+    elsif latest?
+      self.processed = false # force refresh of all dependent ones
+    elsif retried?
+      self.processed = true # retried are considered to be already processed
+    end
+  end
+
   state_machine :status do
     event :process do
       transition [:skipped, :manual] => :created
@@ -136,19 +160,13 @@ class CommitStatus < ApplicationRecord
     end
 
     after_transition do |commit_status, transition|
-      next unless commit_status.project
       next if transition.loopback?
+      next if commit_status.processed?
+      next unless commit_status.project
 
       commit_status.run_after_commit do
-        if pipeline_id
-          if complete? || manual?
-            PipelineProcessWorker.perform_async(pipeline_id, [id])
-          else
-            PipelineUpdateWorker.perform_async(pipeline_id)
-          end
-        end
+        schedule_stage_and_pipeline_update
 
-        StageUpdateWorker.perform_async(stage_id)
         ExpireJobCacheWorker.perform_async(id)
       end
     end
@@ -177,6 +195,11 @@ class CommitStatus < ApplicationRecord
     where(name: names).latest.slow_composite_status || 'success'
   end
 
+  def self.update_as_processed!
+    # Marks items as processed, and increases `lock_version` (Optimisitc Locking)
+    update_all('processed=TRUE, lock_version=COALESCE(lock_version,0)+1')
+  end
+
   def locking_enabled?
     will_save_change_to_status?
   end
@@ -193,6 +216,10 @@ class CommitStatus < ApplicationRecord
     calculate_duration
   end
 
+  def latest?
+    !retried?
+  end
+
   def playable?
     false
   end
@@ -244,4 +271,21 @@ class CommitStatus < ApplicationRecord
       v =~ /\d+/ ? v.to_i : v
     end
   end
+
+  private
+
+  def schedule_stage_and_pipeline_update
+    if Feature.enabled?(:ci_atomic_processing, project)
+      # Atomic Processing requires only single Worker
+      PipelineProcessWorker.perform_async(pipeline_id, [id])
+    else
+      if complete? || manual?
+        PipelineProcessWorker.perform_async(pipeline_id, [id])
+      else
+        PipelineUpdateWorker.perform_async(pipeline_id)
+      end
+
+      StageUpdateWorker.perform_async(stage_id)
+    end
+  end
 end

app/services/ci/pipeline_processing/atomic_processing_service.rb

+# frozen_string_literal: true
+
+module Ci
+  module PipelineProcessing
+    class AtomicProcessingService
+      include Gitlab::Utils::StrongMemoize
+      include ExclusiveLeaseGuard
+
+      attr_reader :pipeline
+
+      DEFAULT_LEASE_TIMEOUT = 1.minute
+      BATCH_SIZE = 20
+
+      def initialize(pipeline)
+        @pipeline = pipeline
+        @collection = AtomicProcessingService::StatusCollection.new(pipeline)
+      end
+
+      def execute
+        return unless pipeline.needs_processing?
+
+        success = try_obtain_lease { process! }
+
+        # re-schedule if we need further processing
+        if success && pipeline.needs_processing?
+          PipelineProcessWorker.perform_async(pipeline.id)
+        end
+
+        success
+      end
+
+      private
+
+      def process!
+        update_stages!
+        update_pipeline!
+        update_statuses_processed!
+
+        true
+      end
+
+      def update_stages!
+        pipeline.stages.ordered.each(&method(:update_stage!))
+      end
+
+      def update_stage!(stage)
+        # Update processables for a given stage in bulk/slices
+        ids = @collection.created_processable_ids_for_stage_position(stage.position)
+        ids.in_groups_of(BATCH_SIZE, false, &method(:update_processables!))
+
+        status = @collection.status_for_stage_position(stage.position)
+        stage.set_status(status)
+      end
+
+      def update_processables!(ids)
+        created_processables = pipeline.processables.for_ids(ids)
+          .with_project_preload
+          .created
+          .latest
+          .ordered_by_stage
+          .select_with_aggregated_needs(project)
+
+        created_processables.each(&method(:update_processable!))
+      end
+
+      def update_pipeline!
+        pipeline.set_status(@collection.status_of_all)
+      end
+
+      def update_statuses_processed!
+        processing = @collection.processing_processables
+        processing.each_slice(BATCH_SIZE) do |slice|
+          pipeline.statuses.match_id_and_lock_version(slice)
+            .update_as_processed!
+        end
+      end
+
+      def update_processable!(processable)
+        status = processable_status(processable)
+        return unless HasStatus::COMPLETED_STATUSES.include?(status)
+
+        # transition status if possible
+        Gitlab::OptimisticLocking.retry_lock(processable) do |subject|
+          Ci::ProcessBuildService.new(project, subject.user)
+            .execute(subject, status)
+
+          # update internal representation of status
+          # to make the status change of processable
+          # to be taken into account during further processing
+          @collection.set_processable_status(
+            processable.id, processable.status, processable.lock_version)
+        end
+      end
+
+      def processable_status(processable)
+        if needs_names = processable.aggregated_needs_names
+          # Processable uses DAG, get status of all dependent needs
+          @collection.status_for_names(needs_names)
+        else
+          # Processable uses Stages, get status of prior stage
+          @collection.status_for_prior_stage_position(processable.stage_idx.to_i)
+        end
+      end
+
+      def project
+        pipeline.project
+      end
+
+      def lease_key
+        "#{super}::pipeline_id:#{pipeline.id}"
+      end
+
+      def lease_timeout
+        DEFAULT_LEASE_TIMEOUT
+      end
+    end
+  end
+end

app/services/ci/pipeline_processing/atomic_processing_service/status_collection.rb

+# frozen_string_literal: true
+
+module Ci
+  module PipelineProcessing
+    class AtomicProcessingService
+      class StatusCollection
+        include Gitlab::Utils::StrongMemoize
+
+        attr_reader :pipeline
+
+        # We use these columns to perform an efficient
+        # calculation of a status
+        STATUSES_COLUMNS = [
+          :id, :name, :status, :allow_failure,
+          :stage_idx, :processed, :lock_version
+        ].freeze
+
+        def initialize(pipeline)
+          @pipeline = pipeline
+          @stage_statuses = {}
+          @prior_stage_statuses = {}
+        end
+
+        # This method updates internal status for given ID
+        def set_processable_status(id, status, lock_version)
+          processable = all_statuses_by_id[id]
+          return unless processable
+
+          processable[:status] = status
+          processable[:lock_version] = lock_version
+        end
+
+        # This methods gets composite status of all processables
+        def status_of_all
+          status_for_array(all_statuses)
+        end
+
+        # This methods gets composite status for processables with given names
+        def status_for_names(names)
+          name_statuses = all_statuses_by_name.slice(*names)
+
+          status_for_array(name_statuses.values)
+        end
+
+        # This methods gets composite status for processables before given stage
+        def status_for_prior_stage_position(position)
+          strong_memoize("status_for_prior_stage_position_#{position}") do
+            stage_statuses = all_statuses_grouped_by_stage_position
+              .select { |stage_position, _| stage_position < position }
+
+            status_for_array(stage_statuses.values.flatten)
+          end
+        end
+
+        # This methods gets a list of processables for a given stage
+        def created_processable_ids_for_stage_position(current_position)
+          all_statuses_grouped_by_stage_position[current_position]
+            .to_a
+            .select { |processable| processable[:status] == 'created' }
+            .map { |processable| processable[:id] }
+        end
+
+        # This methods gets composite status for processables at a given stage
+        def status_for_stage_position(current_position)
+          strong_memoize("status_for_stage_position_#{current_position}") do
+            stage_statuses = all_statuses_grouped_by_stage_position[current_position].to_a
+
+            status_for_array(stage_statuses.flatten)
+          end
+        end
+
+        # This method returns a list of all processable, that are to be processed
+        def processing_processables
+          all_statuses.lazy.reject { |status| status[:processed] }
+        end
+
+        private
+
+        def status_for_array(statuses)
+          result = Gitlab::Ci::Status::Composite
+            .new(statuses)
+            .status
+          result || 'success'
+        end
+
+        def all_statuses_grouped_by_stage_position
+          strong_memoize(:all_statuses_by_order) do
+            all_statuses.group_by { |status| status[:stage_idx].to_i }
+          end
+        end
+
+        def all_statuses_by_id
+          strong_memoize(:all_statuses_by_id) do
+            all_statuses.map do |row|
+              [row[:id], row]
+            end.to_h
+          end
+        end
+
+        def all_statuses_by_name
+          strong_memoize(:statuses_by_name) do
+            all_statuses.map do |row|
+              [row[:name], row]
+            end.to_h
+          end
+        end
+
+        # rubocop: disable CodeReuse/ActiveRecord
+        def all_statuses
+          # We fetch all relevant data in one go.
+          #
+          # This is more efficient than relying
+          # on PostgreSQL to calculate composite status
+          # for us
+          #
+          # Since we need to reprocess everything
+          # we can fetch all of them and do processing
+          # ourselves.
+          strong_memoize(:all_statuses) do
+            raw_statuses = pipeline
+              .statuses
+              .latest
+              .ordered_by_stage
+              .pluck(*STATUSES_COLUMNS)
+
+            raw_statuses.map do |row|
+              STATUSES_COLUMNS.zip(row).to_h
+            end
+          end
+        end
+        # rubocop: enable CodeReuse/ActiveRecord
+      end
+    end
+  end
+end

app/services/ci/pipeline_processing/legacy_processing_service.rb

@@ -18,7 +18,7 @@ module Ci
         # only when the another job has finished
         success = process_builds_with_needs(trigger_build_ids) || success
 
-        @pipeline.update_status
+        @pipeline.update_legacy_status
 
         success
       end

app/services/ci/process_pipeline_service.rb

@@ -11,10 +11,16 @@ module Ci
     def execute(trigger_build_ids = nil)
       update_retried
 
+      if Feature.enabled?(:ci_atomic_processing, pipeline.project)
+        Ci::PipelineProcessing::AtomicProcessingService
+          .new(pipeline)
+          .execute
+      else
         Ci::PipelineProcessing::LegacyProcessingService
           .new(pipeline)
           .execute(trigger_build_ids)
       end
+    end
 
     private
 

app/services/ci/retry_build_service.rb

@@ -11,7 +11,7 @@ module Ci
       reprocess!(build).tap do |new_build|
         build.pipeline.mark_as_processable_after_stage(build.stage_idx)
 
-        new_build.enqueue!
+        Gitlab::OptimisticLocking.retry_lock(new_build, &:enqueue)
 
         MergeRequests::AddTodoWhenBuildFailsService
           .new(project, current_user)
@@ -31,15 +31,17 @@ module Ci
 
       attributes.push([:user, current_user])
 
-      build.retried = true
-
       Ci::Build.transaction do
         # mark all other builds of that name as retried
         build.pipeline.builds.latest
           .where(name: build.name)
-          .update_all(retried: true)
+          .update_all(retried: true, processed: true)
 
-        create_build!(attributes)
+        create_build!(attributes).tap do
+          # mark existing object as retried/processed without a reload
+          build.retried = true
+          build.processed = true
+        end
       end
     end
     # rubocop: enable CodeReuse/ActiveRecord
@@ -49,6 +51,7 @@ module Ci
     def create_build!(attributes)
       build = project.builds.new(Hash[attributes])
       build.deployment = ::Gitlab::Ci::Pipeline::Seed::Deployment.new(build).to_resource
+      build.retried = false
       build.save!
       build
     end

app/workers/pipeline_update_worker.rb

@@ -7,10 +7,7 @@ class PipelineUpdateWorker
   queue_namespace :pipeline_processing
   latency_sensitive_worker!
 
-  # rubocop: disable CodeReuse/ActiveRecord
   def perform(pipeline_id)
-    Ci::Pipeline.find_by(id: pipeline_id)
-      .try(:update_status)
+    Ci::Pipeline.find_by_id(pipeline_id)&.update_legacy_status
   end
-  # rubocop: enable CodeReuse/ActiveRecord
 end

app/workers/stage_update_worker.rb

@@ -7,11 +7,7 @@ class StageUpdateWorker
   queue_namespace :pipeline_processing
   latency_sensitive_worker!
 
-  # rubocop: disable CodeReuse/ActiveRecord
   def perform(stage_id)
-    Ci::Stage.find_by(id: stage_id).try do |stage|
-      stage.update_status
+    Ci::Stage.find_by_id(stage_id)&.update_legacy_status
   end
-  end
-  # rubocop: enable CodeReuse/ActiveRecord
 end

changelogs/unreleased/improve-pipeline-processing.yml

+---
+title: Implement Atomic Processing that updates status of builds, stages and pipelines in one go
+merge_request: 20229
+author:
+type: performance

db/fixtures/development/14_pipelines.rb

@@ -57,7 +57,7 @@ class Gitlab::Seeder::Pipelines
       BUILDS.each { |opts| build_create!(pipeline, opts) }
       EXTERNAL_JOBS.each { |opts| commit_status_create!(pipeline, opts) }
       pipeline.update_duration
-      pipeline.update_status
+      pipeline.update_legacy_status
     end
   end
 

db/fixtures/development/17_cycle_analytics.rb

@@ -187,7 +187,7 @@ class Gitlab::Seeder::CycleAnalytics
 
       pipeline.builds.each(&:enqueue) # make sure all pipelines in pending state
       pipeline.builds.each(&:run!)
-      pipeline.update_status
+      pipeline.update_legacy_status
     end
   end
 
@@ -208,7 +208,7 @@ class Gitlab::Seeder::CycleAnalytics
       job = merge_request.head_pipeline.builds.where.not(environment: nil).last
 
       job.success!
-      job.pipeline.update_status
+      job.pipeline.update_legacy_status
     end
   end
 end

db/migrate/20191115114032_add_processed_to_ci_builds.rb

+# frozen_string_literal: true
+
+class AddProcessedToCiBuilds < ActiveRecord::Migration[5.2]
+  DOWNTIME = false
+
+  def change
+    add_column :ci_builds, :processed, :boolean
+  end
+end

db/schema.rb

@@ -697,6 +697,7 @@ ActiveRecord::Schema.define(version: 2020_01_14_204949) do
     t.integer "upstream_pipeline_id"
     t.bigint "resource_group_id"
     t.datetime_with_timezone "waiting_for_resource_at"
+    t.boolean "processed"
     t.index ["artifacts_expire_at"], name: "index_ci_builds_on_artifacts_expire_at", where: "(artifacts_file <> ''::text)"
     t.index ["auto_canceled_by_id"], name: "index_ci_builds_on_auto_canceled_by_id"
     t.index ["commit_id", "artifacts_expire_at", "id"], name: "index_ci_builds_on_commit_id_and_artifacts_expireatandidpartial", where: "(((type)::text = 'Ci::Build'::text) AND ((retried = false) OR (retried IS NULL)) AND ((name)::text = ANY (ARRAY[('sast'::character varying)::text, ('dependency_scanning'::character varying)::text, ('sast:container'::character varying)::text, ('container_scanning'::character varying)::text, ('dast'::character varying)::text])))"

ee/spec/services/ci/process_pipeline_service_spec.rb

@@ -19,7 +19,16 @@ describe Ci::ProcessPipelineService, '#execute' do
   end
 
   describe 'cross-project pipelines' do
+    using RSpec::Parameterized::TableSyntax
+
+    where(:ci_atomic_processing) do
+      [true, false]
+    end
+
+    with_them do
       before do
+        stub_feature_flags(ci_atomic_processing: ci_atomic_processing)
+
         create_processable(:build, name: 'test', stage: 'test')
         create_processable(:bridge, :variables,  name: 'cross',
                                                 stage: 'build',
@@ -29,12 +38,14 @@ describe Ci::ProcessPipelineService, '#execute' do
         stub_ci_pipeline_to_return_yaml_file
       end
 
-    it 'creates a downstream cross-project pipeline', :sidekiq_might_not_need_inline do
+      it 'creates a downstream cross-project pipeline', :sidekiq do
         service.execute
+        Sidekiq::Worker.drain_all
 
         expect_statuses(%w[test pending], %w[cross created], %w[deploy created])
 
         update_build_status(:test, :success)
+        Sidekiq::Worker.drain_all
 
         expect_statuses(%w[test success], %w[cross success], %w[deploy pending])
 
@@ -45,13 +56,14 @@ describe Ci::ProcessPipelineService, '#execute' do
           .to include(key: 'BRIDGE', value: 'cross', public: false, masked: false)
       end
     end
-
-  def expect_statuses(*statuses)
-    statuses.each do |name, status|
-      pipeline.statuses.find_by(name: name).yield_self do |build|
-        expect(build.status).to eq status
-      end
   end
+
+  def expect_statuses(*expected)
+    statuses = pipeline.statuses
+      .where(name: expected.map(&:first))
+      .pluck(:name, :status)
+
+    expect(statuses).to contain_exactly(*expected)
   end
 
   def update_build_status(name, status)

lib/gitlab/import_export/import_export.yml

@@ -240,6 +240,7 @@ excluded_attributes:
     - :upstream_pipeline_id
     - :resource_group_id
     - :waiting_for_resource_at
+    - :processed
   sentry_issue:
     - :issue_id
   push_event_payload:

spec/features/projects/badges/coverage_spec.rb

@@ -63,7 +63,7 @@ describe 'test coverage badge' do
 
     create(:ci_pipeline, opts).tap do |pipeline|
       yield pipeline
-      pipeline.update_status
+      pipeline.update_legacy_status
     end
   end
 

spec/lib/gitlab/badge/coverage/report_spec.rb

@@ -102,7 +102,7 @@ describe Gitlab::Badge::Coverage::Report do
 
     create(:ci_pipeline, opts).tap do |pipeline|
       yield pipeline
-      pipeline.update_status
+      pipeline.update_legacy_status
     end
   end
 end

spec/lib/gitlab/import_export/all_models.yml

@@ -216,6 +216,7 @@ stages:
 - project
 - pipeline
 - statuses
+- processables
 - builds
 - bridges
 statuses:

spec/lib/gitlab/import_export/safe_model_attributes.yml

@@ -333,6 +333,7 @@ CommitStatus:
 - scheduled_at
 - upstream_pipeline_id
 - interruptible
+- processed
 Ci::Variable:
 - id
 - project_id

spec/models/ci/build_trace_chunk_spec.rb

@@ -604,7 +604,7 @@ describe Ci::BuildTraceChunk, :clean_gitlab_redis_shared_state do
     context 'when traces are archived' do
       let(:subject) do
         project.builds.each do |build|
-          build.success!
+          build.reset.success!
         end
       end
 

spec/models/ci/pipeline_spec.rb

@@ -1008,22 +1008,22 @@ describe Ci::Pipeline, :mailer do
       end
     end
 
-    describe '#duration', :sidekiq_might_not_need_inline do
+    describe '#duration', :sidekiq_inline do
       context 'when multiple builds are finished' do
         before do
           travel_to(current + 30) do
             build.run!
-            build.success!
+            build.reload.success!
             build_b.run!
             build_c.run!
           end
 
           travel_to(current + 40) do
-            build_b.drop!
+            build_b.reload.drop!
           end
 
           travel_to(current + 70) do
-            build_c.success!
+            build_c.reload.success!
           end
         end
 
@@ -1044,7 +1044,7 @@ describe Ci::Pipeline, :mailer do
           end
 
           travel_to(current + 5.minutes) do
-            build.success!
+            build.reload.success!
           end
         end
 
@@ -1585,6 +1585,30 @@ describe Ci::Pipeline, :mailer do
     end
   end
 
+  describe '#needs_processing?' do
+    using RSpec::Parameterized::TableSyntax
+
+    subject { pipeline.needs_processing? }
+
+    where(:processed, :result) do
+      nil   | true
+      false | true
+      true  | false
+    end
+
+    with_them do
+      let(:build) do
+        create(:ci_build, :success, pipeline: pipeline, name: 'rubocop')
+      end
+
+      before do
+        build.update_column(:processed, processed)
+      end
+
+      it { is_expected.to eq(result) }
+    end
+  end
+
   shared_context 'with some outdated pipelines' do
     before do
       create_pipeline(:canceled, 'ref', 'A', project)
@@ -1785,7 +1809,7 @@ describe Ci::Pipeline, :mailer do
     it { is_expected.not_to include('created', 'waiting_for_resource', 'preparing', 'pending') }
   end
 
-  describe '#status', :sidekiq_might_not_need_inline do
+  describe '#status', :sidekiq_inline do
     let(:build) do
       create(:ci_build, :created, pipeline: pipeline, name: 'test')
     end
@@ -1826,7 +1850,7 @@ describe Ci::Pipeline, :mailer do
     context 'on run' do
       before do
         build.enqueue
-        build.run
+        build.reload.run
       end
 
       it { is_expected.to eq('running') }
@@ -1885,7 +1909,7 @@ describe Ci::Pipeline, :mailer do
       it 'updates does not change pipeline status' do
         expect(pipeline.statuses.latest.slow_composite_status).to be_nil
 
-        expect { pipeline.update_status }
+        expect { pipeline.update_legacy_status }
           .to change { pipeline.reload.status }
           .from('created')
           .to('skipped')
@@ -1898,7 +1922,7 @@ describe Ci::Pipeline, :mailer do
       end
 
       it 'updates pipeline status to running' do
-        expect { pipeline.update_status }
+        expect { pipeline.update_legacy_status }
           .to change { pipeline.reload.status }
           .from('created')
           .to('running')
@@ -1911,7 +1935,7 @@ describe Ci::Pipeline, :mailer do
       end
 
       it 'updates pipeline status to scheduled' do
-        expect { pipeline.update_status }
+        expect { pipeline.update_legacy_status }
           .to change { pipeline.reload.status }
           .from('created')
           .to('scheduled')
@@ -1926,7 +1950,7 @@ describe Ci::Pipeline, :mailer do
       end
 
       it 'raises an exception' do
-        expect { pipeline.update_status }
+        expect { pipeline.update_legacy_status }
           .to raise_error(HasStatus::UnknownStatusError)
       end
     end
@@ -2214,11 +2238,11 @@ describe Ci::Pipeline, :mailer do
         stub_full_request(hook.url, method: :post)
       end
 
-      context 'with multiple builds', :sidekiq_might_not_need_inline do
+      context 'with multiple builds', :sidekiq_inline do
         context 'when build is queued' do
           before do
-            build_a.enqueue
-            build_b.enqueue
+            build_a.reload.enqueue
+            build_b.reload.enqueue
           end
 
           it 'receives a pending event once' do
@@ -2228,10 +2252,10 @@ describe Ci::Pipeline, :mailer do
 
         context 'when build is run' do
           before do
-            build_a.enqueue
-            build_a.run
-            build_b.enqueue
-            build_b.run
+            build_a.reload.enqueue
+            build_a.reload.run!
+            build_b.reload.enqueue
+            build_b.reload.run!
           end
 
           it 'receives a running event once' do
@@ -2292,6 +2316,7 @@ describe Ci::Pipeline, :mailer do
              :created,
              pipeline: pipeline,
              name: name,
+             stage: "stage:#{stage_idx}",
              stage_idx: stage_idx)
     end
   end

spec/models/ci/processable_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Ci::Processable do
+  set(:project) { create(:project) }
+  set(:pipeline) { create(:ci_pipeline, project: project) }
+
+  describe '#aggregated_needs_names' do
+    let(:with_aggregated_needs) { pipeline.processables.select_with_aggregated_needs(project) }
+
+    context 'with created status' do
+      let!(:processable) { create(:ci_build, :created, project: project, pipeline: pipeline) }
+
+      context 'with needs' do
+        before do
+          create(:ci_build_need, build: processable, name: 'test1')
+          create(:ci_build_need, build: processable, name: 'test2')
+        end
+
+        it 'returns all processables' do
+          expect(with_aggregated_needs).to contain_exactly(processable)
+        end
+
+        it 'returns all needs' do
+          expect(with_aggregated_needs.first.aggregated_needs_names).to contain_exactly('test1', 'test2')
+        end
+
+        context 'with ci_dag_support disabled' do
+          before do
+            stub_feature_flags(ci_dag_support: false)
+          end
+
+          it 'returns all processables' do
+            expect(with_aggregated_needs).to contain_exactly(processable)
+          end
+
+          it 'returns empty needs' do
+            expect(with_aggregated_needs.first.aggregated_needs_names).to be_nil
+          end
+        end
+      end
+
+      context 'without needs' do
+        it 'returns all processables' do
+          expect(with_aggregated_needs).to contain_exactly(processable)
+        end
+
+        it 'returns empty needs' do
+          expect(with_aggregated_needs.first.aggregated_needs_names).to be_nil
+        end
+      end
+    end
+  end
+end

spec/models/ci/stage_spec.rb

@@ -63,7 +63,7 @@ describe Ci::Stage, :models do
       end
 
       it 'updates stage status correctly' do
-        expect { stage.update_status }
+        expect { stage.update_legacy_status }
           .to change { stage.reload.status }
           .to eq 'running'
       end
@@ -87,7 +87,7 @@ describe Ci::Stage, :models do
       end
 
       it 'updates status to skipped' do
-        expect { stage.update_status }
+        expect { stage.update_legacy_status }
           .to change { stage.reload.status }
           .to eq 'skipped'
       end
@@ -99,7 +99,7 @@ describe Ci::Stage, :models do
       end
 
       it 'updates status to scheduled' do
-        expect { stage.update_status }
+        expect { stage.update_legacy_status }
           .to change { stage.reload.status }
           .to 'scheduled'
       end
@@ -111,7 +111,7 @@ describe Ci::Stage, :models do
       end
 
       it 'updates status to waiting for resource' do
-        expect { stage.update_status }
+        expect { stage.update_legacy_status }
           .to change { stage.reload.status }
           .to 'waiting_for_resource'
       end
@@ -119,7 +119,7 @@ describe Ci::Stage, :models do
 
     context 'when stage is skipped because is empty' do
       it 'updates status to skipped' do
-        expect { stage.update_status }
+        expect { stage.update_legacy_status }
           .to change { stage.reload.status }
           .to eq('skipped')
       end
@@ -133,7 +133,7 @@ describe Ci::Stage, :models do
       it 'retries a lock to update a stage status' do
         stage.lock_version = 100
 
-        stage.update_status
+        stage.update_legacy_status
 
         expect(stage.reload).to be_failed
       end
@@ -147,7 +147,7 @@ describe Ci::Stage, :models do
       end
 
       it 'raises an exception' do
-        expect { stage.update_status }
+        expect { stage.update_legacy_status }
           .to raise_error(HasStatus::UnknownStatusError)
       end
     end
@@ -179,7 +179,7 @@ describe Ci::Stage, :models do
                                  stage_id: stage.id,
                                  status: status)
 
-          stage.update_status
+          stage.update_legacy_status
         end
       end
 
@@ -196,7 +196,7 @@ describe Ci::Stage, :models do
                           status: :failed,
                           allow_failure: true)
 
-        stage.update_status
+        stage.update_legacy_status
       end
 
       it 'is passed with warnings' do
@@ -243,7 +243,7 @@ describe Ci::Stage, :models do
         it 'recalculates index before updating status' do
           expect(stage.reload.position).to be_nil
 
-          stage.update_status
+          stage.update_legacy_status
 
           expect(stage.reload.position).to eq 10
         end
@@ -253,7 +253,7 @@ describe Ci::Stage, :models do
         it 'fallbacks to zero' do
           expect(stage.reload.position).to be_nil
 
-          stage.update_status
+          stage.update_legacy_status
 
           expect(stage.reload.position).to eq 0
         end

spec/models/commit_status_spec.rb

@@ -63,6 +63,42 @@ describe CommitStatus do
     end
   end
 
+  describe '#processed' do
+    subject { commit_status.processed }
+
+    context 'when ci_atomic_processing is disabled' do
+      before do
+        stub_feature_flags(ci_atomic_processing: false)
+
+        commit_status.save!
+      end
+
+      it { is_expected.to be_nil }
+    end
+
+    context 'when ci_atomic_processing is enabled' do
+      before do
+        stub_feature_flags(ci_atomic_processing: true)
+      end
+
+      context 'status is latest' do
+        before do
+          commit_status.update!(retried: false, status: :pending)
+        end
+
+        it { is_expected.to be_falsey }
+      end
+
+      context 'status is retried' do
+        before do
+          commit_status.update!(retried: true, status: :pending)
+        end
+
+        it { is_expected.to be_truthy }
+      end
+    end
+  end
+
   describe '#started?' do
     subject { commit_status.started? }
 

spec/services/ci/create_pipeline_service_spec.rb

@@ -362,11 +362,11 @@ describe Ci::CreatePipelineService do
 
             context 'when build that is not marked as interruptible is running' do
               it 'cancels running outdated pipelines', :sidekiq_might_not_need_inline do
-                pipeline_on_previous_commit
-                  .builds
-                  .find_by_name('build_2_1')
-                  .tap(&:enqueue!)
-                  .run!
+                build_2_1 = pipeline_on_previous_commit
+                  .builds.find_by_name('build_2_1')
+
+                build_2_1.enqueue!
+                build_2_1.reset.run!
 
                 pipeline
 
@@ -377,12 +377,12 @@ describe Ci::CreatePipelineService do
           end
 
           context 'when an uninterruptible build is running' do
-            it 'does not cancel running outdated pipelines', :sidekiq_might_not_need_inline do
-              pipeline_on_previous_commit
-                .builds
-                .find_by_name('build_3_1')
-                .tap(&:enqueue!)
-                .run!
+            it 'does not cancel running outdated pipelines', :sidekiq_inline do
+              build_3_1 = pipeline_on_previous_commit
+                .builds.find_by_name('build_3_1')
+
+              build_3_1.enqueue!
+              build_3_1.reset.run!
 
               pipeline
 

spec/services/ci/pipeline_processing/atomic_processing_service/status_collection_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Ci::PipelineProcessing::AtomicProcessingService::StatusCollection do
+  using RSpec::Parameterized::TableSyntax
+
+  set(:pipeline) { create(:ci_pipeline) }
+  set(:build_a) { create(:ci_build, :success, name: 'build-a', stage: 'build', stage_idx: 0, pipeline: pipeline) }
+  set(:build_b) { create(:ci_build, :failed, name: 'build-b', stage: 'build', stage_idx: 0, pipeline: pipeline) }
+  set(:test_a) { create(:ci_build, :running, name: 'test-a', stage: 'test', stage_idx: 1, pipeline: pipeline) }
+  set(:test_b) { create(:ci_build, :pending, name: 'test-b', stage: 'test', stage_idx: 1, pipeline: pipeline) }
+  set(:deploy) { create(:ci_build, :created, name: 'deploy', stage: 'deploy', stage_idx: 2, pipeline: pipeline) }
+
+  let(:collection) { described_class.new(pipeline) }
+
+  describe '#set_processable_status' do
+    it 'does update existing status of processable' do
+      collection.set_processable_status(test_a.id, 'success', 100)
+
+      expect(collection.status_for_names(['test-a'])).to eq('success')
+    end
+
+    it 'ignores a missing processable' do
+      collection.set_processable_status(-1, 'failed', 100)
+    end
+  end
+
+  describe '#status_of_all' do
+    it 'returns composite status of the collection' do
+      expect(collection.status_of_all).to eq('running')
+    end
+  end
+
+  describe '#status_for_names' do
+    where(:names, :status) do
+      %w[build-a]         | 'success'
+      %w[build-a build-b] | 'failed'
+      %w[build-a test-a]  | 'running'
+    end
+
+    with_them do
+      it 'returns composite status of given names' do
+        expect(collection.status_for_names(names)).to eq(status)
+      end
+    end
+  end
+
+  describe '#status_for_prior_stage_position' do
+    where(:stage, :status) do
+      0 | 'success'
+      1 | 'failed'
+      2 | 'running'
+    end
+
+    with_them do
+      it 'returns composite status for processables in prior stages' do
+        expect(collection.status_for_prior_stage_position(stage)).to eq(status)
+      end
+    end
+  end
+
+  describe '#status_for_stage_position' do
+    where(:stage, :status) do
+      0 | 'failed'
+      1 | 'running'
+      2 | 'created'
+    end
+
+    with_them do
+      it 'returns composite status for processables at a given stages' do
+        expect(collection.status_for_stage_position(stage)).to eq(status)
+      end
+    end
+  end
+
+  describe '#created_processable_ids_for_stage_position' do
+    it 'returns IDs of processables at a given stage position' do
+      expect(collection.created_processable_ids_for_stage_position(0)).to be_empty
+      expect(collection.created_processable_ids_for_stage_position(1)).to be_empty
+      expect(collection.created_processable_ids_for_stage_position(2)).to contain_exactly(deploy.id)
+    end
+  end
+
+  describe '#processing_processables' do
+    it 'returns processables marked as processing' do
+      expect(collection.processing_processables.map { |processable| processable[:id]} )
+        .to contain_exactly(build_a.id, build_b.id, test_a.id, test_b.id, deploy.id)
+    end
+  end
+end

spec/services/ci/pipeline_processing/atomic_processing_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+require_relative 'shared_processing_service.rb'
+
+describe Ci::PipelineProcessing::AtomicProcessingService do
+  before do
+    stub_feature_flags(ci_atomic_processing: true)
+  end
+
+  it_behaves_like 'Pipeline Processing Service'
+end

spec/services/ci/pipeline_processing/legacy_processing_service_spec.rb

@@ -4,5 +4,9 @@ require 'spec_helper'
 require_relative 'shared_processing_service.rb'
 
 describe Ci::PipelineProcessing::LegacyProcessingService do
+  before do
+    stub_feature_flags(ci_atomic_processing: false)
+  end
+
   it_behaves_like 'Pipeline Processing Service'
 end

spec/services/ci/pipeline_processing/shared_processing_service.rb

@@ -879,19 +879,27 @@ shared_examples 'Pipeline Processing Service' do
   end
 
   def succeed_pending
-    builds.pending.map(&:success)
+    builds.pending.each do |build|
+      build.reset.success
+    end
   end
 
   def succeed_running_or_pending
-    pipeline.builds.running_or_pending.each(&:success)
+    pipeline.builds.running_or_pending.each do |build|
+      build.reset.success
+    end
   end
 
   def fail_running_or_pending
-    pipeline.builds.running_or_pending.each(&:drop)
+    pipeline.builds.running_or_pending.each do |build|
+      build.reset.drop
+    end
   end
 
   def cancel_running_or_pending
-    pipeline.builds.running_or_pending.each(&:cancel)
+    pipeline.builds.running_or_pending.each do |build|
+      build.reset.cancel
+    end
   end
 
   def play_manual_action(name)
@@ -911,11 +919,15 @@ shared_examples 'Pipeline Processing Service' do
   end
 
   def create_build(name, **opts)
-    create(:ci_build, :created, pipeline: pipeline, name: name, **opts)
+    create(:ci_build, :created, pipeline: pipeline, name: name, **with_stage_opts(opts))
   end
 
   def successful_build(name, **opts)
-    create(:ci_build, :success, pipeline: pipeline, name: name, **opts)
+    create(:ci_build, :success, pipeline: pipeline, name: name, **with_stage_opts(opts))
+  end
+
+  def with_stage_opts(opts)
+    { stage: "stage-#{opts[:stage_idx].to_i}" }.merge(opts)
   end
 
   def delayed_options

spec/services/ci/retry_build_service_spec.rb

@@ -45,7 +45,8 @@ describe Ci::RetryBuildService do
        user_id auto_canceled_by_id retried failure_reason
        sourced_pipelines artifacts_file_store artifacts_metadata_store
        metadata runner_session trace_chunks upstream_pipeline_id
-       artifacts_file artifacts_metadata artifacts_size commands resource resource_group_id].freeze
+       artifacts_file artifacts_metadata artifacts_size commands
+       resource resource_group_id processed].freeze
 
   shared_examples 'build duplication' do
     let(:another_pipeline) { create(:ci_empty_pipeline, project: project) }
@@ -202,12 +203,13 @@ describe Ci::RetryBuildService do
 
       it 'does not enqueue the new build' do
         expect(new_build).to be_created
+        expect(new_build).not_to be_processed
       end
 
-      it 'does mark old build as retried in the database and on the instance' do
+      it 'does mark old build as retried' do
         expect(new_build).to be_latest
         expect(build).to be_retried
-        expect(build.reload).to be_retried
+        expect(build).to be_processed
       end
 
       context 'when build with deployment is retried' do

spec/services/ci/retry_pipeline_service_spec.rb

@@ -330,7 +330,7 @@ describe Ci::RetryPipelineService, '#execute' do
                       stage: "stage_#{stage_num}",
                       stage_idx: stage_num,
                       pipeline: pipeline, **opts) do |build|
-      pipeline.update_status
+      pipeline.update_legacy_status
     end
   end
 end

spec/workers/pipeline_update_worker_spec.rb

@@ -8,7 +8,7 @@ describe PipelineUpdateWorker do
       let(:pipeline) { create(:ci_pipeline) }
 
       it 'updates pipeline status' do
-        expect_any_instance_of(Ci::Pipeline).to receive(:update_status)
+        expect_any_instance_of(Ci::Pipeline).to receive(:set_status).with('skipped')
 
         described_class.new.perform(pipeline.id)
       end

spec/workers/stage_update_worker_spec.rb

@@ -8,7 +8,7 @@ describe StageUpdateWorker do
       let(:stage) { create(:ci_stage_entity) }
 
       it 'updates stage status' do
-        expect_any_instance_of(Ci::Stage).to receive(:update_status)
+        expect_any_instance_of(Ci::Stage).to receive(:set_status).with('skipped')
 
         described_class.new.perform(stage.id)
       end