From 311b0269b4eb9839fa63f80c8d7a58f32b8138a0 Mon Sep 17 00:00:00 2001
From: GitLab Bot <gitlab-bot@gitlab.com>
Date: Thu, 18 Nov 2021 13:16:36 +0000
Subject: Add latest changes from gitlab-org/gitlab@14-5-stable-ee

---
 .../usage_trends/counter_job_worker_spec.rb        |   3 +-
 .../ci/ref_delete_unlock_artifacts_worker_spec.rb  |  34 +-
 ...ign_resource_from_resource_group_worker_spec.rb |   4 +
 .../check_prometheus_health_worker_spec.rb         |  19 -
 .../check_prometheus_health_worker_spec.rb         |  19 +
 spec/workers/concerns/application_worker_spec.rb   | 381 ++++++++++++++++++---
 .../cleanup_container_repository_worker_spec.rb    |   5 +-
 .../drop_detached_partitions_worker_spec.rb        |   8 +-
 .../database/partition_management_worker_spec.rb   |   9 +-
 .../image_ttl_group_policy_worker_spec.rb          |   4 +-
 .../deployments/archive_in_project_worker_spec.rb  |  18 +
 spec/workers/email_receiver_worker_spec.rb         |   9 +
 spec/workers/emails_on_push_worker_spec.rb         |  37 ++
 spec/workers/every_sidekiq_worker_spec.rb          |   2 +
 .../create_external_cross_reference_worker_spec.rb | 128 +++++++
 spec/workers/issue_rebalancing_worker_spec.rb      |  16 +-
 spec/workers/issues/placement_worker_spec.rb       | 151 ++++++++
 spec/workers/issues/rebalancing_worker_spec.rb     |  90 +++++
 ...eschedule_stuck_issue_rebalances_worker_spec.rb |  26 ++
 .../loose_foreign_keys/cleanup_worker_spec.rb      | 153 +++++++++
 .../namespaces/invite_team_email_worker_spec.rb    |  27 ++
 .../packages/maven/metadata/sync_worker_spec.rb    |   3 +-
 spec/workers/post_receive_spec.rb                  |   8 -
 .../propagate_integration_group_worker_spec.rb     |   2 +-
 ...e_integration_inherit_descendant_worker_spec.rb |   4 +-
 .../propagate_integration_project_worker_spec.rb   |   2 +-
 .../ssh_keys/expired_notification_worker_spec.rb   |   6 +-
 .../workers/tasks_to_be_done/create_worker_spec.rb |  36 ++
 .../users/deactivate_dormant_users_worker_spec.rb  |  36 +-
 29 files changed, 1136 insertions(+), 104 deletions(-)
 delete mode 100644 spec/workers/clusters/applications/check_prometheus_health_worker_spec.rb
 create mode 100644 spec/workers/clusters/integrations/check_prometheus_health_worker_spec.rb
 create mode 100644 spec/workers/deployments/archive_in_project_worker_spec.rb
 create mode 100644 spec/workers/integrations/create_external_cross_reference_worker_spec.rb
 create mode 100644 spec/workers/issues/placement_worker_spec.rb
 create mode 100644 spec/workers/issues/rebalancing_worker_spec.rb
 create mode 100644 spec/workers/issues/reschedule_stuck_issue_rebalances_worker_spec.rb
 create mode 100644 spec/workers/loose_foreign_keys/cleanup_worker_spec.rb
 create mode 100644 spec/workers/namespaces/invite_team_email_worker_spec.rb
 create mode 100644 spec/workers/tasks_to_be_done/create_worker_spec.rb

(limited to 'spec/workers')

diff --git a/spec/workers/analytics/usage_trends/counter_job_worker_spec.rb b/spec/workers/analytics/usage_trends/counter_job_worker_spec.rb
index dd180229d12..c45ec20fe5a 100644
--- a/spec/workers/analytics/usage_trends/counter_job_worker_spec.rb
+++ b/spec/workers/analytics/usage_trends/counter_job_worker_spec.rb
@@ -11,7 +11,8 @@ RSpec.describe Analytics::UsageTrends::CounterJobWorker do
   let(:job_args) { [users_measurement_identifier, user_1.id, user_2.id, recorded_at] }
 
   before do
-    allow(::Analytics::UsageTrends::Measurement.connection).to receive(:transaction_open?).and_return(false)
+    allow(::ApplicationRecord.connection).to receive(:transaction_open?).and_return(false)
+    allow(::Ci::ApplicationRecord.connection).to receive(:transaction_open?).and_return(false) if ::Ci::ApplicationRecord.connection_class?
   end
 
   include_examples 'an idempotent worker' do
diff --git a/spec/workers/ci/ref_delete_unlock_artifacts_worker_spec.rb b/spec/workers/ci/ref_delete_unlock_artifacts_worker_spec.rb
index f510852e753..fe4bc2421a4 100644
--- a/spec/workers/ci/ref_delete_unlock_artifacts_worker_spec.rb
+++ b/spec/workers/ci/ref_delete_unlock_artifacts_worker_spec.rb
@@ -4,7 +4,9 @@ require 'spec_helper'
 
 RSpec.describe Ci::RefDeleteUnlockArtifactsWorker do
   describe '#perform' do
-    subject(:perform) { described_class.new.perform(project_id, user_id, ref) }
+    subject(:perform) { worker.perform(project_id, user_id, ref) }
+
+    let(:worker) { described_class.new }
 
     let(:ref) { 'refs/heads/master' }
 
@@ -40,6 +42,36 @@ RSpec.describe Ci::RefDeleteUnlockArtifactsWorker do
 
             expect(service).to have_received(:execute).with(ci_ref)
           end
+
+          context 'when a locked pipeline with persisted artifacts exists' do
+            let!(:pipeline) { create(:ci_pipeline, :with_persisted_artifacts, ref: 'master', project: project, locked: :artifacts_locked) }
+
+            context 'with ci_update_unlocked_job_artifacts disabled' do
+              before do
+                stub_feature_flags(ci_update_unlocked_job_artifacts: false)
+              end
+
+              it 'logs the correct extra metadata' do
+                expect(worker).to receive(:log_extra_metadata_on_done).with(:unlocked_pipelines, 1)
+                expect(worker).to receive(:log_extra_metadata_on_done).with(:unlocked_job_artifacts, 0)
+
+                perform
+              end
+            end
+
+            context 'with ci_update_unlocked_job_artifacts enabled' do
+              before do
+                stub_feature_flags(ci_update_unlocked_job_artifacts: true)
+              end
+
+              it 'logs the correct extra metadata' do
+                expect(worker).to receive(:log_extra_metadata_on_done).with(:unlocked_pipelines, 1)
+                expect(worker).to receive(:log_extra_metadata_on_done).with(:unlocked_job_artifacts, 2)
+
+                perform
+              end
+            end
+          end
         end
 
         context 'when ci ref does not exist for the given project' do
diff --git a/spec/workers/ci/resource_groups/assign_resource_from_resource_group_worker_spec.rb b/spec/workers/ci/resource_groups/assign_resource_from_resource_group_worker_spec.rb
index 650be1e84a9..be7f7ef5c8c 100644
--- a/spec/workers/ci/resource_groups/assign_resource_from_resource_group_worker_spec.rb
+++ b/spec/workers/ci/resource_groups/assign_resource_from_resource_group_worker_spec.rb
@@ -9,6 +9,10 @@ RSpec.describe Ci::ResourceGroups::AssignResourceFromResourceGroupWorker do
     expect(described_class.get_deduplicate_strategy).to eq(:until_executed)
   end
 
+  it 'has an option to reschedule once if deduplicated' do
+    expect(described_class.get_deduplication_options).to include({ if_deduplicated: :reschedule_once })
+  end
+
   describe '#perform' do
     subject { worker.perform(resource_group_id) }
 
