Add latest changes from gitlab-org/gitlab@14-8-stable-eev14.8.0-rc42

26 files changed, 556 insertions, 118 deletions

diff --git a/spec/workers/auto_devops/disable_worker_spec.rb b/spec/workers/auto_devops/disable_worker_spec.rb
index 239f4b09f5c..e1de97e0ce5 100644
--- a/spec/workers/auto_devops/disable_worker_spec.rb
+++ b/spec/workers/auto_devops/disable_worker_spec.rb
@@ -26,7 +26,7 @@ RSpec.describe AutoDevops::DisableWorker, '#perform' do
     let(:namespace) { create(:namespace, owner: owner) }
     let(:project) { create(:project, :repository, :auto_devops, namespace: namespace) }
 
-    it 'sends an email to pipeline user and project owner' do
+    it 'sends an email to pipeline user and project owner(s)' do
       expect(NotificationService).to receive_message_chain(:new, :autodevops_disabled).with(pipeline, [user.email, owner.email])
 
       subject.perform(pipeline.id)
diff --git a/spec/workers/background_migration/ci_database_worker_spec.rb b/spec/workers/background_migration/ci_database_worker_spec.rb
new file mode 100644
index 00000000000..82c562c4042
--- /dev/null
+++ b/spec/workers/background_migration/ci_database_worker_spec.rb
@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BackgroundMigration::CiDatabaseWorker, :clean_gitlab_redis_shared_state, if: Gitlab::Database.has_config?(:ci) do
+  it_behaves_like 'it runs background migration jobs', 'ci'
+end
diff --git a/spec/workers/background_migration_worker_spec.rb b/spec/workers/background_migration_worker_spec.rb
index 4297e55ca6c..1558c3c9250 100644
--- a/spec/workers/background_migration_worker_spec.rb
+++ b/spec/workers/background_migration_worker_spec.rb
@@ -3,5 +3,5 @@
 require 'spec_helper'
 
 RSpec.describe BackgroundMigrationWorker, :clean_gitlab_redis_shared_state do
-  it_behaves_like 'it runs background migration jobs', 'main', :background_migration_database_health_reschedules
+  it_behaves_like 'it runs background migration jobs', 'main'
 end
diff --git a/spec/workers/ci/delete_objects_worker_spec.rb b/spec/workers/ci/delete_objects_worker_spec.rb
index 52d90d7667a..3d985dffdc5 100644
--- a/spec/workers/ci/delete_objects_worker_spec.rb
+++ b/spec/workers/ci/delete_objects_worker_spec.rb
@@ -6,15 +6,16 @@ RSpec.describe Ci::DeleteObjectsWorker do
   let(:worker) { described_class.new }
 
   it { expect(described_class.idempotent?).to be_truthy }
+  it { is_expected.to respond_to(:max_running_jobs) }
+  it { is_expected.to respond_to(:remaining_work_count) }
+  it { is_expected.to respond_to(:perform_work) }
 
   describe '#perform' do
     it 'executes a service' do
-      allow(worker).to receive(:max_running_jobs).and_return(25)
-
       expect_next_instance_of(Ci::DeleteObjectsService) do |instance|
         expect(instance).to receive(:execute)
         expect(instance).to receive(:remaining_batches_count)
-          .with(max_batch_count: 25)
+          .with(max_batch_count: 20)
           .once
           .and_call_original
       end
@@ -22,30 +23,4 @@ RSpec.describe Ci::DeleteObjectsWorker do
       worker.perform
     end
   end
-
-  describe '#max_running_jobs' do
-    using RSpec::Parameterized::TableSyntax
-
-    before do
-      stub_feature_flags(
-        ci_delete_objects_medium_concurrency: medium,
-        ci_delete_objects_high_concurrency: high
-      )
-    end
-
-    subject(:max_running_jobs) { worker.max_running_jobs }
-
-    where(:medium, :high, :expected) do
-      false | false | 2
-      true  | false | 20
-      true  | true  | 20
-      false | true  | 50
-    end
-
-    with_them do
-      it 'sets up concurrency depending on the feature flag' do
-        expect(max_running_jobs).to eq(expected)
-      end
-    end
-  end
 end
diff --git a/spec/workers/ci/external_pull_requests/create_pipeline_worker_spec.rb b/spec/workers/ci/external_pull_requests/create_pipeline_worker_spec.rb
index 116a0e4d035..a637ac088ff 100644
--- a/spec/workers/ci/external_pull_requests/create_pipeline_worker_spec.rb
+++ b/spec/workers/ci/external_pull_requests/create_pipeline_worker_spec.rb
@@ -4,7 +4,7 @@ require 'spec_helper'
 
 RSpec.describe Ci::ExternalPullRequests::CreatePipelineWorker do
   let_it_be(:project) { create(:project, :auto_devops, :repository) }
-  let_it_be(:user) { project.owner }
+  let_it_be(:user) { project.first_owner }
   let_it_be(:external_pull_request) do
     branch = project.repository.branches.last
     create(:external_pull_request, project: project, source_branch: branch.name, source_sha: branch.target)
diff --git a/spec/workers/cleanup_container_repository_worker_spec.rb b/spec/workers/cleanup_container_repository_worker_spec.rb
index 6ae4308bd46..6723ea2049d 100644
--- a/spec/workers/cleanup_container_repository_worker_spec.rb
+++ b/spec/workers/cleanup_container_repository_worker_spec.rb
@@ -5,7 +5,7 @@ require 'spec_helper'
 RSpec.describe CleanupContainerRepositoryWorker, :clean_gitlab_redis_shared_state do
   let(:repository) { create(:container_repository) }
   let(:project) { repository.project }
