Add latest changes from gitlab-org/gitlab@14-6-stable-eev14.6.0-rc42

20 files changed, 376 insertions, 339 deletions

diff --git a/spec/workers/background_migration_worker_spec.rb b/spec/workers/background_migration_worker_spec.rb
index 7892eb89e80..4297e55ca6c 100644
--- a/spec/workers/background_migration_worker_spec.rb
+++ b/spec/workers/background_migration_worker_spec.rb
@@ -3,148 +3,5 @@
 require 'spec_helper'
 
 RSpec.describe BackgroundMigrationWorker, :clean_gitlab_redis_shared_state do
-  let(:worker) { described_class.new }
-
-  describe '.minimum_interval' do
-    it 'returns 2 minutes' do
-      expect(described_class.minimum_interval).to eq(2.minutes.to_i)
-    end
-  end
-
-  describe '#perform' do
-    before do
-      allow(worker).to receive(:jid).and_return(1)
-      allow(worker).to receive(:always_perform?).and_return(false)
-    end
-
-    it 'can run scheduled job and retried job concurrently' do
-      expect(Gitlab::BackgroundMigration)
-        .to receive(:perform)
-        .with('Foo', [10, 20])
-        .exactly(2).time
-
-      worker.perform('Foo', [10, 20])
-      worker.perform('Foo', [10, 20], described_class::MAX_LEASE_ATTEMPTS - 1)
-    end
-
-    context 'when lease can be obtained' do
-      before do
-        expect(Gitlab::BackgroundMigration)
-          .to receive(:perform)
-          .with('Foo', [10, 20])
-      end
-
-      it 'performs a background migration' do
-        worker.perform('Foo', [10, 20])
-      end
-
-      context 'when lease_attempts is 1' do
-        it 'performs a background migration' do
-          worker.perform('Foo', [10, 20], 1)
-        end
-      end
-    end
-
-    context 'when lease not obtained (migration of same class was performed recently)' do
-      before do
-        expect(Gitlab::BackgroundMigration).not_to receive(:perform)
-
-        worker.lease_for('Foo', false).try_obtain
-      end
-
-      it 'reschedules the migration and decrements the lease_attempts' do
-        expect(described_class)
-          .to receive(:perform_in)
-          .with(a_kind_of(Numeric), 'Foo', [10, 20], 4)
-
-        worker.perform('Foo', [10, 20], 5)
-      end
-
-      context 'when lease_attempts is 1' do
-        before do
-          worker.lease_for('Foo', true).try_obtain
-        end
-
-        it 'reschedules the migration and decrements the lease_attempts' do
-          expect(described_class)
-            .to receive(:perform_in)
-            .with(a_kind_of(Numeric), 'Foo', [10, 20], 0)
-
-          worker.perform('Foo', [10, 20], 1)
-        end
-      end
-
-      context 'when lease_attempts is 0' do
-        before do
-          worker.lease_for('Foo', true).try_obtain
-        end
-
-        it 'gives up performing the migration' do
-          expect(described_class).not_to receive(:perform_in)
-          expect(Sidekiq.logger).to receive(:warn).with(
-            class: 'Foo',
-            message: 'Job could not get an exclusive lease after several tries. Giving up.',
-            job_id: 1)
-
-          worker.perform('Foo', [10, 20], 0)
-        end
-      end
-    end
-
-    context 'when database is not healthy' do
-      before do
-        allow(worker).to receive(:healthy_database?).and_return(false)
-      end
-
-      it 'reschedules a migration if the database is not healthy' do
-        expect(described_class)
-          .to receive(:perform_in)
-          .with(a_kind_of(Numeric), 'Foo', [10, 20], 4)
-
-        worker.perform('Foo', [10, 20])
-      end
-
-      context 'when lease_attempts is 0' do
-        it 'gives up performing the migration' do
-          expect(described_class).not_to receive(:perform_in)
-          expect(Sidekiq.logger).to receive(:warn).with(
-            class: 'Foo',
-            message: 'Database was unhealthy after several tries. Giving up.',
-            job_id: 1)
-
-          worker.perform('Foo', [10, 20], 0)
-        end
-      end
-    end
-
-    it 'sets the class that will be executed as the caller_id' do
-      expect(Gitlab::BackgroundMigration).to receive(:perform) do
-        expect(Gitlab::ApplicationContext.current).to include('meta.caller_id' => 'Foo')
-      end
-
-      worker.perform('Foo', [10, 20])
-    end
-  end
-
-  describe '#healthy_database?' do
-    context 'when replication lag is too great' do
-      it 'returns false' do
-        allow(Postgresql::ReplicationSlot)
-          .to receive(:lag_too_great?)
-          .and_return(true)
-
-        expect(worker.healthy_database?).to eq(false)
-      end
-
-      context 'when replication lag is small enough' do
-        it 'returns true' do
-          allow(Postgresql::ReplicationSlot)
-            .to receive(:lag_too_great?)
-            .and_return(false)
-
-          expect(worker.healthy_database?).to eq(true)
-        end
-      end
-    end
-  end
+  it_behaves_like 'it runs background migration jobs', 'main', :background_migration_database_health_reschedules
 end