diff --git a/spec/workers/clusters/applications/check_prometheus_health_worker_spec.rb b/spec/workers/clusters/applications/check_prometheus_health_worker_spec.rb
deleted file mode 100644
index fb779bf3b01..00000000000
--- a/spec/workers/clusters/applications/check_prometheus_health_worker_spec.rb
+++ /dev/null
@@ -1,19 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe Clusters::Applications::CheckPrometheusHealthWorker, '#perform' do
-  subject { described_class.new.perform }
-
-  it 'triggers health service' do
-    cluster = create(:cluster)
-    allow(Gitlab::Monitor::DemoProjects).to receive(:primary_keys)
-    allow(Clusters::Cluster).to receive_message_chain(:with_application_prometheus, :with_project_http_integrations).and_return([cluster])
-
-    service_instance = instance_double(Clusters::Applications::PrometheusHealthCheckService)
-    expect(Clusters::Applications::PrometheusHealthCheckService).to receive(:new).with(cluster).and_return(service_instance)
-    expect(service_instance).to receive(:execute)
-
-    subject
-  end
-end
diff --git a/spec/workers/clusters/integrations/check_prometheus_health_worker_spec.rb b/spec/workers/clusters/integrations/check_prometheus_health_worker_spec.rb
new file mode 100644
index 00000000000..6f70870bd09
--- /dev/null
+++ b/spec/workers/clusters/integrations/check_prometheus_health_worker_spec.rb
@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Clusters::Integrations::CheckPrometheusHealthWorker, '#perform' do
+  subject { described_class.new.perform }
+
+  it 'triggers health service' do
+    cluster = create(:cluster)
+    allow(Gitlab::Monitor::DemoProjects).to receive(:primary_keys)
+    allow(Clusters::Cluster).to receive_message_chain(:with_integration_prometheus, :with_project_http_integrations).and_return([cluster])
+
+    service_instance = instance_double(Clusters::Integrations::PrometheusHealthCheckService)
+    expect(Clusters::Integrations::PrometheusHealthCheckService).to receive(:new).with(cluster).and_return(service_instance)
+    expect(service_instance).to receive(:execute)
+
+    subject
+  end
+end
diff --git a/spec/workers/concerns/application_worker_spec.rb b/spec/workers/concerns/application_worker_spec.rb
index af038c81b9e..fbf39b3c7cd 100644
--- a/spec/workers/concerns/application_worker_spec.rb
+++ b/spec/workers/concerns/application_worker_spec.rb
@@ -285,48 +285,38 @@ RSpec.describe ApplicationWorker do
     end
   end
 
-  describe '.bulk_perform_async' do
-    before do
-      stub_const(worker.name, worker)
+  context 'different kinds of push_bulk' do
+    shared_context 'disable the `sidekiq_push_bulk_in_batches` feature flag' do
+      before do
+        stub_feature_flags(sidekiq_push_bulk_in_batches: false)
+      end
     end
 
-    it 'enqueues jobs in bulk' do
-      Sidekiq::Testing.fake! do
-        worker.bulk_perform_async([['Foo', [1]], ['Foo', [2]]])
-
-        expect(worker.jobs.count).to eq 2
-        expect(worker.jobs).to all(include('enqueued_at'))
+    shared_context 'set safe limit beyond the number of jobs to be enqueued' do
+      before do
+        stub_const("#{described_class}::SAFE_PUSH_BULK_LIMIT", args.count + 1)
       end
     end
-  end
 
-  describe '.bulk_perform_in' do
-    before do
-      stub_const(worker.name, worker)
+    shared_context 'set safe limit below the number of jobs to be enqueued' do
+      before do
+        stub_const("#{described_class}::SAFE_PUSH_BULK_LIMIT", 2)
+      end
     end
 
-    context 'when delay is valid' do
-      it 'correctly schedules jobs' do
-        Sidekiq::Testing.fake! do
-          worker.bulk_perform_in(1.minute, [['Foo', [1]], ['Foo', [2]]])
+    shared_examples_for 'returns job_id of all enqueued jobs' do
+      let(:job_id_regex) { /[0-9a-f]{12}/ }
 
-          expect(worker.jobs.count).to eq 2
-          expect(worker.jobs).to all(include('at'))
-        end
-      end
-    end
+      it 'returns job_id of all enqueued jobs' do
+        job_ids = perform_action
 
-    context 'when delay is invalid' do
-      it 'raises an ArgumentError exception' do
-        expect { worker.bulk_perform_in(-60, [['Foo']]) }
-          .to raise_error(ArgumentError)
+        expect(job_ids.count).to eq(args.count)
+        expect(job_ids).to all(match(job_id_regex))
       end
     end
 
-    context 'with batches' do
-      let(:batch_delay) { 1.minute }
-
-      it 'correctly schedules jobs' do
+    shared_examples_for 'enqueues the jobs in a batched fashion, with each batch enqueing jobs as per the set safe limit' do
+      it 'enqueues the jobs in a batched fashion, with each batch enqueing jobs as per the set safe limit' do
         expect(Sidekiq::Client).to(
           receive(:push_bulk).with(hash_including('args' => [['Foo', [1]], ['Foo', [2]]]))
                              .ordered
@@ -337,29 +327,318 @@ RSpec.describe ApplicationWorker do
                              .and_call_original)
         expect(Sidekiq::Client).to(
           receive(:push_bulk).with(hash_including('args' => [['Foo', [5]]]))
-                             .ordered
-                             .and_call_original)
+                            .ordered
+                            .and_call_original)
 
-        worker.bulk_perform_in(
-          1.minute,
-          [['Foo', [1]], ['Foo', [2]], ['Foo', [3]], ['Foo', [4]], ['Foo', [5]]],
-          batch_size: 2, batch_delay: batch_delay)
-
-        expect(worker.jobs.count).to eq 5
-        expect(worker.jobs[0]['at']).to eq(worker.jobs[1]['at'])
-        expect(worker.jobs[2]['at']).to eq(worker.jobs[3]['at'])
-        expect(worker.jobs[2]['at'] - worker.jobs[1]['at']).to eq(batch_delay)
-        expect(worker.jobs[4]['at'] - worker.jobs[3]['at']).to eq(batch_delay)
-      end
-
-      context 'when batch_size is invalid' do
-        it 'raises an ArgumentError exception' do
-          expect do
-            worker.bulk_perform_in(1.minute,
-                                   [['Foo']],
-                                   batch_size: -1, batch_delay: batch_delay)
-          end.to raise_error(ArgumentError)
+        perform_action
+
+        expect(worker.jobs.count).to eq args.count
+        expect(worker.jobs).to all(include('enqueued_at'))
+      end
+    end
+
+    shared_examples_for 'enqueues jobs in one go' do
+      it 'enqueues jobs in one go' do
+        expect(Sidekiq::Client).to(
+          receive(:push_bulk).with(hash_including('args' => args)).once.and_call_original)
+        expect(Sidekiq.logger).not_to receive(:info)
+
+        perform_action
+
+        expect(worker.jobs.count).to eq args.count
+        expect(worker.jobs).to all(include('enqueued_at'))
+      end
+    end
+
+    shared_examples_for 'logs bulk insertions' do
+      it 'logs arguments and job IDs' do
+        worker.log_bulk_perform_async!
+
+        expect(Sidekiq.logger).to(
+          receive(:info).with(hash_including('class' => worker.name, 'args_list' => args)).once.and_call_original)
+        expect(Sidekiq.logger).to(
+          receive(:info).with(hash_including('class' => worker.name, 'jid_list' => anything)).once.and_call_original)
+
+        perform_action
+      end
+    end
+
+    before do
+      stub_const(worker.name, worker)
+    end
+
+    let(:args) do
+      [
+        ['Foo', [1]],
+        ['Foo', [2]],
+        ['Foo', [3]],
+        ['Foo', [4]],
+        ['Foo', [5]]
+      ]
+    end
+
+    describe '.bulk_perform_async' do
+      shared_examples_for 'does not schedule the jobs for any specific time' do
+        it 'does not schedule the jobs for any specific time' do
+          perform_action
+
+          expect(worker.jobs).to all(exclude('at'))
+        end
+      end
+
+      subject(:perform_action) do
+        worker.bulk_perform_async(args)
+      end
+
+      context 'push_bulk in safe limit batches' do
+        context 'when the number of jobs to be enqueued does not exceed the safe limit' do
+          include_context 'set safe limit beyond the number of jobs to be enqueued'
+
+          it_behaves_like 'enqueues jobs in one go'
+          it_behaves_like 'logs bulk insertions'
+          it_behaves_like 'returns job_id of all enqueued jobs'
+          it_behaves_like 'does not schedule the jobs for any specific time'
         end
