Add latest changes from gitlab-org/gitlab@14-4-stable-eev14.4.0-rc42

46 files changed, 569 insertions, 150 deletions

diff --git a/app/workers/all_queues.yml b/app/workers/all_queues.yml
index 955674b52a4..c7ce2eb8d00 100644
--- a/app/workers/all_queues.yml
+++ b/app/workers/all_queues.yml
@@ -30,6 +30,15 @@
   :weight: 1
   :idempotent: true
   :tags: []
+- :name: authorized_project_update:authorized_project_update_project_recalculate_per_user
+  :worker_name: AuthorizedProjectUpdate::ProjectRecalculatePerUserWorker
+  :feature_category: :authentication_and_authorization
+  :has_external_dependencies:
+  :urgency: :high
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: authorized_project_update:authorized_project_update_user_refresh_from_replica
   :worker_name: AuthorizedProjectUpdate::UserRefreshFromReplicaWorker
   :feature_category: :authentication_and_authorization
@@ -46,7 +55,7 @@
   :urgency: :low
   :resource_boundary: :unknown
   :weight: 1
-  :idempotent:
+  :idempotent: true
   :tags: []
 - :name: authorized_project_update:authorized_project_update_user_refresh_with_low_urgency
   :worker_name: AuthorizedProjectUpdate::UserRefreshWithLowUrgencyWorker
@@ -185,7 +194,7 @@
   :tags: []
 - :name: cronjob:ci_delete_unit_tests
   :worker_name: Ci::DeleteUnitTestsWorker
-  :feature_category: :continuous_integration
+  :feature_category: :code_testing
   :has_external_dependencies:
   :urgency: :low
   :resource_boundary: :unknown
@@ -194,7 +203,7 @@
   :tags: []
 - :name: cronjob:ci_pipeline_artifacts_expire_artifacts
   :worker_name: Ci::PipelineArtifacts::ExpireArtifactsWorker
-  :feature_category: :continuous_integration
+  :feature_category: :build_artifacts
   :has_external_dependencies:
   :urgency: :low
   :resource_boundary: :unknown
@@ -219,6 +228,24 @@
   :weight: 1
   :idempotent: true
   :tags: []
+- :name: cronjob:ci_stuck_builds_drop_running
+  :worker_name: Ci::StuckBuilds::DropRunningWorker
+  :feature_category: :continuous_integration
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
+- :name: cronjob:ci_stuck_builds_drop_scheduled
+  :worker_name: Ci::StuckBuilds::DropScheduledWorker
+  :feature_category: :continuous_integration
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: cronjob:container_expiration_policy
   :worker_name: ContainerExpirationPolicyWorker
   :feature_category: :container_registry
@@ -255,6 +282,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
@@ -275,7 +311,7 @@
   :tags: []
 - :name: cronjob:expire_build_artifacts
   :worker_name: ExpireBuildArtifactsWorker
-  :feature_category: :continuous_integration
+  :feature_category: :build_artifacts
   :has_external_dependencies:
   :urgency: :low
   :resource_boundary: :unknown
@@ -347,7 +383,7 @@
   :tags: []
 - :name: cronjob:namespaces_in_product_marketing_emails
   :worker_name: Namespaces::InProductMarketingEmailsWorker
-  :feature_category: :subgroups
+  :feature_category: :experimentation_activation
   :has_external_dependencies:
   :urgency: :low
   :resource_boundary: :unknown
@@ -557,7 +593,7 @@
   :feature_category: :continuous_integration
   :has_external_dependencies:
   :urgency: :low
-  :resource_boundary: :cpu
+  :resource_boundary: :unknown
   :weight: 1
   :idempotent:
   :tags: []
@@ -642,6 +678,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
@@ -1418,7 +1472,7 @@
   :urgency: :high
   :resource_boundary: :unknown
   :weight: 3
-  :idempotent:
+  :idempotent: true
   :tags: []
 - :name: pipeline_cache:expire_pipeline_cache
   :worker_name: ExpirePipelineCacheWorker
@@ -1474,6 +1528,15 @@
   :weight: 3
   :idempotent:
   :tags: []
+- :name: pipeline_default:ci_create_downstream_pipeline
+  :worker_name: Ci::CreateDownstreamPipelineWorker
+  :feature_category: :continuous_integration
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :cpu
+  :weight: 3
+  :idempotent:
+  :tags: []
 - :name: pipeline_default:ci_drop_pipeline
   :worker_name: Ci::DropPipelineWorker
   :feature_category: :continuous_integration
@@ -1890,7 +1953,7 @@
   :tags: []
 - :name: default
   :worker_name:
-  :feature_category:
+  :feature_category: :not_owned
   :has_external_dependencies:
   :urgency:
   :resource_boundary:
@@ -2044,7 +2107,7 @@
   :tags: []
 - :name: expire_build_instance_artifacts
   :worker_name: ExpireBuildInstanceArtifactsWorker
-  :feature_category: :continuous_integration
+  :feature_category: :build_artifacts
   :has_external_dependencies:
   :urgency: :low
   :resource_boundary: :unknown
@@ -2215,7 +2278,7 @@
   :tags: []
 - :name: mailers
   :worker_name: ActionMailer::MailDeliveryJob
-  :feature_category: :issue_tracking
+  :feature_category: :not_owned
   :has_external_dependencies:
   :urgency: low
   :resource_boundary:
@@ -2314,7 +2377,7 @@
   :tags: []
 - :name: namespaces_onboarding_issue_created
   :worker_name: Namespaces::OnboardingIssueCreatedWorker
-  :feature_category: :issue_tracking
+  :feature_category: :onboarding
   :has_external_dependencies:
   :urgency: :low
   :resource_boundary: :unknown
@@ -2323,7 +2386,7 @@
   :tags: []
 - :name: namespaces_onboarding_pipeline_created
   :worker_name: Namespaces::OnboardingPipelineCreatedWorker
-  :feature_category: :subgroups
+  :feature_category: :onboarding
   :has_external_dependencies:
   :urgency: :low
   :resource_boundary: :unknown
