Make ScheduleMergeRequestCleanupRefsWorker idempotent

Also log additional information and reduce the number of jobs
scheduled per minute from 300 to 180.

app/services/merge_requests/cleanup_refs_service.rb

@@ -32,7 +32,10 @@ module MergeRequests
       return error('Failed to cache merge ref sha.') unless cache_merge_ref_sha
 
       delete_refs
-      success if update_schedule
+
+      return error('Failed to update schedule.') unless update_schedule
+
+      success
     end
 
     private

app/workers/all_queues.yml

@@ -385,7 +385,7 @@
   :urgency: :low
   :resource_boundary: :unknown
   :weight: 1
-  :idempotent:
+  :idempotent: true
   :tags: []
 - :name: cronjob:schedule_migrate_external_diffs
   :feature_category: :source_code_management

app/workers/schedule_merge_request_cleanup_refs_worker.rb

 # frozen_string_literal: true
 
-class ScheduleMergeRequestCleanupRefsWorker # rubocop:disable Scalability/IdempotentWorker
+class ScheduleMergeRequestCleanupRefsWorker
   include ApplicationWorker
   include CronjobQueue # rubocop:disable Scalability/CronWorkerContext
 
   feature_category :source_code_management
+  idempotent!
 
-  # Based on existing data, MergeRequestCleanupRefsWorker can run for ~190ms per
-  # job and this is scheduled per minute. This means that 300 jobs can be performed
-  # but since there are some spikes from time time, it's better to give it some
-  # allowance.
-  LIMIT = 300
+  # Based on existing data, MergeRequestCleanupRefsWorker can run 3 jobs per
+  # second. This means that 180 jobs can be performed but since there are some
+  # spikes from time time, it's better to give it some allowance.
+  LIMIT = 180
   DELAY = 10.seconds
-  BATCH_SIZE = 50
+  BATCH_SIZE = 30
 
   def perform
     return if Gitlab::Database.read_only?
@@ -20,5 +20,7 @@ class ScheduleMergeRequestCleanupRefsWorker # rubocop:disable Scalability/Idempo
     ids = MergeRequest::CleanupSchedule.scheduled_merge_request_ids(LIMIT).map { |id| [id] }
 
     MergeRequestCleanupRefsWorker.bulk_perform_in(DELAY, ids, batch_size: BATCH_SIZE) # rubocop:disable Scalability/BulkPerformWithContext
+
+    log_extra_metadata_on_done(:merge_requests_count, ids.size)
   end
 end

spec/services/merge_requests/cleanup_refs_service_spec.rb

@@ -91,6 +91,23 @@ RSpec.describe MergeRequests::CleanupRefsService do
         it_behaves_like 'service that does not clean up merge request refs'
       end
 
+      context 'when cleanup schedule fails to update' do
+        before do
+          allow(merge_request.cleanup_schedule).to receive(:update).and_return(false)
+        end
+
+        it 'creates keep around ref and deletes merge request refs' do
+          old_ref_head = ref_head
+
+          aggregate_failures do
+            expect(result[:status]).to eq(:error)
+            expect(kept_around?(old_ref_head)).to be_truthy
+            expect(ref_head).to be_nil
+            expect(merge_request.cleanup_schedule.completed_at).not_to be_present
+          end
+        end
+      end
+
       context 'when merge request is not scheduled to be cleaned up yet' do
         before do
           merge_request.cleanup_schedule.update!(scheduled_at: 1.day.from_now)

spec/workers/schedule_merge_request_cleanup_refs_worker_spec.rb

@@ -20,16 +20,20 @@ RSpec.describe ScheduleMergeRequestCleanupRefsWorker do
       worker.perform
     end
 
-    it 'schedules MergeRequestCleanupRefsWorker to be performed by batch' do
-      expect(MergeRequestCleanupRefsWorker)
-        .to receive(:bulk_perform_in)
-        .with(
-          described_class::DELAY,
-          [[1], [2], [3], [4]],
-          batch_size: described_class::BATCH_SIZE
-        )
+    include_examples 'an idempotent worker' do
+      it 'schedules MergeRequestCleanupRefsWorker to be performed by batch' do
+        expect(MergeRequestCleanupRefsWorker)
+          .to receive(:bulk_perform_in)
+          .with(
+            described_class::DELAY,
+            [[1], [2], [3], [4]],
+            batch_size: described_class::BATCH_SIZE
+          )
 
-      worker.perform
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:merge_requests_count, 4)
+
+        worker.perform
+      end
     end
   end
 end