Add the logic of marking subsequent jobs as processable when playing job

app/services/ci/after_requeue_job_service.rb

+# frozen_string_literal: true
+
+module Ci
+  class AfterRequeueJobService < ::BaseService
+    def execute(processable)
+      process_subsequent_jobs(processable)
+      reset_ancestor_bridges(processable)
+    end
+
+    private
+
+    def process_subsequent_jobs(processable)
+      processable.pipeline.processables.skipped.after_stage(processable.stage_idx).find_each do |processable|
+        process(processable)
+      end
+    end
+
+    def reset_ancestor_bridges(processable)
+      processable.pipeline.reset_ancestor_bridges!
+    end
+
+    def process(processable)
+      Gitlab::OptimisticLocking.retry_lock(processable, name: 'ci_requeue_job') do |processable|
+        processable.process(current_user)
+      end
+    end
+  end
+end

app/services/ci/play_bridge_service.rb

@@ -8,6 +8,10 @@ module Ci
       bridge.tap do |bridge|
         bridge.user = current_user
         bridge.enqueue!
+
+        next unless ::Feature.enabled?(:ci_fix_pipeline_status_for_dag_needs_manual, project, default_enabled: :yaml)
+
+        AfterRequeueJobService.new(project, current_user).execute(bridge)
       end
     end
   end

app/services/ci/play_build_service.rb

@@ -12,7 +12,13 @@ module Ci
       # Try to enqueue the build, otherwise create a duplicate.
       #
       if build.enqueue
-        build.tap { |action| action.update(user: current_user, job_variables_attributes: job_variables_attributes || []) }
+        build.tap do |build|
+          build.update(user: current_user, job_variables_attributes: job_variables_attributes || [])
+
+          next unless ::Feature.enabled?(:ci_fix_pipeline_status_for_dag_needs_manual, project, default_enabled: :yaml)
+
+          AfterRequeueJobService.new(project, current_user).execute(build)
+        end
       else
         Ci::Build.retry(build, current_user)
       end

app/services/ci/retry_build_service.rb

@@ -2,8 +2,6 @@
 
 module Ci
   class RetryBuildService < ::BaseService
