Make ProjectUpdateRepositoryStorageWorker idempotent

Adds specs to ensure storage move state is not changed when run
multiple times.

app/services/projects/update_repository_storage_service.rb

@@ -14,6 +14,8 @@ module Projects
     end
 
     def execute
+      return ServiceResponse.success unless repository_storage_move.scheduled?
+
       repository_storage_move.start!
 
       raise SameFilesystemError if same_filesystem?(repository.storage, destination_storage_name)

app/workers/all_queues.yml

@@ -1530,7 +1530,7 @@
   :urgency: :low
   :resource_boundary: :unknown
   :weight: 1
-  :idempotent: 
+  :idempotent: true
   :tags: []
 - :name: prometheus_create_default_alerts
   :feature_category: :incident_management

app/workers/project_update_repository_storage_worker.rb

 # frozen_string_literal: true
 
-class ProjectUpdateRepositoryStorageWorker # rubocop:disable Scalability/IdempotentWorker
+class ProjectUpdateRepositoryStorageWorker
   include ApplicationWorker
 
+  idempotent!
   feature_category :gitaly
 
   def perform(project_id, new_repository_storage_key, repository_storage_move_id = nil)

changelogs/unreleased/make_repository_moves_job_idempotent.yml

+---
+title: Make ProjectUpdateRepositoryStorageWorker idempotent
+merge_request: 35483
+author:
+type: fixed

spec/factories/project_repository_storage_moves.rb

@@ -14,5 +14,13 @@ FactoryBot.define do
     trait :started do
       state { ProjectRepositoryStorageMove.state_machines[:state].states[:started].value }
     end
+
+    trait :finished do
+      state { ProjectRepositoryStorageMove.state_machines[:state].states[:finished].value }
+    end
+
+    trait :failed do
+      state { ProjectRepositoryStorageMove.state_machines[:state].states[:failed].value }
+    end
   end
 end

spec/services/projects/update_repository_storage_service_spec.rb

@@ -121,6 +121,30 @@ RSpec.describe Projects::UpdateRepositoryStorageService do
           expect(project.reload_pool_repository).to be_nil
         end
       end
+
+      context 'when the repository move is finished' do
+        let(:repository_storage_move) { create(:project_repository_storage_move, :finished, project: project, destination_storage_name: destination) }
+
+        it 'is idempotent' do
+          expect do
+            result = subject.execute
+
+            expect(result).to be_success
+          end.not_to change(repository_storage_move, :state)
+        end
+      end
+
+      context 'when the repository move is failed' do
+        let(:repository_storage_move) { create(:project_repository_storage_move, :failed, project: project, destination_storage_name: destination) }
+
+        it 'is idempotent' do
+          expect do
+            result = subject.execute
+
+            expect(result).to be_success
+          end.not_to change(repository_storage_move, :state)
+        end
+      end
     end
 
     context 'with wiki repository' do

spec/workers/project_update_repository_storage_worker_spec.rb

 # frozen_string_literal: true
 
 require 'spec_helper'
-require 'securerandom'
 
 RSpec.describe ProjectUpdateRepositoryStorageWorker do
   let(:project) { create(:project, :repository) }