diff --git a/app/helpers/application_settings_helper.rb b/app/helpers/application_settings_helper.rb
index 472adc8adb498950b9c754e094ed625e64d85158..321a6e9395e1e1e0748d6408e4e8559bc59d1d60 100644
--- a/app/helpers/application_settings_helper.rb
+++ b/app/helpers/application_settings_helper.rb
@@ -404,6 +404,7 @@ def visible_attributes
       :wiki_page_max_content_bytes,
       :container_registry_delete_tags_service_timeout,
       :rate_limiting_response_text,
+      :package_registry_cleanup_policies_worker_capacity,
       :container_registry_expiration_policies_worker_capacity,
       :container_registry_cleanup_tags_service_max_list_size,
       :container_registry_import_max_tags_count,
diff --git a/app/models/application_setting.rb b/app/models/application_setting.rb
index 9f7222fd2d86020110f66558b8238c1cb93c97e3..17b46f929c3617ca9b3fa997fb47c405a7780176 100644
--- a/app/models/application_setting.rb
+++ b/app/models/application_setting.rb
@@ -399,6 +399,7 @@ def self.kroki_formats_attributes
             numericality: { only_integer: true, greater_than_or_equal_to: 0 }
 
   validates :packages_cleanup_package_file_worker_capacity,
+            :package_registry_cleanup_policies_worker_capacity,
             allow_nil: false,
             numericality: { only_integer: true, greater_than_or_equal_to: 0 }
 
diff --git a/app/models/application_setting_implementation.rb b/app/models/application_setting_implementation.rb
index 17a9e4fdc75cace489b835e5200ea731506aac18..e9a0a1561216bc4a41a3dd4841ca7a2662fceeaf 100644
--- a/app/models/application_setting_implementation.rb
+++ b/app/models/application_setting_implementation.rb
@@ -217,6 +217,7 @@ def defaults
         user_show_add_ssh_key_message: true,
         valid_runner_registrars: VALID_RUNNER_REGISTRAR_TYPES,
         wiki_page_max_content_bytes: 50.megabytes,
+        package_registry_cleanup_policies_worker_capacity: 2,
         container_registry_delete_tags_service_timeout: 250,
         container_registry_expiration_policies_worker_capacity: 4,
         container_registry_cleanup_tags_service_max_list_size: 200,
diff --git a/app/models/packages/cleanup/policy.rb b/app/models/packages/cleanup/policy.rb
index c0585a52e6ebea8c3ba71f3918aa8f218c1de51f..35f58f3680d0f8c335433bd50c38017f0f282011 100644
--- a/app/models/packages/cleanup/policy.rb
+++ b/app/models/packages/cleanup/policy.rb
@@ -23,6 +23,17 @@ def self.active
         where.not(keep_n_duplicated_package_files: 'all')
       end
 
+      def self.with_packages
+        exists_select = ::Packages::Package.installable
+                         .where('packages_packages.project_id = packages_cleanup_policies.project_id')
+                         .select(1)
+        where('EXISTS (?)', exists_select)
+      end
+
+      def self.runnable
+        runnable_schedules.with_packages.order(next_run_at: :asc)
+      end
+
       def set_next_run_at
         # fixed cadence of 12 hours
         self.next_run_at = Time.zone.now + 12.hours
diff --git a/app/workers/all_queues.yml b/app/workers/all_queues.yml
index 0e22895a1b86e3a23b6ddff996d0c803c4bc1d04..966a1202db2cf3e0df3d744b3d704a983388569d 100644
--- a/app/workers/all_queues.yml
+++ b/app/workers/all_queues.yml
@@ -1461,6 +1461,15 @@
   :weight: 1
   :idempotent: false
   :tags: []
+- :name: package_cleanup:packages_cleanup_execute_policy
+  :worker_name: Packages::Cleanup::ExecutePolicyWorker
+  :feature_category: :package_registry
+  :has_external_dependencies: false
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: package_cleanup:packages_cleanup_package_file
   :worker_name: Packages::CleanupPackageFileWorker
   :feature_category: :package_registry