-  let(:user) { project.owner }
+  let(:user) { project.first_owner }
 
   subject { described_class.new }
 
diff --git a/spec/workers/concerns/application_worker_spec.rb b/spec/workers/concerns/application_worker_spec.rb
index 85731de2a45..95d9b982fc4 100644
--- a/spec/workers/concerns/application_worker_spec.rb
+++ b/spec/workers/concerns/application_worker_spec.rb
@@ -247,45 +247,6 @@ RSpec.describe ApplicationWorker do
     end
   end
 
-  describe '.perform_async' do
-    using RSpec::Parameterized::TableSyntax
-
-    where(:primary_only?, :skip_scheduling_ff, :data_consistency, :schedules_job?) do
-      true  | false | :sticky  | false
-      true  | false | :delayed | false
-      true  | false | :always  | false
-      true  | true  | :sticky  | false
-      true  | true  | :delayed | false
-      true  | true  | :always  | false
-      false | false | :sticky  | true
-      false | false | :delayed | true
-      false | false | :always  | false
-      false | true  | :sticky  | false
-      false | true  | :delayed | false
-      false | true  | :always  | false
-    end
-
-    before do
-      stub_const(worker.name, worker)
-      worker.data_consistency(data_consistency)
-
-      allow(Gitlab::Database::LoadBalancing).to receive(:primary_only?).and_return(primary_only?)
-      stub_feature_flags(skip_scheduling_workers_for_replicas: skip_scheduling_ff)
-    end
-
-    with_them do
-      it 'schedules or enqueues the job correctly' do
-        if schedules_job?
-          expect(worker).to receive(:perform_in).with(described_class::DEFAULT_DELAY_INTERVAL.seconds, 123)
-        else
-          expect(worker).not_to receive(:perform_in)
-        end
-
-        worker.perform_async(123)
-      end
-    end
-  end
-
   context 'different kinds of push_bulk' do
     shared_context 'set safe limit beyond the number of jobs to be enqueued' do
       before do
diff --git a/spec/workers/container_expiration_policy_worker_spec.rb b/spec/workers/container_expiration_policy_worker_spec.rb
index ebf80041151..2cfb613865d 100644
--- a/spec/workers/container_expiration_policy_worker_spec.rb
+++ b/spec/workers/container_expiration_policy_worker_spec.rb
@@ -60,12 +60,11 @@ RSpec.describe ContainerExpirationPolicyWorker do
       context 'with container expiration policies' do
         let_it_be(:container_expiration_policy, reload: true) { create(:container_expiration_policy, :runnable) }
         let_it_be(:container_repository) { create(:container_repository, project: container_expiration_policy.project) }
-        let_it_be(:user) { container_expiration_policy.project.owner }
 
         context 'a valid policy' do
           it 'runs the policy' do
             expect(ContainerExpirationPolicyService)
-              .to receive(:new).with(container_expiration_policy.project, user).and_call_original
+              .to receive(:new).with(container_expiration_policy.project, nil).and_call_original
             expect(CleanupContainerRepositoryWorker).to receive(:perform_async).once.and_call_original
 
             expect { subject }.not_to raise_error
@@ -102,7 +101,7 @@ RSpec.describe ContainerExpirationPolicyWorker do
           end
 
           it 'disables the policy and tracks an error' do
-            expect(ContainerExpirationPolicyService).not_to receive(:new).with(container_expiration_policy, user)
+            expect(ContainerExpirationPolicyService).not_to receive(:new).with(container_expiration_policy, nil)
             expect(Gitlab::ErrorTracking).to receive(:log_exception).with(instance_of(described_class::InvalidPolicyError), container_expiration_policy_id: container_expiration_policy.id)
 
             expect { subject }.to change { container_expiration_policy.reload.enabled }.from(true).to(false)