@@ -2332,7 +2395,7 @@
   :tags: []
 - :name: namespaces_onboarding_progress
   :worker_name: Namespaces::OnboardingProgressWorker
-  :feature_category: :product_analytics
+  :feature_category: :onboarding
   :has_external_dependencies:
   :urgency: :low
   :resource_boundary: :cpu
@@ -2341,7 +2404,7 @@
   :tags: []
 - :name: namespaces_onboarding_user_added
   :worker_name: Namespaces::OnboardingUserAddedWorker
-  :feature_category: :users
+  :feature_category: :onboarding
   :has_external_dependencies:
   :urgency: :low
   :resource_boundary: :unknown
@@ -2411,15 +2474,6 @@
   :weight: 1
   :idempotent:
   :tags: []
-- :name: pages_remove
-  :worker_name: PagesRemoveWorker
-  :feature_category: :pages
-  :has_external_dependencies:
-  :urgency: :low
-  :resource_boundary: :unknown
-  :weight: 1
-  :idempotent:
-  :tags: []
 - :name: pages_transfer
   :worker_name: PagesTransferWorker
   :feature_category: :pages
diff --git a/app/workers/authorized_project_update/project_recalculate_per_user_worker.rb b/app/workers/authorized_project_update/project_recalculate_per_user_worker.rb
new file mode 100644
index 00000000000..352c82e5021
--- /dev/null
+++ b/app/workers/authorized_project_update/project_recalculate_per_user_worker.rb
@@ -0,0 +1,25 @@
+# frozen_string_literal: true
+
+module AuthorizedProjectUpdate
+  class ProjectRecalculatePerUserWorker < ProjectRecalculateWorker
+    data_consistency :always
+
+    feature_category :authentication_and_authorization
+    urgency :high
+    queue_namespace :authorized_project_update
+
+    deduplicate :until_executing, including_scheduled: true
+    idempotent!
+
+    def perform(project_id, user_id)
+      project = Project.find_by_id(project_id)
+      user = User.find_by_id(user_id)
+
+      return unless project && user
+
+      in_lock(lock_key(project), ttl: 10.seconds) do
+        AuthorizedProjectUpdate::ProjectRecalculatePerUserService.new(project, user).execute
+      end
+    end
+  end
+end
diff --git a/app/workers/authorized_project_update/project_recalculate_worker.rb b/app/workers/authorized_project_update/project_recalculate_worker.rb
index 4d350d95e7e..3d073f18622 100644
--- a/app/workers/authorized_project_update/project_recalculate_worker.rb
+++ b/app/workers/authorized_project_update/project_recalculate_worker.rb
@@ -26,7 +26,9 @@ module AuthorizedProjectUpdate
     private
 
     def lock_key(project)
-      "#{self.class.name.underscore}/projects/#{project.id}"
+      # The self.class.name.underscore value is hardcoded here as the prefix, so that the same
+      # lock_key for this superclass will be used by the ProjectRecalculatePerUserWorker subclass.
+      "authorized_project_update/project_recalculate_worker/projects/#{project.id}"
     end
   end
 end
diff --git a/app/workers/authorized_project_update/user_refresh_from_replica_worker.rb b/app/workers/authorized_project_update/user_refresh_from_replica_worker.rb
index 48e3d0837c7..daebb23baae 100644
--- a/app/workers/authorized_project_update/user_refresh_from_replica_worker.rb
+++ b/app/workers/authorized_project_update/user_refresh_from_replica_worker.rb
@@ -30,8 +30,6 @@ module AuthorizedProjectUpdate
     # does not allow us to deduplicate these jobs.
     # https://gitlab.com/gitlab-org/gitlab/-/issues/325291
     def use_replica_if_available(&block)
-      return yield unless ::Gitlab::Database::LoadBalancing.enable?
-
       ::Gitlab::Database::LoadBalancing::Session.current.use_replicas_for_read_queries(&block)
     end
 
diff --git a/app/workers/authorized_project_update/user_refresh_over_user_range_worker.rb b/app/workers/authorized_project_update/user_refresh_over_user_range_worker.rb
index ab4d9c13422..f5327449242 100644
--- a/app/workers/authorized_project_update/user_refresh_over_user_range_worker.rb
+++ b/app/workers/authorized_project_update/user_refresh_over_user_range_worker.rb
@@ -19,11 +19,10 @@ module AuthorizedProjectUpdate
     feature_category :authentication_and_authorization
     urgency :low
     queue_namespace :authorized_project_update
-    # This job will not be deduplicated since it is marked with
-    # `data_consistency :delayed` and not `idempotent!`
-    # See https://gitlab.com/gitlab-org/gitlab/-/issues/325291
+
     deduplicate :until_executing, including_scheduled: true
     data_consistency :delayed
+    idempotent!
 
     def perform(start_user_id, end_user_id)
       User.where(id: start_user_id..end_user_id).find_each do |user| # rubocop: disable CodeReuse/ActiveRecord
diff --git a/app/workers/bulk_import_worker.rb b/app/workers/bulk_import_worker.rb
index fa255d064cc..d560ebcc6e6 100644
--- a/app/workers/bulk_import_worker.rb
+++ b/app/workers/bulk_import_worker.rb
@@ -26,7 +26,7 @@ class BulkImportWorker # rubocop:disable Scalability/IdempotentWorker
     created_entities.first(next_batch_size).each do |entity|
       entity.create_pipeline_trackers!
 
-      BulkImports::ExportRequestWorker.perform_async(entity.id) if entity.group_entity?
+      BulkImports::ExportRequestWorker.perform_async(entity.id)
       BulkImports::EntityWorker.perform_async(entity.id)
 
       entity.start!
diff --git a/app/workers/bulk_imports/export_request_worker.rb b/app/workers/bulk_imports/export_request_worker.rb
index d5f7215b08a..8bc0acc9b22 100644
--- a/app/workers/bulk_imports/export_request_worker.rb
+++ b/app/workers/bulk_imports/export_request_worker.rb
@@ -10,8 +10,6 @@ module BulkImports
     worker_has_external_dependencies!
     feature_category :importers
 
