Fix the order of subsequent jobs when requeuing a job

When we requeue a job, we need to process subsequent skipped jobs,
moreover, we need to do this in an order. Previously, we had an order
by stage_idx but after introducing same-stage jobs, this disappeared.

This is behind a FF ci_order_subsequent_jobs_by_stage.

app/services/ci/after_requeue_job_service.rb

@@ -3,39 +3,50 @@
 module Ci
   class AfterRequeueJobService < ::BaseService
     def execute(processable)
-      process_subsequent_jobs(processable)
-      reset_source_bridge(processable)
+      @processable = processable
+
+      process_subsequent_jobs
+      reset_source_bridge
     end
 
     private
 
-    def process_subsequent_jobs(processable)
-      (stage_dependent_jobs(processable) | needs_dependent_jobs(processable))
-      .each do |processable|
-        process(processable)
+    def process_subsequent_jobs
+      dependent_jobs.each do |job|
+        process(job)
       end
     end
 
-    def reset_source_bridge(processable)
-      processable.pipeline.reset_source_bridge!(current_user)
+    def reset_source_bridge
+      @processable.pipeline.reset_source_bridge!(current_user)
+    end
+
+    def dependent_jobs
+      if ::Feature.enabled?(:ci_order_subsequent_jobs_by_stage, @processable.pipeline.project, default_enabled: :yaml)
+        stage_dependent_jobs
+          .or(needs_dependent_jobs.except(:preload))
+          .ordered_by_stage
+      else
+        stage_dependent_jobs | needs_dependent_jobs
+      end
     end
 
-    def process(processable)
-      Gitlab::OptimisticLocking.retry_lock(processable, name: 'ci_requeue_job') do |processable|
-        processable.process(current_user)
+    def process(job)
+      Gitlab::OptimisticLocking.retry_lock(job, name: 'ci_requeue_job') do |job|
+        job.process(current_user)
       end
     end
 
-    def skipped_jobs(processable)
-      processable.pipeline.processables.skipped
+    def stage_dependent_jobs
+      skipped_jobs.after_stage(@processable.stage_idx)
     end
 
-    def stage_dependent_jobs(processable)
-      skipped_jobs(processable).after_stage(processable.stage_idx)
+    def needs_dependent_jobs
+      skipped_jobs.scheduling_type_dag.with_needs([@processable.name])
     end
 
-    def needs_dependent_jobs(processable)
-      skipped_jobs(processable).scheduling_type_dag.with_needs([processable.name])
+    def skipped_jobs
+      @skipped_jobs ||= @processable.pipeline.processables.skipped
     end
   end
 end

config/feature_flags/development/ci_order_subsequent_jobs_by_stage.yml

+---
+name: ci_order_subsequent_jobs_by_stage
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/77528
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/349977
+milestone: '14.7'
+type: development
+group: group::pipeline authoring
+default_enabled: false

spec/factories/ci/builds.rb

@@ -5,6 +5,7 @@ FactoryBot.define do
     name { 'test' }
     add_attribute(:protected) { false }
     created_at { 'Di 29. Okt 09:50:00 CET 2013' }
+    scheduling_type { 'stage' }
     pending
 
     options do
@@ -33,6 +34,8 @@ FactoryBot.define do
     end
 
     trait :dependent do
+      scheduling_type { 'dag' }
+
       transient do
         sequence(:needed_name) { |n| "dependency #{n}" }
         needed { association(:ci_build, name: needed_name, pipeline: pipeline) }

spec/services/ci/after_requeue_job_service_spec.rb

@@ -8,46 +8,56 @@ RSpec.describe Ci::AfterRequeueJobService do
 
   let(:pipeline) { create(:ci_pipeline, project: project) }
 