diff --git a/spec/workers/container_registry/migration/enqueuer_worker_spec.rb b/spec/workers/container_registry/migration/enqueuer_worker_spec.rb
new file mode 100644
index 00000000000..12c14c35365
--- /dev/null
+++ b/spec/workers/container_registry/migration/enqueuer_worker_spec.rb
@@ -0,0 +1,178 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe ContainerRegistry::Migration::EnqueuerWorker, :aggregate_failures do
+  let_it_be_with_reload(:container_repository) { create(:container_repository, created_at: 2.days.ago) }
+
+  let(:worker) { described_class.new }
+
+  before do
+    stub_container_registry_config(enabled: true)
+    stub_application_setting(container_registry_import_created_before: 1.day.ago)
+    stub_container_registry_tags(repository: container_repository.path, tags: %w(tag1 tag2 tag3), with_manifest: true)
+  end
+
+  describe '#perform' do
+    subject { worker.perform }
+
+    shared_examples 'no action' do
+      it 'does not queue or change any repositories' do
+        subject
+
+        expect(container_repository.reload).to be_default
+      end
+    end
+
+    shared_examples 're-enqueuing based on capacity' do
+      context 'below capacity' do
+        before do
+          allow(ContainerRegistry::Migration).to receive(:capacity).and_return(9999)
+        end
+
+        it 're-enqueues the worker' do
+          expect(ContainerRegistry::Migration::EnqueuerWorker).to receive(:perform_async)
+
+          subject
+        end
+      end
+
+      context 'above capacity' do
+        before do
+          allow(ContainerRegistry::Migration).to receive(:capacity).and_return(-1)
+        end
+
+        it 'does not re-enqueue the worker' do
+          expect(ContainerRegistry::Migration::EnqueuerWorker).not_to receive(:perform_async)
+
+          subject
+        end
+      end
+    end
+
+    context 'with qualified repository' do
+      it 'starts the pre-import for the next qualified repository' do
+        method = worker.method(:next_repository)
+        allow(worker).to receive(:next_repository) do
+          next_qualified_repository = method.call
+          allow(next_qualified_repository).to receive(:migration_pre_import).and_return(:ok)
+          next_qualified_repository
+        end
+
+        expect(worker).to receive(:log_extra_metadata_on_done)
+          .with(:container_repository_id, container_repository.id)
+        expect(worker).to receive(:log_extra_metadata_on_done)
+          .with(:import_type, 'next')
+
+        subject
+
+        expect(container_repository.reload).to be_pre_importing
+      end
+
+      it_behaves_like 're-enqueuing based on capacity'
+    end
+
+    context 'migrations are disabled' do
+      before do
+        allow(ContainerRegistry::Migration).to receive(:enabled?).and_return(false)
+      end
+
+      it_behaves_like 'no action'
+    end
+
+    context 'above capacity' do
+      before do
+        create(:container_repository, :importing)
+        create(:container_repository, :importing)
+        allow(ContainerRegistry::Migration).to receive(:capacity).and_return(1)
+      end
+
+      it_behaves_like 'no action'
+
+      it 'does not re-enqueue the worker' do
+        expect(ContainerRegistry::Migration::EnqueuerWorker).not_to receive(:perform_async)
+
+        subject
+      end
+    end
+
+    context 'too soon before previous completed import step' do
+      before do
+        create(:container_repository, :import_done, migration_import_done_at: 1.minute.ago)
+        allow(ContainerRegistry::Migration).to receive(:enqueue_waiting_time).and_return(1.hour)
+      end
+
+      it_behaves_like 'no action'
+    end
+
+    context 'when an aborted import is available' do
+      let_it_be(:aborted_repository) { create(:container_repository, :import_aborted) }
+
+      it 'retries the import for the aborted repository' do
+        method = worker.method(:next_aborted_repository)
+        allow(worker).to receive(:next_aborted_repository) do
+          next_aborted_repository = method.call
+          allow(next_aborted_repository).to receive(:migration_import).and_return(:ok)
+          allow(next_aborted_repository.gitlab_api_client).to receive(:import_status).and_return('import_failed')
+          next_aborted_repository
+        end
+
+        expect(worker).to receive(:log_extra_metadata_on_done)
+          .with(:container_repository_id, aborted_repository.id)
+        expect(worker).to receive(:log_extra_metadata_on_done)
+          .with(:import_type, 'retry')
+
+        subject
+
+        expect(aborted_repository.reload).to be_importing
+        expect(container_repository.reload).to be_default
+      end
+
+      it_behaves_like 're-enqueuing based on capacity'
+    end
+
+    context 'when no repository qualifies' do
+      include_examples 'an idempotent worker' do
+        before do
+          allow(ContainerRepository).to receive(:ready_for_import).and_return(ContainerRepository.none)
+        end
+
+        it_behaves_like 'no action'
+      end
+    end
+
+    context 'over max tag count' do
+      before do
+        stub_application_setting(container_registry_import_max_tags_count: 2)
+      end
+
+      it 'skips the repository' do
+        subject
+
+        expect(container_repository.reload).to be_import_skipped
+        expect(container_repository.migration_skipped_reason).to eq('too_many_tags')
+        expect(container_repository.migration_skipped_at).not_to be_nil
+      end
+
+      it_behaves_like 're-enqueuing based on capacity'
+    end
+
+    context 'when an error occurs' do
+      before do
+        allow(ContainerRegistry::Migration).to receive(:max_tags_count).and_raise(StandardError)
+      end
+
+      it 'aborts the import' do
+        expect(Gitlab::ErrorTracking).to receive(:log_exception).with(
+          instance_of(StandardError),
+          next_repository_id: container_repository.id,
+          next_aborted_repository_id: nil
+        )
+
+        subject
+
+        expect(container_repository.reload).to be_import_aborted
+      end
+    end
+  end
+end
diff --git a/spec/workers/container_registry/migration/guard_worker_spec.rb b/spec/workers/container_registry/migration/guard_worker_spec.rb
new file mode 100644
index 00000000000..7d1df320d4e
--- /dev/null
+++ b/spec/workers/container_registry/migration/guard_worker_spec.rb
@@ -0,0 +1,162 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe ContainerRegistry::Migration::GuardWorker, :aggregate_failures do
+  include_context 'container registry client'
+
+  let(:worker) { described_class.new }
+
+  describe '#perform' do
+    let(:pre_importing_migrations) { ::ContainerRepository.with_migration_states(:pre_importing) }
+    let(:pre_import_done_migrations) { ::ContainerRepository.with_migration_states(:pre_import_done) }
+    let(:importing_migrations) { ::ContainerRepository.with_migration_states(:importing) }
+    let(:import_aborted_migrations) { ::ContainerRepository.with_migration_states(:import_aborted) }
+    let(:import_done_migrations) { ::ContainerRepository.with_migration_states(:import_done) }
+
+    subject { worker.perform }
+
+    before do
+      stub_container_registry_config(enabled: true, api_url: registry_api_url, key: 'spec/fixtures/x509_certificate_pk.key')
+      allow(::ContainerRegistry::Migration).to receive(:max_step_duration).and_return(5.minutes)
+    end
+
+    context 'on gitlab.com' do
+      before do
+        allow(::Gitlab).to receive(:com?).and_return(true)
+      end
+
+      shared_examples 'not aborting any migration' do
+        it 'will not abort the migration' do
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:stale_migrations_count, 1)
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:aborted_stale_migrations_count, 0)
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:long_running_stale_migration_container_repository_ids, [stale_migration.id])
+
+          expect { subject }
+              .to not_change(pre_importing_migrations, :count)
+              .and not_change(pre_import_done_migrations, :count)
+              .and not_change(importing_migrations, :count)
+              .and not_change(import_done_migrations, :count)
+              .and not_change(import_aborted_migrations, :count)
+              .and not_change { stale_migration.reload.migration_state }
+              .and not_change { ongoing_migration.migration_state }
+        end
+      end
+
+      context 'with no stale migrations' do
+        it_behaves_like 'an idempotent worker'
+
+        it 'will not update any migration state' do
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:stale_migrations_count, 0)
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:aborted_stale_migrations_count, 0)
+
+          expect { subject }
+            .to not_change(pre_importing_migrations, :count)
+            .and not_change(pre_import_done_migrations, :count)
+            .and not_change(importing_migrations, :count)
+            .and not_change(import_aborted_migrations, :count)
+        end
+      end
+
+      context 'with pre_importing stale migrations' do
+        let(:ongoing_migration) { create(:container_repository, :pre_importing) }
+        let(:stale_migration) { create(:container_repository, :pre_importing, migration_pre_import_started_at: 35.minutes.ago) }
+        let(:import_status) { 'test' }
+
+        before do
+          allow_next_instance_of(ContainerRegistry::GitlabApiClient) do |client|
+            allow(client).to receive(:import_status).and_return(import_status)
+          end
+        end
+
+        it 'will abort the migration' do
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:stale_migrations_count, 1)
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:aborted_stale_migrations_count, 1)
+
+          expect { subject }
+              .to change(pre_importing_migrations, :count).by(-1)
+              .and not_change(pre_import_done_migrations, :count)
+              .and not_change(importing_migrations, :count)
+              .and not_change(import_done_migrations, :count)
+              .and change(import_aborted_migrations, :count).by(1)
+              .and change { stale_migration.reload.migration_state }.from('pre_importing').to('import_aborted')
+              .and not_change { ongoing_migration.migration_state }
+        end
+
+        context 'the client returns pre_import_in_progress' do
+          let(:import_status) { 'pre_import_in_progress' }
+
+          it_behaves_like 'not aborting any migration'
+        end
+      end
+
+      context 'with pre_import_done stale migrations' do
+        let(:ongoing_migration) { create(:container_repository, :pre_import_done) }
+        let(:stale_migration) { create(:container_repository, :pre_import_done, migration_pre_import_done_at: 35.minutes.ago) }
+
+        before do
+          allow(::ContainerRegistry::Migration).to receive(:max_step_duration).and_return(5.minutes)
+        end
+
+        it 'will abort the migration' do
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:stale_migrations_count, 1)
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:aborted_stale_migrations_count, 1)
+
+          expect { subject }
+              .to not_change(pre_importing_migrations, :count)
+              .and change(pre_import_done_migrations, :count).by(-1)
+              .and not_change(importing_migrations, :count)
+              .and not_change(import_done_migrations, :count)
+              .and change(import_aborted_migrations, :count).by(1)
+              .and change { stale_migration.reload.migration_state }.from('pre_import_done').to('import_aborted')
+              .and not_change { ongoing_migration.migration_state }
+        end
+      end
+
+      context 'with importing stale migrations' do
+        let(:ongoing_migration) { create(:container_repository, :importing) }
+        let(:stale_migration) { create(:container_repository, :importing, migration_import_started_at: 35.minutes.ago) }
+        let(:import_status) { 'test' }
+
+        before do
+          allow_next_instance_of(ContainerRegistry::GitlabApiClient) do |client|
+            allow(client).to receive(:import_status).and_return(import_status)
+          end
+        end
+
+        it 'will abort the migration' do
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:stale_migrations_count, 1)
+          expect(worker).to receive(:log_extra_metadata_on_done).with(:aborted_stale_migrations_count, 1)
+
+          expect { subject }
+              .to not_change(pre_importing_migrations, :count)
+              .and not_change(pre_import_done_migrations, :count)
+              .and change(importing_migrations, :count).by(-1)
+              .and not_change(import_done_migrations, :count)
+              .and change(import_aborted_migrations, :count).by(1)
+              .and change { stale_migration.reload.migration_state }.from('importing').to('import_aborted')
+              .and not_change { ongoing_migration.migration_state }
+        end
+
+        context 'the client returns import_in_progress' do
+          let(:import_status) { 'import_in_progress' }
+
+          it_behaves_like 'not aborting any migration'
+        end
+      end
+    end
+
+    context 'not on gitlab.com' do
+      before do
+        allow(::Gitlab).to receive(:com?).and_return(false)
+      end
+
+      it 'is a no op' do
+        expect(::ContainerRepository).not_to receive(:with_stale_migration)
+        expect(worker).not_to receive(:log_extra_metadata_on_done)
+
+        subject
+      end
+    end
+  end
+end
diff --git a/spec/workers/container_registry/migration/observer_worker_spec.rb b/spec/workers/container_registry/migration/observer_worker_spec.rb
new file mode 100644
index 00000000000..fec6640d7ec
--- /dev/null
+++ b/spec/workers/container_registry/migration/observer_worker_spec.rb
@@ -0,0 +1,57 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe ContainerRegistry::Migration::ObserverWorker, :aggregate_failures do
+  let(:worker) { described_class.new }
+
+  describe '#perform' do
+    subject { worker.perform }
+
+    context 'when the migration feature flag is disabled' do
+      before do
+        stub_feature_flags(container_registry_migration_phase2_enabled: false)
+      end
+
+      it 'does nothing' do
+        expect(worker).not_to receive(:log_extra_metadata_on_done)
+
+        subject
+      end
+    end
+
+    context 'when the migration is enabled' do
+      before do
+        create_list(:container_repository, 3)
+        create(:container_repository, :pre_importing)
+        create(:container_repository, :pre_import_done)
+        create_list(:container_repository, 2, :importing)
+        create(:container_repository, :import_aborted)
+        # batch_count is not allowed within a transaction but
+        # all rspec tests run inside of a transaction.
+        # This mocks the false positive.
+        allow(ActiveRecord::Base.connection).to receive(:transaction_open?).and_return(false) # rubocop:disable Database/MultipleDatabases
+      end
+
+      it 'logs all the counts' do
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:default_count, 3)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:pre_importing_count, 1)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:pre_import_done_count, 1)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:importing_count, 2)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:import_done_count, 0)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:import_aborted_count, 1)
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:import_skipped_count, 0)
+
+        subject
+      end
+
+      context 'with load balancing enabled', :db_load_balancing do
+        it 'uses the replica' do
+          expect(Gitlab::Database::LoadBalancing::Session.current).to receive(:use_replicas_for_read_queries).and_call_original
+
+          subject
+        end
+      end
+    end
+  end
+end
diff --git a/spec/workers/delete_container_repository_worker_spec.rb b/spec/workers/delete_container_repository_worker_spec.rb
index b8363a2f81a..ec040eab2d4 100644
--- a/spec/workers/delete_container_repository_worker_spec.rb
+++ b/spec/workers/delete_container_repository_worker_spec.rb
@@ -5,7 +5,7 @@ require 'spec_helper'
 RSpec.describe DeleteContainerRepositoryWorker do
   let(:registry) { create(:container_repository) }
   let(:project) { registry.project }
