Dependency Proxy TTL workers

app/models/application_setting.rb

@@ -364,6 +364,10 @@ class ApplicationSetting < ApplicationRecord
   validates :container_registry_expiration_policies_worker_capacity,
             numericality: { only_integer: true, greater_than_or_equal_to: 0 }
 
+  validates :dependency_proxy_ttl_group_policy_worker_capacity,
+            allow_nil: false,
+            numericality: { only_integer: true, greater_than_or_equal_to: 0 }
+
   validates :invisible_captcha_enabled,
             inclusion: { in: [true, false], message: _('must be a boolean value') }
 

app/models/concerns/ttl_expirable.rb

+# frozen_string_literal: true
+
+module TtlExpirable
+  extend ActiveSupport::Concern
+
+  included do
+    validates :status, presence: true
+
+    enum status: { default: 0, expired: 1, processing: 2, error: 3 }
+
+    scope :updated_before, ->(number_of_days) { where("updated_at <= ?", Time.zone.now - number_of_days.days) }
+    scope :active, -> { where(status: :default) }
+
+    scope :lock_next_by, ->(sort) do
+      order(sort)
+        .limit(1)
+        .lock('FOR UPDATE SKIP LOCKED')
+    end
+  end
+end

app/models/dependency_proxy/blob.rb

@@ -2,15 +2,14 @@
 
 class DependencyProxy::Blob < ApplicationRecord
   include FileStoreMounter
+  include TtlExpirable
+  include EachBatch
 
   belongs_to :group
 
   validates :group, presence: true
   validates :file, presence: true
   validates :file_name, presence: true
-  validates :status, presence: true
-
-  enum status: { default: 0, expired: 1 }
 
   mount_file_store_uploader DependencyProxy::FileUploader
 

app/models/dependency_proxy/image_ttl_group_policy.rb

@@ -8,4 +8,6 @@ class DependencyProxy::ImageTtlGroupPolicy < ApplicationRecord
   validates :group, presence: true
   validates :enabled, inclusion: { in: [true, false] }
   validates :ttl, numericality: { greater_than: 0 }, allow_nil: true
+
+  scope :enabled, -> { where(enabled: true) }
 end

app/models/dependency_proxy/manifest.rb

@@ -2,6 +2,8 @@
 
 class DependencyProxy::Manifest < ApplicationRecord
   include FileStoreMounter
+  include TtlExpirable
+  include EachBatch
 
   belongs_to :group
 
@@ -9,9 +11,6 @@ class DependencyProxy::Manifest < ApplicationRecord
   validates :file, presence: true
   validates :file_name, presence: true
   validates :digest, presence: true
-  validates :status, presence: true
-
-  enum status: { default: 0, expired: 1 }
 
   mount_file_store_uploader DependencyProxy::FileUploader
 

app/services/dependency_proxy/find_or_create_blob_service.rb

@@ -12,7 +12,7 @@ module DependencyProxy
     def execute
       from_cache = true
       file_name = @blob_sha.sub('sha256:', '') + '.gz'
-      blob = @group.dependency_proxy_blobs.find_or_build(file_name)
+      blob = @group.dependency_proxy_blobs.active.find_or_build(file_name)
 
       unless blob.persisted?
         from_cache = false
@@ -30,6 +30,8 @@ module DependencyProxy
         blob.save!
       end
 
+      # Technical debt: change to read_at https://gitlab.com/gitlab-org/gitlab/-/issues/341536
+      blob.touch if from_cache
       success(blob: blob, from_cache: from_cache)
     end
 

app/services/dependency_proxy/find_or_create_manifest_service.rb

@@ -13,11 +13,16 @@ module DependencyProxy
 
     def execute
       @manifest = @group.dependency_proxy_manifests
+                        .active
                         .find_or_initialize_by_file_name_or_digest(file_name: @file_name, digest: @tag)
 
       head_result = DependencyProxy::HeadManifestService.new(@image, @tag, @token).execute
 
-      return success(manifest: @manifest, from_cache: true) if cached_manifest_matches?(head_result)
+      if cached_manifest_matches?(head_result)
+        @manifest.touch
+
+        return success(manifest: @manifest, from_cache: true)
+      end
 
       pull_new_manifest
       respond(from_cache: false)
@@ -46,6 +51,9 @@ module DependencyProxy
 
     def respond(from_cache: true)
       if @manifest.persisted?
+        # Technical debt: change to read_at https://gitlab.com/gitlab-org/gitlab/-/issues/341536
+        @manifest.touch if from_cache
+
         success(manifest: @manifest, from_cache: from_cache)
       else
         error('Failed to download the manifest from the external registry', 503)

app/workers/all_queues.yml

@@ -264,6 +264,15 @@
   :weight: 1
   :idempotent: true
   :tags: []