+
+        context 'when the number of jobs to be enqueued exceeds safe limit' do
+          include_context 'set safe limit below the number of jobs to be enqueued'
+
+          it_behaves_like 'enqueues the jobs in a batched fashion, with each batch enqueing jobs as per the set safe limit'
+          it_behaves_like 'returns job_id of all enqueued jobs'
+          it_behaves_like 'does not schedule the jobs for any specific time'
+        end
+
+        context 'when the feature flag `sidekiq_push_bulk_in_batches` is disabled' do
+          include_context 'disable the `sidekiq_push_bulk_in_batches` feature flag'
+
+          context 'when the number of jobs to be enqueued does not exceed the safe limit' do
+            include_context 'set safe limit beyond the number of jobs to be enqueued'
+
+            it_behaves_like 'enqueues jobs in one go'
+            it_behaves_like 'logs bulk insertions'
+            it_behaves_like 'returns job_id of all enqueued jobs'
+            it_behaves_like 'does not schedule the jobs for any specific time'
+          end
+
+          context 'when the number of jobs to be enqueued exceeds safe limit' do
+            include_context 'set safe limit below the number of jobs to be enqueued'
+
+            it_behaves_like 'enqueues jobs in one go'
+            it_behaves_like 'returns job_id of all enqueued jobs'
+            it_behaves_like 'does not schedule the jobs for any specific time'
+          end
+        end
+      end
+    end
+
+    describe '.bulk_perform_in' do
+      context 'without batches' do
+        shared_examples_for 'schedules all the jobs at a specific time' do
+          it 'schedules all the jobs at a specific time' do
+            perform_action
+
+            worker.jobs.each do |job_detail|
+              expect(job_detail['at']).to be_within(3.seconds).of(expected_scheduled_at_time)
+            end
+          end
+        end
+
+        let(:delay) { 3.minutes }
+        let(:expected_scheduled_at_time) { Time.current.to_i + delay.to_i }
+
+        subject(:perform_action) do
+          worker.bulk_perform_in(delay, args)
+        end
+
+        context 'when the scheduled time falls in the past' do
+          let(:delay) { -60 }
+
+          it 'raises an ArgumentError exception' do
+            expect { perform_action }
+              .to raise_error(ArgumentError)
+          end
+        end
+
+        context 'push_bulk in safe limit batches' do
+          context 'when the number of jobs to be enqueued does not exceed the safe limit' do
+            include_context 'set safe limit beyond the number of jobs to be enqueued'
+
+            it_behaves_like 'enqueues jobs in one go'
+            it_behaves_like 'returns job_id of all enqueued jobs'
+            it_behaves_like 'schedules all the jobs at a specific time'
+          end
+
+          context 'when the number of jobs to be enqueued exceeds safe limit' do
+            include_context 'set safe limit below the number of jobs to be enqueued'
+
+            it_behaves_like 'enqueues the jobs in a batched fashion, with each batch enqueing jobs as per the set safe limit'
+            it_behaves_like 'returns job_id of all enqueued jobs'
+            it_behaves_like 'schedules all the jobs at a specific time'
+          end
+
+          context 'when the feature flag `sidekiq_push_bulk_in_batches` is disabled' do
+            include_context 'disable the `sidekiq_push_bulk_in_batches` feature flag'
+
+            context 'when the number of jobs to be enqueued does not exceed the safe limit' do
+              include_context 'set safe limit beyond the number of jobs to be enqueued'
+
+              it_behaves_like 'enqueues jobs in one go'
+              it_behaves_like 'returns job_id of all enqueued jobs'
+              it_behaves_like 'schedules all the jobs at a specific time'
+            end
+
+            context 'when the number of jobs to be enqueued exceeds safe limit' do
+              include_context 'set safe limit below the number of jobs to be enqueued'
+
+              it_behaves_like 'enqueues jobs in one go'
+              it_behaves_like 'returns job_id of all enqueued jobs'
+              it_behaves_like 'schedules all the jobs at a specific time'
+            end
+          end
+        end
+      end
+
+      context 'with batches' do
+        shared_examples_for 'schedules all the jobs at a specific time, per batch' do
+          it 'schedules all the jobs at a specific time, per batch' do
+            perform_action
+
+            expect(worker.jobs[0]['at']).to eq(worker.jobs[1]['at'])
+            expect(worker.jobs[2]['at']).to eq(worker.jobs[3]['at'])
+            expect(worker.jobs[2]['at'] - worker.jobs[1]['at']).to eq(batch_delay)
+            expect(worker.jobs[4]['at'] - worker.jobs[3]['at']).to eq(batch_delay)
+          end
+        end
+
+        let(:delay) { 1.minute }
+        let(:batch_size) { 2 }
+        let(:batch_delay) { 10.minutes }
+
+        subject(:perform_action) do
+          worker.bulk_perform_in(delay, args, batch_size: batch_size, batch_delay: batch_delay)
+        end
+
+        context 'when the `batch_size` is invalid' do
+          context 'when `batch_size` is 0' do
+            let(:batch_size) { 0 }
+
+            it 'raises an ArgumentError exception' do
+              expect { perform_action }
+                .to raise_error(ArgumentError)
+            end
+          end
+
+          context 'when `batch_size` is negative' do
+            let(:batch_size) { -3 }
+
+            it 'raises an ArgumentError exception' do
+              expect { perform_action }
+                .to raise_error(ArgumentError)
+            end
+          end
+        end
+
+        context 'when the `batch_delay` is invalid' do
+          context 'when `batch_delay` is 0' do
+            let(:batch_delay) { 0.minutes }
+
+            it 'raises an ArgumentError exception' do
+              expect { perform_action }
+                .to raise_error(ArgumentError)
+            end
+          end
+
+          context 'when `batch_delay` is negative' do
+            let(:batch_delay) { -3.minutes }
+
+            it 'raises an ArgumentError exception' do
+              expect { perform_action }
+                .to raise_error(ArgumentError)
+            end
+          end
+        end
+
+        context 'push_bulk in safe limit batches' do
+          context 'when the number of jobs to be enqueued does not exceed the safe limit' do
+            include_context 'set safe limit beyond the number of jobs to be enqueued'
+
+            it_behaves_like 'enqueues jobs in one go'
+            it_behaves_like 'returns job_id of all enqueued jobs'
+            it_behaves_like 'schedules all the jobs at a specific time, per batch'
+          end
+
+          context 'when the number of jobs to be enqueued exceeds safe limit' do
+            include_context 'set safe limit below the number of jobs to be enqueued'
+
+            it_behaves_like 'enqueues the jobs in a batched fashion, with each batch enqueing jobs as per the set safe limit'
+            it_behaves_like 'returns job_id of all enqueued jobs'
+            it_behaves_like 'schedules all the jobs at a specific time, per batch'
+          end
+
+          context 'when the feature flag `sidekiq_push_bulk_in_batches` is disabled' do
+            include_context 'disable the `sidekiq_push_bulk_in_batches` feature flag'
+
+            context 'when the number of jobs to be enqueued does not exceed the safe limit' do
+              include_context 'set safe limit beyond the number of jobs to be enqueued'
+
+              it_behaves_like 'enqueues jobs in one go'
+              it_behaves_like 'returns job_id of all enqueued jobs'
+              it_behaves_like 'schedules all the jobs at a specific time, per batch'
+            end
+
+            context 'when the number of jobs to be enqueued exceeds safe limit' do
+              include_context 'set safe limit below the number of jobs to be enqueued'
+
+              it_behaves_like 'enqueues jobs in one go'
+              it_behaves_like 'returns job_id of all enqueued jobs'
+              it_behaves_like 'schedules all the jobs at a specific time, per batch'
+            end
+          end
+        end
+      end
+    end
+  end
+
+  describe '.with_status' do
+    around do |example|
+      Sidekiq::Testing.fake!(&example)
+    end
+
+    context 'when the worker does have status_expiration set' do
+      let(:status_expiration_worker) do
+        Class.new(worker) do
+          sidekiq_options status_expiration: 3
+        end
+      end
+
+      it 'uses status_expiration from the worker' do
+        status_expiration_worker.with_status.perform_async
+
+        expect(Sidekiq::Queues[status_expiration_worker.queue].first).to include('status_expiration' => 3)
+        expect(Sidekiq::Queues[status_expiration_worker.queue].length).to eq(1)
+      end
+
+      it 'uses status_expiration from the worker without with_status' do
+        status_expiration_worker.perform_async
+
+        expect(Sidekiq::Queues[status_expiration_worker.queue].first).to include('status_expiration' => 3)
+        expect(Sidekiq::Queues[status_expiration_worker.queue].length).to eq(1)
+      end
+    end
+
+    context 'when the worker does not have status_expiration set' do
+      it 'uses the default status_expiration' do
+        worker.with_status.perform_async
+
+        expect(Sidekiq::Queues[worker.queue].first).to include('status_expiration' => Gitlab::SidekiqStatus::DEFAULT_EXPIRATION)
+        expect(Sidekiq::Queues[worker.queue].length).to eq(1)
+      end
+
+      it 'does not set status_expiration without with_status' do
+        worker.perform_async
+
+        expect(Sidekiq::Queues[worker.queue].first).not_to include('status_expiration')
+        expect(Sidekiq::Queues[worker.queue].length).to eq(1)
       end
     end
   end