-  let(:user) { project.owner }
+  let(:user) { project.first_owner }
 
   subject { described_class.new }
 
diff --git a/spec/workers/delete_merged_branches_worker_spec.rb b/spec/workers/delete_merged_branches_worker_spec.rb
index 861ca111b92..056fcb1200d 100644
--- a/spec/workers/delete_merged_branches_worker_spec.rb
+++ b/spec/workers/delete_merged_branches_worker_spec.rb
@@ -13,11 +13,11 @@ RSpec.describe DeleteMergedBranchesWorker do
         expect(instance).to receive(:execute).and_return(true)
       end
 
-      worker.perform(project.id, project.owner.id)
+      worker.perform(project.id, project.first_owner.id)
     end
 
     it "returns false when project was not found" do
-      expect(worker.perform('unknown', project.owner.id)).to be_falsy
+      expect(worker.perform('unknown', project.first_owner.id)).to be_falsy
     end
   end
 end
diff --git a/spec/workers/every_sidekiq_worker_spec.rb b/spec/workers/every_sidekiq_worker_spec.rb
index 4f9c207f976..1cd5d23d8fc 100644
--- a/spec/workers/every_sidekiq_worker_spec.rb
+++ b/spec/workers/every_sidekiq_worker_spec.rb
@@ -135,6 +135,7 @@ RSpec.describe 'Every Sidekiq worker' do
         'AutoDevops::DisableWorker' => 3,
         'AutoMergeProcessWorker' => 3,
         'BackgroundMigrationWorker' => 3,