+- :name: cronjob:dependency_proxy_image_ttl_group_policy
+  :worker_name: DependencyProxy::ImageTtlGroupPolicyWorker
+  :feature_category: :dependency_proxy
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent:
+  :tags: []
 - :name: cronjob:environments_auto_delete_cron
   :worker_name: Environments::AutoDeleteCronWorker
   :feature_category: :continuous_delivery
@@ -651,6 +660,24 @@
   :weight: 1
   :idempotent: true
   :tags: []
+- :name: dependency_proxy_blob:dependency_proxy_cleanup_blob
+  :worker_name: DependencyProxy::CleanupBlobWorker
+  :feature_category: :dependency_proxy
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
+- :name: dependency_proxy_manifest:dependency_proxy_cleanup_manifest
+  :worker_name: DependencyProxy::CleanupManifestWorker
+  :feature_category: :dependency_proxy
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: deployment:deployments_drop_older_deployments
   :worker_name: Deployments::DropOlderDeploymentsWorker
   :feature_category: :continuous_delivery

app/workers/concerns/dependency_proxy/cleanup_worker.rb

+# frozen_string_literal: true
+
+module DependencyProxy
+  module CleanupWorker
+    extend ActiveSupport::Concern
+    include Gitlab::Utils::StrongMemoize
+
+    def perform_work
+      return unless artifact
+
+      log_metadata(artifact)
+
+      artifact.destroy!
+    rescue StandardError
+      artifact&.error!
+    end
+
+    def max_running_jobs
+      ::Gitlab::CurrentSettings.dependency_proxy_ttl_group_policy_worker_capacity
+    end
+
+    def remaining_work_count
+      expired_artifacts.limit(max_running_jobs + 1).count
+    end
+
+    private
+
+    def model
+      raise NotImplementedError
+    end
+
+    def log_metadata
+      raise NotImplementedError
+    end
+
+    def log_cleanup_item
+      raise NotImplementedError
+    end
+
+    def artifact
+      strong_memoize(:artifact) do
+        model.transaction do
+          to_delete = next_item
+
+          if to_delete
+            to_delete.processing!
+            log_cleanup_item(to_delete)
+          end
+
+          to_delete
+        end
+      end
+    end
+
+    def expired_artifacts
+      model.expired
+    end
+
+    def next_item
+      expired_artifacts.lock_next_by(:updated_at).first
+    end
+  end
+end

app/workers/dependency_proxy/cleanup_blob_worker.rb

+# frozen_string_literal: true
+
+module DependencyProxy
+  class CleanupBlobWorker
+    include ApplicationWorker
+    include LimitedCapacity::Worker
+    include Gitlab::Utils::StrongMemoize
+    include DependencyProxy::CleanupWorker
+
+    data_consistency :always
+
+    sidekiq_options retry: 3
+
+    queue_namespace :dependency_proxy_blob
+    feature_category :dependency_proxy
+    urgency :low
+    worker_resource_boundary :unknown
+    idempotent!
+
+    private
+
+    def model
+      DependencyProxy::Blob
+    end
+
+    def log_metadata(blob)
+      log_extra_metadata_on_done(:dependency_proxy_blob_id, blob.id)
+      log_extra_metadata_on_done(:group_id, blob.group_id)
+    end
+
+    def log_cleanup_item(blob)
+      logger.info(
+        structured_payload(
+          group_id: blob.group_id,
+          dependency_proxy_blob_id: blob.id
+        )
+      )
+    end
+  end
+end

app/workers/dependency_proxy/cleanup_manifest_worker.rb

+# frozen_string_literal: true
+
+module DependencyProxy
+  class CleanupManifestWorker
+    include ApplicationWorker
+    include LimitedCapacity::Worker
+    include Gitlab::Utils::StrongMemoize
+    include DependencyProxy::CleanupWorker
+
+    data_consistency :always
+
+    sidekiq_options retry: 3
+
+    queue_namespace :dependency_proxy_manifest
+    feature_category :dependency_proxy
+    urgency :low
+    worker_resource_boundary :unknown
+    idempotent!
+
+    private
+
+    def model
+      DependencyProxy::Manifest
+    end
+
+    def log_metadata(manifest)
+      log_extra_metadata_on_done(:dependency_proxy_manifest_id, manifest.id)
+      log_extra_metadata_on_done(:group_id, manifest.group_id)
+    end
+
+    def log_cleanup_item(manifest)
+      logger.info(
+        structured_payload(
+          group_id: manifest.group_id,
+          dependency_proxy_manifest_id: manifest.id
+        )
+      )
+    end
+  end
+end

app/workers/dependency_proxy/image_ttl_group_policy_worker.rb

