Add latest changes from gitlab-org/gitlab@master

43 files changed, 756 insertions, 116 deletions

diff --git a/app/models/batched_git_ref_updates/deletion.rb b/app/models/batched_git_ref_updates/deletion.rb
new file mode 100644
index 00000000000..61bba8aeba9
--- /dev/null
+++ b/app/models/batched_git_ref_updates/deletion.rb
@@ -0,0 +1,67 @@
+# frozen_string_literal: true
+
+module BatchedGitRefUpdates
+  class Deletion < ApplicationRecord
+    PARTITION_DURATION = 1.day
+
+    include IgnorableColumns
+    include BulkInsertSafe
+    include PartitionedTable
+    include EachBatch
+
+    self.table_name = 'p_batched_git_ref_updates_deletions'
+    self.primary_key = :id
+    self.sequence_name = :to_be_deleted_git_refs_id_seq
+
+    # This column must be ignored otherwise Rails will cache the default value and `bulk_insert!` will start saving
+    # incorrect partition_id.
+    ignore_column :partition_id, remove_with: '3000.0', remove_after: '3000-01-01'
+
+    belongs_to :project, inverse_of: :to_be_deleted_git_refs
+
+    scope :for_partition, ->(partition) { where(partition_id: partition) }
+    scope :for_project, ->(project_id) { where(project_id: project_id) }
+    scope :select_ref_and_identity, -> { select(:ref, :id, arel_table[:partition_id].as('partition')) }
+
+    partitioned_by :partition_id, strategy: :sliding_list,
+      next_partition_if: ->(active_partition) do
+        oldest_record_in_partition = Deletion
+          .select(:id, :created_at)
+          .for_partition(active_partition.value)
+          .order(:id)
+          .limit(1)
+          .take
+
+        oldest_record_in_partition.present? &&
+          oldest_record_in_partition.created_at < PARTITION_DURATION.ago
+      end,
+      detach_partition_if: ->(partition) do
+        !Deletion
+          .for_partition(partition.value)
+          .status_pending
+          .exists?
+      end
+
+    enum status: { pending: 1, processed: 2 }, _prefix: :status
+
+    def self.mark_records_processed(records)
+      update_by_partition(records) do |partitioned_scope|
+        partitioned_scope.update_all(status: :processed)
+      end
+    end
+
+    # Your scope must select_ref_and_identity before calling this method as it relies on partition being explicitly
+    # selected
+    def self.update_by_partition(records)
+      records.group_by(&:partition).each do |partition, records_within_partition|
+        partitioned_scope = status_pending
+          .for_partition(partition)
+          .where(id: records_within_partition.map(&:id))
+
+        yield(partitioned_scope)
+      end
+    end
+
+    private_class_method :update_by_partition
+  end
+end
diff --git a/app/models/ci/persistent_ref.rb b/app/models/ci/persistent_ref.rb
index f713d5952bc..57e2d943a4c 100644
--- a/app/models/ci/persistent_ref.rb
+++ b/app/models/ci/persistent_ref.rb
@@ -11,7 +11,7 @@ module Ci
 
     delegate :project, :sha, to: :pipeline
     delegate :repository, to: :project
-    delegate :ref_exists?, :create_ref, :delete_refs, to: :repository
+    delegate :ref_exists?, :create_ref, :delete_refs, :async_delete_refs, to: :repository
 
     def exist?
       ref_exists?(path)
@@ -42,6 +42,12 @@ module Ci
         .track_exception(e, pipeline_id: pipeline.id)
     end
 
+    def async_delete
+      return unless should_delete?
+
+      async_delete_refs(path)
+    end
+
     def path
       "refs/#{Repository::REF_PIPELINES}/#{pipeline.id}"
     end
diff --git a/app/models/ci/pipeline.rb b/app/models/ci/pipeline.rb
index 97efccf7160..e2f017b1d14 100644
--- a/app/models/ci/pipeline.rb
+++ b/app/models/ci/pipeline.rb
@@ -342,7 +342,9 @@ module Ci
       # This needs to be kept in sync with `Ci::PipelineRef#should_delete?`
       after_transition any => ::Ci::Pipeline.stopped_statuses do |pipeline|
         pipeline.run_after_commit do
-          if Feature.enabled?(:pipeline_cleanup_ref_worker_async, pipeline.project)
+          if Feature.enabled?(:pipeline_delete_gitaly_refs_in_batches, pipeline.project)
+            pipeline.persistent_ref.async_delete
+          elsif Feature.enabled?(:pipeline_cleanup_ref_worker_async, pipeline.project)
             ::Ci::PipelineCleanupRefWorker.perform_async(pipeline.id)
           else
             pipeline.persistent_ref.delete
diff --git a/app/models/merge_request.rb b/app/models/merge_request.rb
index 2773569161d..cc5152df12f 100644
--- a/app/models/merge_request.rb
+++ b/app/models/merge_request.rb
@@ -1537,20 +1537,29 @@ class MergeRequest < ApplicationRecord
   end
 
   def schedule_cleanup_refs(only: :all)
-    if Feature.enabled?(:merge_request_cleanup_ref_worker_async, target_project)
+    if Feature.enabled?(:merge_request_delete_gitaly_refs_in_batches, target_project)
+      async_cleanup_refs(only: only)
+    elsif Feature.enabled?(:merge_request_cleanup_ref_worker_async, target_project)
       MergeRequests::CleanupRefWorker.perform_async(id, only.to_s)
     else
       cleanup_refs(only: only)
     end
   end
 
-  def cleanup_refs(only: :all)
+  def refs_to_cleanup(only: :all)
     target_refs = []
     target_refs << ref_path       if %i[all head].include?(only)
     target_refs << merge_ref_path if %i[all merge].include?(only)
     target_refs << train_ref_path if %i[all train].include?(only)
+    target_refs
+  end
+
+  def cleanup_refs(only: :all)
+    project.repository.delete_refs(*refs_to_cleanup(only: only))
+  end
 
-    project.repository.delete_refs(*target_refs)
+  def async_cleanup_refs(only: :all)
+    project.repository.async_delete_refs(*refs_to_cleanup(only: only))
   end
 
   def self.merge_request_ref?(ref)