diff --git a/spec/workers/container_expiration_policies/cleanup_container_repository_worker_spec.rb b/spec/workers/container_expiration_policies/cleanup_container_repository_worker_spec.rb
index d4126fe688a..cbffb8f3870 100644
--- a/spec/workers/container_expiration_policies/cleanup_container_repository_worker_spec.rb
+++ b/spec/workers/container_expiration_policies/cleanup_container_repository_worker_spec.rb
@@ -82,8 +82,9 @@ RSpec.describe ContainerExpirationPolicies::CleanupContainerRepositoryWorker do
             nil | 10  | nil
             0   | 5   | nil
             10  | 0   | 0
-            10  | 5   | 0.5
-            3   | 10  | (10 / 3.to_f)
+            10  | 5   | 50.0
+            17  | 3   | 17.65
+            3   | 10  | 333.33
           end
 
           with_them do
diff --git a/spec/workers/database/drop_detached_partitions_worker_spec.rb b/spec/workers/database/drop_detached_partitions_worker_spec.rb
index 8693878ddd5..a10fcaaa5d9 100644
--- a/spec/workers/database/drop_detached_partitions_worker_spec.rb
+++ b/spec/workers/database/drop_detached_partitions_worker_spec.rb
@@ -6,21 +6,19 @@ RSpec.describe Database::DropDetachedPartitionsWorker do
   describe '#perform' do
     subject { described_class.new.perform }
 
-    let(:monitoring) { instance_double('PartitionMonitoring', report_metrics: nil) }
-
     before do
       allow(Gitlab::Database::Partitioning).to receive(:drop_detached_partitions)
-      allow(Gitlab::Database::Partitioning::PartitionMonitoring).to receive(:new).and_return(monitoring)
+      allow(Gitlab::Database::Partitioning).to receive(:report_metrics)
     end
 
-    it 'delegates to Partitioning.drop_detached_partitions' do
+    it 'drops detached partitions' do
       expect(Gitlab::Database::Partitioning).to receive(:drop_detached_partitions)
 
       subject
     end
 
     it 'reports partition metrics' do
-      expect(monitoring).to receive(:report_metrics)
+      expect(Gitlab::Database::Partitioning).to receive(:report_metrics)
 
       subject
     end
diff --git a/spec/workers/database/partition_management_worker_spec.rb b/spec/workers/database/partition_management_worker_spec.rb
index 9ded36743a8..e5362e95f48 100644
--- a/spec/workers/database/partition_management_worker_spec.rb
+++ b/spec/workers/database/partition_management_worker_spec.rb
@@ -6,20 +6,19 @@ RSpec.describe Database::PartitionManagementWorker do
   describe '#perform' do
     subject { described_class.new.perform }
 
-    let(:monitoring) { instance_double('PartitionMonitoring', report_metrics: nil) }
-
     before do
-      allow(Gitlab::Database::Partitioning::PartitionMonitoring).to receive(:new).and_return(monitoring)
+      allow(Gitlab::Database::Partitioning).to receive(:sync_partitions)
+      allow(Gitlab::Database::Partitioning).to receive(:report_metrics)
     end
 
-    it 'delegates to Partitioning' do
+    it 'syncs partitions' do
       expect(Gitlab::Database::Partitioning).to receive(:sync_partitions)
 
       subject
     end
 
     it 'reports partition metrics' do
-      expect(monitoring).to receive(:report_metrics)
+      expect(Gitlab::Database::Partitioning).to receive(:report_metrics)
 
       subject
     end
diff --git a/spec/workers/dependency_proxy/image_ttl_group_policy_worker_spec.rb b/spec/workers/dependency_proxy/image_ttl_group_policy_worker_spec.rb
index d3234f4c212..ae0cb097ebf 100644
--- a/spec/workers/dependency_proxy/image_ttl_group_policy_worker_spec.rb
+++ b/spec/workers/dependency_proxy/image_ttl_group_policy_worker_spec.rb
@@ -12,8 +12,8 @@ RSpec.describe DependencyProxy::ImageTtlGroupPolicyWorker do
     subject { worker.perform }
 
     context 'when there are images to expire' do
-      let_it_be_with_reload(:old_blob) { create(:dependency_proxy_blob, group: group, updated_at: 1.year.ago) }
-      let_it_be_with_reload(:old_manifest) { create(:dependency_proxy_manifest, group: group, updated_at: 1.year.ago) }
+      let_it_be_with_reload(:old_blob) { create(:dependency_proxy_blob, group: group, read_at: 1.year.ago) }
+      let_it_be_with_reload(:old_manifest) { create(:dependency_proxy_manifest, group: group, read_at: 1.year.ago) }
       let_it_be_with_reload(:new_blob) { create(:dependency_proxy_blob, group: group) }
       let_it_be_with_reload(:new_manifest) { create(:dependency_proxy_manifest, group: group) }
 
diff --git a/spec/workers/deployments/archive_in_project_worker_spec.rb b/spec/workers/deployments/archive_in_project_worker_spec.rb
new file mode 100644
index 00000000000..6435fe8bea1
--- /dev/null
+++ b/spec/workers/deployments/archive_in_project_worker_spec.rb
@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Deployments::ArchiveInProjectWorker do
+  subject { described_class.new.perform(deployment&.project_id) }
+
+  describe '#perform' do
+    let(:deployment) { create(:deployment, :success) }
+
+    it 'executes Deployments::ArchiveInProjectService' do
+      expect(Deployments::ArchiveInProjectService)
+          .to receive(:new).with(deployment.project, nil).and_call_original
+
+      subject
+    end
+  end
+end
diff --git a/spec/workers/email_receiver_worker_spec.rb b/spec/workers/email_receiver_worker_spec.rb
index 83e13ded7b3..83720ee132b 100644
--- a/spec/workers/email_receiver_worker_spec.rb
+++ b/spec/workers/email_receiver_worker_spec.rb
@@ -37,6 +37,15 @@ RSpec.describe EmailReceiverWorker, :mailer do
           expect(email.to).to eq(["jake@adventuretime.ooo"])
           expect(email.subject).to include("Rejected")
         end
+
+        it 'strips out the body before passing to EmailRejectionMailer' do
+          mail = Mail.new(raw_message)
+          mail.body = nil
+
+          expect(EmailRejectionMailer).to receive(:rejection).with(anything, mail.encoded, anything).and_call_original
+
+          described_class.new.perform(raw_message)
+        end
       end
 
       context 'when the error is Gitlab::Email::AutoGeneratedEmailError' do
diff --git a/spec/workers/emails_on_push_worker_spec.rb b/spec/workers/emails_on_push_worker_spec.rb
index 6c37c422aed..3e313610054 100644
--- a/spec/workers/emails_on_push_worker_spec.rb
+++ b/spec/workers/emails_on_push_worker_spec.rb
@@ -139,6 +139,43 @@ RSpec.describe EmailsOnPushWorker, :mailer do
 
           perform
         end