+# frozen_string_literal: true
+
+module DependencyProxy
+  class ImageTtlGroupPolicyWorker # rubocop:disable Scalability/IdempotentWorker
+    include ApplicationWorker
+    include CronjobQueue # rubocop:disable Scalability/CronWorkerContext
+
+    data_consistency :always
+
+    feature_category :dependency_proxy
+
+    UPDATE_BATCH_SIZE = 100
+
+    def perform
+      DependencyProxy::ImageTtlGroupPolicy.enabled.each do |policy|
+        # Technical Debt: change to read_before https://gitlab.com/gitlab-org/gitlab/-/issues/341536
+        qualified_blobs = policy.group.dependency_proxy_blobs.active.updated_before(policy.ttl)
+        qualified_manifests = policy.group.dependency_proxy_manifests.active.updated_before(policy.ttl)
+
+        enqueue_blob_cleanup_job if expire_artifacts(qualified_blobs, DependencyProxy::Blob)
+        enqueue_manifest_cleanup_job if expire_artifacts(qualified_manifests, DependencyProxy::Manifest)
+      end
+
+      log_counts
+    end
+
+    private
+
+    def expire_artifacts(artifacts, model)
+      rows_updated = false
+
+      artifacts.each_batch(of: UPDATE_BATCH_SIZE) do |batch|
+        rows = batch.update_all(status: :expired)
+        rows_updated ||= rows > 0
+      end
+
+      rows_updated
+    end
+
+    def enqueue_blob_cleanup_job
+      DependencyProxy::CleanupBlobWorker.perform_with_capacity
+    end
+
+    def enqueue_manifest_cleanup_job
+      DependencyProxy::CleanupManifestWorker.perform_with_capacity
+    end
+
+    def log_counts
+      use_replica_if_available do
+        expired_blob_count = DependencyProxy::Blob.expired.count
+        expired_manifest_count = DependencyProxy::Manifest.expired.count
+        processing_blob_count = DependencyProxy::Blob.processing.count
+        processing_manifest_count = DependencyProxy::Manifest.processing.count
+        error_blob_count = DependencyProxy::Blob.error.count
+        error_manifest_count = DependencyProxy::Manifest.error.count
+
+        log_extra_metadata_on_done(:expired_dependency_proxy_blob_count, expired_blob_count)
+        log_extra_metadata_on_done(:expired_dependency_proxy_manifest_count, expired_manifest_count)
+        log_extra_metadata_on_done(:processing_dependency_proxy_blob_count, processing_blob_count)
+        log_extra_metadata_on_done(:processing_dependency_proxy_manifest_count, processing_manifest_count)
+        log_extra_metadata_on_done(:error_dependency_proxy_blob_count, error_blob_count)
+        log_extra_metadata_on_done(:error_dependency_proxy_manifest_count, error_manifest_count)
+      end
+    end
+
+    def use_replica_if_available(&block)
+      ::Gitlab::Database::LoadBalancing::Session.current.use_replicas_for_read_queries(&block)
+    end
+  end
+end

config/initializers/1_settings.rb

@@ -535,6 +535,9 @@ Settings.cron_jobs['namespaces_prune_aggregation_schedules_worker']['job_class']
 Settings.cron_jobs['container_expiration_policy_worker'] ||= Settingslogic.new({})
 Settings.cron_jobs['container_expiration_policy_worker']['cron'] ||= '50 * * * *'
 Settings.cron_jobs['container_expiration_policy_worker']['job_class'] = 'ContainerExpirationPolicyWorker'
+Settings.cron_jobs['image_ttl_group_policy_worker'] ||= Settingslogic.new({})
+Settings.cron_jobs['image_ttl_group_policy_worker']['cron'] ||= '40 0 * * *'
+Settings.cron_jobs['image_ttl_group_policy_worker']['job_class'] = 'DependencyProxy::ImageTtlGroupPolicyWorker'
 Settings.cron_jobs['x509_issuer_crl_check_worker'] ||= Settingslogic.new({})
 Settings.cron_jobs['x509_issuer_crl_check_worker']['cron'] ||= '30 1 * * *'
 Settings.cron_jobs['x509_issuer_crl_check_worker']['job_class'] = 'X509IssuerCrlCheckWorker'

config/sidekiq_queues.yml

@@ -91,6 +91,10 @@
   - 1
 - - dependency_proxy
   - 1
+- - dependency_proxy_blob
+  - 1
+- - dependency_proxy_manifest
+  - 1
 - - deployment
   - 3
 - - design_management_copy_design_collection

db/migrate/20210910014741_add_dependency_proxy_ttl_group_policy_worker_capacity_to_application_settings.rb

+# frozen_string_literal: true
+
+class AddDependencyProxyTtlGroupPolicyWorkerCapacityToApplicationSettings < Gitlab::Database::Migration[1.0]
+  def change
+    add_column :application_settings,
+               :dependency_proxy_ttl_group_policy_worker_capacity,
+               :smallint,
+               default: 2,
+               null: false
+  end
+end