diff --git a/app/services/batched_git_ref_updates/cleanup_scheduler_service.rb b/app/services/batched_git_ref_updates/cleanup_scheduler_service.rb
new file mode 100644
index 00000000000..cf547c0e6b5
--- /dev/null
+++ b/app/services/batched_git_ref_updates/cleanup_scheduler_service.rb
@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+
+module BatchedGitRefUpdates
+  class CleanupSchedulerService
+    include Gitlab::ExclusiveLeaseHelpers
+    MAX_PROJECTS = 10_000
+    BATCH_SIZE = 100
+    LOCK_TIMEOUT = 10.minutes
+
+    def execute
+      total_projects = 0
+
+      in_lock(self.class.name, retries: 0, ttl: LOCK_TIMEOUT) do
+        Deletion.status_pending.distinct_each_batch(column: :project_id, of: BATCH_SIZE) do |deletions|
+          ProjectCleanupWorker.bulk_perform_async_with_contexts(
+            deletions,
+            arguments_proc: ->(deletion) { deletion.project_id },
+            context_proc: ->(_) { {} } # No project context because loading the project is wasteful
+          )
+
+          total_projects += deletions.count
+          break if total_projects >= MAX_PROJECTS
+        end
+      end
+
+      { total_projects: total_projects }
+    end
+  end
+end
diff --git a/app/services/batched_git_ref_updates/project_cleanup_service.rb b/app/services/batched_git_ref_updates/project_cleanup_service.rb
new file mode 100644
index 00000000000..f9518cad975
--- /dev/null
+++ b/app/services/batched_git_ref_updates/project_cleanup_service.rb
@@ -0,0 +1,44 @@
+# frozen_string_literal: true
+
+module BatchedGitRefUpdates
+  class ProjectCleanupService
+    include ::Gitlab::ExclusiveLeaseHelpers
+
+    LOCK_TIMEOUT = 10.minutes
+    GITALY_BATCH_SIZE = 100
+    QUERY_BATCH_SIZE = 1000
+    MAX_DELETES = 10_000
+
+    def initialize(project_id)
+      @project_id = project_id
+    end
+
+    def execute
+      total_deletes = 0
+
+      in_lock("#{self.class}/#{@project_id}", retries: 0, ttl: LOCK_TIMEOUT) do
+        project = Project.find_by_id(@project_id)
+        break unless project
+
+        Deletion
+          .status_pending
+          .for_project(@project_id)
+          .select_ref_and_identity
+          .each_batch(of: QUERY_BATCH_SIZE) do |batch|
+          refs = batch.map(&:ref)
+
+          refs.each_slice(GITALY_BATCH_SIZE) do |refs_to_delete|
+            project.repository.delete_refs(*refs_to_delete)
+          end
+
+          total_deletes += refs.count
+          Deletion.mark_records_processed(batch)
+
+          break if total_deletes >= MAX_DELETES
+        end
+      end
+
+      { total_deletes: total_deletes }
+    end
+  end
+end
diff --git a/app/services/git/base_hooks_service.rb b/app/services/git/base_hooks_service.rb
index f9280be7ee2..6a8e4d17859 100644
--- a/app/services/git/base_hooks_service.rb
+++ b/app/services/git/base_hooks_service.rb
@@ -83,7 +83,6 @@ module Git
 
     def enqueue_notify_kas
       return unless Gitlab::Kas.enabled?
-      return unless Feature.enabled?(:notify_kas_on_git_push, project)
 
       Clusters::Agents::NotifyGitPushWorker.perform_async(project.id)
     end
diff --git a/app/services/merge_requests/merge_to_ref_service.rb b/app/services/merge_requests/merge_to_ref_service.rb
index acd3bc36e1d..8b79feb5e0f 100644
--- a/app/services/merge_requests/merge_to_ref_service.rb
+++ b/app/services/merge_requests/merge_to_ref_service.rb
@@ -13,13 +13,12 @@ module MergeRequests
   class MergeToRefService < MergeRequests::MergeBaseService
     extend ::Gitlab::Utils::Override
 
-    def execute(merge_request, cache_merge_to_ref_calls = false)
+    def execute(merge_request)
       @merge_request = merge_request
 
       error_check!
 
-      commit_id = commit(cache_merge_to_ref_calls)
-
+      commit_id = extracted_merge_to_ref
       raise_error('Conflicts detected during merge') unless commit_id
 
       commit = project.commit(commit_id)
@@ -56,16 +55,6 @@ module MergeRequests
       params[:first_parent_ref] || merge_request.target_branch_ref
     end
 