-    GROUP_EXPORTED_URL_PATH = "/groups/%s/export_relations"
-
     def perform(entity_id)
       entity = BulkImports::Entity.find(entity_id)
 
@@ -21,8 +19,7 @@ module BulkImports
     private
 
     def request_export(entity)
-      http_client(entity.bulk_import.configuration)
-        .post(GROUP_EXPORTED_URL_PATH % entity.encoded_source_full_path)
+      http_client(entity.bulk_import.configuration).post(entity.export_relations_url_path)
     end
 
     def http_client(configuration)
diff --git a/app/workers/bulk_imports/pipeline_worker.rb b/app/workers/bulk_imports/pipeline_worker.rb
index 760a309a381..35633b55489 100644
--- a/app/workers/bulk_imports/pipeline_worker.rb
+++ b/app/workers/bulk_imports/pipeline_worker.rb
@@ -16,7 +16,7 @@ module BulkImports
 
     def perform(pipeline_tracker_id, stage, entity_id)
       pipeline_tracker = ::BulkImports::Tracker
-        .with_status(:created)
+        .with_status(:created, :started)
         .find_by_id(pipeline_tracker_id)
 
       if pipeline_tracker.present?
@@ -59,18 +59,35 @@ module BulkImports
       pipeline_tracker.pipeline_class.new(context).run
 
       pipeline_tracker.finish!
+    rescue BulkImports::NetworkError => e
+      if e.retriable?(pipeline_tracker)
+        logger.error(
+          worker: self.class.name,
+          entity_id: pipeline_tracker.entity.id,
+          pipeline_name: pipeline_tracker.pipeline_name,
+          message: "Retrying error: #{e.message}"
+        )
+
+        reenqueue(pipeline_tracker, delay: e.retry_delay)
+      else
+        fail_tracker(pipeline_tracker, e)
+      end
     rescue StandardError => e
+      fail_tracker(pipeline_tracker, e)
+    end
+
+    def fail_tracker(pipeline_tracker, exception)
       pipeline_tracker.update!(status_event: 'fail_op', jid: jid)
 
       logger.error(
         worker: self.class.name,
         entity_id: pipeline_tracker.entity.id,
         pipeline_name: pipeline_tracker.pipeline_name,
-        message: e.message
+        message: exception.message
       )
 
       Gitlab::ErrorTracking.track_exception(
-        e,
+        exception,
         entity_id: pipeline_tracker.entity.id,
         pipeline_name: pipeline_tracker.pipeline_name
       )
@@ -88,8 +105,13 @@ module BulkImports
       (Time.zone.now - pipeline_tracker.entity.created_at) > Pipeline::NDJSON_EXPORT_TIMEOUT
     end
 
-    def reenqueue(pipeline_tracker)
-      self.class.perform_in(NDJSON_PIPELINE_PERFORM_DELAY, pipeline_tracker.id, pipeline_tracker.stage, pipeline_tracker.entity.id)
+    def reenqueue(pipeline_tracker, delay: NDJSON_PIPELINE_PERFORM_DELAY)
+      self.class.perform_in(
+        delay,
+        pipeline_tracker.id,
+        pipeline_tracker.stage,
+        pipeline_tracker.entity.id
+      )
     end
   end
 end
diff --git a/app/workers/ci/build_finished_worker.rb b/app/workers/ci/build_finished_worker.rb
index 3bca3015988..f047ba8fde5 100644
--- a/app/workers/ci/build_finished_worker.rb
+++ b/app/workers/ci/build_finished_worker.rb
@@ -15,13 +15,13 @@ module Ci
 
     ARCHIVE_TRACES_IN = 2.minutes.freeze
 
-    # rubocop: disable CodeReuse/ActiveRecord
     def perform(build_id)
-      Ci::Build.find_by(id: build_id).try do |build|
-        process_build(build)
-      end
+      return unless build = Ci::Build.find_by(id: build_id) # rubocop: disable CodeReuse/ActiveRecord
+      return unless build.project
+      return if build.project.pending_delete?
+
+      process_build(build)
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
     private
 
diff --git a/app/workers/ci/create_downstream_pipeline_worker.rb b/app/workers/ci/create_downstream_pipeline_worker.rb
new file mode 100644
index 00000000000..6d4cd2539c1
--- /dev/null
+++ b/app/workers/ci/create_downstream_pipeline_worker.rb
@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+module Ci
+  class CreateDownstreamPipelineWorker # rubocop:disable Scalability/IdempotentWorker
+    include ::ApplicationWorker
+    include ::PipelineQueue
+
+    sidekiq_options retry: 3
+    worker_resource_boundary :cpu
+
+    def perform(bridge_id)
+      ::Ci::Bridge.find_by_id(bridge_id).try do |bridge|
+        ::Ci::CreateDownstreamPipelineService
+          .new(bridge.project, bridge.user)
+          .execute(bridge)
+      end
+    end
+  end
+end
diff --git a/app/workers/ci/delete_unit_tests_worker.rb b/app/workers/ci/delete_unit_tests_worker.rb
index d5bb72ce80c..01d909773d2 100644
--- a/app/workers/ci/delete_unit_tests_worker.rb
+++ b/app/workers/ci/delete_unit_tests_worker.rb
@@ -10,7 +10,7 @@ module Ci
     include CronjobQueue
     # rubocop:enable Scalability/CronWorkerContext
 
-    feature_category :continuous_integration
+    feature_category :code_testing
     idempotent!
 
     def perform
diff --git a/app/workers/ci/pipeline_artifacts/expire_artifacts_worker.rb b/app/workers/ci/pipeline_artifacts/expire_artifacts_worker.rb
index 2af07cf6f93..cde3d71286e 100644
--- a/app/workers/ci/pipeline_artifacts/expire_artifacts_worker.rb
+++ b/app/workers/ci/pipeline_artifacts/expire_artifacts_worker.rb
@@ -14,7 +14,7 @@ module Ci
 
       deduplicate :until_executed, including_scheduled: true
       idempotent!
