Revert deduplication config and use an exclusive lease

Use until_executing again so that the job can re enqueue itself at the
end of the #perform method.
Use an exclusive lease to disallow the potential situation where 2 jobs
pick up the same image repository

app/workers/container_registry/migration/enqueuer_worker.rb

@@ -6,17 +6,24 @@ module ContainerRegistry
       include ApplicationWorker
       include CronjobQueue # rubocop:disable Scalability/CronWorkerContext
       include Gitlab::Utils::StrongMemoize
+      include ExclusiveLeaseGuard
+
+      DEFAULT_LEASE_TIMEOUT = 1.hour.to_i.freeze
 
       data_consistency :always
       feature_category :container_registry
       urgency :low
-      deduplicate :until_executed, including_scheduled: true
+      deduplicate :until_executing, including_scheduled: true
       idempotent!
 
       def perform
-        return unless runnable?
+        re_enqueue = false
+        try_obtain_lease do
+          break unless runnable?
 
-        re_enqueue_if_capacity if handle_aborted_migration || handle_next_migration
+          re_enqueue = handle_aborted_migration || handle_next_migration
+        end
+        re_enqueue_if_capacity if re_enqueue
       end
 
       private
@@ -143,6 +150,16 @@ module ContainerRegistry
         log_extra_metadata_on_done(:container_repository_id, repository&.id)
         log_extra_metadata_on_done(:container_repository_path, repository&.path)
       end
+
+      # used by ExclusiveLeaseGuard
+      def lease_key
+        'container_registry:migration:enqueuer_worker'
+      end
+
+      # used by ExclusiveLeaseGuard
+      def lease_timeout
+        DEFAULT_LEASE_TIMEOUT
+      end
     end
   end
 end

spec/workers/container_registry/migration/enqueuer_worker_spec.rb

@@ -2,7 +2,9 @@
 
 require 'spec_helper'
 
-RSpec.describe ContainerRegistry::Migration::EnqueuerWorker, :aggregate_failures do
+RSpec.describe ContainerRegistry::Migration::EnqueuerWorker, :aggregate_failures, :clean_gitlab_redis_shared_state do
+  include ExclusiveLeaseHelpers
+
   let_it_be_with_reload(:container_repository) { create(:container_repository, created_at: 2.days.ago) }
 
   let(:worker) { described_class.new }
@@ -224,6 +226,21 @@ RSpec.describe ContainerRegistry::Migration::EnqueuerWorker, :aggregate_failures
       end
     end
 
+    context 'with the exclusive lease taken' do
+      let(:lease_key) { worker.send(:lease_key) }
+
+      before do
+        stub_exclusive_lease_taken(lease_key, timeout: 1.hour)
+      end
+
+      it 'does not perform' do
+        expect(worker).not_to receive(:runnable?)
+        expect(worker).not_to receive(:re_enqueue_if_capacity)
+
+        subject
+      end
+    end
+
     def expect_log_extra_metadata(metadata)
       metadata.each do |key, value|
         expect(worker).to receive(:log_extra_metadata_on_done).with(key, value)