diff --git a/app/workers/packages/cleanup/execute_policy_worker.rb b/app/workers/packages/cleanup/execute_policy_worker.rb
new file mode 100644
index 0000000000000000000000000000000000000000..59f0f0250c89eb033bc34629477f08d5f78ed029
--- /dev/null
+++ b/app/workers/packages/cleanup/execute_policy_worker.rb
@@ -0,0 +1,72 @@
+# frozen_string_literal: true
+
+module Packages
+  module Cleanup
+    class ExecutePolicyWorker
+      include ApplicationWorker
+      include LimitedCapacity::Worker
+      include Gitlab::Utils::StrongMemoize
+
+      data_consistency :always
+      queue_namespace :package_cleanup
+      feature_category :package_registry
+      urgency :low
+      worker_resource_boundary :unknown
+      idempotent!
+
+      COUNTS_KEYS = %i[
+        marked_package_files_total_count
+        unique_package_id_and_file_name_total_count
+      ].freeze
+
+      def perform_work
+        return unless next_policy
+
+        log_extra_metadata_on_done(:project_id, next_policy.project_id)
+        result = ::Packages::Cleanup::ExecutePolicyService.new(next_policy).execute
+
+        if result.success?
+          timeout = !!result.payload[:timeout]
+          counts = result.payload[:counts]
+          log_extra_metadata_on_done(:execution_timeout, timeout)
+          COUNTS_KEYS.each do |count_key|
+            log_extra_metadata_on_done(count_key, counts[count_key])
+          end
+        end
+      end
+
+      def remaining_work_count
+        ::Packages::Cleanup::Policy.runnable
+                                   .limit(max_running_jobs + 1)
+                                   .count
+      end
+
+      def max_running_jobs
+        ::Gitlab::CurrentSettings.package_registry_cleanup_policies_worker_capacity
+      end
+
+      private
+
+      def next_policy
+        strong_memoize(:next_policy) do
+          ::Packages::Cleanup::Policy.transaction do
+            # the #lock call is specific to this worker
+            # rubocop: disable CodeReuse/ActiveRecord
+            policy = ::Packages::Cleanup::Policy.runnable
+                                                .limit(1)
+                                                .lock('FOR UPDATE SKIP LOCKED')
+                                                .first
+            # rubocop: enable CodeReuse/ActiveRecord
+
+            next nil unless policy
+
+            policy.set_next_run_at
+            policy.save!
+
+            policy
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/app/workers/packages/cleanup_package_registry_worker.rb b/app/workers/packages/cleanup_package_registry_worker.rb
index a849e055b64b1827c1285d2b606f435b6dd6a3ea..5f14102b5a176f8cbd402b9b5fcb783a2d5d0843 100644
--- a/app/workers/packages/cleanup_package_registry_worker.rb
+++ b/app/workers/packages/cleanup_package_registry_worker.rb
@@ -12,6 +12,7 @@ class CleanupPackageRegistryWorker
 
     def perform
       enqueue_package_file_cleanup_job if Packages::PackageFile.pending_destruction.exists?
+      enqueue_cleanup_policy_jobs if Packages::Cleanup::Policy.runnable.exists?
 
       log_counts
     end
@@ -22,6 +23,10 @@ def enqueue_package_file_cleanup_job
       Packages::CleanupPackageFileWorker.perform_with_capacity
     end
 
+    def enqueue_cleanup_policy_jobs
+      Packages::Cleanup::ExecutePolicyWorker.perform_with_capacity
+    end
+
     def log_counts
       use_replica_if_available do
         pending_destruction_package_files_count = Packages::PackageFile.pending_destruction.count
@@ -31,6 +36,9 @@ def log_counts
         log_extra_metadata_on_done(:pending_destruction_package_files_count, pending_destruction_package_files_count)
         log_extra_metadata_on_done(:processing_package_files_count, processing_package_files_count)
         log_extra_metadata_on_done(:error_package_files_count, error_package_files_count)
+
+        pending_cleanup_policies_count = Packages::Cleanup::Policy.runnable.count
+        log_extra_metadata_on_done(:pending_cleanup_policies_count, pending_cleanup_policies_count)
       end
     end
 
