Update container expiration policy worker

with a loopless mode. In this mode, the worker will not
query and loop through all the available cleanup policies.

This mode is controlled by a feature flag:
`container_registry_expiration_policies_loopless`

app/workers/container_expiration_policy_worker.rb

@@ -29,6 +29,7 @@ class ContainerExpirationPolicyWorker # rubocop:disable Scalability/IdempotentWo
 
   def perform_throttled
     try_obtain_lease do
+      unless loopless_enabled?
         with_runnable_policy do |policy|
           ContainerExpirationPolicy.transaction do
             policy.schedule_next_run!
@@ -38,6 +39,7 @@ class ContainerExpirationPolicyWorker # rubocop:disable Scalability/IdempotentWo
                                end
           end
         end
+      end
 
       ContainerExpirationPolicies::CleanupContainerRepositoryWorker.perform_with_capacity
     end
@@ -75,6 +77,10 @@ class ContainerExpirationPolicyWorker # rubocop:disable Scalability/IdempotentWo
     Feature.enabled?(:container_registry_expiration_policies_throttling)
   end
 
+  def loopless_enabled?
+    Feature.enabled?(:container_registry_expiration_policies_loopless)
+  end
+
   def lease_timeout
     5.hours
   end

config/feature_flags/development/container_registry_expiration_policies_loopless.yml

+---
+name: container_registry_expiration_policies_loopless
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/56962
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/325273
+milestone: '13.11'
+type: development
+group: group::package
+default_enabled: false

spec/workers/container_expiration_policy_worker_spec.rb

@@ -11,7 +11,7 @@ RSpec.describe ContainerExpirationPolicyWorker do
   describe '#perform' do
     subject { worker.perform }
 
-    RSpec.shared_examples 'not executing any policy' do
+    shared_examples 'not executing any policy' do
       it 'does not run any policy' do
         expect(ContainerExpirationPolicyService).not_to receive(:new)
 
@@ -19,6 +19,21 @@ RSpec.describe ContainerExpirationPolicyWorker do
       end
     end
 
+    shared_examples 'handling a taken exclusive lease' do
+      context 'with exclusive lease taken' do
+        before do
+          stub_exclusive_lease_taken(worker.lease_key, timeout: 5.hours)
+        end
+
+        it 'does not do anything' do
+          expect(ContainerExpirationPolicies::CleanupContainerRepositoryWorker).not_to receive(:perform_with_capacity)
+          expect(worker).not_to receive(:runnable_policies)
+
+          expect { subject }.not_to change { ContainerRepository.cleanup_scheduled.count }
+        end
+      end
+    end
+
     context 'With no container expiration policies' do
       it 'does not execute any policies' do
         expect(ContainerRepository).not_to receive(:for_project_id)
@@ -27,10 +42,24 @@ RSpec.describe ContainerExpirationPolicyWorker do
       end
     end
 
+    context 'with throttling enabled' do
+      before do
+        stub_feature_flags(container_registry_expiration_policies_throttling: true)
+      end
+
+      context 'with loopless disabled' do
+        before do
+          stub_feature_flags(container_registry_expiration_policies_loopless: false)
+        end
+
         context 'with container expiration policies' do
           let_it_be(:container_expiration_policy) { create(:container_expiration_policy, :runnable) }
           let_it_be(:container_repository) { create(:container_repository, project: container_expiration_policy.project) }
 
+          before do
+            expect(worker).to receive(:with_runnable_policy).and_call_original
+          end
+
           context 'with a valid container expiration policy' do
             it 'schedules the next run' do
               expect { subject }.to change { container_expiration_policy.reload.next_run_at }
@@ -77,16 +106,22 @@ RSpec.describe ContainerExpirationPolicyWorker do
           end
         end
 
-    context 'with exclusive lease taken' do
+        it_behaves_like 'handling a taken exclusive lease'
+      end
+
+      context 'with loopless enabled' do
         before do
-        stub_exclusive_lease_taken(worker.lease_key, timeout: 5.hours)
+          stub_feature_flags(container_registry_expiration_policies_loopless: true)
+          expect(worker).not_to receive(:with_runnable_policy)
         end
 
-      it 'does not execute any policy' do
-        expect(ContainerExpirationPolicies::CleanupContainerRepositoryWorker).not_to receive(:perform_with_capacity)
-        expect(worker).not_to receive(:runnable_policies)
+        it 'calls the limited capacity worker' do
+          expect(ContainerExpirationPolicies::CleanupContainerRepositoryWorker).to receive(:perform_with_capacity)
 
-        expect { subject }.not_to change { ContainerRepository.cleanup_scheduled.count }
+          subject
+        end
+
+        it_behaves_like 'handling a taken exclusive lease'
       end
     end