-    def commit(cache_merge_to_ref_calls = false)
-      if cache_merge_to_ref_calls
-        Rails.cache.fetch(cache_key, expires_in: 1.day) do
-          extracted_merge_to_ref
-        end
-      else
-        extracted_merge_to_ref
-      end
-    end
-
     def extracted_merge_to_ref
       repository.merge_to_ref(current_user,
         source_sha: source,
@@ -76,9 +65,5 @@ module MergeRequests
     rescue Gitlab::Git::PreReceiveError, Gitlab::Git::CommandError => error
       raise MergeError, error.message
     end
-
-    def cache_key
-      [:merge_to_ref_service, project.full_path, merge_request.target_branch_sha, merge_request.source_branch_sha]
-    end
   end
 end
diff --git a/app/views/projects/_wiki.html.haml b/app/views/projects/_wiki.html.haml
index e82e0972d82..82cfb0435c7 100644
--- a/app/views/projects/_wiki.html.haml
+++ b/app/views/projects/_wiki.html.haml
@@ -7,7 +7,7 @@
   .landing{ class: [('row-content-block row p-0 align-items-center' if can_create_wiki), ('content-block' unless can_create_wiki)] }
     .col-12.col-md-3.p-0
       .svg-content
-        = image_tag 'illustrations/wiki_login_empty.svg'
+        = image_tag 'illustrations/empty-state/empty-wiki-md.svg'
     .col-12.col-md-9.text-center.text-md-left.pl-md-0.pl-sm-3.mb-4
       %h4
         = _("This project does not have a wiki homepage yet")
diff --git a/app/views/projects/confluences/show.html.haml b/app/views/projects/confluences/show.html.haml
index 283408ffa63..cdf3562a8ed 100644
--- a/app/views/projects/confluences/show.html.haml
+++ b/app/views/projects/confluences/show.html.haml
@@ -1,7 +1,7 @@
 - breadcrumb_title _('Confluence')
 - page_title _('Confluence')
 - add_page_specific_style 'page_bundles/wiki'
-= render layout: 'shared/empty_states/wikis_layout', locals: { image_path: 'illustrations/wiki_login_empty.svg' } do
+= render layout: 'shared/empty_states/wikis_layout', locals: { image_path: 'illustrations/empty-state/empty-wiki-md.svg' } do
   %h4
     = s_('WikiEmpty|Confluence is enabled')
   %p
diff --git a/app/views/projects/integrations/shimos/show.html.haml b/app/views/projects/integrations/shimos/show.html.haml
index e6cd8c15809..165e414f75b 100644
--- a/app/views/projects/integrations/shimos/show.html.haml
+++ b/app/views/projects/integrations/shimos/show.html.haml
@@ -1,7 +1,7 @@
 - breadcrumb_title s_('Shimo|Shimo Workspace')
 - page_title s_('Shimo|Shimo Workspace')
 - add_page_specific_style 'page_bundles/wiki'
-= render layout: 'shared/empty_states/wikis_layout', locals: { image_path: 'illustrations/wiki_login_empty.svg' } do
+= render layout: 'shared/empty_states/wikis_layout', locals: { image_path: 'illustrations/empty-state/empty-wiki-md.svg' } do
   %h4
     = s_('Shimo|Shimo Workspace integration is enabled')
   %p
diff --git a/app/views/shared/empty_states/_wikis.html.haml b/app/views/shared/empty_states/_wikis.html.haml
index 7e1b270be3a..567c4a2d444 100644
--- a/app/views/shared/empty_states/_wikis.html.haml
+++ b/app/views/shared/empty_states/_wikis.html.haml
@@ -6,7 +6,7 @@
   - create_path = wiki_page_path(@wiki, params[:id], view: 'create')
   - create_link = link_button_to s_('WikiEmpty|Create your first page'), create_path, title: s_('WikiEmpty|Create your first page'), data: { qa_selector: 'create_first_page_link' }, variant: :confirm
 
-  = render layout: layout_path, locals: { image_path: 'illustrations/wiki_login_empty.svg' } do
+  = render layout: layout_path, locals: { image_path: 'illustrations/empty-state/empty-wiki-md.svg' } do
     %h4.text-left
       = messages.dig(:writable, :title)
     %p.text-left
@@ -20,7 +20,7 @@
 - elsif @project && can?(current_user, :read_issue, @project)
   - issues_link = link_to s_('WikiEmptyIssueMessage|issue tracker'), project_issues_path(@project)
 
-  = render layout: layout_path, locals: { image_path: 'illustrations/wiki_logout_empty.svg' } do
+  = render layout: layout_path, locals: { image_path: 'illustrations/empty-state/empty-wiki-md.svg' } do
     %h4
       = messages.dig(:issuable, :title)
     %p.text-left
@@ -29,7 +29,7 @@
       = link_button_to s_('WikiEmpty|Suggest wiki improvement'), new_project_issue_path(@project), title: s_('WikiEmptyIssueMessage|Suggest wiki improvement'), variant: :confirm
 
 - else
-  = render layout: layout_path, locals: { image_path: 'illustrations/wiki_logout_empty.svg' } do
+  = render layout: layout_path, locals: { image_path: 'illustrations/empty-state/empty-wiki-md.svg' } do
     %h4
       = messages.dig(:readonly, :title)
     %p
diff --git a/app/views/shared/empty_states/_wikis_layout.html.haml b/app/views/shared/empty_states/_wikis_layout.html.haml
index 0b7034838ed..03054c959fd 100644
--- a/app/views/shared/empty_states/_wikis_layout.html.haml
+++ b/app/views/shared/empty_states/_wikis_layout.html.haml
@@ -1,6 +1,6 @@
 .row.empty-state.empty-state-wiki
   .col-12
-    .svg-content{ data: { qa_selector: 'svg_content' } }
+    .svg-content.svg-150{ data: { qa_selector: 'svg_content' } }
       = image_tag image_path
   .col-12
     .text-content.text-center
diff --git a/app/workers/all_queues.yml b/app/workers/all_queues.yml
index e2d199b9e51..e41c85d1550 100644
--- a/app/workers/all_queues.yml
+++ b/app/workers/all_queues.yml
@@ -219,6 +219,15 @@
   :weight: 1
   :idempotent: true
   :tags: []
+- :name: cronjob:batched_git_ref_updates_cleanup_scheduler
+  :worker_name: BatchedGitRefUpdates::CleanupSchedulerWorker
+  :feature_category: :gitaly
+  :has_external_dependencies: false
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: cronjob:bulk_imports_stuck_import
   :worker_name: BulkImports::StuckImportWorker
   :feature_category: :importers
@@ -2307,6 +2316,15 @@
   :weight: 1
   :idempotent: false
   :tags: []
+- :name: batched_git_ref_updates_project_cleanup
+  :worker_name: BatchedGitRefUpdates::ProjectCleanupWorker
+  :feature_category: :gitaly
+  :has_external_dependencies: false
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: bitbucket_server_import_advance_stage
   :worker_name: Gitlab::BitbucketServerImport::AdvanceStageWorker
   :feature_category: :importers
diff --git a/app/workers/batched_git_ref_updates/cleanup_scheduler_worker.rb b/app/workers/batched_git_ref_updates/cleanup_scheduler_worker.rb
new file mode 100644
index 00000000000..9c50e319be0
--- /dev/null
+++ b/app/workers/batched_git_ref_updates/cleanup_scheduler_worker.rb
@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+module BatchedGitRefUpdates
+  class CleanupSchedulerWorker
+    include ApplicationWorker
+    # Ignore RuboCop as the context is added in the service
+    include CronjobQueue # rubocop:disable Scalability/CronWorkerContext
+
+    idempotent!
+    data_consistency :sticky
+
+    feature_category :gitaly
+
+    def perform
+      stats = CleanupSchedulerService.new.execute
+
+      log_extra_metadata_on_done(:stats, stats)
+    end
+  end
+end
diff --git a/app/workers/batched_git_ref_updates/project_cleanup_worker.rb b/app/workers/batched_git_ref_updates/project_cleanup_worker.rb
new file mode 100644
index 00000000000..b2b1df29430
--- /dev/null
+++ b/app/workers/batched_git_ref_updates/project_cleanup_worker.rb
@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+module BatchedGitRefUpdates
+  class ProjectCleanupWorker
+    include ApplicationWorker
+
+    idempotent!
+    data_consistency :delayed
+
+    feature_category :gitaly
+
+    def perform(project_id)
+      stats = ProjectCleanupService.new(project_id).execute
+
+      log_extra_metadata_on_done(:stats, stats)
+    end
+  end
+end
diff --git a/app/workers/clusters/agents/notify_git_push_worker.rb b/app/workers/clusters/agents/notify_git_push_worker.rb
index d2994bb9144..db1de0b3518 100644
--- a/app/workers/clusters/agents/notify_git_push_worker.rb
+++ b/app/workers/clusters/agents/notify_git_push_worker.rb
@@ -14,7 +14,6 @@ module Clusters
 
       def perform(project_id)
         return unless project = ::Project.find_by_id(project_id)
-        return unless Feature.enabled?(:notify_kas_on_git_push, project)
 
         Gitlab::Kas::Client.new.send_git_push_event(project: project)
       end
diff --git a/config/feature_flags/development/merge_request_delete_gitaly_refs_in_batches.yml b/config/feature_flags/development/merge_request_delete_gitaly_refs_in_batches.yml
new file mode 100644
index 00000000000..0f0b81dbbd2
--- /dev/null
+++ b/config/feature_flags/development/merge_request_delete_gitaly_refs_in_batches.yml
@@ -0,0 +1,8 @@
+---
+name: merge_request_delete_gitaly_refs_in_batches
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/125333
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/416969
+milestone: '16.3'
+type: development
+group: group::gitaly
+default_enabled: false
diff --git a/config/feature_flags/development/notify_kas_on_git_push.yml b/config/feature_flags/development/pipeline_delete_gitaly_refs_in_batches.yml
index 32806418bce..66e23fa424a 100644
--- a/config/feature_flags/development/notify_kas_on_git_push.yml
+++ b/config/feature_flags/development/pipeline_delete_gitaly_refs_in_batches.yml
@@ -1,8 +1,8 @@
 ---
-name: notify_kas_on_git_push
-introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/119168
-rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/410429
-milestone: '16.0'
+name: pipeline_delete_gitaly_refs_in_batches
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/125333
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/416969
+milestone: '16.3'
 type: development
-group: group::environments
-default_enabled: true
+group: group::gitaly
+default_enabled: false
diff --git a/config/initializers/1_settings.rb b/config/initializers/1_settings.rb
index 6c7561ffb89..813d596df5f 100644
--- a/config/initializers/1_settings.rb
+++ b/config/initializers/1_settings.rb
@@ -664,6 +664,9 @@ Settings.cron_jobs['inactive_projects_deletion_cron_worker']['job_class'] = 'Pro
 Settings.cron_jobs['loose_foreign_keys_cleanup_worker'] ||= {}
 Settings.cron_jobs['loose_foreign_keys_cleanup_worker']['cron'] ||= '*/1 * * * *'
 Settings.cron_jobs['loose_foreign_keys_cleanup_worker']['job_class'] = 'LooseForeignKeys::CleanupWorker'
+Settings.cron_jobs['batched_git_ref_updates_cleanup_scheduler_worker'] ||= {}
+Settings.cron_jobs['batched_git_ref_updates_cleanup_scheduler_worker']['cron'] ||= '*/1 * * * *'
+Settings.cron_jobs['batched_git_ref_updates_cleanup_scheduler_worker']['job_class'] = 'BatchedGitRefUpdates::CleanupSchedulerWorker'
 Settings.cron_jobs['ci_runner_versions_reconciliation_worker'] ||= {}
 Settings.cron_jobs['ci_runner_versions_reconciliation_worker']['cron'] ||= '@daily'
 Settings.cron_jobs['ci_runner_versions_reconciliation_worker']['job_class'] = 'Ci::Runners::ReconcileExistingRunnerVersionsCronWorker'
diff --git a/config/initializers/postgres_partitioning.rb b/config/initializers/postgres_partitioning.rb
index 1fbe038f0bd..bfd737baec9 100644
--- a/config/initializers/postgres_partitioning.rb
+++ b/config/initializers/postgres_partitioning.rb
@@ -7,7 +7,8 @@ Gitlab::Database::Partitioning.register_models(
     LooseForeignKeys::DeletedRecord,
     Gitlab::Database::BackgroundMigration::BatchedJobTransitionLog,
     Ci::RunnerManagerBuild,
-    Ci::JobAnnotation
+    Ci::JobAnnotation,
+    BatchedGitRefUpdates::Deletion
   ])
 
 if Gitlab.ee?