-      feature_category :continuous_integration
+      feature_category :build_artifacts
 
       def perform
         service = ::Ci::PipelineArtifacts::DestroyAllExpiredService.new
diff --git a/app/workers/ci/stuck_builds/drop_running_worker.rb b/app/workers/ci/stuck_builds/drop_running_worker.rb
new file mode 100644
index 00000000000..db571fdc38d
--- /dev/null
+++ b/app/workers/ci/stuck_builds/drop_running_worker.rb
@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+module Ci
+  module StuckBuilds
+    class DropRunningWorker
+      include ApplicationWorker
+      include ExclusiveLeaseGuard
+
+      idempotent!
+
+      # rubocop:disable Scalability/CronWorkerContext
+      # This is an instance-wide cleanup query, so there's no meaningful
+      # scope to consider this in the context of.
+      include CronjobQueue
+      # rubocop:enable Scalability/CronWorkerContext
+
+      data_consistency :always
+
+      feature_category :continuous_integration
+
+      def perform
+        try_obtain_lease do
+          Ci::StuckBuilds::DropRunningService.new.execute
+        end
+      end
+
+      private
+
+      def lease_timeout
+        30.minutes
+      end
+    end
+  end
+end
diff --git a/app/workers/ci/stuck_builds/drop_scheduled_worker.rb b/app/workers/ci/stuck_builds/drop_scheduled_worker.rb
new file mode 100644
index 00000000000..923841771cf
--- /dev/null
+++ b/app/workers/ci/stuck_builds/drop_scheduled_worker.rb
@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+module Ci
+  module StuckBuilds
+    class DropScheduledWorker
+      include ApplicationWorker
+      include ExclusiveLeaseGuard
+
+      idempotent!
+
+      # rubocop:disable Scalability/CronWorkerContext
+      # This is an instance-wide cleanup query, so there's no meaningful
+      # scope to consider this in the context of.
+      include CronjobQueue
+      # rubocop:enable Scalability/CronWorkerContext
+
+      data_consistency :always
+
+      feature_category :continuous_integration
+
+      def perform
+        try_obtain_lease do
+          Ci::StuckBuilds::DropScheduledService.new.execute
+        end
+      end
+
+      private
+
+      def lease_timeout
+        30.minutes
+      end
+    end
+  end
+end
diff --git a/app/workers/cleanup_container_repository_worker.rb b/app/workers/cleanup_container_repository_worker.rb
index 9adc026ced2..7274ecf62f9 100644
--- a/app/workers/cleanup_container_repository_worker.rb
+++ b/app/workers/cleanup_container_repository_worker.rb
@@ -28,8 +28,8 @@ class CleanupContainerRepositoryWorker
     end
 
     result = Projects::ContainerRepository::CleanupTagsService
-      .new(project, current_user, params)
-      .execute(container_repository)
+      .new(container_repository, current_user, params)
+      .execute
 
     if run_by_container_expiration_policy? && result[:status] == :success
       container_repository.reset_expiration_policy_started_at!
diff --git a/app/workers/concerns/dependency_proxy/cleanup_worker.rb b/app/workers/concerns/dependency_proxy/cleanup_worker.rb
new file mode 100644
index 00000000000..b668634f233
--- /dev/null
+++ b/app/workers/concerns/dependency_proxy/cleanup_worker.rb
@@ -0,0 +1,63 @@
+# 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
diff --git a/app/workers/concerns/gitlab/github_import/object_importer.rb b/app/workers/concerns/gitlab/github_import/object_importer.rb
index a377b7a2000..e1f404b250d 100644
--- a/app/workers/concerns/gitlab/github_import/object_importer.rb
+++ b/app/workers/concerns/gitlab/github_import/object_importer.rb
@@ -26,8 +26,7 @@ module Gitlab
         object = representation_class.from_json_hash(hash)
 
         # To better express in the logs what object is being imported.
-        self.github_id = object.attributes.fetch(:github_id)
-
+        self.github_identifiers = object.github_identifiers
         info(project.id, message: 'starting importer')
 
         importer_class.new(object, project, client).execute
@@ -35,10 +34,10 @@ module Gitlab
         Gitlab::GithubImport::ObjectCounter.increment(project, object_type, :imported)
 
         info(project.id, message: 'importer finished')
-      rescue KeyError => e
+      rescue NoMethodError => e
         # This exception will be more useful in development when a new
         # Representation is created but the developer forgot to add a
-        # `:github_id` field.
+        # `:github_identifiers` field.
         Gitlab::Import::ImportFailureService.track(
           project_id: project.id,
           error_source: importer_class.name,
@@ -72,7 +71,7 @@ module Gitlab
 
       private
 
-      attr_accessor :github_id
+      attr_accessor :github_identifiers
 
       def info(project_id, extra = {})
         Logger.info(log_attributes(project_id, extra))
@@ -82,7 +81,7 @@ module Gitlab
         extra.merge(
           project_id: project_id,
           importer: importer_class.name,
-          github_id: github_id
+          github_identifiers: github_identifiers
         )
       end
     end
diff --git a/app/workers/concerns/worker_attributes.rb b/app/workers/concerns/worker_attributes.rb
index eebea30655c..6f91418e38c 100644
--- a/app/workers/concerns/worker_attributes.rb
+++ b/app/workers/concerns/worker_attributes.rb
@@ -46,8 +46,14 @@ module WorkerAttributes
       set_class_attribute(:feature_category, :not_owned)
     end
 
+    # Special case: if a worker is not owned, get the feature category
+    # (if present) from the calling context.
     def get_feature_category
-      get_class_attribute(:feature_category)
+      feature_category = get_class_attribute(:feature_category)
+
+      return feature_category unless feature_category == :not_owned
+
+      Gitlab::ApplicationContext.current_context_attribute('meta.feature_category') || feature_category
     end
 
     def feature_category_not_owned?
diff --git a/app/workers/container_expiration_policies/cleanup_container_repository_worker.rb b/app/workers/container_expiration_policies/cleanup_container_repository_worker.rb
index 433ed5e0ea4..69f5906f54c 100644
--- a/app/workers/container_expiration_policies/cleanup_container_repository_worker.rb
+++ b/app/workers/container_expiration_policies/cleanup_container_repository_worker.rb
@@ -21,6 +21,7 @@ module ContainerExpirationPolicies
       cleanup_tags_service_original_size
       cleanup_tags_service_before_truncate_size
       cleanup_tags_service_after_truncate_size
+      cleanup_tags_service_cached_tags_count
       cleanup_tags_service_before_delete_size
       cleanup_tags_service_deleted_size
     ].freeze