+        'BackgroundMigration::CiDatabaseWorker' => 3,
         'BuildFinishedWorker' => 3,
         'BuildHooksWorker' => 3,
         'BuildQueueWorker' => 3,
@@ -348,6 +349,7 @@ RSpec.describe 'Every Sidekiq worker' do
         'Namespaces::OnboardingPipelineCreatedWorker' => 3,
         'Namespaces::OnboardingProgressWorker' => 3,
         'Namespaces::OnboardingUserAddedWorker' => 3,
+        'Namespaces::RefreshRootStatisticsWorker' => 3,
         'Namespaces::RootStatisticsWorker' => 3,
         'Namespaces::ScheduleAggregationWorker' => 3,
         'NetworkPolicyMetricsWorker' => 3,
@@ -371,7 +373,6 @@ RSpec.describe 'Every Sidekiq worker' do
         'PagesDomainSslRenewalWorker' => 3,
         'PagesDomainVerificationWorker' => 3,
         'PagesTransferWorker' => 3,
-        'PagesUpdateConfigurationWorker' => 1,
         'PagesWorker' => 3,
         'PersonalAccessTokens::Groups::PolicyWorker' => 3,
         'PersonalAccessTokens::Instance::PolicyWorker' => 3,
@@ -459,7 +460,8 @@ RSpec.describe 'Every Sidekiq worker' do
         'WebHooks::DestroyWorker' => 3,
         'WebHooks::LogExecutionWorker' => 3,
         'Wikis::GitGarbageCollectWorker' => false,