diff --git a/config/sidekiq_queues.yml b/config/sidekiq_queues.yml
index 7fe4db435a6..4f6889bf445 100644
--- a/config/sidekiq_queues.yml
+++ b/config/sidekiq_queues.yml
@@ -77,6 +77,8 @@
   - 1
 - - batched_background_migrations
   - 1
+- - batched_git_ref_updates_project_cleanup
+  - 1
 - - bitbucket_server_import_advance_stage
   - 1
 - - bitbucket_server_import_import_lfs_object
diff --git a/db/docs/p_batched_git_ref_updates_deletions.yml b/db/docs/p_batched_git_ref_updates_deletions.yml
new file mode 100644
index 00000000000..a09672f38da
--- /dev/null
+++ b/db/docs/p_batched_git_ref_updates_deletions.yml
@@ -0,0 +1,11 @@
+---
+table_name: p_batched_git_ref_updates_deletions
+classes:
+- BatchedGitRefUpdates::Deletion
+feature_categories:
+- gitaly
+description: Acts as a queue for refs that need to be deleted in Gitaly. This allows
+  us to batch deletes rather than sending them one at a time.
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/125333
+milestone: '16.3'
+gitlab_schema: gitlab_main
diff --git a/db/migrate/20230704233431_create_table_batched_git_ref_updates_deletions.rb b/db/migrate/20230704233431_create_table_batched_git_ref_updates_deletions.rb
new file mode 100644
index 00000000000..4df413a4a5e
--- /dev/null
+++ b/db/migrate/20230704233431_create_table_batched_git_ref_updates_deletions.rb
@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+class CreateTableBatchedGitRefUpdatesDeletions < Gitlab::Database::Migration[2.1]
+  enable_lock_retries!
+
+  def up
+    options = {
+      primary_key: [:id, :partition_id],
+      options: 'PARTITION BY LIST (partition_id)',
+      if_not_exists: true
+    }
+
+    create_table(:p_batched_git_ref_updates_deletions, **options) do |t|
+      t.bigserial :id, null: false
+      # Do not bother with foreign key as it provides not benefit and has a performance cost. These get cleaned up over
+      # time anyway.
+      t.bigint :project_id, null: false
+      t.bigint :partition_id, null: false, default: 1
+      t.timestamps_with_timezone null: false
+      t.integer :status, null: false, default: 1, limit: 2
+      t.text :ref, null: false, limit: 1024
+
+      t.index [:project_id, :id], where: 'status = 1',
+        name: :idx_deletions_on_project_id_and_id_where_pending
+    end
+
+    connection.execute(<<~SQL)
+      CREATE TABLE IF NOT EXISTS gitlab_partitions_dynamic.p_batched_git_ref_updates_deletions_1
+        PARTITION OF p_batched_git_ref_updates_deletions
+        FOR VALUES IN (1);
+    SQL
+  end
+
+  def down
+    drop_table :p_batched_git_ref_updates_deletions
+  end
+end
diff --git a/db/schema_migrations/20230704233431 b/db/schema_migrations/20230704233431
new file mode 100644
index 00000000000..7b0af2a2a3f
--- /dev/null
+++ b/db/schema_migrations/20230704233431
@@ -0,0 +1 @@
+d07a8cdd97cc221d64257fdcab34c3e44593f6c3a430523a2e73c96fa12c8ba1
\ No newline at end of file
diff --git a/db/structure.sql b/db/structure.sql
index b556427e299..0b8bcde15f8 100644
--- a/db/structure.sql
+++ b/db/structure.sql
@@ -537,6 +537,18 @@ CREATE TABLE loose_foreign_keys_deleted_records (
 )
 PARTITION BY LIST (partition);
 