@@ -147,13 +148,27 @@ module ContainerExpirationPolicies
         log_extra_metadata_on_done(field, value)
       end
 
+      log_truncate(result)
+      log_cache_ratio(result)
+      log_extra_metadata_on_done(:running_jobs_count, running_jobs_count)
+    end
+
+    def log_cache_ratio(result)
+      tags_count = result.payload[:cleanup_tags_service_after_truncate_size]
+      cached_tags_count = result.payload[:cleanup_tags_service_cached_tags_count]
+
+      return unless tags_count && cached_tags_count && tags_count != 0
+
+      log_extra_metadata_on_done(:cleanup_tags_service_cache_hit_ratio, cached_tags_count / tags_count.to_f)
+    end
+
+    def log_truncate(result)
       before_truncate_size = result.payload[:cleanup_tags_service_before_truncate_size]
       after_truncate_size = result.payload[:cleanup_tags_service_after_truncate_size]
       truncated = before_truncate_size &&
                     after_truncate_size &&
                     before_truncate_size != after_truncate_size
       log_extra_metadata_on_done(:cleanup_tags_service_truncated, !!truncated)
-      log_extra_metadata_on_done(:running_jobs_count, running_jobs_count)
     end
 
     def policy
diff --git a/app/workers/container_expiration_policy_worker.rb b/app/workers/container_expiration_policy_worker.rb
index a791fe5d350..5fcbd74ddad 100644
--- a/app/workers/container_expiration_policy_worker.rb
+++ b/app/workers/container_expiration_policy_worker.rb
@@ -45,8 +45,6 @@ class ContainerExpirationPolicyWorker # rubocop:disable Scalability/IdempotentWo
   # not perfomed with a delay
   # https://gitlab.com/gitlab-org/gitlab/-/merge_requests/63635#note_603771207
   def use_replica_if_available(&blk)
-    return yield unless ::Gitlab::Database::LoadBalancing.enable?
-
     ::Gitlab::Database::LoadBalancing::Session.current.use_replicas_for_read_queries(&blk)
   end
 
diff --git a/app/workers/create_note_diff_file_worker.rb b/app/workers/create_note_diff_file_worker.rb
index 4bea4fc872e..8481fd0a2ab 100644
--- a/app/workers/create_note_diff_file_worker.rb
+++ b/app/workers/create_note_diff_file_worker.rb
@@ -10,8 +10,10 @@ class CreateNoteDiffFileWorker # rubocop:disable Scalability/IdempotentWorker
   feature_category :code_review
 
   def perform(diff_note_id)
-    diff_note = DiffNote.find(diff_note_id)
+    return unless diff_note_id.present?
 
-    diff_note.create_diff_file
+    diff_note = DiffNote.find_by_id(diff_note_id) # rubocop: disable CodeReuse/ActiveRecord
+
+    diff_note&.create_diff_file
   end
 end
diff --git a/app/workers/database/drop_detached_partitions_worker.rb b/app/workers/database/drop_detached_partitions_worker.rb
index f9c8ce57a36..1e4dc20a0d2 100644
--- a/app/workers/database/drop_detached_partitions_worker.rb
+++ b/app/workers/database/drop_detached_partitions_worker.rb
@@ -10,7 +10,7 @@ module Database
     idempotent!
 
     def perform
-      Gitlab::Database::Partitioning::DetachedPartitionDropper.new.perform
+      Gitlab::Database::Partitioning.drop_detached_partitions
     ensure
       Gitlab::Database::Partitioning::PartitionMonitoring.new.report_metrics
     end
diff --git a/app/workers/dependency_proxy/cleanup_blob_worker.rb b/app/workers/dependency_proxy/cleanup_blob_worker.rb
new file mode 100644
index 00000000000..054bc5854a3
--- /dev/null
+++ b/app/workers/dependency_proxy/cleanup_blob_worker.rb
@@ -0,0 +1,40 @@
+# 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
diff --git a/app/workers/dependency_proxy/cleanup_manifest_worker.rb b/app/workers/dependency_proxy/cleanup_manifest_worker.rb
new file mode 100644
index 00000000000..1186efa2034
--- /dev/null
+++ b/app/workers/dependency_proxy/cleanup_manifest_worker.rb
@@ -0,0 +1,40 @@
+# 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
diff --git a/app/workers/dependency_proxy/image_ttl_group_policy_worker.rb b/app/workers/dependency_proxy/image_ttl_group_policy_worker.rb
new file mode 100644
index 00000000000..fed469e6dc8
--- /dev/null
+++ b/app/workers/dependency_proxy/image_ttl_group_policy_worker.rb
@@ -0,0 +1,70 @@
+# 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
diff --git a/app/workers/expire_build_artifacts_worker.rb b/app/workers/expire_build_artifacts_worker.rb
index 65d387f73ed..295703cc1c3 100644
--- a/app/workers/expire_build_artifacts_worker.rb
+++ b/app/workers/expire_build_artifacts_worker.rb
@@ -10,7 +10,7 @@ class ExpireBuildArtifactsWorker # rubocop:disable Scalability/IdempotentWorker
   include CronjobQueue
   # rubocop:enable Scalability/CronWorkerContext
 