+
+        context 'when SMIME signing is enabled' do
+          include SmimeHelper
+
+          before :context do
+            @root_ca = generate_root
+            @cert = generate_cert(signer_ca: @root_ca)
+          end
+
+          let(:root_certificate) do
+            Gitlab::X509::Certificate.new(@root_ca[:key], @root_ca[:cert])
+          end
+
+          let(:certificate) do
+            Gitlab::X509::Certificate.new(@cert[:key], @cert[:cert])
+          end
+
+          before do
+            allow(Gitlab::Email::Hook::SmimeSignatureInterceptor).to receive(:certificate).and_return(certificate)
+
+            Mail.register_interceptor(Gitlab::Email::Hook::SmimeSignatureInterceptor)
+          end
+
+          after do
+            Mail.unregister_interceptor(Gitlab::Email::Hook::SmimeSignatureInterceptor)
+          end
+
+          it 'does not sign the email multiple times' do
+            perform
+
+            ActionMailer::Base.deliveries.each do |mail|
+              expect(mail.header['Content-Type'].value).to match('multipart/signed').and match('protocol="application/x-pkcs7-signature"')
+
+              expect(mail.to_s.scan(/Content-Disposition: attachment;\r\n filename=smime.p7s/).size).to eq(1)
+            end
+          end
+        end
       end
 
       context "when recipients are invalid" do
diff --git a/spec/workers/every_sidekiq_worker_spec.rb b/spec/workers/every_sidekiq_worker_spec.rb
index 9a4b27997e9..d00243672f9 100644
--- a/spec/workers/every_sidekiq_worker_spec.rb
+++ b/spec/workers/every_sidekiq_worker_spec.rb
@@ -316,6 +316,8 @@ RSpec.describe 'Every Sidekiq worker' do
         'IssuableExportCsvWorker' => 3,
         'IssuePlacementWorker' => 3,
         'IssueRebalancingWorker' => 3,
+        'Issues::PlacementWorker' => 3,
+        'Issues::RebalancingWorker' => 3,
         'IterationsUpdateStatusWorker' => 3,
         'JiraConnect::SyncBranchWorker' => 3,
         'JiraConnect::SyncBuildsWorker' => 3,
diff --git a/spec/workers/integrations/create_external_cross_reference_worker_spec.rb b/spec/workers/integrations/create_external_cross_reference_worker_spec.rb
new file mode 100644
index 00000000000..61723f44aa5
--- /dev/null
+++ b/spec/workers/integrations/create_external_cross_reference_worker_spec.rb
@@ -0,0 +1,128 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Integrations::CreateExternalCrossReferenceWorker do
+  include AfterNextHelpers
+  using RSpec::Parameterized::TableSyntax
+
+  let_it_be(:project) { create(:jira_project, :repository) }
+  let_it_be(:author) { create(:user) }
+  let_it_be(:commit) { project.commit }
+  let_it_be(:issue) { create(:issue, project: project) }
+  let_it_be(:merge_request) { create(:merge_request, source_project: project, target_project: project) }
+  let_it_be(:note) { create(:note, project: project) }
+  let_it_be(:snippet) { create(:project_snippet, project: project) }
+
+  let(:project_id) { project.id }
+  let(:external_issue_id) { 'JIRA-123' }
+  let(:mentionable_type) { 'Issue' }
+  let(:mentionable_id) { issue.id }
+  let(:author_id) { author.id }
+  let(:job_args) { [project_id, external_issue_id, mentionable_type, mentionable_id, author_id] }
+
+  def perform
+    described_class.new.perform(*job_args)
+  end
+
+  before do
+    allow(Project).to receive(:find_by_id).and_return(project)
+  end
+
+  it_behaves_like 'an idempotent worker' do
+    before do
+      allow(project.external_issue_tracker).to receive(:create_cross_reference_note)
+    end
+
+    it 'can run multiple times with the same arguments' do
+      subject
+
+      expect(project.external_issue_tracker).to have_received(:create_cross_reference_note)
+        .exactly(worker_exec_times).times
+    end
+  end
+
+  it 'has the `until_executed` deduplicate strategy' do
+    expect(described_class.get_deduplicate_strategy).to eq(:until_executed)
+    expect(described_class.get_deduplication_options).to include({ including_scheduled: true })
+  end
+
+  # These are the only models where we currently support cross-references,
+  # although this should be expanded to all `Mentionable` models.
+  # See https://gitlab.com/gitlab-org/gitlab/-/issues/343975
+  where(:mentionable_type, :mentionable_id) do
+    'Commit'       | lazy { commit.id }
+    'Issue'        | lazy { issue.id }
+    'MergeRequest' | lazy { merge_request.id }
+    'Note'         | lazy { note.id }
+    'Snippet'      | lazy { snippet.id }
+  end
+
+  with_them do
+    it 'creates a cross reference' do
+      expect(project.external_issue_tracker).to receive(:create_cross_reference_note).with(
+        be_a(ExternalIssue).and(have_attributes(id: external_issue_id, project: project)),
+        be_a(mentionable_type.constantize).and(have_attributes(id: mentionable_id)),
+        be_a(User).and(have_attributes(id: author_id))
+      )
+
+      perform
+    end
+  end
+
+  describe 'error handling' do
+    shared_examples 'does not create a cross reference' do
+      it 'does not create a cross reference' do
+        expect(project).not_to receive(:external_issue_tracker) if project
+
+        perform
+      end
+    end
+
+    context 'project_id does not exist' do
+      let(:project_id) { non_existing_record_id }
+      let(:project) { nil }
+
+      it_behaves_like 'does not create a cross reference'
+    end
+
+    context 'author_id does not exist' do
+      let(:author_id) { non_existing_record_id }
+
+      it_behaves_like 'does not create a cross reference'
+    end
+
+    context 'mentionable_id does not exist' do
+      let(:mentionable_id) { non_existing_record_id }
+
+      it_behaves_like 'does not create a cross reference'
+    end
+
+    context 'mentionable_type is not a Mentionable' do
+      let(:mentionable_type) { 'User' }
+
+      before do
+        expect(Gitlab::ErrorTracking).to receive(:track_and_raise_for_dev_exception).with(kind_of(ArgumentError))
+      end
+
+      it_behaves_like 'does not create a cross reference'
+    end
+
+    context 'mentionable_type is not a defined constant' do
+      let(:mentionable_type) { 'FooBar' }
+
+      before do
+        expect(Gitlab::ErrorTracking).to receive(:track_and_raise_for_dev_exception).with(kind_of(ArgumentError))
+      end
+
+      it_behaves_like 'does not create a cross reference'
+    end
+
+    context 'mentionable is a Commit and mentionable_id does not exist' do
+      let(:mentionable_type) { 'Commit' }
+      let(:mentionable_id) { non_existing_record_id }
+
+      it_behaves_like 'does not create a cross reference'
+    end
+  end
+end
diff --git a/spec/workers/issue_rebalancing_worker_spec.rb b/spec/workers/issue_rebalancing_worker_spec.rb
index cba42a1577e..cfb19af05b3 100644
--- a/spec/workers/issue_rebalancing_worker_spec.rb
+++ b/spec/workers/issue_rebalancing_worker_spec.rb
@@ -2,7 +2,7 @@
 
 require 'spec_helper'
 
-RSpec.describe IssueRebalancingWorker do
+RSpec.describe IssueRebalancingWorker, :clean_gitlab_redis_shared_state do
   describe '#perform' do
     let_it_be(:group) { create(:group) }
     let_it_be(:project) { create(:project, group: group) }
