1 files changed, 579 insertions, 159 deletions

diff --git a/spec/workers/container_registry/migration/enqueuer_worker_spec.rb b/spec/workers/container_registry/migration/enqueuer_worker_spec.rb
index 81fa28dc603..a57a9e3b2e8 100644
--- a/spec/workers/container_registry/migration/enqueuer_worker_spec.rb
+++ b/spec/workers/container_registry/migration/enqueuer_worker_spec.rb
@@ -23,273 +23,669 @@ RSpec.describe ContainerRegistry::Migration::EnqueuerWorker, :aggregate_failures
 
     shared_examples 'no action' do
       it 'does not queue or change any repositories' do
+        expect(worker).not_to receive(:handle_next_migration)
+        expect(worker).not_to receive(:handle_aborted_migration)
+
         subject
 
         expect(container_repository.reload).to be_default
       end
     end
 
-    shared_examples 're-enqueuing based on capacity' do |capacity_limit: 4|
-      context 'below capacity' do
+    context 'with container_registry_migration_phase2_enqueuer_loop disabled' do
+      before do
+        stub_feature_flags(container_registry_migration_phase2_enqueuer_loop: false)
+      end
+
+      shared_examples 're-enqueuing based on capacity' do |capacity_limit: 4|
+        context 'below capacity' do
+          before do
+            allow(ContainerRegistry::Migration).to receive(:capacity).and_return(capacity_limit)
+          end
+
+          it 're-enqueues the worker' do
+            expect(described_class).to receive(:perform_async)
+            expect(described_class).to receive(:perform_in).with(7.seconds)
+
+            subject
+          end
+
+          context 'enqueue_twice feature flag disabled' do
+            before do
+              stub_feature_flags(container_registry_migration_phase2_enqueue_twice: false)
+            end
+
+            it 'only enqueues the worker once' do
+              expect(described_class).to receive(:perform_async)
+              expect(described_class).not_to receive(:perform_in)
+
+              subject
+            end
+          end
+        end
+
+        context 'above capacity' do
+          before do
+            allow(ContainerRegistry::Migration).to receive(:capacity).and_return(-1)
+          end
+
+          it 'does not re-enqueue the worker' do
+            expect(described_class).not_to receive(:perform_async)
+            expect(described_class).not_to receive(:perform_in).with(7.seconds)
+
+            subject
+          end
+        end
+      end
+
+      context 'with qualified repository' do
         before do
-          allow(ContainerRegistry::Migration).to receive(:capacity).and_return(capacity_limit)
+          allow_worker(on: :next_repository) do |repository|
+            allow(repository).to receive(:migration_pre_import).and_return(:ok)
+          end
         end
 
-        it 're-enqueues the worker' do
-          expect(described_class).to receive(:perform_async)
+        shared_examples 'starting the next import' do
+          it 'starts the pre-import for the next qualified repository' do
+            expect_log_extra_metadata(
+              import_type: 'next',
+              container_repository_id: container_repository.id,
+              container_repository_path: container_repository.path,
+              container_repository_migration_state: 'pre_importing'
+            )
 
-          subject
+            expect { subject }.to make_queries_matching(/LIMIT 2/)
+
+            expect(container_repository.reload).to be_pre_importing
+          end
+        end
+
+        it_behaves_like 'starting the next import'
+
+        context 'when the new pre-import maxes out the capacity' do
+          before do
+            # set capacity to 10
+            stub_feature_flags(
+              container_registry_migration_phase2_capacity_25: false
+            )
+
+            # Plus 2 created above gives 9 importing repositories
+            create_list(:container_repository, 7, :importing)
+          end
+
+          it 'does not re-enqueue the worker' do
+            expect(described_class).not_to receive(:perform_async)
+            expect(described_class).not_to receive(:perform_in)
+
+            subject
+          end
+        end
+
+        it_behaves_like 're-enqueuing based on capacity'
+
+        context 'max tag count is 0' do
+          before do
+            stub_application_setting(container_registry_import_max_tags_count: 0)
+            # Add 8 tags to the next repository
+            stub_container_registry_tags(
+              repository: container_repository.path, tags: %w(a b c d e f g h), with_manifest: true
+            )
+          end
+
+          it_behaves_like 'starting the next import'
+        end
+      end
+
+      context 'migrations are disabled' do
+        before do
+          allow(ContainerRegistry::Migration).to receive(:enabled?).and_return(false)
+        end
+
+        it_behaves_like 'no action' do
+          before do
+            expect_log_extra_metadata(migration_enabled: false)
+          end
         end
       end
 
       context 'above capacity' do
         before do