-  feature_category :continuous_integration
+  feature_category :build_artifacts
 
   def perform
     service = Ci::JobArtifacts::DestroyAllExpiredService.new
diff --git a/app/workers/expire_build_instance_artifacts_worker.rb b/app/workers/expire_build_instance_artifacts_worker.rb
index 96378acca08..77b8f59e365 100644
--- a/app/workers/expire_build_instance_artifacts_worker.rb
+++ b/app/workers/expire_build_instance_artifacts_worker.rb
@@ -7,7 +7,7 @@ class ExpireBuildInstanceArtifactsWorker # rubocop:disable Scalability/Idempoten
 
   sidekiq_options retry: 3
 
-  feature_category :continuous_integration
+  feature_category :build_artifacts
 
   # rubocop: disable CodeReuse/ActiveRecord
   def perform(build_id)
diff --git a/app/workers/expire_job_cache_worker.rb b/app/workers/expire_job_cache_worker.rb
index 401fe1dc1e5..7374f650546 100644
--- a/app/workers/expire_job_cache_worker.rb
+++ b/app/workers/expire_job_cache_worker.rb
@@ -10,11 +10,9 @@ class ExpireJobCacheWorker # rubocop:disable Scalability/IdempotentWorker
 
   queue_namespace :pipeline_cache
   urgency :high
-  # This worker should be idempotent, but we're switching to data_consistency
-  # :sticky and there is an ongoing incompatibility, so it needs to be disabled for
-  # now. The following line can be uncommented and this comment removed once
-  # https://gitlab.com/gitlab-org/gitlab/-/issues/325291 is resolved.
-  # idempotent!
+
+  deduplicate :until_executing, including_scheduled: true
+  idempotent!
 
   # rubocop: disable CodeReuse/ActiveRecord
   def perform(job_id)
diff --git a/app/workers/gitlab/github_import/stage/finish_import_worker.rb b/app/workers/gitlab/github_import/stage/finish_import_worker.rb
index 006b79dbff4..5197c1e1e88 100644
--- a/app/workers/gitlab/github_import/stage/finish_import_worker.rb
+++ b/app/workers/gitlab/github_import/stage/finish_import_worker.rb
@@ -18,36 +18,28 @@ module Gitlab
 
         # project - An instance of Project.
         def import(_, project)
+          @project = project
           project.after_import
-          report_import_time(project)
+          report_import_time
         end
 
-        def report_import_time(project)
-          duration = Time.zone.now - project.created_at
+        private
 
-          histogram.observe({ project: project.full_path }, duration)
-          counter.increment
+        attr_reader :project
+
+        def report_import_time
+          metrics.track_finished_import
 
           info(
             project.id,
             message: "GitHub project import finished",
-            duration_s: duration.round(2),
+            duration_s: metrics.duration.round(2),
             object_counts: ::Gitlab::GithubImport::ObjectCounter.summary(project)
           )
         end
 
-        def histogram
-          @histogram ||= Gitlab::Metrics.histogram(
-            :github_importer_total_duration_seconds,
-            'Total time spent importing GitHub projects, in seconds'
-          )
-        end
-
-        def counter
-          @counter ||= Gitlab::Metrics.counter(
-            :github_importer_imported_projects,
-            'The number of imported GitHub projects'
-          )
+        def metrics
+          @metrics ||= Gitlab::Import::Metrics.new(:github_importer, project)
         end
       end
     end
diff --git a/app/workers/gitlab/github_import/stage/import_base_data_worker.rb b/app/workers/gitlab/github_import/stage/import_base_data_worker.rb
index 715c39caf42..cc6a2255160 100644
--- a/app/workers/gitlab/github_import/stage/import_base_data_worker.rb
+++ b/app/workers/gitlab/github_import/stage/import_base_data_worker.rb
@@ -31,6 +31,22 @@ module Gitlab
           project.import_state.refresh_jid_expiration
 
           ImportPullRequestsWorker.perform_async(project.id)
+        rescue StandardError => e
+          Gitlab::Import::ImportFailureService.track(
+            project_id: project.id,
+            error_source: self.class.name,
+            exception: e,
+            fail_import: abort_on_failure,
+            metrics: true
+          )
+
+          raise(e)
+        end
+
+        private
+
+        def abort_on_failure
+          true
         end
       end
     end
diff --git a/app/workers/gitlab/github_import/stage/import_pull_requests_worker.rb b/app/workers/gitlab/github_import/stage/import_pull_requests_worker.rb
index d76d36531d1..71d0247bae0 100644
--- a/app/workers/gitlab/github_import/stage/import_pull_requests_worker.rb
+++ b/app/workers/gitlab/github_import/stage/import_pull_requests_worker.rb
@@ -27,6 +27,22 @@ module Gitlab
             { waiter.key => waiter.jobs_remaining },
             :pull_requests_merged_by
           )
+        rescue StandardError => e
+          Gitlab::Import::ImportFailureService.track(
+            project_id: project.id,
+            error_source: self.class.name,
+            exception: e,
+            fail_import: abort_on_failure,
+            metrics: true
+          )
+
+          raise(e)
+        end
+
+        private
+
+        def abort_on_failure
+          true
         end
       end
     end
diff --git a/app/workers/gitlab/github_import/stage/import_repository_worker.rb b/app/workers/gitlab/github_import/stage/import_repository_worker.rb
index 227b7c304b0..3e914cc7590 100644
--- a/app/workers/gitlab/github_import/stage/import_repository_worker.rb
+++ b/app/workers/gitlab/github_import/stage/import_repository_worker.rb
@@ -33,6 +33,17 @@ module Gitlab
           counter.increment
 
           ImportBaseDataWorker.perform_async(project.id)
+
+        rescue StandardError => e
+          Gitlab::Import::ImportFailureService.track(
+            project_id: project.id,
+            error_source: self.class.name,
+            exception: e,
+            fail_import: abort_on_failure,
+            metrics: true
+          )
+
+          raise(e)
         end
 
         def counter
