Add locking to pipeline add job service

It prevents the race condition when adding a job to a pipeline
in small frequencies.

It's behind a FF ci_pipeline_add_job_with_lock.

app/services/ci/pipelines/add_job_service.rb

@@ -3,21 +3,33 @@
 module Ci
   module Pipelines
     class AddJobService
+      include ::Gitlab::ExclusiveLeaseHelpers
+
       attr_reader :pipeline
 
       def initialize(pipeline)
         @pipeline = pipeline
 
-        raise ArgumentError, "Pipeline must be persisted for this service to be used" unless @pipeline.persisted?
+        raise ArgumentError, "Pipeline must be persisted for this service to be used" unless pipeline.persisted?
       end
 
       def execute!(job, &block)
         assign_pipeline_attributes(job)
 
-        Ci::Pipeline.transaction do
-          yield(job)
+        if Feature.enabled?(:ci_pipeline_add_job_with_lock, pipeline.project, default_enabled: :yaml)
+          in_lock("ci:pipelines:#{pipeline.id}:add-job", ttl: LOCK_TIMEOUT, sleep_sec: LOCK_SLEEP, retries: LOCK_RETRIES) do
+            Ci::Pipeline.transaction do
+              yield(job)
+
+              job.update_older_statuses_retried! if Feature.enabled?(:ci_fix_commit_status_retried, pipeline.project, default_enabled: :yaml)
+            end
+          end
+        else
+          Ci::Pipeline.transaction do
+            yield(job)
 
-          job.update_older_statuses_retried! if Feature.enabled?(:ci_fix_commit_status_retried, @pipeline.project, default_enabled: :yaml)
+            job.update_older_statuses_retried! if Feature.enabled?(:ci_fix_commit_status_retried, pipeline.project, default_enabled: :yaml)
+          end
         end
 
         ServiceResponse.success(payload: { job: job })
@@ -27,10 +39,14 @@ module Ci
 
       private
 
+      LOCK_TIMEOUT = 1.minute
+      LOCK_SLEEP = 0.5.seconds
+      LOCK_RETRIES = 20
+
       def assign_pipeline_attributes(job)
-        job.pipeline = @pipeline
-        job.project = @pipeline.project
-        job.ref = @pipeline.ref
+        job.pipeline = pipeline
+        job.project = pipeline.project
+        job.ref = pipeline.ref
       end
     end
   end

config/feature_flags/development/ci_pipeline_add_job_with_lock.yml

+---
+name: ci_pipeline_add_job_with_lock
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/65754
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/337628
+milestone: '14.2'
+type: development
+group: group::pipeline authoring
+default_enabled: false

spec/services/ci/pipelines/add_job_service_spec.rb

@@ -3,6 +3,8 @@
 require 'spec_helper'
 
 RSpec.describe Ci::Pipelines::AddJobService do
+  include ExclusiveLeaseHelpers
+
   let_it_be(:pipeline) { create(:ci_pipeline) }
 
   let(:job) { build(:ci_build) }
@@ -68,5 +70,38 @@ RSpec.describe Ci::Pipelines::AddJobService do
         execute
       end
     end
+
+    context 'exclusive lock' do
+      let(:lock_uuid) { 'test' }
+      let(:lock_key) { "ci:pipelines:#{pipeline.id}:add-job" }
+      let(:lock_timeout) { 1.minute }
+
+      before do
+        # "Please stub a default value first if message might be received with other args as well."
+        allow(Gitlab::ExclusiveLease).to receive(:new).and_call_original
+      end
+
+      it 'uses exclusive lock' do
+        lease = stub_exclusive_lease(lock_key, lock_uuid, timeout: lock_timeout)
+        expect(lease).to receive(:try_obtain)
+        expect(lease).to receive(:cancel)
+
+        expect(execute).to be_success
+        expect(execute.payload[:job]).to eq(job)
+      end
+
+      context 'when the FF ci_pipeline_add_job_with_lock is disabled' do
+        before do
+          stub_feature_flags(ci_pipeline_add_job_with_lock: false)
+        end
+
+        it 'does not use exclusive lock' do
+          expect(Gitlab::ExclusiveLease).not_to receive(:new).with(lock_key, timeout: lock_timeout)
+
+          expect(execute).to be_success
+          expect(execute.payload[:job]).to eq(job)
+        end
+      end
+    end
   end
 end