-          allow(ContainerRegistry::Migration).to receive(:capacity).and_return(-1)
+          create(:container_repository, :importing)
+          create(:container_repository, :importing)
+          allow(ContainerRegistry::Migration).to receive(:capacity).and_return(1)
+        end
+
+        it_behaves_like 'no action' do
+          before do
+            expect_log_extra_metadata(below_capacity: false, max_capacity_setting: 1)
+          end
         end
 
         it 'does not re-enqueue the worker' do
-          expect(described_class).not_to receive(:perform_async)
+          expect(ContainerRegistry::Migration::EnqueuerWorker).not_to receive(:perform_async)
+          expect(ContainerRegistry::Migration::EnqueuerWorker).not_to receive(:perform_in)
 
           subject
         end
       end
-    end
 
-    context 'with qualified repository' do
-      before do
-        method = worker.method(:next_repository)
-        allow(worker).to receive(:next_repository) do
-          next_qualified_repository = method.call
-          allow(next_qualified_repository).to receive(:migration_pre_import).and_return(:ok)
-          next_qualified_repository
+      context 'too soon before previous completed import step' do
+        where(:state, :timestamp) do
+          :import_done     | :migration_import_done_at
+          :pre_import_done | :migration_pre_import_done_at
+          :import_aborted  | :migration_aborted_at
+          :import_skipped  | :migration_skipped_at
         end
-      end
 
-      it 'starts the pre-import for the next qualified repository' do
-        expect_log_extra_metadata(
-          import_type: 'next',
-          container_repository_id: container_repository.id,
-          container_repository_path: container_repository.path,
-          container_repository_migration_state: 'pre_importing'
-        )
+        with_them do
+          before do
+            allow(ContainerRegistry::Migration).to receive(:enqueue_waiting_time).and_return(45.minutes)
+            create(:container_repository, state, timestamp => 1.minute.ago)
+          end
 
-        subject
+          it_behaves_like 'no action' do
+            before do
+              expect_log_extra_metadata(waiting_time_passed: false, current_waiting_time_setting: 45.minutes)
+            end
+          end
+        end
 
-        expect(container_repository.reload).to be_pre_importing
+        context 'when last completed repository has nil timestamps' do
+          before do
+            allow(ContainerRegistry::Migration).to receive(:enqueue_waiting_time).and_return(45.minutes)
+            create(:container_repository, migration_state: 'import_done')
+          end
+
+          it 'continues to try the next import' do
+            expect { subject }.to change { container_repository.reload.migration_state }
+          end
+        end
       end
 
-      context 'when the new pre-import maxes out the capacity' do
-        before do
-          # set capacity to 10
-          stub_feature_flags(
-            container_registry_migration_phase2_capacity_25: false
-          )
+      context 'when an aborted import is available' do
+        let_it_be(:aborted_repository) { create(:container_repository, :import_aborted) }
 
-          # Plus 2 created above gives 9 importing repositories
-          create_list(:container_repository, 7, :importing)
+        context 'with a successful registry request' do
+          before do
+            allow_worker(on: :next_aborted_repository) do |repository|
+              allow(repository).to receive(:migration_import).and_return(:ok)
+              allow(repository.gitlab_api_client).to receive(:import_status).and_return('import_failed')
+            end
+          end
+
+          it 'retries the import for the aborted repository' do
+            expect_log_extra_metadata(
+              import_type: 'retry',
+              container_repository_id: aborted_repository.id,
+              container_repository_path: aborted_repository.path,
+              container_repository_migration_state: 'importing'
+            )
+
+            subject
+
+            expect(aborted_repository.reload).to be_importing
+            expect(container_repository.reload).to be_default
+          end
+
+          it_behaves_like 're-enqueuing based on capacity'
         end
 