db/migrate/20210910015047_add_app_settings_dep_proxy_ttl_worker_capacity_check_constraint.rb

+# frozen_string_literal: true
+
+class AddAppSettingsDepProxyTtlWorkerCapacityCheckConstraint < Gitlab::Database::Migration[1.0]
+  CONSTRAINT_NAME = 'app_settings_dep_proxy_ttl_policies_worker_capacity_positive'
+
+  disable_ddl_transaction!
+
+  def up
+    add_check_constraint :application_settings, 'dependency_proxy_ttl_group_policy_worker_capacity >= 0', CONSTRAINT_NAME
+  end
+
+  def down
+    remove_check_constraint :application_settings, CONSTRAINT_NAME
+  end
+end

db/migrate/20210913224558_update_dependency_proxy_manifests_uniqueness_constraint.rb

+# frozen_string_literal: true
+
+class UpdateDependencyProxyManifestsUniquenessConstraint < Gitlab::Database::Migration[1.0]
+  disable_ddl_transaction!
+
+  NEW_INDEX_NAME = 'index_dep_prox_manifests_on_group_id_file_name_and_status'
+  OLD_INDEX_NAME = 'index_dependency_proxy_manifests_on_group_id_and_file_name'
+
+  def up
+    add_concurrent_index :dependency_proxy_manifests, [:group_id, :file_name, :status], unique: true, name: NEW_INDEX_NAME
+    remove_concurrent_index_by_name :dependency_proxy_manifests, OLD_INDEX_NAME
+  end
+
+  def down
+    add_concurrent_index :dependency_proxy_manifests, [:group_id, :file_name], unique: true, name: OLD_INDEX_NAME
+    remove_concurrent_index_by_name :dependency_proxy_manifests, NEW_INDEX_NAME
+  end
+end

db/migrate/20210914172202_add_status_index_to_dependency_proxy_tables.rb

+# frozen_string_literal: true
+
+class AddStatusIndexToDependencyProxyTables < Gitlab::Database::Migration[1.0]
+  MANIFEST_INDEX_NAME = 'index_dependency_proxy_manifests_on_status'
+  BLOB_INDEX_NAME = 'index_dependency_proxy_blobs_on_status'
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_index :dependency_proxy_manifests, :status, name: MANIFEST_INDEX_NAME
+    add_concurrent_index :dependency_proxy_blobs, :status, name: BLOB_INDEX_NAME
+  end
+
+  def down
+    remove_concurrent_index_by_name :dependency_proxy_manifests, MANIFEST_INDEX_NAME
+    remove_concurrent_index_by_name :dependency_proxy_blobs, BLOB_INDEX_NAME
+  end
+end

db/migrate/20210928171122_add_group_id_status_id_index_to_dependency_proxy_tables.rb

+# frozen_string_literal: true
+
+class AddGroupIdStatusIdIndexToDependencyProxyTables < Gitlab::Database::Migration[1.0]
+  MANIFEST_INDEX_NAME = 'index_dependency_proxy_manifests_on_group_id_status_and_id'
+  BLOB_INDEX_NAME = 'index_dependency_proxy_blobs_on_group_id_status_and_id'
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_index :dependency_proxy_manifests, [:group_id, :status, :id], name: MANIFEST_INDEX_NAME
+    add_concurrent_index :dependency_proxy_blobs, [:group_id, :status, :id], name: BLOB_INDEX_NAME
+  end
+
+  def down
+    remove_concurrent_index_by_name :dependency_proxy_manifests, MANIFEST_INDEX_NAME
+    remove_concurrent_index_by_name :dependency_proxy_blobs, BLOB_INDEX_NAME
+  end
+end

db/schema_migrations/20210910014741

+e6342d440d398980470f4dd018c5df56d0b5d4df11caa7ba5dd2e92578dbf678
\ No newline at end of file

db/schema_migrations/20210910015047

+d0b2ee97781a5d3c671b855fb6be844431a73584be47ba35d83c7e8cfec69bcb
\ No newline at end of file

db/schema_migrations/20210913224558

+377af41414793d7e52ffbb1fd60f2f19c58cd63bb0e85192983b5bfe98515ae8
\ No newline at end of file

db/schema_migrations/20210914172202

+2ab67d4cc17d0fdf01b5861a46d6ec51d1e76e7e88209b0964a884edd22cc63d
\ No newline at end of file

db/schema_migrations/20210928171122

+f257ff9896e2d90ced39c2c010df1d4b74badae046651a190585c9c47342d119
\ No newline at end of file

db/structure.sql