diff --git a/app/workers/issue_placement_worker.rb b/app/workers/issue_placement_worker.rb
index e0c4502ed1a..22e2a8e95f4 100644
--- a/app/workers/issue_placement_worker.rb
+++ b/app/workers/issue_placement_worker.rb
@@ -31,7 +31,7 @@ class IssuePlacementWorker
     # while preserving creation order.
     to_place = Issue
       .relative_positioning_query_base(issue)
-      .where(relative_position: nil)
+      .with_null_relative_position
       .order({ created_at: :asc }, { id: :asc })
       .limit(QUERY_LIMIT + 1)
       .to_a
diff --git a/app/workers/namespaces/in_product_marketing_emails_worker.rb b/app/workers/namespaces/in_product_marketing_emails_worker.rb
index 49e65d59e83..470fba1227d 100644
--- a/app/workers/namespaces/in_product_marketing_emails_worker.rb
+++ b/app/workers/namespaces/in_product_marketing_emails_worker.rb
@@ -8,7 +8,7 @@ module Namespaces
 
     include CronjobQueue # rubocop:disable Scalability/CronWorkerContext
 
-    feature_category :subgroups
+    feature_category :experimentation_activation
     urgency :low
 
     def perform
diff --git a/app/workers/namespaces/onboarding_issue_created_worker.rb b/app/workers/namespaces/onboarding_issue_created_worker.rb
index 81d105ab19c..aab5767e0f1 100644
--- a/app/workers/namespaces/onboarding_issue_created_worker.rb
+++ b/app/workers/namespaces/onboarding_issue_created_worker.rb
@@ -8,7 +8,7 @@ module Namespaces
 
     sidekiq_options retry: 3
 
-    feature_category :issue_tracking
+    feature_category :onboarding
     urgency :low
 
     deduplicate :until_executing
diff --git a/app/workers/namespaces/onboarding_pipeline_created_worker.rb b/app/workers/namespaces/onboarding_pipeline_created_worker.rb
index f9a6b734586..4172e286474 100644
--- a/app/workers/namespaces/onboarding_pipeline_created_worker.rb
+++ b/app/workers/namespaces/onboarding_pipeline_created_worker.rb
@@ -8,7 +8,7 @@ module Namespaces
 
     sidekiq_options retry: 3
 
-    feature_category :subgroups
+    feature_category :onboarding
     urgency :low
 
     deduplicate :until_executing
diff --git a/app/workers/namespaces/onboarding_progress_worker.rb b/app/workers/namespaces/onboarding_progress_worker.rb
index b77db1aec5e..77a31d85a9a 100644
--- a/app/workers/namespaces/onboarding_progress_worker.rb
+++ b/app/workers/namespaces/onboarding_progress_worker.rb
@@ -8,7 +8,7 @@ module Namespaces
 
     sidekiq_options retry: 3
 
-    feature_category :product_analytics
+    feature_category :onboarding
     worker_resource_boundary :cpu
     urgency :low
 
diff --git a/app/workers/namespaces/onboarding_user_added_worker.rb b/app/workers/namespaces/onboarding_user_added_worker.rb
index 6a189e81b95..4d17cf9a6e2 100644
--- a/app/workers/namespaces/onboarding_user_added_worker.rb
+++ b/app/workers/namespaces/onboarding_user_added_worker.rb
@@ -8,7 +8,7 @@ module Namespaces
 
     sidekiq_options retry: 3
 
-    feature_category :users
+    feature_category :onboarding
     urgency :low
 
     idempotent!
diff --git a/app/workers/packages/composer/cache_cleanup_worker.rb b/app/workers/packages/composer/cache_cleanup_worker.rb
index 19babf63967..c80d6ea45d8 100644
--- a/app/workers/packages/composer/cache_cleanup_worker.rb
+++ b/app/workers/packages/composer/cache_cleanup_worker.rb
@@ -14,19 +14,7 @@ module Packages
       idempotent!
 
       def perform
-        ::Packages::Composer::CacheFile.without_namespace.find_in_batches do |cache_files|
-          cache_files.each(&:destroy)
-        rescue ActiveRecord::RecordNotFound
-          # ignore. likely due to object already being deleted.
-        end
-
-        ::Packages::Composer::CacheFile.expired.find_in_batches do |cache_files|
-          cache_files.each(&:destroy)
-        rescue ActiveRecord::RecordNotFound
-          # ignore. likely due to object already being deleted.
-        end
-      rescue StandardError => e
-        Gitlab::ErrorTracking.log_exception(e)
+        # no-op: to be removed after 14.5 https://gitlab.com/gitlab-org/gitlab/-/issues/333694
       end
     end
   end
diff --git a/app/workers/packages/composer/cache_update_worker.rb b/app/workers/packages/composer/cache_update_worker.rb
index 874993a1325..5600af6ce24 100644
--- a/app/workers/packages/composer/cache_update_worker.rb
+++ b/app/workers/packages/composer/cache_update_worker.rb
@@ -7,20 +7,14 @@ module Packages
 
       data_consistency :always
 
-      sidekiq_options retry: 3
+      sidekiq_options retry: false
 
       feature_category :package_registry
 
       idempotent!
 
-      def perform(project_id, package_name, last_page_sha)
-        project = Project.find_by_id(project_id)
-
-        return unless project
-
-        Gitlab::Composer::Cache.new(project: project, name: package_name, last_page_sha: last_page_sha).execute
-      rescue StandardError => e
-        Gitlab::ErrorTracking.log_exception(e, project_id: project_id)
+      def perform(*args)
+        # no-op: to be removed after 14.5 https://gitlab.com/gitlab-org/gitlab/-/issues/333694
       end
     end
   end
diff --git a/app/workers/packages/debian/generate_distribution_worker.rb b/app/workers/packages/debian/generate_distribution_worker.rb
index b9b157d25d2..1eff3ea02dd 100644
--- a/app/workers/packages/debian/generate_distribution_worker.rb
+++ b/app/workers/packages/debian/generate_distribution_worker.rb
@@ -2,7 +2,7 @@
 
 module Packages
   module Debian