-        it 'does not re-enqueue the worker' do
-          expect(described_class).not_to receive(:perform_async)
+        context 'when an error occurs' do
+          it 'does not abort that migration' do
+            allow_worker(on: :next_aborted_repository) do |repository|
+              allow(repository).to receive(:retry_aborted_migration).and_raise(StandardError)
+            end
 
-          subject
+            expect_log_extra_metadata(
+              import_type: 'retry',
+              container_repository_id: aborted_repository.id,
+              container_repository_path: aborted_repository.path,
+              container_repository_migration_state: 'import_aborted'
+            )
+
+            subject
+
+            expect(aborted_repository.reload).to be_import_aborted
+            expect(container_repository.reload).to be_default
+          end
         end
       end
 
-      it_behaves_like 're-enqueuing based on capacity'
-    end
+      context 'when no repository qualifies' do
+        include_examples 'an idempotent worker' do
+          before do
+            allow(ContainerRepository).to receive(:ready_for_import).and_return(ContainerRepository.none)
+          end
 
-    context 'migrations are disabled' do
-      before do
-        allow(ContainerRegistry::Migration).to receive(:enabled?).and_return(false)
+          it_behaves_like 'no action'
+        end
       end
 
-      it_behaves_like 'no action' do
+      context 'over max tag count' do
         before do
-          expect_log_extra_metadata(migration_enabled: false)
+          stub_application_setting(container_registry_import_max_tags_count: 2)
         end
-      end
-    end
 
-    context 'above capacity' do
-      before do
-        create(:container_repository, :importing)
-        create(:container_repository, :importing)
-        allow(ContainerRegistry::Migration).to receive(:capacity).and_return(1)
+        it 'skips the repository' do
+          expect_log_extra_metadata(
+            import_type: 'next',
+            container_repository_id: container_repository.id,
+            container_repository_path: container_repository.path,
+            container_repository_migration_state: 'import_skipped',
+            tags_count_too_high: true,
+            max_tags_count_setting: 2
+          )
+
+          subject
+
+          expect(container_repository.reload).to be_import_skipped
+          expect(container_repository.migration_skipped_reason).to eq('too_many_tags')
+          expect(container_repository.migration_skipped_at).not_to be_nil
+        end
+
+        context 're-enqueuing' do
+          before do
+            # skipping will also re-enqueue, so we isolate the capacity behavior here
+            allow_worker(on: :next_repository) do |repository|
+              allow(repository).to receive(:skip_import).and_return(true)
+            end
+          end
+
+          it_behaves_like 're-enqueuing based on capacity', capacity_limit: 3
+        end
       end
 
-      it_behaves_like 'no action' do
+      context 'when an error occurs' do
         before do
-          expect_log_extra_metadata(below_capacity: false, max_capacity_setting: 1)
+          allow(ContainerRegistry::Migration).to receive(:max_tags_count).and_raise(StandardError)
         end
-      end
 
-      it 'does not re-enqueue the worker' do
-        expect(ContainerRegistry::Migration::EnqueuerWorker).not_to receive(:perform_async)
+        it 'aborts the import' do
+          expect_log_extra_metadata(
+            import_type: 'next',
+            container_repository_id: container_repository.id,
+            container_repository_path: container_repository.path,
+            container_repository_migration_state: 'import_aborted'
+          )
+
+          expect(Gitlab::ErrorTracking).to receive(:log_exception).with(
+            instance_of(StandardError),
+            next_repository_id: container_repository.id
+          )
 
-        subject
+          subject
+
+          expect(container_repository.reload).to be_import_aborted
+        end
       end
-    end
 