-        'X509CertificateRevokeWorker' => 3
+        'X509CertificateRevokeWorker' => 3,
+        'ComplianceManagement::MergeRequests::ComplianceViolationsWorker' => 3
       }
     end
 
diff --git a/spec/workers/groups/update_statistics_worker_spec.rb b/spec/workers/groups/update_statistics_worker_spec.rb
new file mode 100644
index 00000000000..7fc166ed300
--- /dev/null
+++ b/spec/workers/groups/update_statistics_worker_spec.rb
@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Groups::UpdateStatisticsWorker do
+  let_it_be(:group) { create(:group) }
+
+  let(:statistics) { %w(wiki_size) }
+
+  subject(:worker) { described_class.new }
+
+  describe '#perform' do
+    it 'updates the group statistics' do
+      expect(Groups::UpdateStatisticsService).to receive(:new)
+        .with(group, statistics: statistics)
+        .and_call_original
+
+      worker.perform(group.id, statistics)
+    end
+
+    context 'when group id does not exist' do
+      it 'ends gracefully' do
+        expect(Groups::UpdateStatisticsService).not_to receive(:new)
+
+        expect { worker.perform(non_existing_record_id, statistics) }.not_to raise_error
+      end
+    end
+  end
+end
diff --git a/spec/workers/loose_foreign_keys/cleanup_worker_spec.rb b/spec/workers/loose_foreign_keys/cleanup_worker_spec.rb
index 497f95cf34d..6f4389a7541 100644
--- a/spec/workers/loose_foreign_keys/cleanup_worker_spec.rb
+++ b/spec/workers/loose_foreign_keys/cleanup_worker_spec.rb
@@ -141,16 +141,6 @@ RSpec.describe LooseForeignKeys::CleanupWorker do
     end
   end
 
-  context 'when the loose_foreign_key_cleanup feature flag is off' do
-    before do
-      stub_feature_flags(loose_foreign_key_cleanup: false)
-    end
-
-    it 'does nothing' do
-      expect { described_class.new.perform }.not_to change { LooseForeignKeys::DeletedRecord.status_processed.count }
-    end
-  end
-
   describe 'multi-database support' do
     where(:current_minute, :configured_base_models, :expected_connection) do
       2 | { main: ApplicationRecord, ci: Ci::ApplicationRecord } | ApplicationRecord.connection
diff --git a/spec/workers/namespaces/process_sync_events_worker_spec.rb b/spec/workers/namespaces/process_sync_events_worker_spec.rb
index 59be1fffdb4..c15a74a2934 100644
--- a/spec/workers/namespaces/process_sync_events_worker_spec.rb
+++ b/spec/workers/namespaces/process_sync_events_worker_spec.rb
@@ -7,10 +7,12 @@ RSpec.describe Namespaces::ProcessSyncEventsWorker do
   let!(:group2) { create(:group) }
   let!(:group3) { create(:group) }
 
+  subject(:worker) { described_class.new }
+
   include_examples 'an idempotent worker'
 
   describe '#perform' do
-    subject(:perform) { described_class.new.perform }
+    subject(:perform) { worker.perform }
 
     before do
       group2.update!(parent: group1)
@@ -28,5 +30,13 @@ RSpec.describe Namespaces::ProcessSyncEventsWorker do
         an_object_having_attributes(namespace_id: group3.id, traversal_ids: [group1.id, group2.id, group3.id])
       )
     end
+
+    it 'logs the service result', :aggregate_failures do
+      expect(worker).to receive(:log_extra_metadata_on_done).with(:estimated_total_events, 5)
+      expect(worker).to receive(:log_extra_metadata_on_done).with(:consumable_events, 5)
+      expect(worker).to receive(:log_extra_metadata_on_done).with(:processed_events, 5)
+
+      perform
+    end
   end
 end
diff --git a/spec/workers/namespaces/update_root_statistics_worker_spec.rb b/spec/workers/namespaces/update_root_statistics_worker_spec.rb
new file mode 100644
index 00000000000..a525904b757
--- /dev/null
+++ b/spec/workers/namespaces/update_root_statistics_worker_spec.rb
@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Namespaces::UpdateRootStatisticsWorker do
+  let(:namespace_id) { 123 }
+
+  let(:event) do
+    Projects::ProjectDeletedEvent.new(data: { project_id: 1, namespace_id: namespace_id })
+  end
+
+  subject { consume_event(event) }
+
+  def consume_event(event)
+    described_class.new.perform(event.class.name, event.data)
+  end
+
+  it 'enqueues ScheduleAggregationWorker' do
+    expect(Namespaces::ScheduleAggregationWorker).to receive(:perform_async).with(namespace_id)
+
+    subject
+  end
+end
diff --git a/spec/workers/pages_update_configuration_worker_spec.rb b/spec/workers/pages_update_configuration_worker_spec.rb
deleted file mode 100644
index af71f6b3cca..00000000000
--- a/spec/workers/pages_update_configuration_worker_spec.rb
+++ /dev/null
@@ -1,12 +0,0 @@
-# frozen_string_literal: true
-require "spec_helper"
-
-RSpec.describe PagesUpdateConfigurationWorker do
-  let_it_be(:project) { create(:project) }
-
-  describe "#perform" do
-    it "does not break" do
-      expect { subject.perform(-1) }.not_to raise_error
-    end
-  end
-end
diff --git a/spec/workers/pipeline_schedule_worker_spec.rb b/spec/workers/pipeline_schedule_worker_spec.rb
index f59d8ad4615..4a7db0eca56 100644
--- a/spec/workers/pipeline_schedule_worker_spec.rb
+++ b/spec/workers/pipeline_schedule_worker_spec.rb
@@ -103,4 +103,14 @@ RSpec.describe PipelineScheduleWorker do
       expect { subject }.not_to raise_error
     end
   end