@@ -10344,7 +10344,9 @@ CREATE TABLE application_settings (
     throttle_authenticated_deprecated_api_requests_per_period integer DEFAULT 3600 NOT NULL,
     throttle_authenticated_deprecated_api_period_in_seconds integer DEFAULT 3600 NOT NULL,
     throttle_authenticated_deprecated_api_enabled boolean DEFAULT false NOT NULL,
+    dependency_proxy_ttl_group_policy_worker_capacity smallint 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_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_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)),
@@ -24837,11 +24839,19 @@ CREATE INDEX index_dep_ci_build_trace_sections_on_project_id ON dep_ci_build_tra
 
 CREATE INDEX index_dep_ci_build_trace_sections_on_section_name_id ON dep_ci_build_trace_sections USING btree (section_name_id);
 
+CREATE UNIQUE INDEX index_dep_prox_manifests_on_group_id_file_name_and_status ON dependency_proxy_manifests USING btree (group_id, file_name, status);
+
 CREATE INDEX index_dependency_proxy_blobs_on_group_id_and_file_name ON dependency_proxy_blobs USING btree (group_id, file_name);
 
+CREATE INDEX index_dependency_proxy_blobs_on_group_id_status_and_id ON dependency_proxy_blobs USING btree (group_id, status, id);
+
+CREATE INDEX index_dependency_proxy_blobs_on_status ON dependency_proxy_blobs USING btree (status);
+
 CREATE INDEX index_dependency_proxy_group_settings_on_group_id ON dependency_proxy_group_settings USING btree (group_id);
 
-CREATE UNIQUE INDEX index_dependency_proxy_manifests_on_group_id_and_file_name ON dependency_proxy_manifests USING btree (group_id, file_name);
+CREATE INDEX index_dependency_proxy_manifests_on_group_id_status_and_id ON dependency_proxy_manifests USING btree (group_id, status, id);
+
+CREATE INDEX index_dependency_proxy_manifests_on_status ON dependency_proxy_manifests USING btree (status);
 
 CREATE INDEX index_deploy_key_id_on_protected_branch_push_access_levels ON protected_branch_push_access_levels USING btree (deploy_key_id);
 

spec/factories/dependency_proxy.rb

@@ -6,6 +6,11 @@ FactoryBot.define do
     size { 1234 }
     file { fixture_file_upload('spec/fixtures/dependency_proxy/a3ed95caeb02ffe68cdd9fd84406680ae93d633cb16422d00e8a7c22955b46d4.gz') }
     file_name { 'a3ed95caeb02ffe68cdd9fd84406680ae93d633cb16422d00e8a7c22955b46d4.gz' }
+    status { :default }
+
+    trait :expired do
+      status { :expired }
+    end
   end
 
   factory :dependency_proxy_manifest, class: 'DependencyProxy::Manifest' do
@@ -13,7 +18,12 @@ FactoryBot.define do
     size { 1234 }
     file { fixture_file_upload('spec/fixtures/dependency_proxy/manifest') }
     digest { 'sha256:d0710affa17fad5f466a70159cc458227bd25d4afb39514ef662ead3e6c99515' }
-    file_name { 'alpine:latest.json' }
+    sequence(:file_name) { |n| "alpine:latest#{n}.json" }
     content_type { 'application/vnd.docker.distribution.manifest.v2+json' }
+    status { :default }
+
+    trait :expired do
+      status { :expired }
+    end
   end
 end

spec/factories/dependency_proxy/image_ttl_group_policies.rb

@@ -6,5 +6,9 @@ FactoryBot.define do
 
     enabled { true }
     ttl { 90 }
+
+    trait :disabled do
+      enabled { false }
+    end
   end
 end

spec/models/application_setting_spec.rb

@@ -77,6 +77,9 @@ RSpec.describe ApplicationSetting do
     it { is_expected.to validate_numericality_of(:container_registry_cleanup_tags_service_max_list_size).only_integer.is_greater_than_or_equal_to(0) }
     it { is_expected.to validate_numericality_of(:container_registry_expiration_policies_worker_capacity).only_integer.is_greater_than_or_equal_to(0) }
 
+    it { is_expected.to validate_numericality_of(:dependency_proxy_ttl_group_policy_worker_capacity).only_integer.is_greater_than_or_equal_to(0) }
+    it { is_expected.not_to allow_value(nil).for(:dependency_proxy_ttl_group_policy_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) }

spec/models/dependency_proxy/blob_spec.rb

@@ -2,17 +2,16 @@
 require 'spec_helper'
 
 RSpec.describe DependencyProxy::Blob, type: :model do
+  it_behaves_like 'ttl_expirable'
+
   describe 'relationships' do
     it { is_expected.to belong_to(:group) }
   end
 
-  it_behaves_like 'having unique enum values'
-
   describe 'validations' do
     it { is_expected.to validate_presence_of(:group) }
     it { is_expected.to validate_presence_of(:file) }
     it { is_expected.to validate_presence_of(:file_name) }
-    it { is_expected.to validate_presence_of(:status) }
   end
 
   describe '.total_size' do

spec/models/dependency_proxy/image_ttl_group_policy_spec.rb