diff --git a/spec/workers/build_hooks_worker_spec.rb b/spec/workers/build_hooks_worker_spec.rb
index 5f7e7e5fb00..a69e188b441 100644
--- a/spec/workers/build_hooks_worker_spec.rb
+++ b/spec/workers/build_hooks_worker_spec.rb
@@ -23,14 +23,6 @@ RSpec.describe BuildHooksWorker do
     end
   end
 
-  describe '.perform_async' do
-    it 'delays scheduling a job by calling perform_in with default delay' do
-      expect(described_class).to receive(:perform_in).with(ApplicationWorker::DEFAULT_DELAY_INTERVAL.second, 123)
-
-      described_class.perform_async(123)
-    end
-  end
-
   it_behaves_like 'worker with data consistency',
                   described_class,
                   data_consistency: :delayed
diff --git a/spec/workers/bulk_imports/entity_worker_spec.rb b/spec/workers/bulk_imports/entity_worker_spec.rb
index deae15a3ca2..ce45299c7f7 100644
--- a/spec/workers/bulk_imports/entity_worker_spec.rb
+++ b/spec/workers/bulk_imports/entity_worker_spec.rb
@@ -14,96 +14,118 @@ RSpec.describe BulkImports::EntityWorker do
     )
   end
 
-  it 'enqueues the first stage pipelines work' do
-    expect_next_instance_of(Gitlab::Import::Logger) do |logger|
-      expect(logger)
-        .to receive(:info)
-        .with(
-          worker: described_class.name,
-          entity_id: entity.id,
-          current_stage: nil
-        )
-    end
+  let(:job_args) { entity.id }
 
-    expect(BulkImports::PipelineWorker)
-      .to receive(:perform_async)
-      .with(
-        pipeline_tracker.id,
-        pipeline_tracker.stage,
-        entity.id
-      )
+  it 'updates pipeline trackers to enqueued state when selected' do
+    worker = BulkImports::EntityWorker.new
 
-    subject.perform(entity.id)
-  end
+    next_tracker = worker.send(:next_pipeline_trackers_for, entity.id).first
 
-  it 'do not enqueue a new pipeline job if the current stage still running' do
-    expect(BulkImports::PipelineWorker)
-      .not_to receive(:perform_async)
+    next_tracker.reload
 
-    subject.perform(entity.id, 0)
-  end
-
-  it 'enqueues the next stage pipelines when the current stage is finished' do
-    next_stage_pipeline_tracker = create(
-      :bulk_import_tracker,
-      entity: entity,
-      pipeline_name: 'Stage1::Pipeline',
-      stage: 1
-    )
+    expect(next_tracker.enqueued?).to be_truthy
 
-    pipeline_tracker.fail_op!
+    expect(worker.send(:next_pipeline_trackers_for, entity.id))
+      .not_to include(next_tracker)
+  end
 
-    expect_next_instance_of(Gitlab::Import::Logger) do |logger|
-      expect(logger)
-        .to receive(:info)
+  include_examples 'an idempotent worker' do
+    it 'enqueues the first stage pipelines work' do
+      expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+        # the worker runs twice but only executes once
+        expect(logger)
+          .to receive(:info).twice
+          .with(
+            worker: described_class.name,
+            entity_id: entity.id,
+            current_stage: nil
+          )
+      end
+
+      expect(BulkImports::PipelineWorker)
+        .to receive(:perform_async)
         .with(
-          worker: described_class.name,
-          entity_id: entity.id,
-          current_stage: 0
+          pipeline_tracker.id,
+          pipeline_tracker.stage,
+          entity.id
         )
+
+      subject
     end
 
-    expect(BulkImports::PipelineWorker)
-      .to receive(:perform_async)
-      .with(
-        next_stage_pipeline_tracker.id,
-        next_stage_pipeline_tracker.stage,
-        entity.id
-      )
+    it 'logs and tracks the raised exceptions' do
+      exception = StandardError.new('Error!')
+
+      expect(BulkImports::PipelineWorker)
+        .to receive(:perform_async)
+              .and_raise(exception)
+
+      expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+        expect(logger)
+          .to receive(:info).twice
+          .with(
+            worker: described_class.name,
+            entity_id: entity.id,
+            current_stage: nil
+          )
+
+        expect(logger)
+          .to receive(:error)
+          .with(
+            worker: described_class.name,
+            entity_id: entity.id,
+            current_stage: nil,
+            error_message: 'Error!'
+          )
+      end
+
+      expect(Gitlab::ErrorTracking)
+        .to receive(:track_exception)
+              .with(exception, entity_id: entity.id)
+
+      subject
+    end
 
-    subject.perform(entity.id, 0)
-  end
+    context 'in first stage' do
+      let(:job_args) { [entity.id, 0] }
 