@@ -35,6 +35,20 @@ RSpec.describe IssueRebalancingWorker 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/placement_worker_spec.rb b/spec/workers/issues/placement_worker_spec.rb
new file mode 100644
index 00000000000..694cdd2ef37
--- /dev/null
+++ b/spec/workers/issues/placement_worker_spec.rb
@@ -0,0 +1,151 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Issues::PlacementWorker do
+  describe '#perform' do
+    let_it_be(:time) { Time.now.utc }
+    let_it_be(:group) { create(:group) }
+    let_it_be(:project) { create(:project, group: group) }
+    let_it_be(:author) { create(:user) }
+    let_it_be(:common_attrs) { { author: author, project: project } }
+    let_it_be(:unplaced) { common_attrs.merge(relative_position: nil) }
+    let_it_be_with_reload(:issue) { create(:issue, **unplaced, created_at: time) }
+    let_it_be_with_reload(:issue_a) { create(:issue, **unplaced, created_at: time - 1.minute) }
+    let_it_be_with_reload(:issue_b) { create(:issue, **unplaced, created_at: time - 2.minutes) }
+    let_it_be_with_reload(:issue_c) { create(:issue, **unplaced, created_at: time + 1.minute) }
+    let_it_be_with_reload(:issue_d) { create(:issue, **unplaced, created_at: time + 2.minutes) }
+    let_it_be_with_reload(:issue_e) { create(:issue, **common_attrs, relative_position: 10, created_at: time + 1.minute) }
+    let_it_be_with_reload(:issue_f) { create(:issue, **unplaced, created_at: time + 1.minute) }
+
+    let_it_be(:irrelevant) { create(:issue, relative_position: nil, created_at: time) }
+
+    shared_examples 'running the issue placement worker' do
+      let(:issue_id) { issue.id }
+      let(:project_id) { project.id }
+
+      it 'places all issues created at most 5 minutes before this one at the end, most recent last' do
+        expect { run_worker }.not_to change { irrelevant.reset.relative_position }
+
+        expect(project.issues.order_by_relative_position)
+          .to eq([issue_e, issue_b, issue_a, issue, issue_c, issue_f, issue_d])
+        expect(project.issues.where(relative_position: nil)).not_to exist
+      end
+
+      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)
+
+        run_worker
+      end
+
+      context 'there are more than QUERY_LIMIT unplaced issues' do
+        before_all do
+          # Ensure there are more than N issues in this set
+          n = described_class::QUERY_LIMIT
+          create_list(:issue, n - 5, **unplaced)
+        end
+
+        it 'limits the sweep to QUERY_LIMIT records, and reschedules placement' do
+          expect(Issue).to receive(:move_nulls_to_end)
+                             .with(have_attributes(count: described_class::QUERY_LIMIT))
+                             .and_call_original
+
+          expect(described_class).to receive(:perform_async).with(nil, project.id)
+
+          run_worker
+
+          expect(project.issues.where(relative_position: nil)).to exist
+        end
+
+        it 'is eventually correct' do
+          prefix = project.issues.where.not(relative_position: nil).order(:relative_position).to_a
+          moved = project.issues.where.not(id: prefix.map(&:id))
+
+          run_worker
+
+          expect(project.issues.where(relative_position: nil)).to exist
+
+          run_worker
+
+          expect(project.issues.where(relative_position: nil)).not_to exist
+          expect(project.issues.order(:relative_position)).to eq(prefix + moved.order(:created_at, :id))
+        end
+      end
+
+      context 'we are passed bad IDs' do
+        let(:issue_id) { non_existing_record_id }
+        let(:project_id) { non_existing_record_id }
+
+        def max_positions_by_project
+          Issue
+            .group(:project_id)
+            .pluck(:project_id, Issue.arel_table[:relative_position].maximum.as('max_relative_position'))
+            .to_h
+        end
+
+        it 'does move any issues to the end' do
+          expect { run_worker }.not_to change { max_positions_by_project }
+        end
+
+        context 'the project_id refers to an empty project' do
+          let!(:project_id) { create(:project).id }
+
+          it 'does move any issues to the end' do
+            expect { run_worker }.not_to change { max_positions_by_project }
+          end
+        end
+      end
+
+      it 'anticipates the failure to place the issues, and schedules rebalancing' do
+        allow(Issue).to receive(:move_nulls_to_end) { raise RelativePositioning::NoSpaceLeft }
+
+        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)
+
+        run_worker
+      end
+    end
+
+    context 'passing an issue ID' do
+      def run_worker
+        described_class.new.perform(issue_id)
+      end
+
+      let(:worker_arguments) { { issue_id: issue_id, project_id: nil } }
+
+      it_behaves_like 'running the issue placement worker'
+
+      context 'when block_issue_repositioning is enabled' do
+        let(:issue_id) { issue.id }
+        let(:project_id) { project.id }
+
+        before do
+          stub_feature_flags(block_issue_repositioning: group)
+        end
+
+        it 'does not run repositioning tasks' do
+          expect { run_worker }.not_to change { issue.reset.relative_position }
+        end
+      end
+    end
+
+    context 'passing a project ID' do
+      def run_worker
+        described_class.new.perform(nil, project_id)
+      end
+
+      let(:worker_arguments) { { issue_id: nil, project_id: project_id } }
+
+      it_behaves_like 'running the issue placement worker'
+    end
+  end
+
+  it 'has the `until_executed` deduplicate strategy' do
+    expect(described_class.get_deduplicate_strategy).to eq(:until_executed)
+    expect(described_class.get_deduplication_options).to include({ including_scheduled: true })
+  end
+end
diff --git a/spec/workers/issues/rebalancing_worker_spec.rb b/spec/workers/issues/rebalancing_worker_spec.rb
new file mode 100644
index 00000000000..438edd85f66
--- /dev/null
+++ b/spec/workers/issues/rebalancing_worker_spec.rb
@@ -0,0 +1,90 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Issues::RebalancingWorker do
+  describe '#perform' do
+    let_it_be(:group) { create(:group) }
+    let_it_be(:project) { create(:project, group: group) }
+    let_it_be(:issue) { create(:issue, project: project) }
+
+    shared_examples 'running the worker' do
+      it 'runs an instance of Issues::RelativePositionRebalancingService' do
+        service = double(execute: nil)
+        service_param = arguments.second.present? ? kind_of(Project.id_in([project]).class) : kind_of(group&.all_projects.class)
+
+        expect(Issues::RelativePositionRebalancingService).to receive(:new).with(service_param).and_return(service)
+
+        described_class.new.perform(*arguments)
+      end
+
+      it 'anticipates there being too many concurent rebalances' do
+        service = double
+        service_param = arguments.second.present? ? kind_of(Project.id_in([project]).class) : kind_of(group&.all_projects.class)
+
+        allow(service).to receive(:execute).and_raise(Issues::RelativePositionRebalancingService::TooManyConcurrentRebalances)
+        expect(Issues::RelativePositionRebalancingService).to receive(:new).with(service_param).and_return(service)
+        expect(Gitlab::ErrorTracking).to receive(:log_exception).with(Issues::RelativePositionRebalancingService::TooManyConcurrentRebalances, include(project_id: arguments.second, root_namespace_id: arguments.third))
+
+        described_class.new.perform(*arguments)
+      end
+
+      it 'takes no action if the value is nil' do
+        expect(Issues::RelativePositionRebalancingService).not_to receive(:new)
+        expect(Gitlab::ErrorTracking).not_to receive(:log_exception)
+
+        described_class.new.perform # all arguments are nil
+      end
+    end
+
+    shared_examples 'safely handles non-existent ids' do
+      it 'anticipates the inability to find the issue' do
+        expect(Gitlab::ErrorTracking).to receive(:log_exception).with(ArgumentError, include(project_id: arguments.second, root_namespace_id: arguments.third))
+        expect(Issues::RelativePositionRebalancingService).not_to receive(:new)
+
+        described_class.new.perform(*arguments)
+      end
+    end
+
+    context 'without root_namespace param' do
+      it_behaves_like 'running the worker' do
+        let(:arguments) { [-1, project.id] }
+      end
+
+      it_behaves_like 'safely handles non-existent ids' do
+        let(:arguments) { [nil, -1] }
+      end
+
+      include_examples 'an idempotent worker' do
+        let(:job_args) { [-1, project.id] }
+      end
+
+      include_examples 'an idempotent worker' do
+        let(:job_args) { [nil, -1] }
+      end
+    end
+
+    context 'with root_namespace param' do
+      it_behaves_like 'running the worker' do
+        let(:arguments) { [nil, nil, group.id] }
+      end
+
+      it_behaves_like 'safely handles non-existent ids' do
+        let(:arguments) { [nil, nil, -1] }
+      end
+
+      include_examples 'an idempotent worker' do
+        let(:job_args) { [nil, nil, group.id] }
+      end
+
+      include_examples 'an idempotent worker' do
+        let(:job_args) { [nil, nil, -1] }
+      end
+    end
+  end
+
+  it 'has the `until_executed` deduplicate strategy' do
+    expect(described_class.get_deduplicate_strategy).to eq(:until_executed)
+    expect(described_class.get_deduplication_options).to include({ including_scheduled: true })
+  end
+end
diff --git a/spec/workers/issues/reschedule_stuck_issue_rebalances_worker_spec.rb b/spec/workers/issues/reschedule_stuck_issue_rebalances_worker_spec.rb
new file mode 100644
index 00000000000..02d1241d2ba
--- /dev/null
+++ b/spec/workers/issues/reschedule_stuck_issue_rebalances_worker_spec.rb
@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Issues::RescheduleStuckIssueRebalancesWorker, :clean_gitlab_redis_shared_state do
+  let_it_be(:group) { create(:group) }
+  let_it_be(:project) { create(:project, group: group) }
+
+  subject(:worker) { described_class.new }
+
+  describe '#perform' do
+    it 'does not schedule a rebalance' do
+      expect(IssueRebalancingWorker).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
+
+      worker.perform
+    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
new file mode 100644
index 00000000000..544be2a69a6
--- /dev/null
+++ b/spec/workers/loose_foreign_keys/cleanup_worker_spec.rb
@@ -0,0 +1,153 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe LooseForeignKeys::CleanupWorker do
+  include MigrationsHelpers
+
+  def create_table_structure
+    migration = ActiveRecord::Migration.new.extend(Gitlab::Database::MigrationHelpers::LooseForeignKeyHelpers)
+
+    migration.create_table :_test_loose_fk_parent_table_1
+    migration.create_table :_test_loose_fk_parent_table_2
+
+    migration.create_table :_test_loose_fk_child_table_1_1 do |t|
+      t.bigint :parent_id
+    end
+
+    migration.create_table :_test_loose_fk_child_table_1_2 do |t|
+      t.bigint :parent_id_with_different_column
+    end
+
+    migration.create_table :_test_loose_fk_child_table_2_1 do |t|
+      t.bigint :parent_id
+    end
+
+    migration.track_record_deletions(:_test_loose_fk_parent_table_1)
+    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
+  end
+
+  let(:loose_fk_parent_table_1) { table(:_test_loose_fk_parent_table_1) }
+  let(:loose_fk_parent_table_2) { table(:_test_loose_fk_parent_table_2) }
+  let(:loose_fk_child_table_1_1) { table(:_test_loose_fk_child_table_1_1) }
+  let(:loose_fk_child_table_1_2) { table(:_test_loose_fk_child_table_1_2) }
+  let(:loose_fk_child_table_2_1) { table(:_test_loose_fk_child_table_2_1) }
+
+  before(:all) do
+    create_table_structure
+  end
+
+  after(:all) do
+    migration = ActiveRecord::Migration.new
+
+    migration.drop_table :_test_loose_fk_parent_table_1
+    migration.drop_table :_test_loose_fk_parent_table_2
+    migration.drop_table :_test_loose_fk_child_table_1_1
+    migration.drop_table :_test_loose_fk_child_table_1_2
+    migration.drop_table :_test_loose_fk_child_table_2_1
+  end
+
+  before do
+    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)
+
+    parent_record_2 = loose_fk_parent_table_1.create!
+    2.times { loose_fk_child_table_1_1.create!(parent_id: parent_record_2.id) }
+    3.times { loose_fk_child_table_1_2.create!(parent_id_with_different_column: parent_record_2.id) }
+
+    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
+  end
+
+  it 'cleans up all rows' do
+    described_class.new.perform
+
+    expect(loose_fk_child_table_1_1.count).to eq(0)
+    expect(loose_fk_child_table_1_2.where(parent_id_with_different_column: nil).count).to eq(4)
+    expect(loose_fk_child_table_2_1.count).to eq(0)
+  end
+
+  context 'when deleting in batches' do
+    before do
+      stub_const('LooseForeignKeys::CleanupWorker::BATCH_SIZE', 2)
+    end
+
+    it 'cleans up all rows' do
+      expect(LooseForeignKeys::BatchCleanerService).to receive(:new).exactly(:twice).and_call_original
+
+      described_class.new.perform
+
+      expect(loose_fk_child_table_1_1.count).to eq(0)
+      expect(loose_fk_child_table_1_2.where(parent_id_with_different_column: nil).count).to eq(4)
+      expect(loose_fk_child_table_2_1.count).to eq(0)
+    end
+  end
+
+  context 'when the deleted rows count limit have been reached' do
+    def count_deletable_rows
+      loose_fk_child_table_1_1.count + loose_fk_child_table_2_1.count
+    end
+
+    before do
+      stub_const('LooseForeignKeys::ModificationTracker::MAX_DELETES', 2)
+      stub_const('LooseForeignKeys::CleanerService::DELETE_LIMIT', 1)
+    end
+
+    it 'cleans up 2 rows' do
+      expect { described_class.new.perform }.to change { count_deletable_rows }.by(-2)
+    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
+end
diff --git a/spec/workers/namespaces/invite_team_email_worker_spec.rb b/spec/workers/namespaces/invite_team_email_worker_spec.rb
new file mode 100644
index 00000000000..47fdff9a8ef
--- /dev/null
+++ b/spec/workers/namespaces/invite_team_email_worker_spec.rb
@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Namespaces::InviteTeamEmailWorker do
+  let_it_be(:user) { create(:user) }
+  let_it_be(:group) { create(:group) }
+
+  it 'sends the email' do
+    expect(Namespaces::InviteTeamEmailService).to receive(:send_email).with(user, group).once
+    subject.perform(group.id, user.id)
+  end
+
+  context 'when user id is non-existent' do
+    it 'does not send the email' do
+      expect(Namespaces::InviteTeamEmailService).not_to receive(:send_email)
+      subject.perform(group.id, non_existing_record_id)
+    end
+  end
+
+  context 'when group id is non-existent' do
+    it 'does not send the email' do
+      expect(Namespaces::InviteTeamEmailService).not_to receive(:send_email)
+      subject.perform(non_existing_record_id, user.id)
+    end
+  end
+end
diff --git a/spec/workers/packages/maven/metadata/sync_worker_spec.rb b/spec/workers/packages/maven/metadata/sync_worker_spec.rb
index 10482b3e327..4b3cc6f964b 100644
--- a/spec/workers/packages/maven/metadata/sync_worker_spec.rb
+++ b/spec/workers/packages/maven/metadata/sync_worker_spec.rb
@@ -8,6 +8,7 @@ RSpec.describe Packages::Maven::Metadata::SyncWorker, type: :worker do
 
   let(:versions) { %w[1.2 1.1 2.1 3.0-SNAPSHOT] }
   let(:worker) { described_class.new }