@@ -20,4 +20,13 @@ RSpec.describe DependencyProxy::ImageTtlGroupPolicy, type: :model do
       it { is_expected.to validate_numericality_of(:ttl).allow_nil.is_greater_than(0) }
     end
   end
+
+  describe '.enabled' do
+    it 'returns policies that are enabled' do
+      enabled_policy = create(:image_ttl_group_policy)
+      create(:image_ttl_group_policy, :disabled)
+
+      expect(described_class.enabled).to contain_exactly(enabled_policy)
+    end
+  end
 end

spec/models/dependency_proxy/manifest_spec.rb

@@ -2,18 +2,17 @@
 require 'spec_helper'
 
 RSpec.describe DependencyProxy::Manifest, type: :model do
+  it_behaves_like 'ttl_expirable'
+
   describe 'relationships' do
     it { is_expected.to belong_to(:group) }
   end
 
-  it_behaves_like 'having unique enum values'
-
   describe 'validations' do
     it { is_expected.to validate_presence_of(:group) }
     it { is_expected.to validate_presence_of(:file) }
     it { is_expected.to validate_presence_of(:file_name) }
     it { is_expected.to validate_presence_of(:digest) }
-    it { is_expected.to validate_presence_of(:status) }
   end
 
   describe 'file is being stored' do

spec/services/dependency_proxy/find_or_create_blob_service_spec.rb

@@ -4,7 +4,8 @@ require 'spec_helper'
 RSpec.describe DependencyProxy::FindOrCreateBlobService do
   include DependencyProxyHelpers
 
-  let(:blob)  { create(:dependency_proxy_blob) }
+  let_it_be_with_reload(:blob) { create(:dependency_proxy_blob) }
+
   let(:group) { blob.group }
   let(:image) { 'alpine' }
   let(:tag)   { '3.9' }
@@ -17,11 +18,7 @@ RSpec.describe DependencyProxy::FindOrCreateBlobService do
     stub_registry_auth(image, token)
   end
 
-  context 'no cache' do
-    before do
-      stub_blob_download(image, blob_sha)
-    end
-
+  shared_examples 'downloads the remote blob' do
     it 'downloads blob from remote registry if there is no cached one' do
       expect(subject[:status]).to eq(:success)
       expect(subject[:blob]).to be_a(DependencyProxy::Blob)
@@ -30,15 +27,34 @@ RSpec.describe DependencyProxy::FindOrCreateBlobService do
     end
   end
 
+  context 'no cache' do
+    before do
+      stub_blob_download(image, blob_sha)
+    end
+
+    it_behaves_like 'downloads the remote blob'
+  end
+
   context 'cached blob' do
     let(:blob_sha) { blob.file_name.sub('.gz', '') }
 
     it 'uses cached blob instead of downloading one' do
+      expect { subject }.to change { blob.reload.updated_at }
+
       expect(subject[:status]).to eq(:success)
       expect(subject[:blob]).to be_a(DependencyProxy::Blob)
       expect(subject[:blob]).to eq(blob)
       expect(subject[:from_cache]).to eq true
     end
+
+    context 'when the cached blob is expired' do
+      before do
+        blob.update_column(:status, DependencyProxy::Blob.statuses[:expired])
+        stub_blob_download(image, blob_sha)
+      end
+
+      it_behaves_like 'downloads the remote blob'
+    end
   end
 
   context 'no such blob exists remotely' do

spec/services/dependency_proxy/find_or_create_manifest_service_spec.rb

@@ -21,19 +21,19 @@ RSpec.describe DependencyProxy::FindOrCreateManifestService do
   describe '#execute' do
     subject { described_class.new(group, image, tag, token).execute }
 
+    shared_examples 'downloading the manifest' do
+      it 'downloads manifest from remote registry if there is no cached one', :aggregate_failures do
+        expect { subject }.to change { group.dependency_proxy_manifests.count }.by(1)
+        expect(subject[:status]).to eq(:success)
+        expect(subject[:manifest]).to be_a(DependencyProxy::Manifest)
+        expect(subject[:manifest]).to be_persisted
+        expect(subject[:from_cache]).to eq false
+      end
+    end
+
     context 'when no manifest exists' do
       let_it_be(:image) { 'new-image' }
 
-      shared_examples 'downloading the manifest' do
-        it 'downloads manifest from remote registry if there is no cached one', :aggregate_failures do
-          expect { subject }.to change { group.dependency_proxy_manifests.count }.by(1)
-          expect(subject[:status]).to eq(:success)
-          expect(subject[:manifest]).to be_a(DependencyProxy::Manifest)
-          expect(subject[:manifest]).to be_persisted
-          expect(subject[:from_cache]).to eq false
-        end
-      end
-
       context 'successful head request' do
         before do
           stub_manifest_head(image, tag, headers: headers)
