Define deduplication strategy in Resource Group Worker

Currently, the AssignResourceFromResourceGroupWorker
doesn't have any deduplication strategies, so that
multiple jobs with the same argument could be scheduled
simulteniously, which is suspected as a cause of race condition.
This commit adds `until_executed` deduplication strategy
to prevent multiple jobs from running on the same
resource group.

Changelog: fixed

app/workers/all_queues.yml

@@ -1619,7 +1619,7 @@
   :urgency: :low
   :resource_boundary: :unknown
   :weight: 5
-  :idempotent:
+  :idempotent: true
   :tags: []
 - :name: pipeline_processing:pipeline_process
   :worker_name: PipelineProcessWorker

app/workers/ci/resource_groups/assign_resource_from_resource_group_worker.rb

@@ -2,7 +2,10 @@
 
 module Ci
   module ResourceGroups
-    class AssignResourceFromResourceGroupWorker # rubocop:disable Scalability/IdempotentWorker
+    # This worker is to assign a resource to a pipeline job from a resource group
+    # and enqueue the job to be executed by a runner.
+    # See https://docs.gitlab.com/ee/ci/yaml/#resource_group for more information.
+    class AssignResourceFromResourceGroupWorker
       include ApplicationWorker
 
       sidekiq_options retry: 3
@@ -11,6 +14,13 @@ module Ci
       queue_namespace :pipeline_processing
       feature_category :continuous_delivery
 
+      # This worker is idempotent that it produces the same result
+      # as long as the same resource group id is passed as an argument.
+      # Therefore, we can deduplicate the sidekiq jobs until the on-going
+      # assignment process has been finished.
+      idempotent!
+      deduplicate :until_executed
+
       def perform(resource_group_id)
         ::Ci::ResourceGroup.find_by_id(resource_group_id).try do |resource_group|
           Ci::ResourceGroups::AssignResourceFromResourceGroupService.new(resource_group.project, nil)

spec/workers/ci/resource_groups/assign_resource_from_resource_group_worker_spec.rb

@@ -5,15 +5,23 @@ require 'spec_helper'
 RSpec.describe Ci::ResourceGroups::AssignResourceFromResourceGroupWorker do
   let(:worker) { described_class.new }
 
+  it 'has the `until_executed` deduplicate strategy' do
+    expect(described_class.get_deduplicate_strategy).to eq(:until_executed)
+  end
+
   describe '#perform' do
     subject { worker.perform(resource_group_id) }
 
-    context 'when resource group exists' do
     let(:resource_group) { create(:ci_resource_group) }
     let(:resource_group_id) { resource_group.id }
 
+    include_examples 'an idempotent worker' do
+      let(:job_args) { [resource_group_id] }
+    end
+
+    context 'when resource group exists' do
       it 'executes AssignResourceFromResourceGroupService' do
-        expect_next_instance_of(Ci::ResourceGroups::AssignResourceFromResourceGroupService, resource_group.project, nil) do |service|
+        expect_next_instances_of(Ci::ResourceGroups::AssignResourceFromResourceGroupService, 2, resource_group.project, nil) do |service|
           expect(service).to receive(:execute).with(resource_group)
         end
 
@@ -22,7 +30,7 @@ RSpec.describe Ci::ResourceGroups::AssignResourceFromResourceGroupWorker do
     end
 
     context 'when build does not exist' do
-      let(:resource_group_id) { 123 }
+      let(:resource_group_id) { non_existing_record_id }
 
       it 'does not execute AssignResourceFromResourceGroupService' do
         expect(Ci::ResourceGroups::AssignResourceFromResourceGroupService).not_to receive(:new)