-  let!(:build) { create(:ci_build, pipeline: pipeline, stage_idx: 0, name: 'build') }
-  let!(:test1) { create(:ci_build, :success, pipeline: pipeline, stage_idx: 1) }
-  let!(:test2) { create(:ci_build, :skipped, pipeline: pipeline, stage_idx: 1) }
-  let!(:test3) { create(:ci_build, :skipped, :dependent, pipeline: pipeline, stage_idx: 1, needed: build) }
-  let!(:deploy) { create(:ci_build, :skipped, :dependent, pipeline: pipeline, stage_idx: 2, needed: test3) }
-
-  subject(:execute_service) { described_class.new(project, user).execute(build) }
-
-  it 'marks subsequent skipped jobs as processable' do
-    expect(test1.reload).to be_success
-    expect(test2.reload).to be_skipped
-    expect(test3.reload).to be_skipped
-    expect(deploy.reload).to be_skipped
-
-    execute_service
-
-    expect(test1.reload).to be_success
-    expect(test2.reload).to be_created
-    expect(test3.reload).to be_created
-    expect(deploy.reload).to be_created
+  let!(:build1) { create(:ci_build, name: 'build1', pipeline: pipeline, stage_idx: 0) }
+  let!(:test1) { create(:ci_build, :success, name: 'test1', pipeline: pipeline, stage_idx: 1) }
+  let!(:test2) { create(:ci_build, :skipped, name: 'test2', pipeline: pipeline, stage_idx: 1) }
+  let!(:test3) { create(:ci_build, :skipped, :dependent, name: 'test3', pipeline: pipeline, stage_idx: 1, needed: build1) }
+  let!(:deploy) { create(:ci_build, :skipped, :dependent, name: 'deploy', pipeline: pipeline, stage_idx: 2, needed: test3) }
+
+  subject(:execute_service) { described_class.new(project, user).execute(build1) }
+
+  shared_examples 'processing subsequent skipped jobs' do
+    it 'marks subsequent skipped jobs as processable' do
+      expect(test1.reload).to be_success
+      expect(test2.reload).to be_skipped
+      expect(test3.reload).to be_skipped
+      expect(deploy.reload).to be_skipped
+
+      execute_service
+
+      expect(test1.reload).to be_success
+      expect(test2.reload).to be_created
+      expect(test3.reload).to be_created
+      expect(deploy.reload).to be_created
+    end
   end
 
+  it_behaves_like 'processing subsequent skipped jobs'
+
   context 'when there is a job need from the same stage' do
-    let!(:test4) do
+    let!(:build2) do
       create(:ci_build,
              :skipped,
              :dependent,
+             name: 'build2',
              pipeline: pipeline,
              stage_idx: 0,
              scheduling_type: :dag,
-             needed: build)
+             needed: build1)
     end
 
-    it 'marks subsequent skipped jobs as processable' do
-      expect { execute_service }.to change { test4.reload.status }.from('skipped').to('created')
+    shared_examples 'processing the same stage job' do
+      it 'marks subsequent skipped jobs as processable' do
+        expect { execute_service }.to change { build2.reload.status }.from('skipped').to('created')
+      end
     end
+
+    it_behaves_like 'processing subsequent skipped jobs'
+    it_behaves_like 'processing the same stage job'
   end
 
   context 'when the pipeline is a downstream pipeline and the bridge is depended' do
-    let!(:trigger_job) { create(:ci_bridge, :strategy_depend, status: 'success') }
+    let!(:trigger_job) { create(:ci_bridge, :strategy_depend, name: 'trigger_job', status: 'success') }
 
     before do
       create(:ci_sources_pipeline, pipeline: pipeline, source_job: trigger_job)

spec/services/ci/pipeline_processing/atomic_processing_service_spec.rb

@@ -1004,6 +1004,63 @@ RSpec.describe Ci::PipelineProcessing::AtomicProcessingService do
       end
     end
 
+    context 'when the dependency is stage-independent', :sidekiq_inline do
+      let(:config) do
+        <<-EOY
+        stages: [A, B]
+
+        A1:
+          stage: A
+          script: exit 0
+          when: manual
+
+        A2:
+          stage: A
+          script: exit 0
+          needs: [A1]
+
+        B:
+          stage: B
+          needs: [A2]
+          script: exit 0
+        EOY
+      end
+
+      let(:pipeline) do
+        Ci::CreatePipelineService.new(project, user, { ref: 'master' }).execute(:push).payload
+      end
+
+      before do
+        stub_ci_pipeline_yaml_file(config)
+      end
+
+      it 'processes subsequent jobs in the correct order when playing first job' do
+        expect(all_builds_names).to eq(%w[A1 A2 B])
+        expect(all_builds_statuses).to eq(%w[manual skipped skipped])
+
+        play_manual_action('A1')
+
+        expect(all_builds_names).to eq(%w[A1 A2 B])
+        expect(all_builds_statuses).to eq(%w[pending created created])
+      end
+
+      context 'when the FF ci_order_subsequent_jobs_by_stage is disabled' do
+        before do
+          stub_feature_flags(ci_order_subsequent_jobs_by_stage: false)
+        end
+
+        it 'processes subsequent jobs in an incorrect order when playing first job' do
+          expect(all_builds_names).to eq(%w[A1 A2 B])
+          expect(all_builds_statuses).to eq(%w[manual skipped skipped])
+
+          play_manual_action('A1')
+
+          expect(all_builds_names).to eq(%w[A1 A2 B])
+          expect(all_builds_statuses).to eq(%w[pending created skipped])
+        end
+      end
+    end
+
     private
 
     def all_builds