diff --git a/db/migrate/20220713175658_add_packages_cleanup_policies_worker_capacity_to_application_settings.rb b/db/migrate/20220713175658_add_packages_cleanup_policies_worker_capacity_to_application_settings.rb
new file mode 100644
index 0000000000000000000000000000000000000000..8768786410a393e5ded8ce2b68470961f3cf5091
--- /dev/null
+++ b/db/migrate/20220713175658_add_packages_cleanup_policies_worker_capacity_to_application_settings.rb
@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+class AddPackagesCleanupPoliciesWorkerCapacityToApplicationSettings < Gitlab::Database::Migration[2.0]
+  def change
+    add_column :application_settings,
+               :package_registry_cleanup_policies_worker_capacity,
+               :integer,
+               default: 2,
+               null: false
+  end
+end
diff --git a/db/migrate/20220713175737_add_application_settings_packages_cleanup_policies_worker_capacity_constraint.rb b/db/migrate/20220713175737_add_application_settings_packages_cleanup_policies_worker_capacity_constraint.rb
new file mode 100644
index 0000000000000000000000000000000000000000..9aba85570ea1a279a64771186f04d1b97c99083f
--- /dev/null
+++ b/db/migrate/20220713175737_add_application_settings_packages_cleanup_policies_worker_capacity_constraint.rb
@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+class AddApplicationSettingsPackagesCleanupPoliciesWorkerCapacityConstraint < Gitlab::Database::Migration[2.0]
+  CONSTRAINT_NAME = 'app_settings_pkg_registry_cleanup_pol_worker_capacity_gte_zero'
+
+  disable_ddl_transaction!
+
+  def up
+    add_check_constraint :application_settings,
+                         'package_registry_cleanup_policies_worker_capacity >= 0',
+                         CONSTRAINT_NAME
+  end
+
+  def down
+    remove_check_constraint :application_settings, CONSTRAINT_NAME
+  end
+end
diff --git a/db/migrate/20220713175812_add_enabled_policies_index_to_packages_cleanup_policies.rb b/db/migrate/20220713175812_add_enabled_policies_index_to_packages_cleanup_policies.rb
new file mode 100644
index 0000000000000000000000000000000000000000..fe4162e8ac3bf87c6cb856a9b652e8c852875ab6
--- /dev/null
+++ b/db/migrate/20220713175812_add_enabled_policies_index_to_packages_cleanup_policies.rb
@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+class AddEnabledPoliciesIndexToPackagesCleanupPolicies < Gitlab::Database::Migration[2.0]
+  disable_ddl_transaction!
+
+  INDEX_NAME = 'idx_enabled_pkgs_cleanup_policies_on_next_run_at_project_id'
+
+  def up
+    add_concurrent_index :packages_cleanup_policies,
+                         [:next_run_at, :project_id],
+                         where: "keep_n_duplicated_package_files <> 'all'",
+                         name: INDEX_NAME
+  end
+
+  def down
+    remove_concurrent_index_by_name :packages_cleanup_policies, INDEX_NAME
+  end
+end
diff --git a/db/schema_migrations/20220713175658 b/db/schema_migrations/20220713175658
new file mode 100644
index 0000000000000000000000000000000000000000..9b086972336a19e000b1af9e217fc15c8be10469
--- /dev/null
+++ b/db/schema_migrations/20220713175658
@@ -0,0 +1 @@
+6bbaa8006a848a65e866c7836d0b0e28e3c303d28b329f5e12f978dd895e868f
\ No newline at end of file
diff --git a/db/schema_migrations/20220713175737 b/db/schema_migrations/20220713175737
new file mode 100644
index 0000000000000000000000000000000000000000..88f4550ced0cbbfba0de4fe54a2a75b8626591e9
--- /dev/null
+++ b/db/schema_migrations/20220713175737
@@ -0,0 +1 @@
+95a535d8f97ec96df918547aff7947acacbdf37fd0d3656878c9c60d80f3fd02
\ No newline at end of file
diff --git a/db/schema_migrations/20220713175812 b/db/schema_migrations/20220713175812
new file mode 100644
index 0000000000000000000000000000000000000000..13e0279a11e360de367052c9aea5a9d68c65dd29
--- /dev/null
+++ b/db/schema_migrations/20220713175812
@@ -0,0 +1 @@
+41e42a51a0c5b3af8d94edc25e9421a754d6fc517f343bd718b16fd6bfc383f3
\ No newline at end of file
diff --git a/db/structure.sql b/db/structure.sql
index ea704f23f9f9df4282915948a566d9cc24a3d593..3d236e9c747d46b89c0bb0d97dca6147531ef6ef 100644
--- a/db/structure.sql
+++ b/db/structure.sql
@@ -11333,12 +11333,14 @@ CREATE TABLE application_settings (
     error_tracking_api_url text,
     git_rate_limit_users_allowlist text[] DEFAULT '{}'::text[] NOT NULL,
     error_tracking_access_token_encrypted text,
+    package_registry_cleanup_policies_worker_capacity integer DEFAULT 2 NOT NULL,
     CONSTRAINT app_settings_container_reg_cleanup_tags_max_list_size_positive CHECK ((container_registry_cleanup_tags_service_max_list_size >= 0)),
     CONSTRAINT app_settings_container_registry_pre_import_tags_rate_positive CHECK ((container_registry_pre_import_tags_rate >= (0)::numeric)),
     CONSTRAINT app_settings_dep_proxy_ttl_policies_worker_capacity_positive CHECK ((dependency_proxy_ttl_group_policy_worker_capacity >= 0)),
     CONSTRAINT app_settings_ext_pipeline_validation_service_url_text_limit CHECK ((char_length(external_pipeline_validation_service_url) <= 255)),
     CONSTRAINT app_settings_git_rate_limit_users_allowlist_max_usernames CHECK ((cardinality(git_rate_limit_users_allowlist) <= 100)),
     CONSTRAINT app_settings_p_cleanup_package_file_worker_capacity_positive CHECK ((packages_cleanup_package_file_worker_capacity >= 0)),
+    CONSTRAINT app_settings_pkg_registry_cleanup_pol_worker_capacity_gte_zero CHECK ((package_registry_cleanup_policies_worker_capacity >= 0)),
     CONSTRAINT app_settings_registry_exp_policies_worker_capacity_positive CHECK ((container_registry_expiration_policies_worker_capacity >= 0)),
     CONSTRAINT app_settings_yaml_max_depth_positive CHECK ((max_yaml_depth > 0)),
     CONSTRAINT app_settings_yaml_max_size_positive CHECK ((max_yaml_size_bytes > 0)),
@@ -27042,6 +27044,8 @@ CREATE INDEX idx_eaprpb_external_approval_rule_id ON external_approval_rules_pro
 
 CREATE INDEX idx_elastic_reindexing_slices_on_elastic_reindexing_subtask_id ON elastic_reindexing_slices USING btree (elastic_reindexing_subtask_id);
 
+CREATE INDEX idx_enabled_pkgs_cleanup_policies_on_next_run_at_project_id ON packages_cleanup_policies USING btree (next_run_at, project_id) WHERE (keep_n_duplicated_package_files <> 'all'::text);
+
 CREATE UNIQUE INDEX idx_environment_merge_requests_unique_index ON deployment_merge_requests USING btree (environment_id, merge_request_id);
 
 CREATE UNIQUE INDEX idx_external_audit_event_destination_id_key_uniq ON audit_events_streaming_headers USING btree (key, external_audit_event_destination_id);
diff --git a/doc/api/settings.md b/doc/api/settings.md
index e33f3990ddd71aaf72a5daff7dbc3e94c58c71e8..28a33e4deb480bfdc65ab48ea24541c02120c076 100644
--- a/doc/api/settings.md
+++ b/doc/api/settings.md
@@ -175,6 +175,7 @@ Example response:
   "container_registry_expiration_policies_caching": true,
   "container_registry_expiration_policies_worker_capacity": 4,
   "container_registry_token_expire_delay": 5,
+  "package_registry_cleanup_policies_worker_capacity": 2,
   "repository_storages": ["default"],
   "plantuml_enabled": false,
   "plantuml_url": null,
@@ -271,6 +272,7 @@ listed in the descriptions of the relevant settings.
 | `container_registry_expiration_policies_caching`  | boolean | no                                   | Caching during the execution of [cleanup policies](../user/packages/container_registry/reduce_container_registry_storage.md#set-cleanup-limits-to-conserve-resources). |
 | `container_registry_expiration_policies_worker_capacity`  | integer          | no                  | Number of workers for [cleanup policies](../user/packages/container_registry/reduce_container_registry_storage.md#set-cleanup-limits-to-conserve-resources). |
 | `container_registry_token_expire_delay`  | integer          | no                                   | Container Registry token duration in minutes. |
+| `package_registry_cleanup_policies_worker_capacity`  | integer          | no                  | Number of workers assigned to the packages cleanup policies. |
 | `deactivate_dormant_users`               | boolean          | no                                   | Enable [automatic deactivation of dormant users](../user/admin_area/moderate_users.md#automatically-deactivate-dormant-users). |
 | `default_artifacts_expire_in`            | string           | no                                   | Set the default expiration time for each job's artifacts. |
 | `default_branch_name`                    | string           | no                                   | [Instance-level custom initial branch name](../user/project/repository/branches/default.md#instance-level-custom-initial-branch-name) ([introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/225258) in GitLab 13.2). |
diff --git a/spec/models/application_setting_spec.rb b/spec/models/application_setting_spec.rb
index 8b88503a1b9733ef2c1bf19379c3309e53d3270e..0b3521cdd0caa1bb28773e9b295c11c7bc84dd3e 100644
--- a/spec/models/application_setting_spec.rb
+++ b/spec/models/application_setting_spec.rb
@@ -104,6 +104,9 @@
     it { is_expected.to validate_numericality_of(:packages_cleanup_package_file_worker_capacity).only_integer.is_greater_than_or_equal_to(0) }
     it { is_expected.not_to allow_value(nil).for(:packages_cleanup_package_file_worker_capacity) }
 
+    it { is_expected.to validate_numericality_of(:package_registry_cleanup_policies_worker_capacity).only_integer.is_greater_than_or_equal_to(0) }
+    it { is_expected.not_to allow_value(nil).for(:package_registry_cleanup_policies_worker_capacity) }
+
     it { is_expected.to validate_numericality_of(:snippet_size_limit).only_integer.is_greater_than(0) }
     it { is_expected.to validate_numericality_of(:wiki_page_max_content_bytes).only_integer.is_greater_than_or_equal_to(1024) }
     it { is_expected.to validate_presence_of(:max_artifacts_size) }
diff --git a/spec/models/packages/cleanup/policy_spec.rb b/spec/models/packages/cleanup/policy_spec.rb
index 9f4568932d234f57012d252e7d5d86949c21bf83..a37042520e73b0566fd0221c348521203c4f33f8 100644
--- a/spec/models/packages/cleanup/policy_spec.rb
+++ b/spec/models/packages/cleanup/policy_spec.rb
@@ -26,6 +26,30 @@
     it { is_expected.to contain_exactly(active_policy) }
   end
 
+  describe '.with_packages' do
+    let_it_be(:policy_with_packages) { create(:packages_cleanup_policy) }
+    let_it_be(:policy_without_packages) { create(:packages_cleanup_policy) }
+    let_it_be(:package) { create(:package, project: policy_with_packages.project) }
+
+    subject { described_class.with_packages }
+
+    it { is_expected.to contain_exactly(policy_with_packages) }
+  end
+
+  describe '.runnable' do
+    let_it_be(:runnable_policy_with_packages) { create(:packages_cleanup_policy, :runnable) }
+    let_it_be(:runnable_policy_without_packages) { create(:packages_cleanup_policy, :runnable) }
+    let_it_be(:non_runnable_policy_with_packages) { create(:packages_cleanup_policy) }
+    let_it_be(:non_runnable_policy_without_packages) { create(:packages_cleanup_policy) }
+
+    let_it_be(:package1) { create(:package, project: runnable_policy_with_packages.project) }
+    let_it_be(:package2) { create(:package, project: non_runnable_policy_with_packages.project) }
+
+    subject { described_class.runnable }
+
+    it { is_expected.to contain_exactly(runnable_policy_with_packages) }
+  end
+
   describe '#keep_n_duplicated_package_files_disabled?' do
     subject { policy.keep_n_duplicated_package_files_disabled? }
 
diff --git a/spec/workers/every_sidekiq_worker_spec.rb b/spec/workers/every_sidekiq_worker_spec.rb
index 57b0f30aa7469982bb9c1f5d8b0a60e01cb8c5c3..e8ec7c28537fcaa4766cab9dfe00dcf0cbf1dacd 100644
--- a/spec/workers/every_sidekiq_worker_spec.rb
+++ b/spec/workers/every_sidekiq_worker_spec.rb
@@ -357,6 +357,7 @@
         'ObjectStorage::BackgroundMoveWorker' => 5,
         'ObjectStorage::MigrateUploadsWorker' => 3,
         'Packages::CleanupPackageFileWorker' => 0,
+        'Packages::Cleanup::ExecutePolicyWorker' => 0,
         'Packages::Composer::CacheUpdateWorker' => false,
         'Packages::Go::SyncPackagesWorker' => 3,
         'Packages::MarkPackageFilesForDestructionWorker' => 3,
diff --git a/spec/workers/packages/cleanup/execute_policy_worker_spec.rb b/spec/workers/packages/cleanup/execute_policy_worker_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..81fcec1a36024b9e6ac905b14993ef6a65d4433c
--- /dev/null
+++ b/spec/workers/packages/cleanup/execute_policy_worker_spec.rb
@@ -0,0 +1,160 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Packages::Cleanup::ExecutePolicyWorker do
+  let(:worker) { described_class.new }
+
+  describe '#perform_work' do
+    subject(:perform_work) { worker.perform_work }
+
+    shared_examples 'not executing any policy' do
+      it 'is a no op' do
+        expect(::Packages::Cleanup::ExecutePolicyService).not_to receive(:new)
+
+        expect { perform_work }.not_to change { Packages::PackageFile.installable.count }
+      end
+    end
+
+    context 'with no policies' do
+      it_behaves_like 'not executing any policy'
+    end
+
+    context 'with no runnable policies' do
+      let_it_be(:policy) { create(:packages_cleanup_policy) }
+
+      it_behaves_like 'not executing any policy'
+    end
+
+    context 'with runnable policies linked to no packages' do
+      let_it_be(:policy) { create(:packages_cleanup_policy, :runnable) }
+
+      it_behaves_like 'not executing any policy'
+    end
+
+    context 'with runnable policies linked to packages' do
+      let_it_be(:policy) { create(:packages_cleanup_policy, :runnable, keep_n_duplicated_package_files: '1') }
+      let_it_be(:package) { create(:package, project: policy.project) }
+
+      let_it_be(:package_file1) { create(:package_file, file_name: 'test1', package: package) }
+      let_it_be(:package_file2) { create(:package_file, file_name: 'test1', package: package) }
+
+      include_examples 'an idempotent worker' do
+        it 'executes the policy' do
+          expect(::Packages::Cleanup::ExecutePolicyService)
+            .to receive(:new).with(policy).and_call_original
+          expect_log_extra_metadata(:project_id, policy.project_id)
+          expect_log_extra_metadata(:execution_timeout, false)
+          expect_log_extra_metadata(:marked_package_files_total_count, 1)
+          expect_log_extra_metadata(:unique_package_id_and_file_name_total_count, 1)
+
+          expect { perform_work }
+            .to change { package.package_files.installable.count }.by(-1)
+            .and change { policy.reload.next_run_at.future? }.from(false).to(true)
+        end
+
+        context 'with a timeout' do
+          let(:mark_service_response) do
+            ServiceResponse.error(
+              message: 'Timeout',
+              payload: { marked_package_files_count: 1 }
+            )
+          end
+
+          it 'executes the policy partially' do
+            expect_next_instance_of(::Packages::MarkPackageFilesForDestructionService) do |service|
+              expect(service).to receive(:execute).and_return(mark_service_response)
+            end
+
+            expect_log_extra_metadata(:project_id, policy.project_id)
+            expect_log_extra_metadata(:execution_timeout, true)
+            expect_log_extra_metadata(:marked_package_files_total_count, 1)
+            expect_log_extra_metadata(:unique_package_id_and_file_name_total_count, 1)
+
+            expect { perform_work }
+              .to change { policy.reload.next_run_at.future? }.from(false).to(true)
+          end
+        end
+      end
+
+      context 'with several eligible policies' do
+        let_it_be(:policy2) { create(:packages_cleanup_policy, :runnable) }
+        let_it_be(:package2) { create(:package, project: policy2.project) }
+
+        before do
+          policy2.update_column(:next_run_at, 100.years.ago)
+        end
+
+        it 'executes the most urgent policy' do
+          expect(::Packages::Cleanup::ExecutePolicyService)
+            .to receive(:new).with(policy2).and_call_original
+          expect_log_extra_metadata(:project_id, policy2.project_id)
+          expect_log_extra_metadata(:execution_timeout, false)
+          expect_log_extra_metadata(:marked_package_files_total_count, 0)
+          expect_log_extra_metadata(:unique_package_id_and_file_name_total_count, 0)
+
+          expect { perform_work }
+            .to change { policy2.reload.next_run_at.future? }.from(false).to(true)
+            .and not_change { policy.reload.next_run_at }
+        end
+      end
+    end
+
+    context 'with runnable policy linked to packages in a disabled state' do
+      let_it_be(:policy) { create(:packages_cleanup_policy, :runnable, keep_n_duplicated_package_files: 'all') }
+      let_it_be(:package) { create(:package, project: policy.project) }
+
+      it_behaves_like 'not executing any policy'
+    end
+
+    def expect_log_extra_metadata(key, value)
+      expect(worker).to receive(:log_extra_metadata_on_done).with(key, value)
+    end
+  end
+
+  describe '#remaining_work_count' do
+    subject { worker.remaining_work_count}
+
+    context 'with no policies' do
+      it { is_expected.to eq(0) }
+    end
+
+    context 'with no runnable policies' do
+      let_it_be(:policy) { create(:packages_cleanup_policy) }
+
+      it { is_expected.to eq(0) }
+    end
+
+    context 'with runnable policies linked to no packages' do
+      let_it_be(:policy) { create(:packages_cleanup_policy, :runnable) }
+
+      it { is_expected.to eq(0) }
+    end
+
+    context 'with runnable policies linked to packages' do
+      let_it_be(:policy) { create(:packages_cleanup_policy, :runnable) }
+      let_it_be(:package) { create(:package, project: policy.project) }
+
+      it { is_expected.to eq(1) }
+    end
+
+    context 'with runnable policy linked to packages in a disabled state' do
+      let_it_be(:policy) { create(:packages_cleanup_policy, :runnable, keep_n_duplicated_package_files: 'all') }
+      let_it_be(:package) { create(:package, project: policy.project) }
+
+      it { is_expected.to eq(0) }
+    end
+  end
+
+  describe '#max_running_jobs' do
+    let(:capacity) { 50 }
+
+    subject { worker.max_running_jobs }
+
+    before do
+      stub_application_setting(package_registry_cleanup_policies_worker_capacity: capacity)
+    end
+
+    it { is_expected.to eq(capacity) }
+  end
+end
diff --git a/spec/workers/packages/cleanup_package_registry_worker_spec.rb b/spec/workers/packages/cleanup_package_registry_worker_spec.rb
index e43864975f681366f612c329b38bfac9ad4e6d56..e12f2198f663da9dab75e1451b7e45301e524114 100644
--- a/spec/workers/packages/cleanup_package_registry_worker_spec.rb
+++ b/spec/workers/packages/cleanup_package_registry_worker_spec.rb
@@ -5,6 +5,8 @@
 RSpec.describe Packages::CleanupPackageRegistryWorker do
   describe '#perform' do
     let_it_be_with_reload(:package_files) { create_list(:package_file, 2, :pending_destruction) }
+    let_it_be(:policy) { create(:packages_cleanup_policy, :runnable) }
+    let_it_be(:package) { create(:package, project: policy.project) }
 
     let(:worker) { described_class.new }
 
@@ -34,6 +36,28 @@
       end
     end
 
+    context 'with runnable policies' do
+      it_behaves_like 'an idempotent worker'
+
+      it 'queues the cleanup job' do
+        expect(Packages::Cleanup::ExecutePolicyWorker).to receive(:perform_with_capacity)
+
+        perform
+      end
+    end
+
+    context 'with no runnable policies' do
+      before do
+        policy.update_column(:next_run_at, 5.minutes.from_now)
+      end
+
+      it 'does not queue the cleanup job' do
+        expect(Packages::Cleanup::ExecutePolicyWorker).not_to receive(:perform_with_capacity)
+
+        perform
+      end
+    end
+
     describe 'counts logging' do
       let_it_be(:processing_package_file) { create(:package_file, status: :processing) }
 
@@ -41,6 +65,7 @@
         expect(worker).to receive(:log_extra_metadata_on_done).with(:pending_destruction_package_files_count, 2)
         expect(worker).to receive(:log_extra_metadata_on_done).with(:processing_package_files_count, 1)
         expect(worker).to receive(:log_extra_metadata_on_done).with(:error_package_files_count, 0)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:pending_cleanup_policies_count, 1)
 
         perform
       end