+CREATE TABLE p_batched_git_ref_updates_deletions (
+    id bigint NOT NULL,
+    project_id bigint NOT NULL,
+    partition_id bigint DEFAULT 1 NOT NULL,
+    created_at timestamp with time zone NOT NULL,
+    updated_at timestamp with time zone NOT NULL,
+    status smallint DEFAULT 1 NOT NULL,
+    ref text NOT NULL,
+    CONSTRAINT check_f322d53b92 CHECK ((char_length(ref) <= 1024))
+)
+PARTITION BY LIST (partition_id);
+
 CREATE TABLE security_findings (
     id bigint NOT NULL,
     scan_id bigint NOT NULL,
@@ -19440,6 +19452,15 @@ CREATE SEQUENCE organizations_id_seq
 
 ALTER SEQUENCE organizations_id_seq OWNED BY organizations.id;
 
+CREATE SEQUENCE p_batched_git_ref_updates_deletions_id_seq
+    START WITH 1
+    INCREMENT BY 1
+    NO MINVALUE
+    NO MAXVALUE
+    CACHE 1;
+
+ALTER SEQUENCE p_batched_git_ref_updates_deletions_id_seq OWNED BY p_batched_git_ref_updates_deletions.id;
+
 CREATE SEQUENCE p_ci_job_annotations_id_seq
     START WITH 1
     INCREMENT BY 1
@@ -25765,6 +25786,8 @@ ALTER TABLE ONLY organization_users ALTER COLUMN id SET DEFAULT nextval('organiz
 
 ALTER TABLE ONLY organizations ALTER COLUMN id SET DEFAULT nextval('organizations_id_seq'::regclass);
 
+ALTER TABLE ONLY p_batched_git_ref_updates_deletions ALTER COLUMN id SET DEFAULT nextval('p_batched_git_ref_updates_deletions_id_seq'::regclass);
+
 ALTER TABLE ONLY p_ci_builds ALTER COLUMN id SET DEFAULT nextval('ci_builds_id_seq'::regclass);
 
 ALTER TABLE ONLY p_ci_builds_metadata ALTER COLUMN id SET DEFAULT nextval('ci_builds_metadata_id_seq'::regclass);
@@ -28048,6 +28071,9 @@ ALTER TABLE ONLY organization_users
 ALTER TABLE ONLY organizations
     ADD CONSTRAINT organizations_pkey PRIMARY KEY (id);
 
+ALTER TABLE ONLY p_batched_git_ref_updates_deletions
+    ADD CONSTRAINT p_batched_git_ref_updates_deletions_pkey PRIMARY KEY (id, partition_id);
+
 ALTER TABLE ONLY p_ci_job_annotations
     ADD CONSTRAINT p_ci_job_annotations_pkey PRIMARY KEY (id, partition_id);
 
@@ -30007,6 +30033,8 @@ CREATE INDEX idx_container_exp_policies_on_project_id_next_run_at_enabled ON con
 
 CREATE INDEX idx_container_repos_on_exp_cleanup_status_project_id_start_date ON container_repositories USING btree (expiration_policy_cleanup_status, project_id, expiration_policy_started_at);
 
+CREATE INDEX idx_deletions_on_project_id_and_id_where_pending ON ONLY p_batched_git_ref_updates_deletions USING btree (project_id, id) WHERE (status = 1);
+
 CREATE INDEX idx_deployment_clusters_on_cluster_id_and_kubernetes_namespace ON deployment_clusters USING btree (cluster_id, kubernetes_namespace);
 
 CREATE INDEX idx_devops_adoption_segments_namespace_end_time ON analytics_devops_adoption_snapshots USING btree (namespace_id, end_time);
diff --git a/doc/api/group_iterations.md b/doc/api/group_iterations.md
index cae271e6365..1a70c852529 100644
--- a/doc/api/group_iterations.md
+++ b/doc/api/group_iterations.md
@@ -30,6 +30,7 @@ GET /groups/:id/iterations?updated_after=2013-10-02T09%3A24%3A18Z
 | ------------------- | ------- | -------- | ----------- |
 | `state`             | string  | no       | 'Return `opened`, `upcoming`, `current`, `closed`, or `all` iterations.' |
 | `search`            | string  | no       | Return only iterations with a title matching the provided string.                              |
+| `in`                | array of strings | no | Fields in which fuzzy search should be performed with the query given in the argument `search`. The available options are `title` and `cadence_title`. Default is `[title]`. [Introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/350991) in GitLab 16.2. |
 | `include_ancestors` | boolean | no       | Include iterations from parent group and its ancestors. Defaults to `true`.                    |
 | `updated_before`    | datetime | no      | Return only iterations updated before the given datetime. Expected in ISO 8601 format (`2019-03-15T08:00:00Z`). [Introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/378662) in GitLab 15.10. |
 | `updated_after`     | datetime | no      | Return only iterations updated after the given datetime. Expected in ISO 8601 format (`2019-03-15T08:00:00Z`). [Introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/378662) in GitLab 15.10. |
diff --git a/doc/api/iterations.md b/doc/api/iterations.md
index 567a2def09f..14633c3fc83 100644
--- a/doc/api/iterations.md
+++ b/doc/api/iterations.md
@@ -32,6 +32,7 @@ GET /projects/:id/iterations?updated_after=2013-10-02T09%3A24%3A18Z
 | ------------------- | ------- | -------- | ----------- |
 | `state`             | string  | no       | 'Return `opened`, `upcoming`, `current`, `closed`, or `all` iterations.'                       |
 | `search`            | string  | no       | Return only iterations with a title matching the provided string.                              |
+| `in`                | array of strings | no | Fields in which fuzzy search should be performed with the query given in the argument `search`. The available options are `title` and `cadence_title`. Default is `[title]`. [Introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/350991) in GitLab 16.2. |
 | `include_ancestors` | boolean | no       | Include iterations from parent group and its ancestors. Defaults to `true`.                    |
 | `updated_before`    | datetime | no      | Return only iterations updated before the given datetime. Expected in ISO 8601 format (`2019-03-15T08:00:00Z`). [Introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/378662) in GitLab 15.10. |
 | `updated_after`     | datetime | no      | Return only iterations updated after the given datetime. Expected in ISO 8601 format (`2019-03-15T08:00:00Z`). [Introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/378662) in GitLab 15.10. |
diff --git a/doc/user/clusters/agent/gitops.md b/doc/user/clusters/agent/gitops.md
index aff78ed477b..4d33603f770 100644
--- a/doc/user/clusters/agent/gitops.md
+++ b/doc/user/clusters/agent/gitops.md
@@ -78,6 +78,7 @@ For additional repository structure recommendations, see the [Flux documentation
 
 > - [Introduced](https://gitlab.com/gitlab-org/gitlab/-/issues/392852) in GitLab 16.1 with a [flag](../../../administration/feature_flags.md) named `notify_kas_on_git_push`. Disabled by default.
 > - [Enabled on GitLab.com and self-managed](https://gitlab.com/gitlab-org/gitlab/-/merge_requests/126527) in GitLab 16.2.
+> - [Feature flag removed](https://gitlab.com/gitlab-org/gitlab/-/issues/410429) in GitLab 16.3.
 
 Usually, the Flux source controller reconciles Git repositories at configured intervals.
 This can cause delays between a `git push` and the reconciliation of the cluster state, and results in
diff --git a/lib/gitlab/git/repository.rb b/lib/gitlab/git/repository.rb
index cb95af69b00..4d7bd549dc8 100644
--- a/lib/gitlab/git/repository.rb
+++ b/lib/gitlab/git/repository.rb
@@ -750,6 +750,16 @@ module Gitlab
         raise DeleteBranchError, e
       end
 
+      def async_delete_refs(*refs)
+        raise "async_delete_refs only supports project repositories" unless container.is_a?(Project)
+
+        records = refs.map do |ref|
+          BatchedGitRefUpdates::Deletion.new(project_id: container.id, ref: ref, created_at: Time.current, updated_at: Time.current)
+        end
+
+        BatchedGitRefUpdates::Deletion.bulk_insert!(records)
+      end
+
       def delete_refs(...)
         wrapped_gitaly_errors do
           gitaly_delete_refs(...)
diff --git a/spec/db/schema_spec.rb b/spec/db/schema_spec.rb
index bb50c7cf69d..2dfea0c7924 100644
--- a/spec/db/schema_spec.rb
+++ b/spec/db/schema_spec.rb
@@ -89,6 +89,7 @@ RSpec.describe 'Database schema', feature_category: :database do
     oauth_access_tokens: %w[resource_owner_id application_id],
     oauth_applications: %w[owner_id],
     p_ci_builds: %w[project_id runner_id user_id erased_by_id trigger_request_id partition_id],
+    p_batched_git_ref_updates_deletions: %w[project_id partition_id],
     product_analytics_events_experimental: %w[event_id txn_id user_id],
     project_build_artifacts_size_refreshes: %w[last_job_artifact_id],
     project_data_transfers: %w[project_id namespace_id],
diff --git a/spec/models/batched_git_ref_updates/deletion_spec.rb b/spec/models/batched_git_ref_updates/deletion_spec.rb
new file mode 100644
index 00000000000..1679e8977b3
--- /dev/null
+++ b/spec/models/batched_git_ref_updates/deletion_spec.rb
@@ -0,0 +1,125 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BatchedGitRefUpdates::Deletion, feature_category: :gitaly do
+  describe '.mark_records_processed' do
+    let_it_be(:deletion_1) { described_class.create!(project_id: 5, ref: 'refs/test/1') }
+    let_it_be(:deletion_2) { described_class.create!(project_id: 1, ref: 'refs/test/2') }
+    let_it_be(:deletion_3) { described_class.create!(project_id: 3, ref: 'refs/test/3') }
+    let_it_be(:deletion_4) { described_class.create!(project_id: 1, ref: 'refs/test/4') }
+    let_it_be(:deletion_5) { described_class.create!(project_id: 4, ref: 'refs/test/5', status: :processed) }
+
+    it 'updates all records' do
+      expect(described_class.status_pending.count).to eq(4)
+      expect(described_class.status_processed.count).to eq(1)
+
+      deletions = described_class.for_project(1).select_ref_and_identity
+      described_class.mark_records_processed(deletions)
+
+      deletions.each do |deletion|
+        expect(deletion.reload.status).to eq("processed")
+      end
+
+      expect(described_class.status_pending.count).to eq(2)
+      expect(described_class.status_processed.count).to eq(3)
+    end
+  end
+
+  describe 'sliding_list partitioning' do
+    let(:partition_manager) { Gitlab::Database::Partitioning::PartitionManager.new(described_class) }
+
+    describe 'next_partition_if callback' do
+      let(:active_partition) { described_class.partitioning_strategy.active_partition }
+
+      subject(:value) { described_class.partitioning_strategy.next_partition_if.call(active_partition) }
+
+      context 'when the partition is empty' do
+        it { is_expected.to eq(false) }
+      end
+
+      context 'when the partition has records' do
+        before do
+          described_class.create!(project_id: 1, ref: 'refs/test/1', status: :processed)
+          described_class.create!(project_id: 2, ref: 'refs/test/2', status: :pending)
+        end
+
+        it { is_expected.to eq(false) }
+      end
+
+      context 'when the first record of the partition is older than PARTITION_DURATION' do
+        before do
+          described_class.create!(
+            project_id: 1,
+            ref: 'refs/test/1',
+            created_at: (described_class::PARTITION_DURATION + 1.day).ago)
+
+          described_class.create!(project_id: 2, ref: 'refs/test/2')
+        end
+
+        it { is_expected.to eq(true) }
+      end
+    end
+
+    describe 'detach_partition_if callback' do
+      let(:active_partition) { described_class.partitioning_strategy.active_partition }
+
+      subject(:value) { described_class.partitioning_strategy.detach_partition_if.call(active_partition) }
+
+      context 'when the partition contains unprocessed records' do
+        before do
+          described_class.create!(project_id: 1, ref: 'refs/test/1')
+          described_class.create!(project_id: 2, ref: 'refs/test/2', status: :processed)
+        end
+
+        it { is_expected.to eq(false) }
+      end
+
+      context 'when the partition contains only processed records' do
+        before do
+          described_class.create!(project_id: 1, ref: 'refs/test/1', status: :processed)
+          described_class.create!(project_id: 2, ref: 'refs/test/2', status: :processed)
+        end
+
+        it { is_expected.to eq(true) }
+      end
+    end
+
+    describe 'the behavior of the strategy' do
+      it 'moves records to new partitions as time passes', :freeze_time do
+        # We start with partition 1
+        expect(described_class.partitioning_strategy.current_partitions.map(&:value)).to eq([1])
+
+        # it's not a day old yet so no new partitions are created
+        partition_manager.sync_partitions
+
+        expect(described_class.partitioning_strategy.current_partitions.map(&:value)).to eq([1])
+
+        # add one record so the next partition will be created
+        described_class.create!(project_id: 1, ref: 'refs/test/1')
+
+        # after traveling forward a day
+        travel(described_class::PARTITION_DURATION + 1.second)
+
+        # a new partition is created
+        partition_manager.sync_partitions
+
+        expect(described_class.partitioning_strategy.current_partitions.map(&:value)).to contain_exactly(1, 2)
+
+        # and we can insert to the new partition
+        described_class.create!(project_id: 2, ref: 'refs/test/2')
+
+        # after processing old records
+        described_class.mark_records_processed(described_class.for_partition(1).select_ref_and_identity)
+
+        partition_manager.sync_partitions
+
+        # the old one is removed
+        expect(described_class.partitioning_strategy.current_partitions.map(&:value)).to eq([2])
+
+        # and we only have the newly created partition left.
+        expect(described_class.count).to eq(1)
+      end
+    end
+  end
+end
diff --git a/spec/models/ci/persistent_ref_spec.rb b/spec/models/ci/persistent_ref_spec.rb
index ecaa8f59ecf..ed4ea02d8ba 100644
--- a/spec/models/ci/persistent_ref_spec.rb
+++ b/spec/models/ci/persistent_ref_spec.rb
@@ -3,26 +3,40 @@
 require 'spec_helper'
 
 RSpec.describe Ci::PersistentRef do
-  it 'cleans up persistent refs after pipeline finished', :sidekiq_inline do
+  it 'cleans up persistent refs async after pipeline finished' do
     pipeline = create(:ci_pipeline, :running)
 
-    expect(Ci::PipelineCleanupRefWorker).to receive(:perform_async).with(pipeline.id)
-
-    pipeline.succeed!
+    expect { pipeline.succeed! }
+      .to change { ::BatchedGitRefUpdates::Deletion.count }
+      .by(1)
   end
 
-  context 'when pipeline_cleanup_ref_worker_async is disabled' do
+  context 'when pipeline_delete_gitaly_refs_in_batches is disabled' do
     before do
-      stub_feature_flags(pipeline_cleanup_ref_worker_async: false)
+      stub_feature_flags(pipeline_delete_gitaly_refs_in_batches: false)
     end
 
     it 'cleans up persistent refs after pipeline finished' do
       pipeline = create(:ci_pipeline, :running)
 
-      expect(pipeline.persistent_ref).to receive(:delete).once
+      expect(Ci::PipelineCleanupRefWorker).to receive(:perform_async).with(pipeline.id)
 
       pipeline.succeed!
     end
+
+    context 'when pipeline_cleanup_ref_worker_async is disabled' do
+      before do
+        stub_feature_flags(pipeline_cleanup_ref_worker_async: false)
+      end
+
+      it 'cleans up persistent refs after pipeline finished' do
+        pipeline = create(:ci_pipeline, :running)
+
+        expect(pipeline.persistent_ref).to receive(:delete).once
+
+        pipeline.succeed!
+      end
+    end
   end
 
   describe '#exist?' do
diff --git a/spec/models/ci/pipeline_spec.rb b/spec/models/ci/pipeline_spec.rb
index ae3725a0b08..1e3ce8bc656 100644
--- a/spec/models/ci/pipeline_spec.rb
+++ b/spec/models/ci/pipeline_spec.rb
@@ -1328,33 +1328,47 @@ RSpec.describe Ci::Pipeline, :mailer, factory_default: :keep, feature_category:
 
     %w[succeed! drop! cancel! skip! block! delay!].each do |action|
       context "when the pipeline received #{action} event" do
-        it 'deletes a persistent ref asynchronously', :sidekiq_inline do
-          expect(pipeline.persistent_ref).not_to receive(:delete_refs)
-
-          expect(Ci::PipelineCleanupRefWorker).to receive(:perform_async)
-            .with(pipeline.id).and_call_original
-
-          expect_next_instance_of(Ci::PersistentRef) do |persistent_ref|
-            expect(persistent_ref).to receive(:delete_refs)
-              .with("refs/#{Repository::REF_PIPELINES}/#{pipeline.id}").once
-          end
+        it 'deletes a persistent ref asynchronously' do
+          expect(pipeline.persistent_ref).to receive(:async_delete)
+          expect(pipeline.persistent_ref).not_to receive(:delete)
 
           pipeline.public_send(action)
         end
 
-        context 'when pipeline_cleanup_ref_worker_async is disabled' do
+        context 'when pipeline_delete_gitaly_refs_in_batches is disabled' do
           before do
-            stub_feature_flags(pipeline_cleanup_ref_worker_async: false)
+            stub_feature_flags(pipeline_delete_gitaly_refs_in_batches: false)
           end
 
-          it 'deletes a persistent ref synchronously' do
-            expect(Ci::PipelineCleanupRefWorker).not_to receive(:perform_async).with(pipeline.id)
+          it 'deletes a persistent ref asynchronously via ::Ci::PipelineCleanupRefWorker', :sidekiq_inline do
+            expect(pipeline.persistent_ref).not_to receive(:delete_refs)
 
-            expect(pipeline.persistent_ref).to receive(:delete_refs).once
-              .with("refs/#{Repository::REF_PIPELINES}/#{pipeline.id}")
+            expect(Ci::PipelineCleanupRefWorker).to receive(:perform_async)
+              .with(pipeline.id).and_call_original
+
+            expect_next_instance_of(Ci::PersistentRef) do |persistent_ref|
+              expect(persistent_ref).to receive(:delete_refs)
+                .with("refs/#{Repository::REF_PIPELINES}/#{pipeline.id}").once
+            end
 
             pipeline.public_send(action)
           end
+
+          context 'when pipeline_cleanup_ref_worker_async is disabled' do
+            before do
+              stub_feature_flags(pipeline_delete_gitaly_refs_in_batches: false)
+              stub_feature_flags(pipeline_cleanup_ref_worker_async: false)
+            end
+
+            it 'deletes a persistent ref synchronously' do
+              expect(Ci::PipelineCleanupRefWorker).not_to receive(:perform_async).with(pipeline.id)
+
+              expect(pipeline.persistent_ref).to receive(:delete_refs).once
+                .with("refs/#{Repository::REF_PIPELINES}/#{pipeline.id}")
+
+              pipeline.public_send(action)
+            end
+          end
         end
       end
     end
diff --git a/spec/models/merge_request_spec.rb b/spec/models/merge_request_spec.rb
index bf71d289105..1dfd14c0993 100644
--- a/spec/models/merge_request_spec.rb
+++ b/spec/models/merge_request_spec.rb
@@ -5124,24 +5124,39 @@ RSpec.describe MergeRequest, factory_default: :keep, feature_category: :code_rev
 
     let(:merge_request) { build(:merge_request, source_project: create(:project, :repository)) }
 
-    it 'does schedule MergeRequests::CleanupRefWorker' do
-      expect(MergeRequests::CleanupRefWorker).to receive(:perform_async).with(merge_request.id, 'train')
+    it 'deletes refs asynchronously' do
+      expect(merge_request.target_project.repository)
+        .to receive(:async_delete_refs)
+        .with(merge_request.train_ref_path)
 
       subject
     end
 
-    context 'when merge_request_cleanup_ref_worker_async is disabled' do
+    context 'when merge_request_delete_gitaly_refs_in_batches is disabled' do
       before do
-        stub_feature_flags(merge_request_cleanup_ref_worker_async: false)
+        stub_feature_flags(merge_request_delete_gitaly_refs_in_batches: false)
       end
 
-      it 'deletes all refs from the target project' do
-        expect(merge_request.target_project.repository)
-          .to receive(:delete_refs)
-          .with(merge_request.train_ref_path)
+      it 'does schedule MergeRequests::CleanupRefWorker' do
+        expect(MergeRequests::CleanupRefWorker).to receive(:perform_async).with(merge_request.id, 'train')
 
         subject
       end
+
+      context 'when merge_request_cleanup_ref_worker_async is disabled' do
+        before do
+          stub_feature_flags(merge_request_delete_gitaly_refs_in_batches: false)
+          stub_feature_flags(merge_request_cleanup_ref_worker_async: false)
+        end
+
+        it 'deletes all refs from the target project' do
+          expect(merge_request.target_project.repository)
+            .to receive(:delete_refs)
+            .with(merge_request.train_ref_path)
+
+          subject
+        end
+      end
     end
   end
 
diff --git a/spec/services/batched_git_ref_updates/cleanup_scheduler_service_spec.rb b/spec/services/batched_git_ref_updates/cleanup_scheduler_service_spec.rb
new file mode 100644
index 00000000000..50081a5e9e7
--- /dev/null
+++ b/spec/services/batched_git_ref_updates/cleanup_scheduler_service_spec.rb
@@ -0,0 +1,55 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BatchedGitRefUpdates::CleanupSchedulerService, feature_category: :gitaly do
+  let(:service) { described_class.new }
+
+  describe '#execute' do
+    before do
+      BatchedGitRefUpdates::Deletion.create!(project_id: 123, ref: 'ref1')
+      BatchedGitRefUpdates::Deletion.create!(project_id: 123, ref: 'ref2')
+      BatchedGitRefUpdates::Deletion.create!(project_id: 456, ref: 'ref3')
+      BatchedGitRefUpdates::Deletion.create!(project_id: 789, ref: 'ref4', status: :processed)
+    end
+
+    it 'schedules ProjectCleanupWorker for each project in pending BatchedGitRefUpdates::Deletion' do
+      project_ids = []
+      expect(BatchedGitRefUpdates::ProjectCleanupWorker)
+        .to receive(:bulk_perform_async_with_contexts) do |deletions, arguments_proc:, context_proc:| # rubocop:disable Lint/UnusedBlockArgument
+        project_ids += deletions.map(&arguments_proc)
+      end
+
+      service.execute
+
+      expect(project_ids).to contain_exactly(123, 456)
+    end
+
+    it 'returns stats' do
+      stats = service.execute
+
+      expect(stats).to eq({
+        total_projects: 2
+      })
+    end
+
+    it 'acquires a lock to avoid running duplicate instances' do
+      expect(service).to receive(:in_lock) # Mock and don't yield
+        .with(described_class.name, retries: 0, ttl: described_class::LOCK_TIMEOUT)
+      expect(BatchedGitRefUpdates::ProjectCleanupWorker).not_to receive(:bulk_perform_async_with_contexts)
+
+      service.execute
+    end
+
+    it 'limits to MAX_PROJECTS before it stops' do
+      stub_const("#{described_class}::BATCH_SIZE", 1)
+      stub_const("#{described_class}::MAX_PROJECTS", 1)
+
+      stats = service.execute
+
+      expect(stats).to eq({
+        total_projects: 1
+      })
+    end
+  end
+end
diff --git a/spec/services/batched_git_ref_updates/project_cleanup_service_spec.rb b/spec/services/batched_git_ref_updates/project_cleanup_service_spec.rb
new file mode 100644
index 00000000000..dcdfdfade3c
--- /dev/null
+++ b/spec/services/batched_git_ref_updates/project_cleanup_service_spec.rb
@@ -0,0 +1,98 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BatchedGitRefUpdates::ProjectCleanupService, feature_category: :gitaly do
+  let(:service) { described_class.new(project1.id) }
+  let_it_be(:project1) { create(:project, :repository) }
+  let_it_be(:project2) { create(:project, :repository) }
+  let!(:project1_ref1) do
+    BatchedGitRefUpdates::Deletion.create!(project_id: project1.id, ref: 'refs/test/project1-ref1')
+  end
+
+  let!(:project1_ref2) do
+    BatchedGitRefUpdates::Deletion.create!(project_id: project1.id, ref: 'refs/test/project1-ref2')
+  end
+
+  let!(:project1_ref3) do
+    BatchedGitRefUpdates::Deletion.create!(project_id: project1.id, ref: 'refs/test/already-processed',
+      status: :processed)
+  end
+
+  let!(:project2_ref1) do
+    BatchedGitRefUpdates::Deletion.create!(project_id: project2.id, ref: 'refs/test/project2-ref1')
+  end
+
+  describe '#execute' do
+    before do
+      project1.repository.create_ref('HEAD', 'refs/test/ref-to-not-be-deleted')
+      project1.repository.create_ref('HEAD', project1_ref1.ref)
+      project1.repository.create_ref('HEAD', project1_ref2.ref)
+      project1.repository.create_ref('HEAD', 'refs/test/already-processed')
+      project2.repository.create_ref('HEAD', project2_ref1.ref)
+    end
+
+    it 'deletes the named refs in batches for the given project only' do
+      expect(test_refs(project1)).to include(
+        'refs/test/ref-to-not-be-deleted',
+        'refs/test/already-processed',
+        'refs/test/project1-ref1',
+        'refs/test/project1-ref1',
+        'refs/test/project1-ref2')
+
+      service.execute
+
+      expect(test_refs(project1)).to include(
+        'refs/test/already-processed',
+        'refs/test/ref-to-not-be-deleted')
+
+      expect(test_refs(project1)).not_to include(
+        'refs/test/project1-ref1',
+        'refs/test/project1-ref2')
+
+      expect(test_refs(project2)).to include('refs/test/project2-ref1')
+    end
+
+    it 'marks the processed BatchedGitRefUpdates::Deletion as processed' do
+      service.execute
+
+      expect(BatchedGitRefUpdates::Deletion.status_pending.map(&:ref)).to contain_exactly('refs/test/project2-ref1')
+      expect(BatchedGitRefUpdates::Deletion.status_processed.map(&:ref)).to contain_exactly(
+        'refs/test/project1-ref1',
+        'refs/test/project1-ref2',
+        'refs/test/already-processed')
+    end
+
+    it 'returns stats' do
+      result = service.execute
+
+      expect(result[:total_deletes]).to eq(2)
+    end
+
+    it 'acquires a lock for the given project_id to avoid running duplicate instances' do
+      expect(service).to receive(:in_lock) # Mock and don't yield
+        .with("#{described_class}/#{project1.id}", retries: 0, ttl: described_class::LOCK_TIMEOUT)
+
+      expect { service.execute }.not_to change { BatchedGitRefUpdates::Deletion.status_pending.count }
+    end
+
+    it 'does nothing when the project does not exist' do
+      result = described_class.new(non_existing_record_id).execute
+
+      expect(result[:total_deletes]).to eq(0)
+    end
+
+    it 'stops after it reaches limit of deleted refs' do
+      stub_const("#{described_class}::QUERY_BATCH_SIZE", 1)
+      stub_const("#{described_class}::MAX_DELETES", 1)
+
+      result = service.execute
+
+      expect(result[:total_deletes]).to eq(1)
+    end
+
+    def test_refs(project)
+      project.repository.list_refs(['refs/test/']).map(&:name)
+    end
+  end
+end
diff --git a/spec/services/git/base_hooks_service_spec.rb b/spec/services/git/base_hooks_service_spec.rb
index 60883db0cd5..e083c8d7316 100644
--- a/spec/services/git/base_hooks_service_spec.rb
+++ b/spec/services/git/base_hooks_service_spec.rb
@@ -363,17 +363,5 @@ RSpec.describe Git::BaseHooksService, feature_category: :source_code_management
         subject.execute
       end
     end
-
-    context 'when :notify_kas_on_git_push feature flag is disabled' do
-      before do
-        stub_feature_flags(notify_kas_on_git_push: false)
-      end
-
-      it do
-        expect(Clusters::Agents::NotifyGitPushWorker).not_to receive(:perform_async)
-
-        subject.execute
-      end
-    end
   end
 end
diff --git a/spec/services/merge_requests/cleanup_refs_service_spec.rb b/spec/services/merge_requests/cleanup_refs_service_spec.rb
index efb6265e3d8..c818c60ad5f 100644
--- a/spec/services/merge_requests/cleanup_refs_service_spec.rb
+++ b/spec/services/merge_requests/cleanup_refs_service_spec.rb
@@ -18,6 +18,7 @@ RSpec.describe MergeRequests::CleanupRefsService, feature_category: :code_review
 
   describe '#execute' do
     before do
+      stub_feature_flags(merge_request_delete_gitaly_refs_in_batches: false)
       stub_feature_flags(merge_request_cleanup_ref_worker_async: false)
 
       # Need to re-enable this as it's being stubbed in spec_helper for
diff --git a/spec/services/merge_requests/merge_to_ref_service_spec.rb b/spec/services/merge_requests/merge_to_ref_service_spec.rb
index 4e951f1bc85..5105a275fba 100644
--- a/spec/services/merge_requests/merge_to_ref_service_spec.rb
+++ b/spec/services/merge_requests/merge_to_ref_service_spec.rb
@@ -36,29 +36,6 @@ RSpec.describe MergeRequests::MergeToRefService, feature_category: :code_review_
       expect(repository.ref_exists?(target_ref)).to be(true)
       expect(ref_head.id).to eq(result[:commit_id])
     end
-
-    context 'cache_merge_to_ref_calls parameter', :use_clean_rails_memory_store_caching do
-      before do
-        # warm the cache
-        #
-        service.execute(merge_request, true)
-      end
-
-      context 'when true' do
-        it 'caches the response', :request_store do
-          expect { 3.times { service.execute(merge_request, true) } }
-            .not_to change(Gitlab::GitalyClient, :get_request_count)
-        end
-      end
-
-      context 'when false' do
-        it 'does not cache the response', :request_store do
-          expect(Gitlab::GitalyClient).to receive(:call).at_least(3).times.and_call_original
-
-          3.times { service.execute(merge_request, false) }
-        end
-      end
-    end
   end
 
   shared_examples_for 'successfully evaluates pre-condition checks' do
diff --git a/spec/workers/batched_git_ref_updates/cleanup_scheduler_worker_spec.rb b/spec/workers/batched_git_ref_updates/cleanup_scheduler_worker_spec.rb
new file mode 100644
index 00000000000..a52043993b7
--- /dev/null
+++ b/spec/workers/batched_git_ref_updates/cleanup_scheduler_worker_spec.rb
@@ -0,0 +1,31 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BatchedGitRefUpdates::CleanupSchedulerWorker, feature_category: :gitaly do
+  let(:stats) { { total_projects: 456 } }
+  let(:service) { instance_double(BatchedGitRefUpdates::CleanupSchedulerService, execute: stats) }
+  let(:worker) { described_class.new }
+
+  describe '#perform' do
+    before do
+      allow(BatchedGitRefUpdates::CleanupSchedulerService).to receive(:new).and_return(service)
+    end
+
+    it 'delegates to CleanupSchedulerService' do
+      expect(service).to receive(:execute)
+
+      worker.perform
+    end
+
+    it 'logs stats' do
+      worker.perform
+
+      expect(worker.logging_extras).to eq({
+        "extra.batched_git_ref_updates_cleanup_scheduler_worker.stats" => { total_projects: 456 }
+      })
+    end
+  end
+
+  it_behaves_like 'an idempotent worker'
+end
diff --git a/spec/workers/batched_git_ref_updates/project_cleanup_worker_spec.rb b/spec/workers/batched_git_ref_updates/project_cleanup_worker_spec.rb
new file mode 100644
index 00000000000..5442b9dd051
--- /dev/null
+++ b/spec/workers/batched_git_ref_updates/project_cleanup_worker_spec.rb
@@ -0,0 +1,33 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BatchedGitRefUpdates::ProjectCleanupWorker, feature_category: :gitaly do
+  let(:stats) { { total_deletes: 456 } }
+  let(:service) { instance_double(BatchedGitRefUpdates::ProjectCleanupService, execute: stats) }
+  let(:worker) { described_class.new }
+
+  describe '#perform' do
+    before do
+      allow(BatchedGitRefUpdates::ProjectCleanupService).to receive(:new).with(123).and_return(service)
+    end
+
+    it 'delegates to ProjectCleanupService' do
+      expect(service).to receive(:execute)
+
+      worker.perform(123)
+    end
+
+    it 'logs stats' do
+      worker.perform(123)
+
+      expect(worker.logging_extras).to eq({
+        "extra.batched_git_ref_updates_project_cleanup_worker.stats" => { total_deletes: 456 }
+      })
+    end
+  end
+
+  it_behaves_like 'an idempotent worker' do
+    let(:job_args) { [123] }
+  end
+end
diff --git a/spec/workers/clusters/agents/notify_git_push_worker_spec.rb b/spec/workers/clusters/agents/notify_git_push_worker_spec.rb
index 561a66b86e9..c6ef8dc3338 100644
--- a/spec/workers/clusters/agents/notify_git_push_worker_spec.rb
+++ b/spec/workers/clusters/agents/notify_git_push_worker_spec.rb
@@ -25,17 +25,5 @@ RSpec.describe Clusters::Agents::NotifyGitPushWorker, feature_category: :deploym
         expect { subject }.not_to raise_error
       end
     end
-
-    context 'when the :notify_kas_on_git_push feature flag is disabled' do
-      before do
-        stub_feature_flags(notify_kas_on_git_push: false)
-      end
-
-      it 'does not notify KAS' do
-        expect(Gitlab::Kas::Client).not_to receive(:new)
-
-        subject
-      end
-    end
   end
 end