+
+  context 'when the project is missing' do
+    before do
+      project.delete
+    end
+
+    it 'does not raise an exception' do
+      expect { subject }.not_to raise_error
+    end
+  end
 end
diff --git a/spec/workers/post_receive_spec.rb b/spec/workers/post_receive_spec.rb
index 42e39c51a88..9b33e559c71 100644
--- a/spec/workers/post_receive_spec.rb
+++ b/spec/workers/post_receive_spec.rb
@@ -15,7 +15,7 @@ RSpec.describe PostReceive do
   let(:wrongly_encoded_changes) { changes.encode("ISO-8859-1").force_encoding("UTF-8") }
   let(:base64_changes) { Base64.encode64(wrongly_encoded_changes) }
   let(:gl_repository) { "project-#{project.id}" }
-  let(:key) { create(:key, user: project.owner) }
+  let(:key) { create(:key, user: project.first_owner) }
   let!(:key_id) { key.shell_id }
 
   let(:project) do
@@ -47,7 +47,7 @@ RSpec.describe PostReceive do
 
     context 'with PersonalSnippet' do
       let(:gl_repository) { "snippet-#{snippet.id}" }
-      let(:snippet) { create(:personal_snippet, author: project.owner) }
+      let(:snippet) { create(:personal_snippet, author: project.first_owner) }
 
       it 'does not log an error' do
         expect(Gitlab::GitLogger).not_to receive(:error)
@@ -60,7 +60,7 @@ RSpec.describe PostReceive do
 
     context 'with ProjectSnippet' do
       let(:gl_repository) { "snippet-#{snippet.id}" }
-      let(:snippet) { create(:snippet, type: 'ProjectSnippet', project: nil, author: project.owner) }
+      let(:snippet) { create(:snippet, type: 'ProjectSnippet', project: nil, author: project.first_owner) }
 
       it 'returns false and logs an error' do
         expect(Gitlab::GitLogger).to receive(:error).with("POST-RECEIVE: #{error_message}")
@@ -74,7 +74,7 @@ RSpec.describe PostReceive do
       let(:empty_project) { create(:project, :empty_repo) }
 
       before do
-        allow_next(Gitlab::GitPostReceive).to receive(:identify).and_return(empty_project.owner)
+        allow_next(Gitlab::GitPostReceive).to receive(:identify).and_return(empty_project.first_owner)
         # Need to mock here so we can expect calls on project
         allow(Gitlab::GlRepository).to receive(:parse).and_return([empty_project, empty_project, Gitlab::GlRepository::PROJECT])
       end
@@ -128,7 +128,7 @@ RSpec.describe PostReceive do
       let(:push_service) { double(execute: true) }
 
       before do
-        allow_next(Gitlab::GitPostReceive).to receive(:identify).and_return(project.owner)
+        allow_next(Gitlab::GitPostReceive).to receive(:identify).and_return(project.first_owner)
         allow(Gitlab::GlRepository).to receive(:parse).and_return([project, project, Gitlab::GlRepository::PROJECT])
       end
 
@@ -381,7 +381,7 @@ RSpec.describe PostReceive do
       allow(Project).to receive(:find_by).and_return(project)
       expect_next(MergeRequests::PushedBranchesService).to receive(:execute).and_return(%w(tést))
 
-      expect(UpdateMergeRequestsWorker).to receive(:perform_async).with(project.id, project.owner.id, any_args)
+      expect(UpdateMergeRequestsWorker).to receive(:perform_async).with(project.id, project.first_owner.id, any_args)
 
       perform
     end
@@ -461,13 +461,13 @@ RSpec.describe PostReceive do
     end
 
     context 'with PersonalSnippet' do
-      let!(:snippet) { create(:personal_snippet, :repository, author: project.owner) }
+      let!(:snippet) { create(:personal_snippet, :repository, author: project.first_owner) }
 
       it_behaves_like 'snippet changes actions'
     end
 
     context 'with ProjectSnippet' do
-      let!(:snippet) { create(:project_snippet, :repository, project: project, author: project.owner) }
+      let!(:snippet) { create(:project_snippet, :repository, project: project, author: project.first_owner) }
 
       it_behaves_like 'snippet changes actions'
     end
diff --git a/spec/workers/project_destroy_worker_spec.rb b/spec/workers/project_destroy_worker_spec.rb
index 00a4ddac29f..0b0543a5089 100644
--- a/spec/workers/project_destroy_worker_spec.rb
+++ b/spec/workers/project_destroy_worker_spec.rb
@@ -14,7 +14,7 @@ RSpec.describe ProjectDestroyWorker do
 
   describe '#perform' do
     it 'deletes the project' do
-      subject.perform(project.id, project.owner.id, {})
+      subject.perform(project.id, project.first_owner.id, {})
 
       expect(Project.all).not_to include(project)
       expect(Dir.exist?(path)).to be_falsey
@@ -22,7 +22,7 @@ RSpec.describe ProjectDestroyWorker do
 
     it 'does not raise error when project could not be found' do
       expect do
-        subject.perform(-1, project.owner.id, {})
+        subject.perform(-1, project.first_owner.id, {})
       end.not_to raise_error
     end
 