@@ -60,6 +60,8 @@ RSpec.describe DependencyProxy::FindOrCreateManifestService do
 
       shared_examples 'using the cached manifest' do
         it 'uses cached manifest instead of downloading one', :aggregate_failures do
+          expect { subject }.to change { dependency_proxy_manifest.reload.updated_at }
+
           expect(subject[:status]).to eq(:success)
           expect(subject[:manifest]).to be_a(DependencyProxy::Manifest)
           expect(subject[:manifest]).to eq(dependency_proxy_manifest)
@@ -87,6 +89,16 @@ RSpec.describe DependencyProxy::FindOrCreateManifestService do
         end
       end
 
+      context 'when the cached manifest is expired' do
+        before do
+          dependency_proxy_manifest.update_column(:status, DependencyProxy::Manifest.statuses[:expired])
+          stub_manifest_head(image, tag, headers: headers)
+          stub_manifest_download(image, tag, headers: headers)
+        end
+
+        it_behaves_like 'downloading the manifest'
+      end
+
       context 'failed connection' do
         before do
           expect(DependencyProxy::HeadManifestService).to receive(:new).and_raise(Net::OpenTimeout)

spec/support/shared_examples/models/concerns/ttl_expirable_shared_examples.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.shared_examples 'ttl_expirable' do
+  let_it_be(:class_symbol) { described_class.model_name.param_key.to_sym }
+
+  it_behaves_like 'having unique enum values'
+
+  describe 'validations' do
+    it { is_expected.to validate_presence_of(:status) }
+  end
+
+  describe '.updated_before' do
+    # rubocop:disable Rails/SaveBang
+    let_it_be_with_reload(:item1) { create(class_symbol) }
+    let_it_be(:item2) { create(class_symbol) }
+    # rubocop:enable Rails/SaveBang
+
+    before do
+      item1.update_column(:updated_at, 1.month.ago)
+    end
+
+    it 'returns items with created at older than the supplied number of days' do
+      expect(described_class.updated_before(10)).to contain_exactly(item1)
+    end
+  end
+
+  describe '.active' do
+    # rubocop:disable Rails/SaveBang
+    let_it_be(:item1) { create(class_symbol) }
+    let_it_be(:item2) { create(class_symbol, :expired) }
+    let_it_be(:item3) { create(class_symbol, status: :error) }
+    # rubocop:enable Rails/SaveBang
+
+    it 'returns only active items' do
+      expect(described_class.active).to contain_exactly(item1)
+    end
+  end
+
+  describe '.lock_next_by' do
+    let_it_be(:item1) { create(class_symbol, created_at: 1.month.ago, updated_at: 1.day.ago) }
+    let_it_be(:item2) { create(class_symbol, created_at: 1.year.ago, updated_at: 1.year.ago) }
+    let_it_be(:item3) { create(class_symbol, created_at: 2.years.ago, updated_at: 1.month.ago) }
+
+    it 'returns the first item sorted by the argument' do
+      expect(described_class.lock_next_by(:updated_at)).to contain_exactly(item2)
+      expect(described_class.lock_next_by(:created_at)).to contain_exactly(item3)
+    end
+  end
+end

spec/support/shared_examples/workers/concerns/dependency_proxy/cleanup_worker_shared_examples.rb

+# frozen_string_literal: true
+
+RSpec.shared_examples 'dependency_proxy_cleanup_worker' do
+  let_it_be(:group) { create(:group) }
+
+  let(:worker) { described_class.new }
+
+  describe '#perform_work' do
+    subject(:perform_work) { worker.perform_work }
+
+    context 'with no work to do' do
+      it { is_expected.to be_nil }
+    end
+
+    context 'with work to do' do
+      let_it_be(:artifact1) { create(factory_type, :expired, group: group) }
+      let_it_be(:artifact2) { create(factory_type, :expired, group: group, updated_at: 6.months.ago, created_at: 2.years.ago) }
+      let_it_be_with_reload(:artifact3) { create(factory_type, :expired, group: group, updated_at: 1.year.ago, created_at: 1.year.ago) }
+      let_it_be(:artifact4) { create(factory_type, group: group, updated_at: 2.years.ago, created_at: 2.years.ago) }
+
+      it 'deletes the oldest expired artifact based on updated_at', :aggregate_failures do
+        expect(worker).to receive(:log_extra_metadata_on_done).with("#{factory_type}_id".to_sym, artifact3.id)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:group_id, group.id)
+
+        expect { perform_work }.to change { artifact1.class.count }.by(-1)
+      end
+    end
+  end
+
+  describe '#max_running_jobs' do
+    let(:capacity) { 5 }
+
+    subject { worker.max_running_jobs }
+
+    before do
+      stub_application_setting(dependency_proxy_ttl_group_policy_worker_capacity: capacity)
+    end
+
+    it { is_expected.to eq(capacity) }
+  end
+
+  describe '#remaining_work_count' do
+    let_it_be(:expired_artifacts) do
+      (1..3).map do |_|
+        create(factory_type, :expired, group: group)
+      end
+    end
+
+    subject { worker.remaining_work_count }
+
+    it { is_expected.to eq(3) }
+  end
+end