-    class GenerateDistributionWorker # rubocop:disable Scalability/IdempotentWorker
+    class GenerateDistributionWorker
       include ApplicationWorker
 
       data_consistency :always
diff --git a/app/workers/pages_remove_worker.rb b/app/workers/pages_remove_worker.rb
deleted file mode 100644
index 4de99b8654d..00000000000
--- a/app/workers/pages_remove_worker.rb
+++ /dev/null
@@ -1,16 +0,0 @@
-# frozen_string_literal: true
-
-# TODO: remove this worker https://gitlab.com/gitlab-org/gitlab/-/issues/340641
-class PagesRemoveWorker # rubocop:disable Scalability/IdempotentWorker
-  include ApplicationWorker
-
-  data_consistency :always
-
-  sidekiq_options retry: 3
-  feature_category :pages
-  loggable_arguments 0
-
-  def perform(project_id)
-    # no-op
-  end
-end
diff --git a/app/workers/pipeline_hooks_worker.rb b/app/workers/pipeline_hooks_worker.rb
index 322f92d376b..c67f3860a50 100644
--- a/app/workers/pipeline_hooks_worker.rb
+++ b/app/workers/pipeline_hooks_worker.rb
@@ -12,9 +12,10 @@ class PipelineHooksWorker # rubocop:disable Scalability/IdempotentWorker
 
   # rubocop: disable CodeReuse/ActiveRecord
   def perform(pipeline_id)
-    Ci::Pipeline
-      .find_by(id: pipeline_id)
-      .try(:execute_hooks)
+    pipeline = Ci::Pipeline.find_by(id: pipeline_id)
+    return unless pipeline
+
+    Ci::Pipelines::HookService.new(pipeline).execute
   end
   # rubocop: enable CodeReuse/ActiveRecord
 end
diff --git a/app/workers/pipeline_process_worker.rb b/app/workers/pipeline_process_worker.rb
index 9cd471a5ab6..9370b361068 100644
--- a/app/workers/pipeline_process_worker.rb
+++ b/app/workers/pipeline_process_worker.rb
@@ -14,7 +14,7 @@ class PipelineProcessWorker
   loggable_arguments 1
 
   idempotent!
-  deduplicate :until_executing, feature_flag: :ci_idempotent_pipeline_process_worker
+  deduplicate :until_executing
 
   # rubocop: disable CodeReuse/ActiveRecord
   def perform(pipeline_id)
diff --git a/app/workers/run_pipeline_schedule_worker.rb b/app/workers/run_pipeline_schedule_worker.rb
index dd0f14a5cab..12042ebc4f0 100644
--- a/app/workers/run_pipeline_schedule_worker.rb
+++ b/app/workers/run_pipeline_schedule_worker.rb
@@ -27,8 +27,9 @@ class RunPipelineScheduleWorker # rubocop:disable Scalability/IdempotentWorker
                                   user,
                                   ref: schedule.ref)
       .execute!(:schedule, ignore_skip_ci: true, save_on_errors: false, schedule: schedule)
-  rescue Ci::CreatePipelineService::CreateError
-    # no-op. This is a user operation error such as corrupted .gitlab-ci.yml.
+  rescue Ci::CreatePipelineService::CreateError => e
+    # This is a user operation error such as corrupted .gitlab-ci.yml. Log the error for debugging purpose.
+    log_extra_metadata_on_done(:pipeline_creation_error, e)
   rescue StandardError => e
     error(schedule, e)
   end
@@ -37,10 +38,16 @@ class RunPipelineScheduleWorker # rubocop:disable Scalability/IdempotentWorker
 
   def error(schedule, error)
     failed_creation_counter.increment
+    log_error(schedule, error)
+    track_error(schedule, error)
+  end
 
+  def log_error(schedule, error)
     Gitlab::AppLogger.error "Failed to create a scheduled pipeline. " \
                        "schedule_id: #{schedule.id} message: #{error.message}"
+  end
 
+  def track_error(schedule, error)
     Gitlab::ErrorTracking
       .track_and_raise_for_dev_exception(error,
                        issue_url: 'https://gitlab.com/gitlab-org/gitlab-foss/issues/41231',
diff --git a/app/workers/stuck_ci_jobs_worker.rb b/app/workers/stuck_ci_jobs_worker.rb
index a2b2686c8d5..72004f7568c 100644
--- a/app/workers/stuck_ci_jobs_worker.rb
+++ b/app/workers/stuck_ci_jobs_worker.rb
@@ -2,6 +2,7 @@
 
 class StuckCiJobsWorker # rubocop:disable Scalability/IdempotentWorker
   include ApplicationWorker
+  include ExclusiveLeaseGuard
 
   # rubocop:disable Scalability/CronWorkerContext
   # This is an instance-wide cleanup query, so there's no meaningful
@@ -12,25 +13,19 @@ class StuckCiJobsWorker # rubocop:disable Scalability/IdempotentWorker
   data_consistency :always
 
   feature_category :continuous_integration
-  worker_resource_boundary :cpu
-
-  EXCLUSIVE_LEASE_KEY = 'stuck_ci_builds_worker_lease'
 
   def perform
-    return unless try_obtain_lease
-
-    Ci::StuckBuilds::DropService.new.execute
+    Ci::StuckBuilds::DropRunningWorker.perform_in(20.minutes)
+    Ci::StuckBuilds::DropScheduledWorker.perform_in(40.minutes)
 
-    remove_lease
+    try_obtain_lease do
+      Ci::StuckBuilds::DropPendingService.new.execute
+    end
   end
 
   private
 
-  def try_obtain_lease
-    @uuid = Gitlab::ExclusiveLease.new(EXCLUSIVE_LEASE_KEY, timeout: 30.minutes).try_obtain
-  end
-
-  def remove_lease
-    Gitlab::ExclusiveLease.cancel(EXCLUSIVE_LEASE_KEY, @uuid)
+  def lease_timeout
+    30.minutes
   end
 end