diff --git a/spec/workers/projects/git_garbage_collect_worker_spec.rb b/spec/workers/projects/git_garbage_collect_worker_spec.rb
index 7b54d7df4b2..ae567107443 100644
--- a/spec/workers/projects/git_garbage_collect_worker_spec.rb
+++ b/spec/workers/projects/git_garbage_collect_worker_spec.rb
@@ -32,6 +32,21 @@ RSpec.describe Projects::GitGarbageCollectWorker do
 
         subject.perform(*params)
       end
+
+      context 'when deduplication service runs into a GRPC internal error' do
+        before do
+          allow_next_instance_of(::Projects::GitDeduplicationService) do |instance|
+            expect(instance).to receive(:execute).and_raise(GRPC::Internal)
+          end
+        end
+
+        it_behaves_like 'can collect git garbage' do
+          let(:resource) { project }
+          let(:statistics_service_klass) { Projects::UpdateStatisticsService }
+          let(:statistics_keys) { [:repository_size, :lfs_objects_size] }
+          let(:expected_default_lease) { "projects:#{resource.id}" }
+        end
+      end
     end
 
     context 'LFS object garbage collection' do
diff --git a/spec/workers/projects/process_sync_events_worker_spec.rb b/spec/workers/projects/process_sync_events_worker_spec.rb
index 600fbbc6b20..963e0ad1028 100644
--- a/spec/workers/projects/process_sync_events_worker_spec.rb
+++ b/spec/workers/projects/process_sync_events_worker_spec.rb
@@ -6,10 +6,12 @@ RSpec.describe Projects::ProcessSyncEventsWorker do
   let!(:group) { create(:group) }
   let!(:project) { create(:project) }
 
+  subject(:worker) { described_class.new }
+
   include_examples 'an idempotent worker'
 
   describe '#perform' do
-    subject(:perform) { described_class.new.perform }
+    subject(:perform) { worker.perform }
 
     before do
       project.update!(namespace: group)
@@ -24,5 +26,13 @@ RSpec.describe Projects::ProcessSyncEventsWorker do
         an_object_having_attributes(namespace_id: group.id)
       )
     end
+
+    it 'logs the service result', :aggregate_failures do
+      expect(worker).to receive(:log_extra_metadata_on_done).with(:estimated_total_events, 2)
+      expect(worker).to receive(:log_extra_metadata_on_done).with(:consumable_events, 2)
+      expect(worker).to receive(:log_extra_metadata_on_done).with(:processed_events, 2)
+
+      perform
+    end
   end
 end
diff --git a/spec/workers/run_pipeline_schedule_worker_spec.rb b/spec/workers/run_pipeline_schedule_worker_spec.rb
index bb11d1dbb58..846b4455bf9 100644
--- a/spec/workers/run_pipeline_schedule_worker_spec.rb
+++ b/spec/workers/run_pipeline_schedule_worker_spec.rb
@@ -10,12 +10,25 @@ RSpec.describe RunPipelineScheduleWorker do
 
     let(:worker) { described_class.new }
 
-    context 'when a project not found' do
+    context 'when a schedule not found' do
       it 'does not call the Service' do
         expect(Ci::CreatePipelineService).not_to receive(:new)
         expect(worker).not_to receive(:run_pipeline_schedule)
 
-        worker.perform(100000, user.id)
+        worker.perform(non_existing_record_id, user.id)
+      end
+    end
+
+    context 'when a schedule project is missing' do
+      before do
+        project.delete
+      end
+
+      it 'does not call the Service' do
+        expect(Ci::CreatePipelineService).not_to receive(:new)
+        expect(worker).not_to receive(:run_pipeline_schedule)
+
+        worker.perform(pipeline_schedule.id, user.id)
       end
     end
 
@@ -24,7 +37,7 @@ RSpec.describe RunPipelineScheduleWorker do
         expect(Ci::CreatePipelineService).not_to receive(:new)
         expect(worker).not_to receive(:run_pipeline_schedule)
 
-        worker.perform(pipeline_schedule.id, 10000)
+        worker.perform(pipeline_schedule.id, non_existing_record_id)
       end
     end
 
diff --git a/spec/workers/web_hook_worker_spec.rb b/spec/workers/web_hook_worker_spec.rb
index bbb8844a447..dbdf7a2b978 100644
--- a/spec/workers/web_hook_worker_spec.rb
+++ b/spec/workers/web_hook_worker_spec.rb
@@ -19,7 +19,16 @@ RSpec.describe WebHookWorker do
       expect { subject.perform(non_existing_record_id, data, hook_name) }.not_to raise_error
     end
 
-    it 'retrieves recursion detection data, reinstates it, and cleans it from payload', :request_store, :aggregate_failures do
+    it 'retrieves recursion detection data and reinstates it', :request_store, :aggregate_failures do
+      uuid = SecureRandom.uuid
+      params = { recursion_detection_request_uuid: uuid }
+
+      expect_next(WebHookService, project_hook, data.with_indifferent_access, hook_name, anything).to receive(:execute)
+      expect { subject.perform(project_hook.id, data, hook_name, params) }
+        .to change { Gitlab::WebHooks::RecursionDetection::UUID.instance.request_uuid }.to(uuid)
+    end
+
+    it 'retrieves recursion detection data, reinstates it, and cleans it from payload when passed through as data', :request_store, :aggregate_failures do
       uuid = SecureRandom.uuid
       full_data = data.merge({ _gitlab_recursion_detection_request_uuid: uuid })