-    context 'too soon before previous completed import step' do
-      where(:state, :timestamp) do
-        :import_done     | :migration_import_done_at
-        :pre_import_done | :migration_pre_import_done_at
-        :import_aborted  | :migration_aborted_at
-        :import_skipped  | :migration_skipped_at
+      context 'with the exclusive lease taken' do
+        let(:lease_key) { worker.send(:lease_key) }
+
+        before do
+          stub_exclusive_lease_taken(lease_key, timeout: 30.minutes)
+        end
+
+        it 'does not perform' do
+          expect(worker).not_to receive(:runnable?)
+          expect(worker).not_to receive(:re_enqueue_if_capacity)
+
+          subject
+        end
       end
+    end
 
-      with_them do
+    context 'with container_registry_migration_phase2_enqueuer_loop enabled' do
+      context 'migrations are disabled' do
         before do
-          allow(ContainerRegistry::Migration).to receive(:enqueue_waiting_time).and_return(45.minutes)
-          create(:container_repository, state, timestamp => 1.minute.ago)
+          allow(ContainerRegistry::Migration).to receive(:enabled?).and_return(false)
         end
 
         it_behaves_like 'no action' do
           before do
-            expect_log_extra_metadata(waiting_time_passed: false, current_waiting_time_setting: 45.minutes)
+            expect_log_extra_metadata(migration_enabled: false)
           end
         end
       end
 
-      context 'when last completed repository has nil timestamps' do
+      context 'with no repository qualifies' do
+        include_examples 'an idempotent worker' do
+          before do
+            allow(ContainerRepository).to receive(:ready_for_import).and_return(ContainerRepository.none)
+          end
+
+          it_behaves_like 'no action'
+        end
+      end
+
+      context 'when multiple aborted imports are available' do
+        let_it_be(:aborted_repository1) { create(:container_repository, :import_aborted) }
+        let_it_be(:aborted_repository2) { create(:container_repository, :import_aborted) }
+
         before do
-          allow(ContainerRegistry::Migration).to receive(:enqueue_waiting_time).and_return(45.minutes)
-          create(:container_repository, migration_state: 'import_done')
+          container_repository.update!(created_at: 30.seconds.ago)
         end
 
-        it 'continues to try the next import' do
-          expect { subject }.to change { container_repository.reload.migration_state }
+        context 'with successful registry requests' do
+          before do
+            allow_worker(on: :next_aborted_repository) do |repository|
+              allow(repository).to receive(:migration_import).and_return(:ok)
+              allow(repository.gitlab_api_client).to receive(:import_status).and_return('import_failed')
+            end
+          end
+
+          it 'retries the import for the aborted repository' do
+            expect_log_info(
+              [
+                {
+                  import_type: 'retry',
+                  container_repository_id: aborted_repository1.id,
+                  container_repository_path: aborted_repository1.path,
+                  container_repository_migration_state: 'importing'
+                },
+                {
+                  import_type: 'retry',
+                  container_repository_id: aborted_repository2.id,
+                  container_repository_path: aborted_repository2.path,
+                  container_repository_migration_state: 'importing'
+                }
+              ]
+            )
+
+            expect(worker).to receive(:handle_next_migration).and_call_original
+
+            subject
+
+            expect(aborted_repository1.reload).to be_importing
+            expect(aborted_repository2.reload).to be_importing
+          end
+        end
+
+        context 'when an error occurs' do
+          it 'does abort that migration' do
+            allow_worker(on: :next_aborted_repository) do |repository|
+              allow(repository).to receive(:retry_aborted_migration).and_raise(StandardError)
+            end
+
+            expect_log_info(
+              [
+                {
+                  import_type: 'retry',
+                  container_repository_id: aborted_repository1.id,
+                  container_repository_path: aborted_repository1.path,
+                  container_repository_migration_state: 'import_aborted'
+                }
+              ]
+            )
+
+            subject
+
+            expect(aborted_repository1.reload).to be_import_aborted
+            expect(aborted_repository2.reload).to be_import_aborted
+          end
         end
       end
-    end
 
-    context 'when an aborted import is available' do
-      let_it_be(:aborted_repository) { create(:container_repository, :import_aborted) }
+      context 'when multiple qualified repositories are available' do
+        let_it_be(:container_repository2) { create(:container_repository, created_at: 2.days.ago) }
 