+  let(:data_struct) { Struct.new(:release, :latest, :versions, keyword_init: true) }
 
   describe '#perform' do
     let(:user) { create(:user) }
@@ -197,7 +198,7 @@ RSpec.describe Packages::Maven::Metadata::SyncWorker, type: :worker do
   def versions_from(xml_content)
     xml_doc = Nokogiri::XML(xml_content)
 
-    OpenStruct.new(
+    data_struct.new(
       release: xml_doc.xpath('//metadata/versioning/release').first.content,
       latest: xml_doc.xpath('//metadata/versioning/latest').first.content,
       versions: xml_doc.xpath('//metadata/versioning/versions/version').map(&:content)
diff --git a/spec/workers/post_receive_spec.rb b/spec/workers/post_receive_spec.rb
index 039f86f1911..42e39c51a88 100644
--- a/spec/workers/post_receive_spec.rb
+++ b/spec/workers/post_receive_spec.rb
@@ -91,14 +91,6 @@ RSpec.describe PostReceive do
 
         perform
       end
-
-      it 'tracks an event for the empty_repo_upload experiment', :experiment do
-        expect_next_instance_of(EmptyRepoUploadExperiment) do |e|
-          expect(e).to receive(:track_initial_write)
-        end
-
-        perform
-      end
     end
 
     shared_examples 'not updating remote mirrors' do
diff --git a/spec/workers/propagate_integration_group_worker_spec.rb b/spec/workers/propagate_integration_group_worker_spec.rb
index 9d46534df4f..60442438a1d 100644
--- a/spec/workers/propagate_integration_group_worker_spec.rb
+++ b/spec/workers/propagate_integration_group_worker_spec.rb
@@ -22,7 +22,7 @@ RSpec.describe PropagateIntegrationGroupWorker do
       end
 
       context 'with a group integration' do
-        let_it_be(:integration) { create(:redmine_integration, group: group, project: nil) }
+        let_it_be(:integration) { create(:redmine_integration, :group, group: group) }
 
         it 'calls to BulkCreateIntegrationService' do
           expect(BulkCreateIntegrationService).to receive(:new)
diff --git a/spec/workers/propagate_integration_inherit_descendant_worker_spec.rb b/spec/workers/propagate_integration_inherit_descendant_worker_spec.rb
index 8a231d4104c..c9a7bfaa8b6 100644
--- a/spec/workers/propagate_integration_inherit_descendant_worker_spec.rb
+++ b/spec/workers/propagate_integration_inherit_descendant_worker_spec.rb
@@ -5,8 +5,8 @@ require 'spec_helper'
 RSpec.describe PropagateIntegrationInheritDescendantWorker do
   let_it_be(:group) { create(:group) }
   let_it_be(:subgroup) { create(:group, parent: group) }
-  let_it_be(:group_integration) { create(:redmine_integration, group: group, project: nil) }
-  let_it_be(:subgroup_integration) { create(:redmine_integration, group: subgroup, project: nil, inherit_from_id: group_integration.id) }
+  let_it_be(:group_integration) { create(:redmine_integration, :group, group: group) }
+  let_it_be(:subgroup_integration) { create(:redmine_integration, :group, group: subgroup, inherit_from_id: group_integration.id) }
 
   it_behaves_like 'an idempotent worker' do
     let(:job_args) { [group_integration.id, subgroup_integration.id, subgroup_integration.id] }
diff --git a/spec/workers/propagate_integration_project_worker_spec.rb b/spec/workers/propagate_integration_project_worker_spec.rb
index 312631252cc..c7adf1b826f 100644
--- a/spec/workers/propagate_integration_project_worker_spec.rb
+++ b/spec/workers/propagate_integration_project_worker_spec.rb
@@ -22,7 +22,7 @@ RSpec.describe PropagateIntegrationProjectWorker do
       end
 
       context 'with a group integration' do
-        let_it_be(:integration) { create(:redmine_integration, group: group, project: nil) }
+        let_it_be(:integration) { create(:redmine_integration, :group, group: group) }
 
         it 'calls to BulkCreateIntegrationService' do
           expect(BulkCreateIntegrationService).to receive(:new)
diff --git a/spec/workers/ssh_keys/expired_notification_worker_spec.rb b/spec/workers/ssh_keys/expired_notification_worker_spec.rb
index 109d24f03ab..be38391ff8c 100644
--- a/spec/workers/ssh_keys/expired_notification_worker_spec.rb
+++ b/spec/workers/ssh_keys/expired_notification_worker_spec.rb
@@ -20,7 +20,7 @@ RSpec.describe SshKeys::ExpiredNotificationWorker, type: :worker do
         stub_const("SshKeys::ExpiredNotificationWorker::BATCH_SIZE", 5)
       end
 
-      let_it_be_with_reload(:keys) { create_list(:key, 20, expires_at: 3.days.ago, user: user) }
+      let_it_be_with_reload(:keys) { create_list(:key, 20, expires_at: Time.current, user: user) }
 
       it 'updates all keys regardless of batch size' do
         worker.perform
@@ -54,8 +54,8 @@ RSpec.describe SshKeys::ExpiredNotificationWorker, type: :worker do
     context 'when key has expired in the past' do
       let_it_be(:expired_past) { create(:key, expires_at: 1.day.ago, user: user) }
 
-      it 'does update notified column' do
-        expect { worker.perform }.to change { expired_past.reload.expiry_notification_delivered_at }
+      it 'does not update notified column' do
+        expect { worker.perform }.not_to change { expired_past.reload.expiry_notification_delivered_at }
       end
 
       context 'when key has already been notified of expiration' do
diff --git a/spec/workers/tasks_to_be_done/create_worker_spec.rb b/spec/workers/tasks_to_be_done/create_worker_spec.rb
new file mode 100644
index 00000000000..a158872273f
--- /dev/null
+++ b/spec/workers/tasks_to_be_done/create_worker_spec.rb
@@ -0,0 +1,36 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe TasksToBeDone::CreateWorker do
+  let_it_be(:member_task) { create(:member_task, tasks: MemberTask::TASKS.values) }
+  let_it_be(:current_user) { create(:user) }
+
+  let(:assignee_ids) { [1, 2] }
+  let(:job_args) { [member_task.id, current_user.id, assignee_ids] }
+
+  before do
+    member_task.project.group.add_owner(current_user)
+  end
+
+  describe '.perform' do
+    it 'executes the task services for all tasks to be done', :aggregate_failures do
+      MemberTask::TASKS.each_key do |task|
+        service_class = "TasksToBeDone::Create#{task.to_s.camelize}TaskService".constantize
+
+        expect(service_class)
+          .to receive(:new)
+          .with(project: member_task.project, current_user: current_user, assignee_ids: assignee_ids)
+          .and_call_original
+      end
+
+      expect { described_class.new.perform(*job_args) }.to change(Issue, :count).by(3)
+    end
+  end
+
+  include_examples 'an idempotent worker' do
+    it 'creates 3 task issues' do
+      expect { subject }.to change(Issue, :count).by(3)
+    end
+  end
+end
diff --git a/spec/workers/users/deactivate_dormant_users_worker_spec.rb b/spec/workers/users/deactivate_dormant_users_worker_spec.rb
index 934c497c79a..20cd55e19eb 100644
--- a/spec/workers/users/deactivate_dormant_users_worker_spec.rb
+++ b/spec/workers/users/deactivate_dormant_users_worker_spec.rb
@@ -3,6 +3,8 @@
 require 'spec_helper'
 
 RSpec.describe Users::DeactivateDormantUsersWorker do
+  using RSpec::Parameterized::TableSyntax
+
   describe '#perform' do
     let_it_be(:dormant) { create(:user, last_activity_on: User::MINIMUM_INACTIVE_DAYS.days.ago.to_date) }
     let_it_be(:inactive) { create(:user, last_activity_on: nil) }
@@ -22,12 +24,12 @@ RSpec.describe Users::DeactivateDormantUsersWorker do
     context 'when automatic deactivation of dormant users is enabled' do
       before do
         stub_application_setting(deactivate_dormant_users: true)
+        stub_const("#{described_class.name}::PAUSE_SECONDS", 0)
       end
 
       it 'deactivates dormant users' do
         freeze_time do
           stub_const("#{described_class.name}::BATCH_SIZE", 1)
-          stub_const("#{described_class.name}::PAUSE_SECONDS", 0)
 
           expect(worker).to receive(:sleep).twice
 
@@ -37,6 +39,38 @@ RSpec.describe Users::DeactivateDormantUsersWorker do
           expect(User.with_no_activity.count).to eq(0)
         end
       end
+
+      where(:user_type, :expected_state) do
+        :human             | 'deactivated'
+        :support_bot       | 'active'
+        :alert_bot         | 'active'
+        :visual_review_bot | 'active'
+        :service_user      | 'deactivated'
+        :ghost             | 'active'
+        :project_bot       | 'active'
+        :migration_bot     | 'active'
+        :security_bot      | 'active'
+        :automation_bot    | 'active'
+      end
+      with_them do
+        it 'deactivates certain user types' do
+          user = create(:user, user_type: user_type, state: :active, last_activity_on: User::MINIMUM_INACTIVE_DAYS.days.ago.to_date)
+
+          worker.perform
+
+          expect(user.reload.state).to eq(expected_state)
+        end
+      end
+
+      it 'does not deactivate non-active users' do
+        human_user = create(:user, user_type: :human, state: :blocked, last_activity_on: User::MINIMUM_INACTIVE_DAYS.days.ago.to_date)
+        service_user = create(:user, user_type: :service_user, state: :blocked, last_activity_on: User::MINIMUM_INACTIVE_DAYS.days.ago.to_date)
+
+        worker.perform
+
+        expect(human_user.reload.state).to eq('blocked')
+        expect(service_user.reload.state).to eq('blocked')
+      end
     end
 
     context 'when automatic deactivation of dormant users is disabled' do
-- 
cgit v1.2.1