spec/workers/dependency_proxy/cleanup_blob_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe DependencyProxy::CleanupBlobWorker do
+  let_it_be(:factory_type) { :dependency_proxy_blob }
+
+  it_behaves_like 'dependency_proxy_cleanup_worker'
+end

spec/workers/dependency_proxy/cleanup_manifest_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe DependencyProxy::CleanupManifestWorker do
+  let_it_be(:factory_type) { :dependency_proxy_manifest }
+
+  it_behaves_like 'dependency_proxy_cleanup_worker'
+end

spec/workers/dependency_proxy/image_ttl_group_policy_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe DependencyProxy::ImageTtlGroupPolicyWorker do
+  let(:worker) { described_class.new }
+
+  describe '#perform' do
+    let_it_be(:policy) { create(:image_ttl_group_policy) }
+    let_it_be(:group) { policy.group }
+
+    subject { worker.perform }
+
+    context 'when there are images to expire' do
+      let_it_be_with_reload(:old_blob) { create(:dependency_proxy_blob, group: group, updated_at: 1.year.ago) }
+      let_it_be_with_reload(:old_manifest) { create(:dependency_proxy_manifest, group: group, updated_at: 1.year.ago) }
+      let_it_be_with_reload(:new_blob) { create(:dependency_proxy_blob, group: group) }
+      let_it_be_with_reload(:new_manifest) { create(:dependency_proxy_manifest, group: group) }
+
+      it 'calls the limited capacity workers', :aggregate_failures do
+        expect(DependencyProxy::CleanupBlobWorker).to receive(:perform_with_capacity)
+        expect(DependencyProxy::CleanupManifestWorker).to receive(:perform_with_capacity)
+
+        subject
+      end
+
+      it 'updates the old images to expired' do
+        expect { subject }
+          .to change { old_blob.reload.status }.from('default').to('expired')
+          .and change { old_manifest.reload.status }.from('default').to('expired')
+          .and not_change { new_blob.reload.status }
+          .and not_change { new_manifest.reload.status }
+      end
+    end
+
+    context 'when there are no images to expire' do
+      it 'does not do anything', :aggregate_failures do
+        expect(DependencyProxy::CleanupBlobWorker).not_to receive(:perform_with_capacity)
+        expect(DependencyProxy::CleanupManifestWorker).not_to receive(:perform_with_capacity)
+
+        subject
+      end
+    end
+
+    context 'counts logging' do
+      let_it_be(:expired_blob) { create(:dependency_proxy_blob, :expired, group: group) }
+      let_it_be(:expired_blob2) { create(:dependency_proxy_blob, :expired, group: group) }
+      let_it_be(:expired_manifest) { create(:dependency_proxy_manifest, :expired, group: group) }
+      let_it_be(:processing_blob) { create(:dependency_proxy_blob, status: :processing, group: group) }
+      let_it_be(:processing_manifest) { create(:dependency_proxy_manifest, status: :processing, group: group) }
+      let_it_be(:error_blob) { create(:dependency_proxy_blob, status: :error, group: group) }
+      let_it_be(:error_manifest) { create(:dependency_proxy_manifest, status: :error, group: group) }
+
+      it 'logs all the counts', :aggregate_failures do
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:expired_dependency_proxy_blob_count, 2)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:expired_dependency_proxy_manifest_count, 1)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:processing_dependency_proxy_blob_count, 1)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:processing_dependency_proxy_manifest_count, 1)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:error_dependency_proxy_blob_count, 1)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:error_dependency_proxy_manifest_count, 1)
+
+        subject
+      end
+
+      context 'with load balancing enabled', :db_load_balancing do
+        it 'reads the counts from the replica' do
+          expect(Gitlab::Database::LoadBalancing::Session.current).to receive(:use_replicas_for_read_queries).and_call_original
+
+          subject
+        end
+      end
+    end
+  end
+end

spec/workers/every_sidekiq_worker_spec.rb

@@ -198,6 +198,8 @@ RSpec.describe 'Every Sidekiq worker' do
         'DeleteMergedBranchesWorker' => 3,
         'DeleteStoredFilesWorker' => 3,
         'DeleteUserWorker' => 3,
+        'DependencyProxy::CleanupBlobWorker' => 3,
+        'DependencyProxy::CleanupManifestWorker' => 3,
         'Deployments::AutoRollbackWorker' => 3,
         'Deployments::DropOlderDeploymentsWorker' => 3,
         'Deployments::FinishedWorker' => 3,