-      context 'with a successful registry request' do
         before do
-          method = worker.method(:next_aborted_repository)
-          allow(worker).to receive(:next_aborted_repository) do
-            next_aborted_repository = method.call
-            allow(next_aborted_repository).to receive(:migration_import).and_return(:ok)
-            allow(next_aborted_repository.gitlab_api_client).to receive(:import_status).and_return('import_failed')
-            next_aborted_repository
+          allow_worker(on: :next_repository) do |repository|
+            allow(repository).to receive(:migration_pre_import).and_return(:ok)
           end
-        end
 
-        it 'retries the import for the aborted repository' do
-          expect_log_extra_metadata(
-            import_type: 'retry',
-            container_repository_id: aborted_repository.id,
-            container_repository_path: aborted_repository.path,
-            container_repository_migration_state: 'importing'
+          stub_container_registry_tags(
+            repository: container_repository2.path,
+            tags: %w(tag4 tag5 tag6),
+            with_manifest: true
           )
+        end
 
-          subject
+        shared_examples 'starting all the next imports' do
+          it 'starts the pre-import for the next qualified repositories' do
+            expect_log_info(
+              [
+                {
+                  import_type: 'next',
+                  container_repository_id: container_repository.id,
+                  container_repository_path: container_repository.path,
+                  container_repository_migration_state: 'pre_importing'
+                },
+                {
+                  import_type: 'next',
+                  container_repository_id: container_repository2.id,
+                  container_repository_path: container_repository2.path,
+                  container_repository_migration_state: 'pre_importing'
+                }
+              ]
+            )
+
+            expect(worker).to receive(:handle_next_migration).exactly(3).times.and_call_original
+
+            expect { subject }.to make_queries_matching(/LIMIT 2/)
+
+            expect(container_repository.reload).to be_pre_importing
+            expect(container_repository2.reload).to be_pre_importing
+          end
+        end
 
-          expect(aborted_repository.reload).to be_importing
-          expect(container_repository.reload).to be_default
+        it_behaves_like 'starting all the next imports'
+
+        context 'when the new pre-import maxes out the capacity' do
+          before do
+            # set capacity to 10
+            stub_feature_flags(
+              container_registry_migration_phase2_capacity_25: false
+            )
+
+            # Plus 2 created above gives 9 importing repositories
+            create_list(:container_repository, 7, :importing)
+          end
+
+          it 'starts the pre-import only for one qualified repository' do
+            expect_log_info(
+              [
+                {
+                  import_type: 'next',
+                  container_repository_id: container_repository.id,
+                  container_repository_path: container_repository.path,
+                  container_repository_migration_state: 'pre_importing'
+                }
+              ]
+            )
+
+            subject
+
+            expect(container_repository.reload).to be_pre_importing
+            expect(container_repository2.reload).to be_default
+          end
         end
 
-        it_behaves_like 're-enqueuing based on capacity'
+        context 'max tag count is 0' do
+          before do
+            stub_application_setting(container_registry_import_max_tags_count: 0)
+            # Add 8 tags to the next repository
+            stub_container_registry_tags(
+              repository: container_repository.path, tags: %w(a b c d e f g h), with_manifest: true
+            )
+          end
+
+          it_behaves_like 'starting all the next imports'
+        end
+
+        context 'when the deadline is hit' do
+          it 'does not handle the second qualified repository' do
+            expect(worker).to receive(:loop_deadline).and_return(5.seconds.from_now, 2.seconds.ago)
+            expect(worker).to receive(:handle_next_migration).once.and_call_original
+
+            subject
+
+            expect(container_repository.reload).to be_pre_importing
+            expect(container_repository2.reload).to be_default
+          end
+        end
       end
 
-      context 'when an error occurs' do
-        it 'does not abort that migration' do
-          method = worker.method(:next_aborted_repository)
-          allow(worker).to receive(:next_aborted_repository) do
-            next_aborted_repository = method.call
-            allow(next_aborted_repository).to receive(:retry_aborted_migration).and_raise(StandardError)
-            next_aborted_repository
+      context 'when a mix of aborted imports and qualified repositories are available' do
+        let_it_be(:aborted_repository) { create(:container_repository, :import_aborted) }
+
+        before do
+          allow_worker(on: :next_aborted_repository) do |repository|
+            allow(repository).to receive(:migration_import).and_return(:ok)
+            allow(repository.gitlab_api_client).to receive(:import_status).and_return('import_failed')
           end
 
-          expect_log_extra_metadata(
-            import_type: 'retry',
-            container_repository_id: aborted_repository.id,
-            container_repository_path: aborted_repository.path,
-            container_repository_migration_state: 'import_aborted'
+          allow_worker(on: :next_repository) do |repository|
+            allow(repository).to receive(:migration_pre_import).and_return(:ok)
+          end
+        end
+
+        it 'retries the aborted repository and start the migration on the qualified repository' do
+          expect_log_info(
+            [
+              {
+                import_type: 'retry',
+                container_repository_id: aborted_repository.id,
+                container_repository_path: aborted_repository.path,
+                container_repository_migration_state: 'importing'
+              },
+              {
+                import_type: 'next',
+                container_repository_id: container_repository.id,
+                container_repository_path: container_repository.path,
+                container_repository_migration_state: 'pre_importing'
+              }
+            ]
           )
 
           subject
 
-          expect(aborted_repository.reload).to be_import_aborted
-          expect(container_repository.reload).to be_default
+          expect(aborted_repository.reload).to be_importing
+          expect(container_repository.reload).to be_pre_importing
         end
       end
-    end
 
-    context 'when no repository qualifies' do
-      include_examples 'an idempotent worker' do
+      context 'above capacity' do
         before do
-          allow(ContainerRepository).to receive(:ready_for_import).and_return(ContainerRepository.none)
+          create(:container_repository, :importing)
+          create(:container_repository, :importing)
+          allow(ContainerRegistry::Migration).to receive(:capacity).and_return(1)
         end
 
-        it_behaves_like 'no action'
+        it_behaves_like 'no action' do
+          before do
+            expect_log_extra_metadata(below_capacity: false, max_capacity_setting: 1)
+          end
+        end
       end
-    end
 
-    context 'over max tag count' do
-      before do
-        stub_application_setting(container_registry_import_max_tags_count: 2)
-      end
+      context 'too soon before previous completed import step' do
+        where(:state, :timestamp) do
+          :import_done     | :migration_import_done_at
+          :pre_import_done | :migration_pre_import_done_at
+          :import_aborted  | :migration_aborted_at
+          :import_skipped  | :migration_skipped_at
+        end
 
-      it 'skips the repository' do
-        expect_log_extra_metadata(
-          import_type: 'next',
-          container_repository_id: container_repository.id,
-          container_repository_path: container_repository.path,
-          container_repository_migration_state: 'import_skipped',
-          tags_count_too_high: true,
-          max_tags_count_setting: 2
-        )
+        with_them do
+          before do
+            allow(ContainerRegistry::Migration).to receive(:enqueue_waiting_time).and_return(45.minutes)
+            create(:container_repository, state, timestamp => 1.minute.ago)
+          end
 
-        subject
+          it_behaves_like 'no action' do
+            before do
+              expect_log_extra_metadata(waiting_time_passed: false, current_waiting_time_setting: 45.minutes)
+            end
+          end
+        end
 
-        expect(container_repository.reload).to be_import_skipped
-        expect(container_repository.migration_skipped_reason).to eq('too_many_tags')
-        expect(container_repository.migration_skipped_at).not_to be_nil
+        context 'when last completed repository has nil timestamps' do
+          before do
+            allow(ContainerRegistry::Migration).to receive(:enqueue_waiting_time).and_return(45.minutes)
+            create(:container_repository, migration_state: 'import_done')
+          end
+
+          it 'continues to try the next import' do
+            expect { subject }.to change { container_repository.reload.migration_state }
+          end
+        end
       end
 
-      it_behaves_like 're-enqueuing based on capacity', capacity_limit: 3
-    end
+      context 'over max tag count' do
+        before do
+          stub_application_setting(container_registry_import_max_tags_count: 2)
+        end
 
-    context 'when an error occurs' do
-      before do
-        allow(ContainerRegistry::Migration).to receive(:max_tags_count).and_raise(StandardError)
+        it 'skips the repository' do
+          expect_log_info(
+            [
+              {
+                import_type: 'next',
+                container_repository_id: container_repository.id,
+                container_repository_path: container_repository.path,
+                container_repository_migration_state: 'import_skipped',
+                container_repository_migration_skipped_reason: 'too_many_tags'
+              }
+            ]
+          )
+
+          expect(worker).to receive(:handle_next_migration).twice.and_call_original
+          # skipping the migration will re_enqueue the job
+          expect(described_class).to receive(:enqueue_a_job)
+
+          subject
+
+          expect(container_repository.reload).to be_import_skipped
+          expect(container_repository.migration_skipped_reason).to eq('too_many_tags')
+          expect(container_repository.migration_skipped_at).not_to be_nil
+        end
       end
 
-      it 'aborts the import' do
-        expect_log_extra_metadata(
-          import_type: 'next',
-          container_repository_id: container_repository.id,
-          container_repository_path: container_repository.path,
-          container_repository_migration_state: 'import_aborted'
-        )
+      context 'when an error occurs' do
+        before do
+          allow(ContainerRegistry::Migration).to receive(:max_tags_count).and_raise(StandardError)
+        end
 
-        expect(Gitlab::ErrorTracking).to receive(:log_exception).with(
-          instance_of(StandardError),
-          next_repository_id: container_repository.id
-        )
+        it 'aborts the import' do
+          expect_log_info(
+            [
+              {
+                import_type: 'next',
+                container_repository_id: container_repository.id,
+                container_repository_path: container_repository.path,
+                container_repository_migration_state: 'import_aborted'
+              }
+            ]
+          )
 
-        subject
+          expect(Gitlab::ErrorTracking).to receive(:log_exception).with(
+            instance_of(StandardError),
+            next_repository_id: container_repository.id
+          )
 
-        expect(container_repository.reload).to be_import_aborted
-      end
-    end
+          # aborting the migration will re_enqueue the job
+          expect(described_class).to receive(:enqueue_a_job)
 
-    context 'with the exclusive lease taken' do
-      let(:lease_key) { worker.send(:lease_key) }
+          subject
 
-      before do
-        stub_exclusive_lease_taken(lease_key, timeout: 30.minutes)
+          expect(container_repository.reload).to be_import_aborted
+        end
       end
 
-      it 'does not perform' do
-        expect(worker).not_to receive(:runnable?)
-        expect(worker).not_to receive(:re_enqueue_if_capacity)
+      context 'with the exclusive lease taken' do
+        let(:lease_key) { worker.send(:lease_key) }
 
-        subject
+        before do
+          stub_exclusive_lease_taken(lease_key, timeout: 30.minutes)
+        end
+
+        it 'does not perform' do
+          expect(worker).not_to receive(:handle_aborted_migration)
+          expect(worker).not_to receive(:handle_next_migration)
+
+          subject
+        end
       end
     end
 
@@ -298,5 +694,29 @@ RSpec.describe ContainerRegistry::Migration::EnqueuerWorker, :aggregate_failures
         expect(worker).to receive(:log_extra_metadata_on_done).with(key, value)
       end
     end
+
+    def expect_log_info(expected_multiple_arguments)
+      expected_multiple_arguments.each do |extras|
+        expect(worker.logger).to receive(:info).with(worker.structured_payload(extras))
+      end
+    end
+
+    def allow_worker(on:)
+      method_repository = worker.method(on)
+      allow(worker).to receive(on) do
+        repository = method_repository.call
+
+        yield repository if repository
+
+        repository
+      end
+    end
+  end
+
+  describe 'worker attributes' do
+    it 'has deduplication set' do
+      expect(described_class.get_deduplicate_strategy).to eq(:until_executing)
+      expect(described_class.get_deduplication_options).to include(ttl: 30.minutes)
+    end
   end
 end