-  it 'logs and tracks the raised exceptions' do
-    exception = StandardError.new('Error!')
+      it 'do not enqueue a new pipeline job if the current stage still running' do
+        expect(BulkImports::PipelineWorker)
+          .not_to receive(:perform_async)
 
-    expect(BulkImports::PipelineWorker)
-      .to receive(:perform_async)
-      .and_raise(exception)
+        subject
+      end
 
-    expect_next_instance_of(Gitlab::Import::Logger) do |logger|
-      expect(logger)
-        .to receive(:info)
-        .with(
-          worker: described_class.name,
-          entity_id: entity.id,
-          current_stage: nil
+      it 'enqueues the next stage pipelines when the current stage is finished' do
+        next_stage_pipeline_tracker = create(
+          :bulk_import_tracker,
+          entity: entity,
+          pipeline_name: 'Stage1::Pipeline',
+          stage: 1
         )
 
-      expect(logger)
-        .to receive(:error)
-        .with(
-          worker: described_class.name,
-          entity_id: entity.id,
-          current_stage: nil,
-          error_message: 'Error!'
-        )
+        pipeline_tracker.fail_op!
+
+        expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+          expect(logger)
+            .to receive(:info).twice
+            .with(
+              worker: described_class.name,
+              entity_id: entity.id,
+              current_stage: 0
+            )
+        end
+
+        expect(BulkImports::PipelineWorker)
+          .to receive(:perform_async)
+            .with(
+              next_stage_pipeline_tracker.id,
+              next_stage_pipeline_tracker.stage,
+              entity.id
+            )
+
+        subject
+      end
     end
-
-    expect(Gitlab::ErrorTracking)
-      .to receive(:track_exception)
-      .with(exception, entity_id: entity.id)
-
-    subject.perform(entity.id)
   end
 end
diff --git a/spec/workers/bulk_imports/pipeline_worker_spec.rb b/spec/workers/bulk_imports/pipeline_worker_spec.rb
index c902d1f2034..2da9195a6ef 100644
--- a/spec/workers/bulk_imports/pipeline_worker_spec.rb
+++ b/spec/workers/bulk_imports/pipeline_worker_spec.rb
@@ -60,18 +60,8 @@ RSpec.describe BulkImports::PipelineWorker do
       create(
         :bulk_import_tracker,
         entity: entity,
-        pipeline_name: 'FakePipeline'
-      )
-    end
-  end
-
-  it_behaves_like 'successfully runs the pipeline' do
-    let(:pipeline_tracker) do
-      create(
-        :bulk_import_tracker,
-        :started,
-        entity: entity,
-        pipeline_name: 'FakePipeline'
+        pipeline_name: 'FakePipeline',
+        status_event: 'enqueue'
       )
     end
   end
@@ -109,7 +99,8 @@ RSpec.describe BulkImports::PipelineWorker do
       pipeline_tracker = create(
         :bulk_import_tracker,
         entity: entity,
-        pipeline_name: 'InexistentPipeline'
+        pipeline_name: 'InexistentPipeline',
+        status_event: 'enqueue'
       )
 
       expect_next_instance_of(Gitlab::Import::Logger) do |logger|
@@ -150,7 +141,8 @@ RSpec.describe BulkImports::PipelineWorker do
         pipeline_tracker = create(
           :bulk_import_tracker,
           entity: entity,
-          pipeline_name: 'FakePipeline'
+          pipeline_name: 'FakePipeline',
+          status_event: 'enqueue'
         )
 
         exception = BulkImports::NetworkError.new(
@@ -163,7 +155,21 @@ RSpec.describe BulkImports::PipelineWorker do
             .and_raise(exception)
         end
 
-        expect(subject).to receive(:jid).and_return('jid')
+        expect(subject).to receive(:jid).and_return('jid').twice
+
+        expect_any_instance_of(BulkImports::Tracker) do |tracker|
+          expect(tracker).to receive(:retry).and_call_original
+        end
+
+        expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+          expect(logger)
+            .to receive(:info)
+            .with(
+              worker: described_class.name,
+              pipeline_name: 'FakePipeline',
+              entity_id: entity.id
+            )
+        end
 
         expect(described_class)
           .to receive(:perform_in)
@@ -175,6 +181,10 @@ RSpec.describe BulkImports::PipelineWorker do
           )
 
         subject.perform(pipeline_tracker.id, pipeline_tracker.stage, entity.id)
+
+        pipeline_tracker.reload
+
+        expect(pipeline_tracker.enqueued?).to be_truthy
       end
     end
   end
@@ -200,7 +210,8 @@ RSpec.describe BulkImports::PipelineWorker do
       create(
         :bulk_import_tracker,
         entity: entity,
-        pipeline_name: 'NdjsonPipeline'
+        pipeline_name: 'NdjsonPipeline',
+        status_event: 'enqueue'
       )
     end
 
diff --git a/spec/workers/ci/pending_builds/update_group_worker_spec.rb b/spec/workers/ci/pending_builds/update_group_worker_spec.rb
new file mode 100644
index 00000000000..8c6bf018158
--- /dev/null
+++ b/spec/workers/ci/pending_builds/update_group_worker_spec.rb
@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::PendingBuilds::UpdateGroupWorker do
+  describe '#perform' do
+    let(:worker) { described_class.new }
+
+    context 'when a group is not provided' do
+      it 'does not call the service' do
+        expect(::Ci::UpdatePendingBuildService).not_to receive(:new)
+      end
+    end
+
+    context 'when everything is ok' do
+      let(:group) { create(:group) }
+      let(:update_pending_build_service) { instance_double(::Ci::UpdatePendingBuildService) }
+      let(:update_params) { { "namespace_id" => group.id } }
+
+      it 'calls the service' do
+        expect(::Ci::UpdatePendingBuildService).to receive(:new).with(group, update_params).and_return(update_pending_build_service)
+        expect(update_pending_build_service).to receive(:execute)
+
+        worker.perform(group.id, update_params)
+      end
+
+      include_examples 'an idempotent worker' do
+        let(:pending_build) { create(:ci_pending_build) }
+        let(:update_params) { { "namespace_id" => pending_build.namespace_id } }
+        let(:job_args) { [pending_build.namespace_id, update_params] }
+
+        it 'updates the pending builds' do
+          subject
+
+          expect(pending_build.reload.namespace_id).to eq(update_params["namespace_id"])
+        end
+      end
+    end
+  end
+end
diff --git a/spec/workers/ci/pending_builds/update_project_worker_spec.rb b/spec/workers/ci/pending_builds/update_project_worker_spec.rb
new file mode 100644
index 00000000000..4a67127564e
--- /dev/null
+++ b/spec/workers/ci/pending_builds/update_project_worker_spec.rb
@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::PendingBuilds::UpdateProjectWorker do
+  describe '#perform' do
+    let(:worker) { described_class.new }
+
+    context 'when a project is not provided' do
+      it 'does not call the service' do
+        expect(::Ci::UpdatePendingBuildService).not_to receive(:new)
+      end
+    end
+
+    context 'when everything is ok' do
+      let(:project) { create(:project) }
+      let(:group) { create(:group) }
+      let(:update_pending_build_service) { instance_double(::Ci::UpdatePendingBuildService) }
+      let(:update_params) { { "namespace_id" => group.id } }
+
+      it 'calls the service' do
+        expect(::Ci::UpdatePendingBuildService).to receive(:new).with(project, update_params).and_return(update_pending_build_service)
+        expect(update_pending_build_service).to receive(:execute)
+
+        worker.perform(project.id, update_params)
+      end
+
+      include_examples 'an idempotent worker' do
+        let(:pending_build) { create(:ci_pending_build) }
+        let(:job_args) { [pending_build.project_id, update_params] }
+
+        it 'updates the pending builds' do
+          subject
+
+          expect(pending_build.reload.namespace_id).to eq(update_params["namespace_id"])
+        end
+      end
+    end
+  end
+end
diff --git a/spec/workers/concerns/application_worker_spec.rb b/spec/workers/concerns/application_worker_spec.rb
index fbf39b3c7cd..7608b5f49a1 100644
--- a/spec/workers/concerns/application_worker_spec.rb
+++ b/spec/workers/concerns/application_worker_spec.rb
@@ -248,39 +248,40 @@ RSpec.describe ApplicationWorker do
   end
 
   describe '.perform_async' do
-    before do
-      stub_const(worker.name, worker)
-    end
-
-    shared_examples_for 'worker utilizes load balancing capabilities' do |data_consistency|
-      before do
-        worker.data_consistency(data_consistency)
-      end
-
-      it 'call perform_in' do
-        expect(worker).to receive(:perform_in).with(described_class::DEFAULT_DELAY_INTERVAL.seconds, 123)
+    using RSpec::Parameterized::TableSyntax
 
-        worker.perform_async(123)
-      end
+    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
 
-    context 'when workers data consistency is :sticky' do
-      it_behaves_like 'worker utilizes load balancing capabilities', :sticky
-    end
+    before do
+      stub_const(worker.name, worker)
+      worker.data_consistency(data_consistency)
 
-    context 'when workers data consistency is :delayed' do
-      it_behaves_like 'worker utilizes load balancing capabilities', :delayed
+      allow(Gitlab::Database::LoadBalancing).to receive(:primary_only?).and_return(primary_only?)
+      stub_feature_flags(skip_scheduling_workers_for_replicas: skip_scheduling_ff)
     end
 
-    context 'when workers data consistency is :always' do
-      before do
-        worker.data_consistency(:always)
-      end
-
-      it 'does not call perform_in' do
-        expect(worker).not_to receive(:perform_in)
+    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
+        worker.perform_async(123)
       end
     end
   end
diff --git a/spec/workers/create_commit_signature_worker_spec.rb b/spec/workers/create_commit_signature_worker_spec.rb
index d283ff5b732..0e31faf47af 100644
--- a/spec/workers/create_commit_signature_worker_spec.rb
+++ b/spec/workers/create_commit_signature_worker_spec.rb
@@ -143,7 +143,7 @@ RSpec.describe CreateCommitSignatureWorker do
       let(:type) { :X509 }
 
       it 'performs a single query for commit signatures' do
-        expect(X509CommitSignature).to receive(:by_commit_sha).with(commit_shas).once.and_return([])
+        expect(CommitSignatures::X509CommitSignature).to receive(:by_commit_sha).with(commit_shas).once.and_return([])
 
         subject
       end
@@ -153,7 +153,7 @@ RSpec.describe CreateCommitSignatureWorker do
       let(:type) { :PGP }
 
       it 'performs a single query for commit signatures' do
-        expect(GpgSignature).to receive(:by_commit_sha).with(commit_shas).once.and_return([])
+        expect(CommitSignatures::GpgSignature).to receive(:by_commit_sha).with(commit_shas).once.and_return([])
 
         subject
       end
diff --git a/spec/workers/every_sidekiq_worker_spec.rb b/spec/workers/every_sidekiq_worker_spec.rb
index d00243672f9..00b6d2635a5 100644
--- a/spec/workers/every_sidekiq_worker_spec.rb
+++ b/spec/workers/every_sidekiq_worker_spec.rb
@@ -398,7 +398,6 @@ RSpec.describe 'Every Sidekiq worker' do
         'PropagateIntegrationInheritWorker' => 3,
         'PropagateIntegrationProjectWorker' => 3,
         'PropagateIntegrationWorker' => 3,
-        'PropagateServiceTemplateWorker' => 3,
         'PurgeDependencyProxyCacheWorker' => 3,
         'ReactiveCachingWorker' => 3,
         'RebaseWorker' => 3,
diff --git a/spec/workers/issuable_export_csv_worker_spec.rb b/spec/workers/issuable_export_csv_worker_spec.rb
index bcc2420996d..a18d10ad3df 100644
--- a/spec/workers/issuable_export_csv_worker_spec.rb
+++ b/spec/workers/issuable_export_csv_worker_spec.rb
@@ -35,10 +35,15 @@ RSpec.describe IssuableExportCsvWorker do
     end
 
     context 'with params' do
-      let(:params) { { 'test_key' => true } }
+      let(:params) { { 'test_key' => true, 'not' => { 'label_name' => ['SomeLabel'] } } }
 
-      it 'converts controller string keys to symbol keys for IssuesFinder' do
-        expect(IssuesFinder).to receive(:new).with(user, hash_including(test_key: true)).and_call_original
+      it 'allows symbol access for IssuesFinder' do
+        expect(IssuesFinder).to receive(:new).and_wrap_original do |method, user, params|
+          expect(params[:test_key]).to eq(true)
+          expect(params[:not][:label_name]).to eq(['SomeLabel'])
+
+          method.call(user, params)
+        end
 
         subject
       end
diff --git a/spec/workers/issue_placement_worker_spec.rb b/spec/workers/issue_placement_worker_spec.rb
index 50b9d58a5b0..9b5121d98e8 100644
--- a/spec/workers/issue_placement_worker_spec.rb
+++ b/spec/workers/issue_placement_worker_spec.rb
@@ -35,7 +35,7 @@ RSpec.describe IssuePlacementWorker do
       it 'schedules rebalancing if needed' do
         issue_a.update!(relative_position: RelativePositioning::MAX_POSITION)
 
-        expect(IssueRebalancingWorker).to receive(:perform_async).with(nil, nil, project.group.id)
+        expect(Issues::RebalancingWorker).to receive(:perform_async).with(nil, nil, project.group.id)
 
         run_worker
       end
@@ -52,7 +52,7 @@ RSpec.describe IssuePlacementWorker do
                              .with(have_attributes(count: described_class::QUERY_LIMIT))
                              .and_call_original
 
-          expect(described_class).to receive(:perform_async).with(nil, project.id)
+          expect(Issues::PlacementWorker).to receive(:perform_async).with(nil, project.id)
 
           run_worker
 
@@ -101,7 +101,7 @@ RSpec.describe IssuePlacementWorker do
       it 'anticipates the failure to place the issues, and schedules rebalancing' do
         allow(Issue).to receive(:move_nulls_to_end) { raise RelativePositioning::NoSpaceLeft }
 
-        expect(IssueRebalancingWorker).to receive(:perform_async).with(nil, nil, project.group.id)
+        expect(Issues::RebalancingWorker).to receive(:perform_async).with(nil, nil, project.group.id)
         expect(Gitlab::ErrorTracking)
           .to receive(:log_exception)
           .with(RelativePositioning::NoSpaceLeft, worker_arguments)
diff --git a/spec/workers/issues/placement_worker_spec.rb b/spec/workers/issues/placement_worker_spec.rb
index 694cdd2ef37..33fa0b31b72 100644
--- a/spec/workers/issues/placement_worker_spec.rb
+++ b/spec/workers/issues/placement_worker_spec.rb
@@ -35,7 +35,7 @@ RSpec.describe Issues::PlacementWorker do
       it 'schedules rebalancing if needed' do
         issue_a.update!(relative_position: RelativePositioning::MAX_POSITION)
 
-        expect(IssueRebalancingWorker).to receive(:perform_async).with(nil, nil, project.group.id)
+        expect(Issues::RebalancingWorker).to receive(:perform_async).with(nil, nil, project.group.id)
 
         run_worker
       end
diff --git a/spec/workers/issues/rebalancing_worker_spec.rb b/spec/workers/issues/rebalancing_worker_spec.rb
index 438edd85f66..e1c0b348a4f 100644
--- a/spec/workers/issues/rebalancing_worker_spec.rb
+++ b/spec/workers/issues/rebalancing_worker_spec.rb
@@ -35,6 +35,20 @@ RSpec.describe Issues::RebalancingWorker do
 
         described_class.new.perform # all arguments are nil
       end
+
+      it 'does not schedule a new rebalance if it finished under 1h ago' do
+        container_type = arguments.second.present? ? ::Gitlab::Issues::Rebalancing::State::PROJECT : ::Gitlab::Issues::Rebalancing::State::NAMESPACE
+        container_id = arguments.second || arguments.third
+
+        Gitlab::Redis::SharedState.with do |redis|
+          redis.set(::Gitlab::Issues::Rebalancing::State.send(:recently_finished_key, container_type, container_id), true)
+        end
+
+        expect(Issues::RelativePositionRebalancingService).not_to receive(:new)
+        expect(Gitlab::ErrorTracking).not_to receive(:log_exception)
+
+        described_class.new.perform(*arguments)
+      end
     end
 
     shared_examples 'safely handles non-existent ids' do
diff --git a/spec/workers/issues/reschedule_stuck_issue_rebalances_worker_spec.rb b/spec/workers/issues/reschedule_stuck_issue_rebalances_worker_spec.rb
index 02d1241d2ba..6723c425f34 100644
--- a/spec/workers/issues/reschedule_stuck_issue_rebalances_worker_spec.rb
+++ b/spec/workers/issues/reschedule_stuck_issue_rebalances_worker_spec.rb
@@ -10,15 +10,15 @@ RSpec.describe Issues::RescheduleStuckIssueRebalancesWorker, :clean_gitlab_redis
 
   describe '#perform' do
     it 'does not schedule a rebalance' do
-      expect(IssueRebalancingWorker).not_to receive(:perform_async)
+      expect(Issues::RebalancingWorker).not_to receive(:perform_async)
 
       worker.perform
     end
 
     it 'schedules a rebalance in case there are any rebalances started' do
       expect(::Gitlab::Issues::Rebalancing::State).to receive(:fetch_rebalancing_groups_and_projects).and_return([[group.id], [project.id]])
-      expect(IssueRebalancingWorker).to receive(:bulk_perform_async).with([[nil, nil, group.id]]).once
-      expect(IssueRebalancingWorker).to receive(:bulk_perform_async).with([[nil, project.id, nil]]).once
+      expect(Issues::RebalancingWorker).to receive(:bulk_perform_async).with([[nil, nil, group.id]]).once
+      expect(Issues::RebalancingWorker).to receive(:bulk_perform_async).with([[nil, project.id, nil]]).once
 
       worker.perform
     end
diff --git a/spec/workers/loose_foreign_keys/cleanup_worker_spec.rb b/spec/workers/loose_foreign_keys/cleanup_worker_spec.rb
index 544be2a69a6..3c628d036ff 100644
--- a/spec/workers/loose_foreign_keys/cleanup_worker_spec.rb
+++ b/spec/workers/loose_foreign_keys/cleanup_worker_spec.rb
@@ -27,43 +27,40 @@ RSpec.describe LooseForeignKeys::CleanupWorker do
     migration.track_record_deletions(:_test_loose_fk_parent_table_2)
   end
 
-  let!(:parent_model_1) do
-    Class.new(ApplicationRecord) do
-      self.table_name = '_test_loose_fk_parent_table_1'
-
-      include LooseForeignKey
-
-      loose_foreign_key :_test_loose_fk_child_table_1_1, :parent_id, on_delete: :async_delete
-      loose_foreign_key :_test_loose_fk_child_table_1_2, :parent_id_with_different_column, on_delete: :async_nullify
-    end
-  end
-
-  let!(:parent_model_2) do
-    Class.new(ApplicationRecord) do
-      self.table_name = '_test_loose_fk_parent_table_2'
-
-      include LooseForeignKey
-
-      loose_foreign_key :_test_loose_fk_child_table_2_1, :parent_id, on_delete: :async_delete
-    end
-  end
-
-  let!(:child_model_1) do
-    Class.new(ApplicationRecord) do
-      self.table_name = '_test_loose_fk_child_table_1_1'
-    end
-  end
-
-  let!(:child_model_2) do
-    Class.new(ApplicationRecord) do
-      self.table_name = '_test_loose_fk_child_table_1_2'
-    end
-  end
-
-  let!(:child_model_3) do
-    Class.new(ApplicationRecord) do
-      self.table_name = '_test_loose_fk_child_table_2_1'
-    end
+  let(:all_loose_foreign_key_definitions) do
+    {
+      '_test_loose_fk_parent_table_1' => [
+        ActiveRecord::ConnectionAdapters::ForeignKeyDefinition.new(
+          '_test_loose_fk_child_table_1_1',
+          '_test_loose_fk_parent_table_1',
+          {
+            column: 'parent_id',
+            on_delete: :async_delete,
+            gitlab_schema: :gitlab_main
+          }
+        ),
+        ActiveRecord::ConnectionAdapters::ForeignKeyDefinition.new(
+          '_test_loose_fk_child_table_1_2',
+          '_test_loose_fk_parent_table_1',
+          {
+            column: 'parent_id_with_different_column',
+            on_delete: :async_nullify,
+            gitlab_schema: :gitlab_main
+          }
+        )
+      ],
+      '_test_loose_fk_parent_table_2' => [
+        ActiveRecord::ConnectionAdapters::ForeignKeyDefinition.new(
+          '_test_loose_fk_child_table_2_1',
+          '_test_loose_fk_parent_table_2',
+          {
+            column: 'parent_id',
+            on_delete: :async_delete,
+            gitlab_schema: :gitlab_main
+          }
+        )
+      ]
+    }
   end
 
   let(:loose_fk_parent_table_1) { table(:_test_loose_fk_parent_table_1) }
@@ -87,6 +84,8 @@ RSpec.describe LooseForeignKeys::CleanupWorker do
   end
 
   before do
+    allow(Gitlab::Database::LooseForeignKeys).to receive(:definitions_by_table).and_return(all_loose_foreign_key_definitions)
+
     parent_record_1 = loose_fk_parent_table_1.create!
     loose_fk_child_table_1_1.create!(parent_id: parent_record_1.id)
     loose_fk_child_table_1_2.create!(parent_id_with_different_column: parent_record_1.id)
@@ -98,8 +97,8 @@ RSpec.describe LooseForeignKeys::CleanupWorker do
     parent_record_3 = loose_fk_parent_table_2.create!
     5.times { loose_fk_child_table_2_1.create!(parent_id: parent_record_3.id) }
 
-    parent_model_1.delete_all
-    parent_model_2.delete_all
+    loose_fk_parent_table_1.delete_all
+    loose_fk_parent_table_2.delete_all
   end
 
   it 'cleans up all rows' do
diff --git a/spec/workers/namespaces/process_sync_events_worker_spec.rb b/spec/workers/namespaces/process_sync_events_worker_spec.rb
new file mode 100644
index 00000000000..59be1fffdb4
--- /dev/null
+++ b/spec/workers/namespaces/process_sync_events_worker_spec.rb
@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Namespaces::ProcessSyncEventsWorker do
+  let!(:group1) { create(:group) }
+  let!(:group2) { create(:group) }
+  let!(:group3) { create(:group) }
+
+  include_examples 'an idempotent worker'
+
+  describe '#perform' do
+    subject(:perform) { described_class.new.perform }
+
+    before do
+      group2.update!(parent: group1)
+      group3.update!(parent: group2)
+    end
+
+    it 'consumes all sync events' do
+      expect { perform }.to change(Namespaces::SyncEvent, :count).from(5).to(0)
+    end
+
+    it 'syncs namespace hierarchy traversal ids' do
+      expect { perform }.to change(Ci::NamespaceMirror, :all).to contain_exactly(
+        an_object_having_attributes(namespace_id: group1.id, traversal_ids: [group1.id]),
+        an_object_having_attributes(namespace_id: group2.id, traversal_ids: [group1.id, group2.id]),
+        an_object_having_attributes(namespace_id: group3.id, traversal_ids: [group1.id, group2.id, group3.id])
+      )
+    end
+  end
+end
diff --git a/spec/workers/projects/process_sync_events_worker_spec.rb b/spec/workers/projects/process_sync_events_worker_spec.rb
new file mode 100644
index 00000000000..600fbbc6b20
--- /dev/null
+++ b/spec/workers/projects/process_sync_events_worker_spec.rb
@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Projects::ProcessSyncEventsWorker do
+  let!(:group) { create(:group) }
+  let!(:project) { create(:project) }
+
+  include_examples 'an idempotent worker'
+
+  describe '#perform' do
+    subject(:perform) { described_class.new.perform }
+
+    before do
+      project.update!(namespace: group)
+    end
+
+    it 'consumes all sync events' do
+      expect { perform }.to change(Projects::SyncEvent, :count).from(2).to(0)
+    end
+
+    it 'syncs project namespace id' do
+      expect { perform }.to change(Ci::ProjectMirror, :all).to contain_exactly(
+        an_object_having_attributes(namespace_id: group.id)
+      )
+    end
+  end
+end
diff --git a/spec/workers/propagate_integration_worker_spec.rb b/spec/workers/propagate_integration_worker_spec.rb
index 902e3206d35..030caefb833 100644
--- a/spec/workers/propagate_integration_worker_spec.rb
+++ b/spec/workers/propagate_integration_worker_spec.rb
@@ -18,7 +18,7 @@ RSpec.describe PropagateIntegrationWorker do
     end
 
     it 'calls the propagate service with the integration' do
-      expect(Admin::PropagateIntegrationService).to receive(:propagate).with(integration)
+      expect(Integrations::PropagateService).to receive(:propagate).with(integration)
 
       subject.perform(integration.id)
     end
diff --git a/spec/workers/purge_dependency_proxy_cache_worker_spec.rb b/spec/workers/purge_dependency_proxy_cache_worker_spec.rb
index 393745958be..b928104fb58 100644
--- a/spec/workers/purge_dependency_proxy_cache_worker_spec.rb
+++ b/spec/workers/purge_dependency_proxy_cache_worker_spec.rb
@@ -4,18 +4,18 @@ require 'spec_helper'
 
 RSpec.describe PurgeDependencyProxyCacheWorker do
   let_it_be(:user) { create(:admin) }
-  let_it_be(:blob) { create(:dependency_proxy_blob )}
-  let_it_be(:group, reload: true) { blob.group }
-  let_it_be(:manifest) { create(:dependency_proxy_manifest, group: group )}
+  let_it_be_with_refind(:blob) { create(:dependency_proxy_blob )}
+  let_it_be_with_reload(:group) { blob.group }
+  let_it_be_with_refind(:manifest) { create(:dependency_proxy_manifest, group: group )}
   let_it_be(:group_id) { group.id }
 
   subject { described_class.new.perform(user.id, group_id) }
 
   describe '#perform' do
-    shared_examples 'not removing blobs and manifests' do
-      it 'does not remove blobs and manifests', :aggregate_failures do
-        expect { subject }.not_to change { group.dependency_proxy_blobs.size }
-        expect { subject }.not_to change { group.dependency_proxy_manifests.size }
+    shared_examples 'not expiring blobs and manifests' do
+      it 'does not expire blobs and manifests', :aggregate_failures do
+        expect { subject }.not_to change { blob.status }
+        expect { subject }.not_to change { manifest.status }
         expect(subject).to be_nil
       end
     end
@@ -25,39 +25,36 @@ RSpec.describe PurgeDependencyProxyCacheWorker do
         include_examples 'an idempotent worker' do
           let(:job_args) { [user.id, group_id] }
 
-          it 'deletes the blobs and returns ok', :aggregate_failures do
-            expect(group.dependency_proxy_blobs.size).to eq(1)
-            expect(group.dependency_proxy_manifests.size).to eq(1)
-
+          it 'expires the blobs and returns ok', :aggregate_failures do
             subject
 
-            expect(group.dependency_proxy_blobs.size).to eq(0)
-            expect(group.dependency_proxy_manifests.size).to eq(0)
+            expect(blob).to be_expired
+            expect(manifest).to be_expired
           end
         end
       end
 
       context 'when admin mode is disabled' do
-        it_behaves_like 'not removing blobs and manifests'
+        it_behaves_like 'not expiring blobs and manifests'
       end
     end
 
     context 'a non-admin user' do
       let(:user) { create(:user) }
 
-      it_behaves_like 'not removing blobs and manifests'
+      it_behaves_like 'not expiring blobs and manifests'
     end
 
     context 'an invalid user id' do
       let(:user) { double('User', id: 99999 ) }
 
-      it_behaves_like 'not removing blobs and manifests'
+      it_behaves_like 'not expiring blobs and manifests'
     end
 
     context 'an invalid group' do
       let(:group_id) { 99999 }
 
-      it_behaves_like 'not removing blobs and manifests'
+      it_behaves_like 'not expiring blobs and manifests'
     end
   end
 end
diff --git a/spec/workers/todos_destroyer/private_features_worker_spec.rb b/spec/workers/todos_destroyer/private_features_worker_spec.rb
index f346a004670..88d9be051d0 100644
--- a/spec/workers/todos_destroyer/private_features_worker_spec.rb
+++ b/spec/workers/todos_destroyer/private_features_worker_spec.rb
@@ -6,7 +6,7 @@ RSpec.describe TodosDestroyer::PrivateFeaturesWorker do
   it "calls the Todos::Destroy::PrivateFeaturesService with the params it was given" do
     service = double
 
-    expect(::Todos::Destroy::PrivateFeaturesService).to receive(:new).with(100, nil).and_return(service)
+    expect(::Todos::Destroy::UnauthorizedFeaturesService).to receive(:new).with(100, nil).and_return(service)
     expect(service).to receive(:execute)
 
     described_class.new.perform(100)