Make PipelineProcessWorker idempotent with until_executing deduplication

This worker does nothing but calling
Ci::PipelineProcessing::AtomicProcessingService.
And this service is an idempotent service.
It does nothing unless the pipeline needs_processing?

This change is behind a FF ci_idempotent_pipeline_process_worker

app/workers/all_queues.yml

@@ -1620,7 +1620,7 @@
   :urgency: :high
   :resource_boundary: :unknown
   :weight: 5
-  :idempotent:
+  :idempotent: true
   :tags: []
 - :name: pipeline_processing:stage_update
   :worker_name: StageUpdateWorker

app/workers/pipeline_process_worker.rb

 # frozen_string_literal: true
 
-class PipelineProcessWorker # rubocop:disable Scalability/IdempotentWorker
+class PipelineProcessWorker
   include ApplicationWorker
 
   sidekiq_options retry: 3
@@ -10,7 +10,9 @@ class PipelineProcessWorker # rubocop:disable Scalability/IdempotentWorker
   feature_category :continuous_integration
   urgency :high
   loggable_arguments 1
-  data_consistency :delayed, feature_flag: :load_balancing_for_pipeline_process_worker
+
+  idempotent!
+  deduplicate :until_executing, feature_flag: :ci_idempotent_pipeline_process_worker
 
   # rubocop: disable CodeReuse/ActiveRecord
   def perform(pipeline_id)

config/feature_flags/development/load_balancing_for_pipeline_process_worker.yml → config/feature_flags/development/ci_idempotent_pipeline_process_worker.yml

 ---
-name: load_balancing_for_pipeline_process_worker
-introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/61766
-rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/330960
-milestone: '13.12'
+name: ci_idempotent_pipeline_process_worker
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/62410
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/332963
+milestone: '14.0'
 type: development
-group: group::pipeline execution
+group: group::pipeline authoring
 default_enabled: false

spec/workers/pipeline_process_worker_spec.rb

@@ -3,10 +3,44 @@
 require 'spec_helper'
 
 RSpec.describe PipelineProcessWorker do
+  let_it_be(:pipeline) { create(:ci_pipeline) }
+
+  include_examples 'an idempotent worker' do
+    let(:pipeline) { create(:ci_pipeline, :created) }
+    let(:job_args) { [pipeline.id] }
+
+    before do
+      create(:ci_build, :created, pipeline: pipeline)
+    end
+
+    it 'processes the pipeline' do
+      expect(pipeline.status).to eq('created')
+      expect(pipeline.processables.pluck(:status)).to contain_exactly('created')
+
+      subject
+
+      expect(pipeline.reload.status).to eq('pending')
+      expect(pipeline.processables.pluck(:status)).to contain_exactly('pending')
+
+      subject
+
+      expect(pipeline.reload.status).to eq('pending')
+      expect(pipeline.processables.pluck(:status)).to contain_exactly('pending')
+    end
+  end
+
+  context 'when the FF ci_idempotent_pipeline_process_worker is disabled' do
+    before do
+      stub_feature_flags(ci_idempotent_pipeline_process_worker: false)
+    end
+
+    it 'is not deduplicated' do
+      expect(described_class).not_to be_deduplication_enabled
+    end
+  end
+
   describe '#perform' do
     context 'when pipeline exists' do
-      let(:pipeline) { create(:ci_pipeline) }
-
       it 'processes pipeline' do
         expect_any_instance_of(Ci::ProcessPipelineService).to receive(:execute)
 
@@ -16,14 +50,9 @@ RSpec.describe PipelineProcessWorker do
 
     context 'when pipeline does not exist' do
       it 'does not raise exception' do
-        expect { described_class.new.perform(123) }
+        expect { described_class.new.perform(non_existing_record_id) }
           .not_to raise_error
       end
     end
-
-    it_behaves_like 'worker with data consistency',
-                described_class,
-                feature_flag: :load_balancing_for_pipeline_process_worker,
-                data_consistency: :delayed
   end
 end