-    include Gitlab::OptimisticLocking
-
     def self.clone_accessors
       %i[pipeline project ref tag options name
          allow_failure stage stage_id stage_idx trigger_request
@@ -16,10 +14,8 @@ module Ci
       build.ensure_scheduling_type!
 
       reprocess!(build).tap do |new_build|
-        mark_subsequent_stages_as_processable(build)
-        build.pipeline.reset_ancestor_bridges!
-
         Gitlab::OptimisticLocking.retry_lock(new_build, name: 'retry_build', &:enqueue)
+        AfterRequeueJobService.new(project, current_user).execute(build)
 
         ::MergeRequests::AddTodoWhenBuildFailsService
           .new(project, current_user)
@@ -65,12 +61,6 @@ module Ci
       end
       build
     end
-
-    def mark_subsequent_stages_as_processable(build)
-      build.pipeline.processables.skipped.after_stage(build.stage_idx).find_each do |skipped|
-        retry_optimistic_lock(skipped, name: 'ci_retry_build_mark_subsequent_stages') { |build| build.process(current_user) }
-      end
-    end
   end
 end
 

spec/services/ci/after_requeue_job_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::AfterRequeueJobService do
+  let_it_be(:project) { create(:project) }
+  let_it_be(:user) { project.owner }
+
+  let(:pipeline) { create(:ci_pipeline, project: project) }
+
+  let!(:build) { create(:ci_build, pipeline: pipeline, stage_idx: 0) }
+  let!(:test1) { create(:ci_build, :success, pipeline: pipeline, stage_idx: 1) }
+  let!(:test2) { create(:ci_build, :skipped, pipeline: pipeline, stage_idx: 1) }
+
+  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
+
+    execute_service
+
+    expect(test1.reload).to be_success
+    expect(test2.reload).to be_created
+  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') }
+
+    before do
+      create(:ci_sources_pipeline, pipeline: pipeline, source_job: trigger_job)
+    end
+
+    it 'marks source bridge as pending' do
+      expect { execute_service }.to change { trigger_job.reload.status }.from('success').to('pending')
+    end
+  end
+end

spec/services/ci/pipeline_processing/shared_processing_service_tests_with_yaml.rb

@@ -49,7 +49,13 @@ RSpec.shared_context 'Pipeline Processing Service Tests With Yaml' do
       statuses = pipeline.latest_statuses.by_name(job_names).to_a
       expect(statuses.count).to eq(job_names.count) # ensure that we have the same counts
 
-      statuses.each { |status| status.public_send("#{event}!") }
+      statuses.each do |status|
+        if event == 'play'
+          status.play(user)
+        else
+          status.public_send("#{event}!")
+        end
+      end
     end
   end
 end

spec/services/ci/pipeline_processing/test_cases/dag_build_test_two_manual_review_test_staging_production.yml

+config:
+  stages: [build, test, review, deploy, post_deploy]
+
+  build:
+    stage: build
+    script: exit 0
+
+  test:
+    stage: test
+    script: exit 0
+
+  release_test1:
+    stage: test
+    when: manual
+    script: exit 0
+
+  release_test2:
+    stage: test
+    when: manual
+    script: exit 0
+
+  review:
+    stage: review
+    script: exit 0
+    needs: [test, release_test1, release_test2]
+
+  staging:
+    stage: deploy
+    script: exit 0
+    needs: [release_test1]
+
+  production:
+    stage: deploy
+    script: exit 0
+    needs: [release_test2]
+
+  after_deploy:
+    stage: post_deploy
+    script: exit 0
+    needs: [production]
+
+  handle_failure:
+    stage: post_deploy
+    when: on_failure
+    script: exit 0
+
+init:
+  expect:
+    pipeline: pending
+    stages:
+      build: pending
+      test: created
+      review: created
+      deploy: created
+      post_deploy: created
+    jobs:
+      build: pending
+      test: created
+      release_test1: created
+      release_test2: created
+      review: created
+      staging: created
+      production: created
+      after_deploy: created
+      handle_failure: created
+
+transitions:
+  - event: success
+    jobs: [build]
+    expect:
+      pipeline: running
+      stages:
+        build: success
+        test: pending
+        review: skipped
+        deploy: skipped
+        post_deploy: pending
+      jobs:
+        build: success
+        test: pending
+        release_test1: manual
+        release_test2: manual
+        review: skipped
+        staging: skipped
+        production: skipped
+        after_deploy: skipped
+        handle_failure: created
+
+  - event: success
+    jobs: [test]
+    expect:
+      pipeline: success
+      stages:
+        build: success
+        test: success
+        review: skipped
+        deploy: skipped
+        post_deploy: skipped
+      jobs:
+        build: success
+        test: success
+        release_test1: manual
+        release_test2: manual
+        review: skipped
+        staging: skipped
+        production: skipped
+        after_deploy: skipped
+        handle_failure: skipped
+
+  - event: play
+    jobs: [release_test1]
+    expect:
+      pipeline: running
+      stages:
+        build: success
+        test: running
+        review: skipped
+        deploy: pending
+        post_deploy: pending
+      jobs:
+        build: success
+        test: success
+        release_test1: pending
+        release_test2: manual
+        review: skipped
+        staging: created
+        production: skipped
+        after_deploy: skipped
+        handle_failure: created
+
+  - event: success
+    jobs: [release_test1]
+    expect:
+      pipeline: running
+      stages:
+        build: success
+        test: success
+        review: skipped
+        deploy: pending
+        post_deploy: pending
+      jobs:
+        build: success
+        test: success
+        release_test1: success
+        release_test2: manual
+        review: skipped
+        staging: pending
+        production: skipped
+        after_deploy: skipped
+        handle_failure: created
+
+  - event: success
+    jobs: [staging]
+    expect:
+      pipeline: success
+      stages:
+        build: success
+        test: success
+        review: skipped
+        deploy: success
+        post_deploy: skipped
+      jobs:
+        build: success
+        test: success
+        release_test1: success
+        release_test2: manual
+        review: skipped
+        staging: success
+        production: skipped
+        after_deploy: skipped
+        handle_failure: skipped

spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_false.yml

@@ -21,7 +21,7 @@ init:
       deploy: created
 
 transitions:
-  - event: enqueue
+  - event: play
     jobs: [test]
     expect:
       pipeline: pending

spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_false_deploy_always.yml

@@ -22,7 +22,7 @@ init:
       deploy: created
 
 transitions:
-  - event: enqueue
+  - event: play
     jobs: [test]
     expect:
       pipeline: pending

spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_false_deploy_on_failure.yml

@@ -22,7 +22,7 @@ init:
       deploy: created
 
 transitions:
-  - event: enqueue
+  - event: play
     jobs: [test]
     expect:
       pipeline: pending

spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_deploy_on_failure.yml

@@ -22,7 +22,7 @@ init:
       deploy: skipped
 
 transitions:
-  - event: enqueue
+  - event: play
     jobs: [test]
     expect:
       pipeline: pending
@@ -31,7 +31,7 @@ transitions:
         deploy: skipped
       jobs:
         test: pending
-        deploy: skipped
+        deploy: created
 
   - event: drop
     jobs: [test]

spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true.yml → spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_fails.yml

@@ -21,7 +21,7 @@ init:
       deploy: skipped
 
 transitions:
-  - event: enqueue
+  - event: play
     jobs: [test]
     expect:
       pipeline: pending
@@ -30,7 +30,7 @@ transitions:
         deploy: skipped
       jobs:
         test: pending
-        deploy: skipped
+        deploy: created
 
   - event: run
     jobs: [test]
@@ -41,15 +41,26 @@ transitions:
         deploy: skipped
       jobs:
         test: running
-        deploy: skipped
+        deploy: created
 
   - event: drop
     jobs: [test]
+    expect:
+      pipeline: running
+      stages:
+        test: success
+        deploy: pending
+      jobs:
+        test: failed
+        deploy: pending
+
+  - event: success
+    jobs: [deploy]
     expect:
       pipeline: success
       stages:
         test: success
-        deploy: skipped
+        deploy: success
       jobs:
         test: failed
-        deploy: skipped
+        deploy: success

spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_succeeds.yml

+config:
+  test:
+    stage: test
+    when: manual
+    allow_failure: true
+    script: exit 0
+
+  deploy:
+    stage: deploy
+    script: exit 0
+    needs: [test]
+
+init:
+  expect:
+    pipeline: skipped
+    stages:
+      test: skipped
+      deploy: skipped
+    jobs:
+      test: manual
+      deploy: skipped
+
+transitions:
+  - event: play
+    jobs: [test]
+    expect:
+      pipeline: pending
+      stages:
+        test: pending
+        deploy: skipped
+      jobs:
+        test: pending
+        deploy: created
+
+  - event: run
+    jobs: [test]
+    expect:
+      pipeline: running
+      stages:
+        test: running
+        deploy: skipped
+      jobs:
+        test: running
+        deploy: created
+
+  - event: success
+    jobs: [test]
+    expect:
+      pipeline: running
+      stages:
+        test: success
+        deploy: pending
+      jobs:
+        test: success
+        deploy: pending
+
+  - event: success
+    jobs: [deploy]
+    expect:
+      pipeline: success
+      stages:
+        test: success
+        deploy: success
+      jobs:
+        test: success
+        deploy: success

spec/services/ci/pipeline_processing/test_cases/stage_test_manual_allow_failure_false.yml

@@ -20,7 +20,7 @@ init:
       deploy: created
 
 transitions:
-  - event: enqueue
+  - event: play
     jobs: [test]
     expect:
       pipeline: pending

spec/services/ci/pipeline_processing/test_cases/stage_test_manual_allow_failure_true.yml

@@ -31,7 +31,7 @@ transitions:
         test: manual
         deploy: success
 
-  - event: enqueue
+  - event: play
     jobs: [test]
     expect:
       pipeline: running

spec/services/ci/pipeline_processing/test_cases/stage_test_manual_allow_failure_true_deploy_on_failure.yml

@@ -21,7 +21,7 @@ init:
       deploy: skipped
 
 transitions:
-  - event: enqueue
+  - event: play
     jobs: [test]
     expect:
       pipeline: pending
@@ -30,7 +30,7 @@ transitions:
         deploy: skipped
       jobs:
         test: pending
-        deploy: skipped
+        deploy: created
 
   - event: drop
     jobs: [test]

spec/services/ci/play_bridge_service_spec.rb

@@ -35,6 +35,28 @@ RSpec.describe Ci::PlayBridgeService, '#execute' do
       expect(bridge.reload.user).to eq(user)
     end
 
+    context 'when a subsequent job is skipped' do
+      let!(:job) { create(:ci_build, :skipped, pipeline: pipeline, stage_idx: bridge.stage_idx + 1) }
+
+      before do
+        create(:ci_build_need, build: job, name: bridge.name)
+      end
+
+      it 'marks the subsequent job as processable' do
+        expect { execute_service }.to change { job.reload.status }.from('skipped').to('created')
+      end
+
+      context 'when the FF ci_fix_pipeline_status_for_dag_needs_manual is disabled' do
+        before do
+          stub_feature_flags(ci_fix_pipeline_status_for_dag_needs_manual: false)
+        end
+
+        it 'does not change the subsequent job' do
+          expect { execute_service }.not_to change { job.reload.status }.from('skipped')
+        end
+      end
+    end
+
     context 'when bridge is not playable' do
       let(:bridge) { create(:ci_bridge, :failed, pipeline: pipeline, downstream: downstream_project) }
 

spec/services/ci/play_build_service_spec.rb

@@ -61,6 +61,28 @@ RSpec.describe Ci::PlayBuildService, '#execute' do
       expect(build.reload.user).to eq user
     end
 
+    context 'when a subsequent job is skipped' do
+      let!(:job) { create(:ci_build, :skipped, pipeline: pipeline, stage_idx: build.stage_idx + 1) }
+
+      before do
+        create(:ci_build_need, build: job, name: build.name)
+      end
+
+      it 'marks the subsequent job as processable' do
+        expect { service.execute(build) }.to change { job.reload.status }.from('skipped').to('created')
+      end
+
+      context 'when the FF ci_fix_pipeline_status_for_dag_needs_manual is disabled' do
+        before do
+          stub_feature_flags(ci_fix_pipeline_status_for_dag_needs_manual: false)
+        end
+
+        it 'does not change the subsequent job' do
+          expect { service.execute(build) }.not_to change { job.reload.status }.from('skipped')
+        end
+      end
+    end
+
     context 'when variables are supplied' do
       let(:job_variables) do
         [{ key: 'first', secret_value: 'first' },