Add latest changes from gitlab-org/gitlab@14-0-stable-eev14.0.0-rc42

50 files changed, 967 insertions, 1114 deletions

diff --git a/spec/workers/analytics/instance_statistics/count_job_trigger_worker_spec.rb b/spec/workers/analytics/instance_statistics/count_job_trigger_worker_spec.rb
deleted file mode 100644
index da0cbe37400..00000000000
--- a/spec/workers/analytics/instance_statistics/count_job_trigger_worker_spec.rb
+++ /dev/null
@@ -1,17 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe Analytics::InstanceStatistics::CountJobTriggerWorker do
-  it_behaves_like 'an idempotent worker'
-
-  context 'triggers a job for each measurement identifiers' do
-    let(:expected_count) { Analytics::UsageTrends::Measurement.identifier_query_mapping.keys.size }
-
-    it 'triggers CounterJobWorker jobs' do
-      subject.perform
-
-      expect(Analytics::UsageTrends::CounterJobWorker.jobs.count).to eq(expected_count)
-    end
-  end
-end
diff --git a/spec/workers/analytics/instance_statistics/counter_job_worker_spec.rb b/spec/workers/analytics/instance_statistics/counter_job_worker_spec.rb
deleted file mode 100644
index 4994fec44ab..00000000000
--- a/spec/workers/analytics/instance_statistics/counter_job_worker_spec.rb
+++ /dev/null
@@ -1,70 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe Analytics::InstanceStatistics::CounterJobWorker do
-  let_it_be(:user_1) { create(:user) }
-  let_it_be(:user_2) { create(:user) }
-
-  let(:users_measurement_identifier) { ::Analytics::UsageTrends::Measurement.identifiers.fetch(:users) }
-  let(:recorded_at) { Time.zone.now }
-  let(:job_args) { [users_measurement_identifier, user_1.id, user_2.id, recorded_at] }
-
-  before do
-    allow(ActiveRecord::Base.connection).to receive(:transaction_open?).and_return(false)
-  end
-
-  include_examples 'an idempotent worker' do
-    it 'counts a scope and stores the result' do
-      subject
-
-      measurement = Analytics::UsageTrends::Measurement.users.first
-      expect(measurement.recorded_at).to be_like_time(recorded_at)
-      expect(measurement.identifier).to eq('users')
-      expect(measurement.count).to eq(2)
-    end
-  end
-
-  context 'when no records are in the database' do
-    let(:users_measurement_identifier) { ::Analytics::UsageTrends::Measurement.identifiers.fetch(:groups) }
-
-    subject { described_class.new.perform(users_measurement_identifier, nil, nil, recorded_at) }
-
-    it 'sets 0 as the count' do
-      subject
-
-      measurement = Analytics::UsageTrends::Measurement.groups.first
-      expect(measurement.recorded_at).to be_like_time(recorded_at)
-      expect(measurement.identifier).to eq('groups')
-      expect(measurement.count).to eq(0)
-    end
-  end
-
-  it 'does not raise error when inserting duplicated measurement' do
-    subject
-
-    expect { subject }.not_to raise_error
-  end
-
-  it 'does not insert anything when BatchCount returns error' do
-    allow(Gitlab::Database::BatchCount).to receive(:batch_count).and_return(Gitlab::Database::BatchCounter::FALLBACK)
-
-    expect { subject }.not_to change { Analytics::UsageTrends::Measurement.count }
-  end
-
-  context 'when pipelines_succeeded identifier is passed' do
-    let_it_be(:pipeline) { create(:ci_pipeline, :success) }
-
-    let(:successful_pipelines_measurement_identifier) { ::Analytics::UsageTrends::Measurement.identifiers.fetch(:pipelines_succeeded) }
-    let(:job_args) { [successful_pipelines_measurement_identifier, pipeline.id, pipeline.id, recorded_at] }
-
-    it 'counts successful pipelines' do
-      subject
-
-      measurement = Analytics::UsageTrends::Measurement.pipelines_succeeded.first
-      expect(measurement.recorded_at).to be_like_time(recorded_at)
-      expect(measurement.identifier).to eq('pipelines_succeeded')
-      expect(measurement.count).to eq(1)
-    end
-  end
-end
diff --git a/spec/workers/authorized_project_update/project_recalculate_worker_spec.rb b/spec/workers/authorized_project_update/project_recalculate_worker_spec.rb
new file mode 100644
index 00000000000..403793a15e2
--- /dev/null
+++ b/spec/workers/authorized_project_update/project_recalculate_worker_spec.rb
@@ -0,0 +1,68 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe AuthorizedProjectUpdate::ProjectRecalculateWorker do
+  include ExclusiveLeaseHelpers
+
+  let_it_be(:project) { create(:project) }
+
+  subject(:worker) { described_class.new }
+
+  it 'is labeled as high urgency' do
+    expect(described_class.get_urgency).to eq(:high)
+  end
+
+  include_examples 'an idempotent worker' do
+    let(:job_args) { project.id }
+
+    it 'does not change authorizations when run twice' do
+      user = create(:user)
+      project.add_developer(user)
+
+      user.project_authorizations.delete_all
+
+      expect { worker.perform(project.id) }.to change { project.project_authorizations.reload.size }.by(1)
+      expect { worker.perform(project.id) }.not_to change { project.project_authorizations.reload.size }
+    end
+  end
+
+  describe '#perform' do
+    it 'does not fail if the project does not exist' do
+      expect do
+        worker.perform(non_existing_record_id)
+      end.not_to raise_error
+    end
+
+    it 'calls AuthorizedProjectUpdate::ProjectRecalculateService' do
+      expect_next_instance_of(AuthorizedProjectUpdate::ProjectRecalculateService, project) do |service|
+        expect(service).to receive(:execute)
+      end
+
+      worker.perform(project.id)
+    end
+
+    context 'exclusive lease' do
+      let(:lock_key) { "#{described_class.name.underscore}/#{project.root_namespace.id}" }
+      let(:timeout) { 10.seconds }
+
+      context 'when exclusive lease has not been taken' do
+        it 'obtains a new exclusive lease' do
+          expect_to_obtain_exclusive_lease(lock_key, timeout: timeout)
+
+          worker.perform(project.id)
+        end
+      end
+
+      context 'when exclusive lease has already been taken' do
+        before do
+          stub_exclusive_lease_taken(lock_key, timeout: timeout)
+        end
+
+        it 'raises an error' do
+          expect { worker.perform(project.id) }.to raise_error(Gitlab::ExclusiveLeaseHelpers::FailedToObtainLockError)
+        end
+      end
+    end
+  end
+end
diff --git a/spec/workers/authorized_project_update/user_refresh_from_replica_worker_spec.rb b/spec/workers/authorized_project_update/user_refresh_from_replica_worker_spec.rb
new file mode 100644
index 00000000000..cdf2cb493b0
--- /dev/null
+++ b/spec/workers/authorized_project_update/user_refresh_from_replica_worker_spec.rb
@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe AuthorizedProjectUpdate::UserRefreshFromReplicaWorker do
+  it 'is labeled as low urgency' do
+    expect(described_class.get_urgency).to eq(:low)
+  end
+
+  it_behaves_like "refreshes user's project authorizations"
+end
diff --git a/spec/workers/authorized_project_update/user_refresh_over_user_range_worker_spec.rb b/spec/workers/authorized_project_update/user_refresh_over_user_range_worker_spec.rb
index 832d5afd957..7c0c4d5bab4 100644
--- a/spec/workers/authorized_project_update/user_refresh_over_user_range_worker_spec.rb
+++ b/spec/workers/authorized_project_update/user_refresh_over_user_range_worker_spec.rb
@@ -3,7 +3,8 @@
 require 'spec_helper'
 
 RSpec.describe AuthorizedProjectUpdate::UserRefreshOverUserRangeWorker do
-  let(:project) { create(:project) }
+  let_it_be(:project) { create(:project) }
+
   let(:user) { project.namespace.owner }
   let(:start_user_id) { user.id }
   let(:end_user_id) { start_user_id }
@@ -11,56 +12,35 @@ RSpec.describe AuthorizedProjectUpdate::UserRefreshOverUserRangeWorker do
 
   it_behaves_like 'worker with data consistency',
                   described_class,
-                  feature_flag: :delayed_consistency_for_user_refresh_over_range_worker,
                   data_consistency: :delayed
 
   describe '#perform' do
-    context 'when the feature flag `periodic_project_authorization_update_via_replica` is enabled' do
-      before do
-        stub_feature_flags(periodic_project_authorization_update_via_replica: true)
-      end
-
-      context 'checks if project authorization update is required' do
-        it 'checks if a project_authorization refresh is needed for each of the users' do
-          User.where(id: start_user_id..end_user_id).each do |user|
-            expect(AuthorizedProjectUpdate::FindRecordsDueForRefreshService).to(
-              receive(:new).with(user).and_call_original)
-          end
-
-          execute_worker
-        end
-      end
-
-      context 'when there are project authorization records due for either removal or addition for a specific user' do
-        before do
-          user.project_authorizations.delete_all
+    context 'checks if project authorization update is required' do
+      it 'checks if a project_authorization refresh is needed for each of the users' do
+        User.where(id: start_user_id..end_user_id).each do |user|
+          expect(AuthorizedProjectUpdate::FindRecordsDueForRefreshService).to(
+            receive(:new).with(user).and_call_original)
         end
 
-        it 'enqueues a new project authorization update job for the user' do
-          expect(AuthorizedProjectUpdate::UserRefreshWithLowUrgencyWorker).to receive(:perform_async).with(user.id)
+        execute_worker
+      end
+    end
 
-          execute_worker
-        end
+    context 'when there are project authorization records due for either removal or addition for a specific user' do
+      before do
+        user.project_authorizations.delete_all
       end
 
-      context 'when there are no additions or removals to be made to project authorizations for a specific user' do
-        it 'does not enqueue a new project authorization update job for the user' do
-          expect(AuthorizedProjectUpdate::UserRefreshWithLowUrgencyWorker).not_to receive(:perform_async)
+      it 'enqueues a new project authorization update job for the user' do
+        expect(AuthorizedProjectUpdate::UserRefreshWithLowUrgencyWorker).to receive(:perform_async).with(user.id)
 
-          execute_worker
-        end
+        execute_worker
       end
     end
 
-    context 'when the feature flag `periodic_project_authorization_update_via_replica` is disabled' do
-      before do
-        stub_feature_flags(periodic_project_authorization_update_via_replica: false)
-      end
-
-      it 'calls AuthorizedProjectUpdate::RecalculateForUserRangeService' do
-        expect_next_instance_of(AuthorizedProjectUpdate::RecalculateForUserRangeService, start_user_id, end_user_id) do |service|
-          expect(service).to receive(:execute)
-        end
+    context 'when there are no additions or removals to be made to project authorizations for a specific user' do
+      it 'does not enqueue a new project authorization update job for the user' do
+        expect(AuthorizedProjectUpdate::UserRefreshWithLowUrgencyWorker).not_to receive(:perform_async)
 
         execute_worker
       end
diff --git a/spec/workers/build_hooks_worker_spec.rb b/spec/workers/build_hooks_worker_spec.rb
index 8395d8fb0e7..5f7e7e5fb00 100644
--- a/spec/workers/build_hooks_worker_spec.rb
+++ b/spec/workers/build_hooks_worker_spec.rb
@@ -24,18 +24,8 @@ RSpec.describe BuildHooksWorker do
   end
 
   describe '.perform_async' do
-    context 'when delayed_perform_for_build_hooks_worker feature flag is disabled' do
-      before do
-        stub_feature_flags(delayed_perform_for_build_hooks_worker: false)
-      end
-
-      it 'does not call perform_in' do
-        expect(described_class).not_to receive(:perform_in)
-      end
-    end
-
-    it 'delays scheduling a job by calling perform_in' do
-      expect(described_class).to receive(:perform_in).with(described_class::DATA_CONSISTENCY_DELAY.second, 123)
+    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
@@ -43,6 +33,5 @@ RSpec.describe BuildHooksWorker do
 
   it_behaves_like 'worker with data consistency',
                   described_class,
-                  feature_flag: :load_balancing_for_build_hooks_worker,
                   data_consistency: :delayed
 end
diff --git a/spec/workers/build_queue_worker_spec.rb b/spec/workers/build_queue_worker_spec.rb
new file mode 100644
index 00000000000..5f8510abf23
--- /dev/null
+++ b/spec/workers/build_queue_worker_spec.rb
@@ -0,0 +1,31 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BuildQueueWorker do
+  describe '#perform' do
+    context 'when build exists' do
+      let!(:build) { create(:ci_build) }
+
+      it 'ticks runner queue value' do
+        expect_next_instance_of(Ci::UpdateBuildQueueService) do |instance|
+          expect(instance).to receive(:tick).with(build)
+        end
+
+        described_class.new.perform(build.id)
+      end
+    end
+
+    context 'when build does not exist' do
+      it 'does not raise exception' do
+        expect { described_class.new.perform(123) }
+          .not_to raise_error
+      end
+    end
+  end
+
+  it_behaves_like 'worker with data consistency',
+                  described_class,
+                  feature_flag: :load_balancing_for_build_queue_worker,
+                  data_consistency: :sticky
+end
diff --git a/spec/workers/bulk_import_worker_spec.rb b/spec/workers/bulk_import_worker_spec.rb
index 9119394f250..205bf23f36d 100644
--- a/spec/workers/bulk_import_worker_spec.rb
+++ b/spec/workers/bulk_import_worker_spec.rb
@@ -22,6 +22,16 @@ RSpec.describe BulkImportWorker do
       end
     end
 
+    context 'when bulk import is failed' do
+      it 'does nothing' do
+        bulk_import = create(:bulk_import, :failed)
+
+        expect(described_class).not_to receive(:perform_in)
+
+        subject.perform(bulk_import.id)
+      end
+    end
+
     context 'when all entities are processed' do
       it 'marks bulk import as finished' do
         bulk_import = create(:bulk_import, :started)
@@ -34,6 +44,18 @@ RSpec.describe BulkImportWorker do
       end
     end
 
+    context 'when all entities are failed' do
+      it 'marks bulk import as failed' do
+        bulk_import = create(:bulk_import, :started)
+        create(:bulk_import_entity, :failed, bulk_import: bulk_import)
+        create(:bulk_import_entity, :failed, bulk_import: bulk_import)
+
+        subject.perform(bulk_import.id)
+
+        expect(bulk_import.reload.failed?).to eq(true)
+      end
+    end
+
     context 'when maximum allowed number of import entities in progress' do
       it 'reenqueues itself' do
         bulk_import = create(:bulk_import, :started)
diff --git a/spec/workers/bulk_imports/export_request_worker_spec.rb b/spec/workers/bulk_imports/export_request_worker_spec.rb
index f7838279212..8d528011752 100644
--- a/spec/workers/bulk_imports/export_request_worker_spec.rb
+++ b/spec/workers/bulk_imports/export_request_worker_spec.rb
@@ -19,7 +19,7 @@ RSpec.describe BulkImports::ExportRequestWorker do
       it 'requests relations export' do
         expected = "/groups/foo%2Fbar/export_relations"
 
-        expect_next_instance_of(BulkImports::Clients::Http) do |client|
+        expect_next_instance_of(BulkImports::Clients::HTTP) do |client|
           expect(client).to receive(:post).with(expected).twice
         end
 
diff --git a/spec/workers/bulk_imports/pipeline_worker_spec.rb b/spec/workers/bulk_imports/pipeline_worker_spec.rb
index 27151177634..972a4158194 100644
--- a/spec/workers/bulk_imports/pipeline_worker_spec.rb
+++ b/spec/workers/bulk_imports/pipeline_worker_spec.rb
@@ -8,10 +8,16 @@ RSpec.describe BulkImports::PipelineWorker do
       def initialize(_); end
 
       def run; end
+
+      def self.ndjson_pipeline?
+        false
+      end
     end
   end
 
-  let_it_be(:entity) { create(:bulk_import_entity) }
+  let_it_be(:bulk_import) { create(:bulk_import) }
+  let_it_be(:config) { create(:bulk_import_configuration, bulk_import: bulk_import) }
+  let_it_be(:entity) { create(:bulk_import_entity, bulk_import: bulk_import) }
 
   before do
     stub_const('FakePipeline', pipeline_class)
@@ -27,6 +33,7 @@ RSpec.describe BulkImports::PipelineWorker do
     expect(BulkImports::Stage)
       .to receive(:pipeline_exists?)
       .with('FakePipeline')
+      .twice
       .and_return(true)
 
     expect_next_instance_of(Gitlab::Import::Logger) do |logger|
@@ -122,4 +129,117 @@ RSpec.describe BulkImports::PipelineWorker do
       expect(pipeline_tracker.jid).to eq('jid')
     end
   end
+
+  context 'when ndjson pipeline' do
+    let(:ndjson_pipeline) do
+      Class.new do
+        def initialize(_); end
+
+        def run; end
+
+        def self.ndjson_pipeline?
+          true
+        end
+
+        def self.relation
+          'test'
+        end
+      end
+    end
+
+    let(:pipeline_tracker) do
+      create(
+        :bulk_import_tracker,
+        entity: entity,
+        pipeline_name: 'NdjsonPipeline'
+      )
+    end
+
+    before do
+      stub_const('NdjsonPipeline', ndjson_pipeline)
+      allow(BulkImports::Stage)
+        .to receive(:pipeline_exists?)
+        .with('NdjsonPipeline')
+        .and_return(true)
+    end
+
+    it 'runs the pipeline successfully' do
+      allow_next_instance_of(BulkImports::ExportStatus) do |status|
+        allow(status).to receive(:started?).and_return(false)
+        allow(status).to receive(:failed?).and_return(false)
+      end
+
+      subject.perform(pipeline_tracker.id, pipeline_tracker.stage, entity.id)
+
+      expect(pipeline_tracker.reload.status_name).to eq(:finished)
+    end
+
+    context 'when export status is started' do
+      it 'reenqueues pipeline worker' do
+        allow_next_instance_of(BulkImports::ExportStatus) do |status|
+          allow(status).to receive(:started?).and_return(true)
+          allow(status).to receive(:failed?).and_return(false)
+        end
+
+        expect(described_class)
+          .to receive(:perform_in)
+          .with(
+            described_class::NDJSON_PIPELINE_PERFORM_DELAY,
+            pipeline_tracker.id,
+            pipeline_tracker.stage,
+            entity.id
+          )
+
+        subject.perform(pipeline_tracker.id, pipeline_tracker.stage, entity.id)
+      end
+    end
+
+    context 'when job reaches timeout' do
+      it 'marks as failed and logs the error' do
+        old_created_at = entity.created_at
+        entity.update!(created_at: (BulkImports::Pipeline::NDJSON_EXPORT_TIMEOUT + 1.hour).ago)
+
+        expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+          expect(logger)
+            .to receive(:error)
+            .with(
+              worker: described_class.name,
+              pipeline_name: 'NdjsonPipeline',
+              entity_id: entity.id,
+              message: 'Pipeline timeout'
+            )
+        end
+
+        subject.perform(pipeline_tracker.id, pipeline_tracker.stage, entity.id)
+
+        expect(pipeline_tracker.reload.status_name).to eq(:failed)
+
+        entity.update!(created_at: old_created_at)
+      end
+    end
+
+    context 'when export status is failed' do
+      it 'marks as failed and logs the error' do
+        allow_next_instance_of(BulkImports::ExportStatus) do |status|
+          allow(status).to receive(:failed?).and_return(true)
+          allow(status).to receive(:error).and_return('Error!')
+        end
+
+        expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+          expect(logger)
+            .to receive(:error)
+            .with(
+              worker: described_class.name,
+              pipeline_name: 'NdjsonPipeline',
+              entity_id: entity.id,
+              message: 'Error!'
+            )
+        end
+
+        subject.perform(pipeline_tracker.id, pipeline_tracker.stage, entity.id)
+
+        expect(pipeline_tracker.reload.status_name).to eq(:failed)
+      end
+    end
+  end
 end
diff --git a/spec/workers/ci/initial_pipeline_process_worker_spec.rb b/spec/workers/ci/initial_pipeline_process_worker_spec.rb
index 5db9287fe96..5fb8671fd5c 100644
--- a/spec/workers/ci/initial_pipeline_process_worker_spec.rb
+++ b/spec/workers/ci/initial_pipeline_process_worker_spec.rb
@@ -4,7 +4,9 @@ require 'spec_helper'
 
 RSpec.describe Ci::InitialPipelineProcessWorker do
   describe '#perform' do
-    let_it_be(:pipeline) { create(:ci_pipeline, :with_job, status: :created) }
+    let_it_be_with_reload(:pipeline) do
+      create(:ci_pipeline, :with_job, status: :created)
+    end
 
     include_examples 'an idempotent worker' do
       let(:job_args) { pipeline.id }
diff --git a/spec/workers/clusters/applications/activate_service_worker_spec.rb b/spec/workers/clusters/applications/activate_service_worker_spec.rb
index c157c57888e..7b05b76bebc 100644
--- a/spec/workers/clusters/applications/activate_service_worker_spec.rb
+++ b/spec/workers/clusters/applications/activate_service_worker_spec.rb
@@ -8,13 +8,13 @@ RSpec.describe Clusters::Applications::ActivateServiceWorker, '#perform' do
       let(:service_name) { 'prometheus' }
 
       before do
-        create(:clusters_applications_prometheus, :installed, cluster: cluster)
+        create(:clusters_integrations_prometheus, cluster: cluster)
       end
 
       context 'cluster type: group' do
         let(:group) { create(:group) }
         let(:project) { create(:project, group: group) }
-        let(:cluster) { create(:cluster_for_group, :with_installed_helm, groups: [group]) }
+        let(:cluster) { create(:cluster_for_group, groups: [group]) }
 
         it 'ensures Prometheus service is activated' do
           expect { described_class.new.perform(cluster.id, service_name) }
@@ -24,7 +24,7 @@ RSpec.describe Clusters::Applications::ActivateServiceWorker, '#perform' do
 
       context 'cluster type: project' do
         let(:project) { create(:project) }
-        let(:cluster) { create(:cluster, :with_installed_helm, projects: [project]) }
+        let(:cluster) { create(:cluster, projects: [project]) }
 
         it 'ensures Prometheus service is activated' do
           expect { described_class.new.perform(cluster.id, service_name) }
diff --git a/spec/workers/clusters/applications/deactivate_service_worker_spec.rb b/spec/workers/clusters/applications/deactivate_service_worker_spec.rb
index 18cceaaf3b1..4068c5c9eaa 100644
--- a/spec/workers/clusters/applications/deactivate_service_worker_spec.rb
+++ b/spec/workers/clusters/applications/deactivate_service_worker_spec.rb
@@ -6,19 +6,19 @@ RSpec.describe Clusters::Applications::DeactivateServiceWorker, '#perform' do
   context 'cluster exists' do
     describe 'prometheus service' do
       let(:service_name) { 'prometheus' }
-      let!(:application) { create(:clusters_applications_prometheus, :installed, cluster: cluster) }
+      let!(:integration) { create(:clusters_integrations_prometheus, cluster: cluster) }
 
       context 'prometheus service exists' do
         let!(:prometheus_service) { create(:prometheus_service, project: project, manual_configuration: false, active: true) }
 
         before do
-          application.delete # prometheus service before save synchronises active stated with application existance.
+          integration.delete # prometheus service before save synchronises active stated with integration existence.
         end
 
         context 'cluster type: group' do
           let(:group) { create(:group) }
           let(:project) { create(:project, group: group) }
-          let(:cluster) { create(:cluster_for_group, :with_installed_helm, groups: [group]) }
+          let(:cluster) { create(:cluster_for_group, groups: [group]) }
 
           it 'ensures Prometheus service is deactivated' do
             expect { described_class.new.perform(cluster.id, service_name) }
@@ -28,7 +28,7 @@ RSpec.describe Clusters::Applications::DeactivateServiceWorker, '#perform' do
 
         context 'cluster type: project' do
           let(:project) { create(:project) }
-          let(:cluster) { create(:cluster, :with_installed_helm, projects: [project]) }
+          let(:cluster) { create(:cluster, projects: [project]) }
 
           it 'ensures Prometheus service is deactivated' do
             expect { described_class.new.perform(cluster.id, service_name) }
@@ -38,7 +38,7 @@ RSpec.describe Clusters::Applications::DeactivateServiceWorker, '#perform' do
 
         context 'cluster type: instance' do
           let(:project) { create(:project) }
-          let(:cluster) { create(:cluster, :with_installed_helm, :instance) }
+          let(:cluster) { create(:cluster, :instance) }
 
           it 'ensures Prometheus service is deactivated' do
             expect { described_class.new.perform(cluster.id, service_name) }
@@ -50,7 +50,7 @@ RSpec.describe Clusters::Applications::DeactivateServiceWorker, '#perform' do
       context 'prometheus service does not exist' do
         context 'cluster type: project' do
           let(:project) { create(:project) }
-          let(:cluster) { create(:cluster, :with_installed_helm, projects: [project]) }
+          let(:cluster) { create(:cluster, projects: [project]) }
 
           it 'does not raise errors' do
             expect { described_class.new.perform(cluster.id, service_name) }.not_to raise_error
diff --git a/spec/workers/clusters/cleanup/app_worker_spec.rb b/spec/workers/clusters/cleanup/app_worker_spec.rb
deleted file mode 100644
index 661468f037f..00000000000
--- a/spec/workers/clusters/cleanup/app_worker_spec.rb
+++ /dev/null
@@ -1,41 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe Clusters::Cleanup::AppWorker do
-  describe '#perform' do
-    subject { worker_instance.perform(cluster.id) }
-
-    let!(:worker_instance) { described_class.new }
-    let!(:cluster) { create(:cluster, :project, :cleanup_uninstalling_applications, provider_type: :gcp) }
-    let!(:logger) { worker_instance.send(:logger) }
-
-    it_behaves_like 'cluster cleanup worker base specs'
-
-    context 'when exceeded the execution limit' do
-      subject { worker_instance.perform(cluster.id, worker_instance.send(:execution_limit)) }
-
-      let(:worker_instance) { described_class.new }
-      let(:logger) { worker_instance.send(:logger) }
-      let!(:helm) { create(:clusters_applications_helm, :installed, cluster: cluster) }
-      let!(:ingress) { create(:clusters_applications_ingress, :scheduled, cluster: cluster) }
-
-      it 'logs the error' do
-        expect(logger).to receive(:error)
-        .with(
-          hash_including(
-            exception: 'ClusterCleanupMethods::ExceededExecutionLimitError',
-            cluster_id: kind_of(Integer),
-            class_name: described_class.name,
-            applications: "helm:installed,ingress:scheduled",
-            cleanup_status: cluster.cleanup_status_name,
-            event: :failed_to_remove_cluster_and_resources,
-            message: "exceeded execution limit of 10 tries"
-          )
-        )
-
-        subject
-      end
-    end
-  end
-end
diff --git a/spec/workers/concerns/application_worker_spec.rb b/spec/workers/concerns/application_worker_spec.rb
index 5c1a1d3ae8f..29c69ff8b4b 100644
--- a/spec/workers/concerns/application_worker_spec.rb
+++ b/spec/workers/concerns/application_worker_spec.rb
@@ -176,6 +176,40 @@ RSpec.describe ApplicationWorker do
     end
   end
 
+  describe '.perform_async' do
+    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)
+
+        worker.perform_async(123)
+      end
+    end
+
+    context 'when workers data consistency is :sticky' do
+      it_behaves_like 'worker utilizes load balancing capabilities', :sticky
+    end
+
+    context 'when workers data consistency is :delayed' do
+      it_behaves_like 'worker utilizes load balancing capabilities', :delayed
+    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)
+
+        worker.perform_async
+      end
+    end
+  end
+
   describe '.bulk_perform_async' do
     it 'enqueues jobs in bulk' do
       Sidekiq::Testing.fake! do
diff --git a/spec/workers/concerns/worker_attributes_spec.rb b/spec/workers/concerns/worker_attributes_spec.rb
index a654ecbd3e2..d4b17c65f46 100644
--- a/spec/workers/concerns/worker_attributes_spec.rb
+++ b/spec/workers/concerns/worker_attributes_spec.rb
@@ -62,6 +62,12 @@ RSpec.describe WorkerAttributes do
   end
 
   describe '.idempotent!' do
+    it 'sets `idempotent` attribute of the worker class to true' do
+      worker.idempotent!
+
+      expect(worker.send(:class_attributes)[:idempotent]).to eq(true)
+    end
+
     context 'when data consistency is not :always' do
       it 'raise exception' do
         worker.data_consistency(:sticky)
@@ -71,4 +77,66 @@ RSpec.describe WorkerAttributes do
       end
     end
   end
+
+  describe '.idempotent?' do
+    subject(:idempotent?) { worker.idempotent? }
+
+    context 'when the worker is idempotent' do
+      before do
+        worker.idempotent!
+      end
+
+      it { is_expected.to be_truthy }
+    end
+
+    context 'when the worker is not idempotent' do
+      it { is_expected.to be_falsey }
+    end
+  end
+
+  describe '.deduplicate' do
+    it 'sets deduplication_strategy and deduplication_options' do
+      worker.deduplicate(:until_executing, including_scheduled: true)
+
+      expect(worker.send(:class_attributes)[:deduplication_strategy]).to eq(:until_executing)
+      expect(worker.send(:class_attributes)[:deduplication_options]).to eq(including_scheduled: true)
+    end
+  end
+
+  describe '#deduplication_enabled?' do
+    subject(:deduplication_enabled?) { worker.deduplication_enabled? }
+
+    context 'when no feature flag is set' do
+      before do
+        worker.deduplicate(:until_executing)
+      end
+
+      it { is_expected.to eq(true) }
+    end
+
+    context 'when feature flag is set' do
+      before do
+        skip_feature_flags_yaml_validation
+        skip_default_enabled_yaml_check
+
+        worker.deduplicate(:until_executing, feature_flag: :my_feature_flag)
+      end
+
+      context 'when the FF is enabled' do
+        before do
+          stub_feature_flags(my_feature_flag: true)
+        end
+
+        it { is_expected.to eq(true) }
+      end
+
+      context 'when the FF is disabled' do
+        before do
+          stub_feature_flags(my_feature_flag: false)
+        end
+
+        it { is_expected.to eq(false) }
+      end
+    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 04f568515ed..c399697cbe0 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
@@ -85,7 +85,7 @@ RSpec.describe ContainerExpirationPolicies::CleanupContainerRepositoryWorker do
 
       context 'with policy running shortly' do
         before do
-          repository.cleanup_unfinished! if loopless_enabled?
+          repository.cleanup_unfinished!
           policy.update_column(:next_run_at, 1.minute.from_now)
         end
 
@@ -108,371 +108,261 @@ RSpec.describe ContainerExpirationPolicies::CleanupContainerRepositoryWorker do
         it 'skips the repository' do
           expect(ContainerExpirationPolicies::CleanupService).not_to receive(:new)
 
-          if loopless_enabled?
-            expect { subject }
-              .to not_change { ContainerRepository.waiting_for_cleanup.count }
-              .and not_change { repository.reload.expiration_policy_cleanup_status }
-          else
-            expect { subject }.to change { ContainerRepository.waiting_for_cleanup.count }.from(1).to(0)
-            expect(repository.reload.cleanup_unscheduled?).to be_truthy
-          end
+          expect { subject }
+            .to not_change { ContainerRepository.waiting_for_cleanup.count }
+            .and not_change { repository.reload.expiration_policy_cleanup_status }
         end
       end
     end
 
-    context 'with loopless enabled' do
+    context 'with repository in cleanup unscheduled state' do
       before do
-        stub_feature_flags(container_registry_expiration_policies_loopless: true)
+        policy.update_column(:next_run_at, 5.minutes.ago)
       end
 
-      context 'with repository in cleanup unscheduled state' do
-        before do
-          policy.update_column(:next_run_at, 5.minutes.ago)
-        end
+      it_behaves_like 'handling all repository conditions'
+    end
 
-        it_behaves_like 'handling all repository conditions'
+    context 'with repository in cleanup unfinished state' do
+      before do
+        repository.cleanup_unfinished!
       end
 
-      context 'with repository in cleanup unfinished state' do
-        before do
-          repository.cleanup_unfinished!
-        end
+      it_behaves_like 'handling all repository conditions'
+    end
 
-        it_behaves_like 'handling all repository conditions'
-      end
+    context 'container repository selection' do
+      where(:repository_cleanup_status, :repository_policy_status, :other_repository_cleanup_status, :other_repository_policy_status, :expected_selected_repository) do
+        :unscheduled | :disabled     | :unscheduled | :disabled     | :none
+        :unscheduled | :disabled     | :unscheduled | :runnable     | :other_repository
+        :unscheduled | :disabled     | :unscheduled | :not_runnable | :none
 
-      context 'container repository selection' do
-        where(:repository_cleanup_status, :repository_policy_status, :other_repository_cleanup_status, :other_repository_policy_status, :expected_selected_repository) do
-          :unscheduled | :disabled     | :unscheduled | :disabled     | :none
-          :unscheduled | :disabled     | :unscheduled | :runnable     | :other_repository
-          :unscheduled | :disabled     | :unscheduled | :not_runnable | :none
+        :unscheduled | :disabled     | :scheduled   | :disabled     | :none
+        :unscheduled | :disabled     | :scheduled   | :runnable     | :other_repository
+        :unscheduled | :disabled     | :scheduled   | :not_runnable | :none
 
-          :unscheduled | :disabled     | :scheduled   | :disabled     | :none
-          :unscheduled | :disabled     | :scheduled   | :runnable     | :other_repository
-          :unscheduled | :disabled     | :scheduled   | :not_runnable | :none
+        :unscheduled | :disabled     | :unfinished  | :disabled     | :none
+        :unscheduled | :disabled     | :unfinished  | :runnable     | :other_repository
+        :unscheduled | :disabled     | :unfinished  | :not_runnable | :other_repository
 
-          :unscheduled | :disabled     | :unfinished  | :disabled     | :none
-          :unscheduled | :disabled     | :unfinished  | :runnable     | :other_repository
-          :unscheduled | :disabled     | :unfinished  | :not_runnable | :other_repository
+        :unscheduled | :disabled     | :ongoing     | :disabled     | :none
+        :unscheduled | :disabled     | :ongoing     | :runnable     | :none
+        :unscheduled | :disabled     | :ongoing     | :not_runnable | :none
 
-          :unscheduled | :disabled     | :ongoing     | :disabled     | :none
-          :unscheduled | :disabled     | :ongoing     | :runnable     | :none
-          :unscheduled | :disabled     | :ongoing     | :not_runnable | :none
+        :unscheduled | :runnable     | :unscheduled | :disabled     | :repository
+        :unscheduled | :runnable     | :unscheduled | :runnable     | :repository
+        :unscheduled | :runnable     | :unscheduled | :not_runnable | :repository
 
-          :unscheduled | :runnable     | :unscheduled | :disabled     | :repository
-          :unscheduled | :runnable     | :unscheduled | :runnable     | :repository
-          :unscheduled | :runnable     | :unscheduled | :not_runnable | :repository
+        :unscheduled | :runnable     | :scheduled   | :disabled     | :repository
+        :unscheduled | :runnable     | :scheduled   | :runnable     | :repository
+        :unscheduled | :runnable     | :scheduled   | :not_runnable | :repository
 
-          :unscheduled | :runnable     | :scheduled   | :disabled     | :repository
-          :unscheduled | :runnable     | :scheduled   | :runnable     | :repository
-          :unscheduled | :runnable     | :scheduled   | :not_runnable | :repository
+        :unscheduled | :runnable     | :unfinished  | :disabled     | :repository
+        :unscheduled | :runnable     | :unfinished  | :runnable     | :repository
+        :unscheduled | :runnable     | :unfinished  | :not_runnable | :repository
 
-          :unscheduled | :runnable     | :unfinished  | :disabled     | :repository
-          :unscheduled | :runnable     | :unfinished  | :runnable     | :repository
-          :unscheduled | :runnable     | :unfinished  | :not_runnable | :repository
+        :unscheduled | :runnable     | :ongoing     | :disabled     | :repository
+        :unscheduled | :runnable     | :ongoing     | :runnable     | :repository
+        :unscheduled | :runnable     | :ongoing     | :not_runnable | :repository
 
-          :unscheduled | :runnable     | :ongoing     | :disabled     | :repository
-          :unscheduled | :runnable     | :ongoing     | :runnable     | :repository
-          :unscheduled | :runnable     | :ongoing     | :not_runnable | :repository
+        :scheduled   | :disabled     | :unscheduled | :disabled     | :none
+        :scheduled   | :disabled     | :unscheduled | :runnable     | :other_repository
+        :scheduled   | :disabled     | :unscheduled | :not_runnable | :none
 
-          :scheduled   | :disabled     | :unscheduled | :disabled     | :none
-          :scheduled   | :disabled     | :unscheduled | :runnable     | :other_repository
-          :scheduled   | :disabled     | :unscheduled | :not_runnable | :none
+        :scheduled   | :disabled     | :scheduled   | :disabled     | :none
+        :scheduled   | :disabled     | :scheduled   | :runnable     | :other_repository
+        :scheduled   | :disabled     | :scheduled   | :not_runnable | :none
 
-          :scheduled   | :disabled     | :scheduled   | :disabled     | :none
-          :scheduled   | :disabled     | :scheduled   | :runnable     | :other_repository
-          :scheduled   | :disabled     | :scheduled   | :not_runnable | :none
+        :scheduled   | :disabled     | :unfinished  | :disabled     | :none
+        :scheduled   | :disabled     | :unfinished  | :runnable     | :other_repository
+        :scheduled   | :disabled     | :unfinished  | :not_runnable | :other_repository
 
-          :scheduled   | :disabled     | :unfinished  | :disabled     | :none
-          :scheduled   | :disabled     | :unfinished  | :runnable     | :other_repository
-          :scheduled   | :disabled     | :unfinished  | :not_runnable | :other_repository
+        :scheduled   | :disabled     | :ongoing     | :disabled     | :none
+        :scheduled   | :disabled     | :ongoing     | :runnable     | :none
+        :scheduled   | :disabled     | :ongoing     | :not_runnable | :none
 
-          :scheduled   | :disabled     | :ongoing     | :disabled     | :none
-          :scheduled   | :disabled     | :ongoing     | :runnable     | :none
-          :scheduled   | :disabled     | :ongoing     | :not_runnable | :none
+        :scheduled   | :runnable     | :unscheduled | :disabled     | :repository
+        :scheduled   | :runnable     | :unscheduled | :runnable     | :other_repository
+        :scheduled   | :runnable     | :unscheduled | :not_runnable | :repository
 
-          :scheduled   | :runnable     | :unscheduled | :disabled     | :repository
-          :scheduled   | :runnable     | :unscheduled | :runnable     | :other_repository
-          :scheduled   | :runnable     | :unscheduled | :not_runnable | :repository
+        :scheduled   | :runnable     | :scheduled   | :disabled     | :repository
+        :scheduled   | :runnable     | :scheduled   | :runnable     | :repository
+        :scheduled   | :runnable     | :scheduled   | :not_runnable | :repository
 
-          :scheduled   | :runnable     | :scheduled   | :disabled     | :repository
-          :scheduled   | :runnable     | :scheduled   | :runnable     | :repository
-          :scheduled   | :runnable     | :scheduled   | :not_runnable | :repository
+        :scheduled   | :runnable     | :unfinished  | :disabled     | :repository
+        :scheduled   | :runnable     | :unfinished  | :runnable     | :repository
+        :scheduled   | :runnable     | :unfinished  | :not_runnable | :repository
 
-          :scheduled   | :runnable     | :unfinished  | :disabled     | :repository
-          :scheduled   | :runnable     | :unfinished  | :runnable     | :repository
-          :scheduled   | :runnable     | :unfinished  | :not_runnable | :repository
+        :scheduled   | :runnable     | :ongoing     | :disabled     | :repository
+        :scheduled   | :runnable     | :ongoing     | :runnable     | :repository
+        :scheduled   | :runnable     | :ongoing     | :not_runnable | :repository
 
-          :scheduled   | :runnable     | :ongoing     | :disabled     | :repository
-          :scheduled   | :runnable     | :ongoing     | :runnable     | :repository
-          :scheduled   | :runnable     | :ongoing     | :not_runnable | :repository
+        :scheduled   | :not_runnable | :unscheduled | :disabled     | :none
+        :scheduled   | :not_runnable | :unscheduled | :runnable     | :other_repository
+        :scheduled   | :not_runnable | :unscheduled | :not_runnable | :none
 
-          :scheduled   | :not_runnable | :unscheduled | :disabled     | :none
-          :scheduled   | :not_runnable | :unscheduled | :runnable     | :other_repository
-          :scheduled   | :not_runnable | :unscheduled | :not_runnable | :none
+        :scheduled   | :not_runnable | :scheduled   | :disabled     | :none
+        :scheduled   | :not_runnable | :scheduled   | :runnable     | :other_repository
+        :scheduled   | :not_runnable | :scheduled   | :not_runnable | :none
 
-          :scheduled   | :not_runnable | :scheduled   | :disabled     | :none
-          :scheduled   | :not_runnable | :scheduled   | :runnable     | :other_repository
-          :scheduled   | :not_runnable | :scheduled   | :not_runnable | :none
+        :scheduled   | :not_runnable | :unfinished  | :disabled     | :none
+        :scheduled   | :not_runnable | :unfinished  | :runnable     | :other_repository
+        :scheduled   | :not_runnable | :unfinished  | :not_runnable | :other_repository
 
-          :scheduled   | :not_runnable | :unfinished  | :disabled     | :none
-          :scheduled   | :not_runnable | :unfinished  | :runnable     | :other_repository
-          :scheduled   | :not_runnable | :unfinished  | :not_runnable | :other_repository
+        :scheduled   | :not_runnable | :ongoing     | :disabled     | :none
+        :scheduled   | :not_runnable | :ongoing     | :runnable     | :none
+        :scheduled   | :not_runnable | :ongoing     | :not_runnable | :none
 
-          :scheduled   | :not_runnable | :ongoing     | :disabled     | :none
-          :scheduled   | :not_runnable | :ongoing     | :runnable     | :none
-          :scheduled   | :not_runnable | :ongoing     | :not_runnable | :none
+        :unfinished  | :disabled     | :unscheduled | :disabled     | :none
+        :unfinished  | :disabled     | :unscheduled | :runnable     | :other_repository
+        :unfinished  | :disabled     | :unscheduled | :not_runnable | :none
 
-          :unfinished  | :disabled     | :unscheduled | :disabled     | :none
-          :unfinished  | :disabled     | :unscheduled | :runnable     | :other_repository
-          :unfinished  | :disabled     | :unscheduled | :not_runnable | :none
+        :unfinished  | :disabled     | :scheduled   | :disabled     | :none
+        :unfinished  | :disabled     | :scheduled   | :runnable     | :other_repository
+        :unfinished  | :disabled     | :scheduled   | :not_runnable | :none
 
-          :unfinished  | :disabled     | :scheduled   | :disabled     | :none
-          :unfinished  | :disabled     | :scheduled   | :runnable     | :other_repository
-          :unfinished  | :disabled     | :scheduled   | :not_runnable | :none
+        :unfinished  | :disabled     | :unfinished  | :disabled     | :none
+        :unfinished  | :disabled     | :unfinished  | :runnable     | :other_repository
+        :unfinished  | :disabled     | :unfinished  | :not_runnable | :other_repository
 
-          :unfinished  | :disabled     | :unfinished  | :disabled     | :none
-          :unfinished  | :disabled     | :unfinished  | :runnable     | :other_repository
-          :unfinished  | :disabled     | :unfinished  | :not_runnable | :other_repository
+        :unfinished  | :disabled     | :ongoing     | :disabled     | :none
+        :unfinished  | :disabled     | :ongoing     | :runnable     | :none
+        :unfinished  | :disabled     | :ongoing     | :not_runnable | :none
 
-          :unfinished  | :disabled     | :ongoing     | :disabled     | :none
-          :unfinished  | :disabled     | :ongoing     | :runnable     | :none
-          :unfinished  | :disabled     | :ongoing     | :not_runnable | :none
+        :unfinished  | :runnable     | :unscheduled | :disabled     | :repository
+        :unfinished  | :runnable     | :unscheduled | :runnable     | :other_repository
+        :unfinished  | :runnable     | :unscheduled | :not_runnable | :repository
+
+        :unfinished  | :runnable     | :scheduled   | :disabled     | :repository
+        :unfinished  | :runnable     | :scheduled   | :runnable     | :other_repository
+        :unfinished  | :runnable     | :scheduled   | :not_runnable | :repository
 
-          :unfinished  | :runnable     | :unscheduled | :disabled     | :repository
-          :unfinished  | :runnable     | :unscheduled | :runnable     | :other_repository
-          :unfinished  | :runnable     | :unscheduled | :not_runnable | :repository
-
-          :unfinished  | :runnable     | :scheduled   | :disabled     | :repository
-          :unfinished  | :runnable     | :scheduled   | :runnable     | :other_repository
-          :unfinished  | :runnable     | :scheduled   | :not_runnable | :repository
+        :unfinished  | :runnable     | :unfinished  | :disabled     | :repository
+        :unfinished  | :runnable     | :unfinished  | :runnable     | :repository
+        :unfinished  | :runnable     | :unfinished  | :not_runnable | :repository
 
-          :unfinished  | :runnable     | :unfinished  | :disabled     | :repository
-          :unfinished  | :runnable     | :unfinished  | :runnable     | :repository
-          :unfinished  | :runnable     | :unfinished  | :not_runnable | :repository
+        :unfinished  | :runnable     | :ongoing     | :disabled     | :repository
+        :unfinished  | :runnable     | :ongoing     | :runnable     | :repository
+        :unfinished  | :runnable     | :ongoing     | :not_runnable | :repository
 
-          :unfinished  | :runnable     | :ongoing     | :disabled     | :repository
-          :unfinished  | :runnable     | :ongoing     | :runnable     | :repository
-          :unfinished  | :runnable     | :ongoing     | :not_runnable | :repository
+        :unfinished  | :not_runnable | :unscheduled | :disabled     | :repository
+        :unfinished  | :not_runnable | :unscheduled | :runnable     | :other_repository
+        :unfinished  | :not_runnable | :unscheduled | :not_runnable | :repository
 
-          :unfinished  | :not_runnable | :unscheduled | :disabled     | :repository
-          :unfinished  | :not_runnable | :unscheduled | :runnable     | :other_repository
-          :unfinished  | :not_runnable | :unscheduled | :not_runnable | :repository
+        :unfinished  | :not_runnable | :scheduled   | :disabled     | :repository
+        :unfinished  | :not_runnable | :scheduled   | :runnable     | :other_repository
+        :unfinished  | :not_runnable | :scheduled   | :not_runnable | :repository
 
-          :unfinished  | :not_runnable | :scheduled   | :disabled     | :repository
-          :unfinished  | :not_runnable | :scheduled   | :runnable     | :other_repository
-          :unfinished  | :not_runnable | :scheduled   | :not_runnable | :repository
+        :unfinished  | :not_runnable | :unfinished  | :disabled     | :repository
+        :unfinished  | :not_runnable | :unfinished  | :runnable     | :repository
+        :unfinished  | :not_runnable | :unfinished  | :not_runnable | :repository
 
-          :unfinished  | :not_runnable | :unfinished  | :disabled     | :repository
-          :unfinished  | :not_runnable | :unfinished  | :runnable     | :repository
-          :unfinished  | :not_runnable | :unfinished  | :not_runnable | :repository
+        :unfinished  | :not_runnable | :ongoing     | :disabled     | :repository
+        :unfinished  | :not_runnable | :ongoing     | :runnable     | :repository
+        :unfinished  | :not_runnable | :ongoing     | :not_runnable | :repository
 
-          :unfinished  | :not_runnable | :ongoing     | :disabled     | :repository
-          :unfinished  | :not_runnable | :ongoing     | :runnable     | :repository
-          :unfinished  | :not_runnable | :ongoing     | :not_runnable | :repository
+        :ongoing     | :disabled     | :unscheduled | :disabled     | :none
+        :ongoing     | :disabled     | :unscheduled | :runnable     | :other_repository
+        :ongoing     | :disabled     | :unscheduled | :not_runnable | :none
 
-          :ongoing     | :disabled     | :unscheduled | :disabled     | :none
-          :ongoing     | :disabled     | :unscheduled | :runnable     | :other_repository
-          :ongoing     | :disabled     | :unscheduled | :not_runnable | :none
+        :ongoing     | :disabled     | :scheduled   | :disabled     | :none
+        :ongoing     | :disabled     | :scheduled   | :runnable     | :other_repository
+        :ongoing     | :disabled     | :scheduled   | :not_runnable | :none
 
-          :ongoing     | :disabled     | :scheduled   | :disabled     | :none
-          :ongoing     | :disabled     | :scheduled   | :runnable     | :other_repository
-          :ongoing     | :disabled     | :scheduled   | :not_runnable | :none
+        :ongoing     | :disabled     | :unfinished  | :disabled     | :none
+        :ongoing     | :disabled     | :unfinished  | :runnable     | :other_repository
+        :ongoing     | :disabled     | :unfinished  | :not_runnable | :other_repository
 
-          :ongoing     | :disabled     | :unfinished  | :disabled     | :none
-          :ongoing     | :disabled     | :unfinished  | :runnable     | :other_repository
-          :ongoing     | :disabled     | :unfinished  | :not_runnable | :other_repository
+        :ongoing     | :disabled     | :ongoing     | :disabled     | :none
+        :ongoing     | :disabled     | :ongoing     | :runnable     | :none
+        :ongoing     | :disabled     | :ongoing     | :not_runnable | :none
 
-          :ongoing     | :disabled     | :ongoing     | :disabled     | :none
-          :ongoing     | :disabled     | :ongoing     | :runnable     | :none
-          :ongoing     | :disabled     | :ongoing     | :not_runnable | :none
+        :ongoing     | :runnable     | :unscheduled | :disabled     | :none
+        :ongoing     | :runnable     | :unscheduled | :runnable     | :other_repository
+        :ongoing     | :runnable     | :unscheduled | :not_runnable | :none
 
-          :ongoing     | :runnable     | :unscheduled | :disabled     | :none
-          :ongoing     | :runnable     | :unscheduled | :runnable     | :other_repository
-          :ongoing     | :runnable     | :unscheduled | :not_runnable | :none
+        :ongoing     | :runnable     | :scheduled   | :disabled     | :none
+        :ongoing     | :runnable     | :scheduled   | :runnable     | :other_repository
+        :ongoing     | :runnable     | :scheduled   | :not_runnable | :none
 
-          :ongoing     | :runnable     | :scheduled   | :disabled     | :none
-          :ongoing     | :runnable     | :scheduled   | :runnable     | :other_repository
-          :ongoing     | :runnable     | :scheduled   | :not_runnable | :none
+        :ongoing     | :runnable     | :unfinished  | :disabled     | :none
+        :ongoing     | :runnable     | :unfinished  | :runnable     | :other_repository
+        :ongoing     | :runnable     | :unfinished  | :not_runnable | :other_repository
 
-          :ongoing     | :runnable     | :unfinished  | :disabled     | :none
-          :ongoing     | :runnable     | :unfinished  | :runnable     | :other_repository
-          :ongoing     | :runnable     | :unfinished  | :not_runnable | :other_repository
+        :ongoing     | :runnable     | :ongoing     | :disabled     | :none
+        :ongoing     | :runnable     | :ongoing     | :runnable     | :none
+        :ongoing     | :runnable     | :ongoing     | :not_runnable | :none
 
-          :ongoing     | :runnable     | :ongoing     | :disabled     | :none
-          :ongoing     | :runnable     | :ongoing     | :runnable     | :none
-          :ongoing     | :runnable     | :ongoing     | :not_runnable | :none
+        :ongoing     | :not_runnable | :unscheduled | :disabled     | :none
+        :ongoing     | :not_runnable | :unscheduled | :runnable     | :other_repository
+        :ongoing     | :not_runnable | :unscheduled | :not_runnable | :none
 
-          :ongoing     | :not_runnable | :unscheduled | :disabled     | :none
-          :ongoing     | :not_runnable | :unscheduled | :runnable     | :other_repository
-          :ongoing     | :not_runnable | :unscheduled | :not_runnable | :none
+        :ongoing     | :not_runnable | :scheduled   | :disabled     | :none
+        :ongoing     | :not_runnable | :scheduled   | :runnable     | :other_repository
+        :ongoing     | :not_runnable | :scheduled   | :not_runnable | :none
 
-          :ongoing     | :not_runnable | :scheduled   | :disabled     | :none
-          :ongoing     | :not_runnable | :scheduled   | :runnable     | :other_repository
-          :ongoing     | :not_runnable | :scheduled   | :not_runnable | :none
+        :ongoing     | :not_runnable | :unfinished  | :disabled     | :none
+        :ongoing     | :not_runnable | :unfinished  | :runnable     | :other_repository
+        :ongoing     | :not_runnable | :unfinished  | :not_runnable | :other_repository
 
-          :ongoing     | :not_runnable | :unfinished  | :disabled     | :none
-          :ongoing     | :not_runnable | :unfinished  | :runnable     | :other_repository
-          :ongoing     | :not_runnable | :unfinished  | :not_runnable | :other_repository
+        :ongoing     | :not_runnable | :ongoing     | :disabled     | :none
+        :ongoing     | :not_runnable | :ongoing     | :runnable     | :none
+        :ongoing     | :not_runnable | :ongoing     | :not_runnable | :none
+      end
 
-          :ongoing     | :not_runnable | :ongoing     | :disabled     | :none
-          :ongoing     | :not_runnable | :ongoing     | :runnable     | :none
-          :ongoing     | :not_runnable | :ongoing     | :not_runnable | :none
+      with_them do
+        before do
+          update_container_repository(repository, repository_cleanup_status, repository_policy_status)
+          update_container_repository(other_repository, other_repository_cleanup_status, other_repository_policy_status)
         end
 
-        with_them do
-          before do
-            update_container_repository(repository, repository_cleanup_status, repository_policy_status)
-            update_container_repository(other_repository, other_repository_cleanup_status, other_repository_policy_status)
-          end
-
-          subject { worker.send(:container_repository) }
-
-          if params[:expected_selected_repository] == :none
-            it 'does not select any repository' do
-              expect(subject).to eq(nil)
-            end
-          else
-            it 'does select a repository' do
-              selected_repository = expected_selected_repository == :repository ? repository : other_repository
+        subject { worker.send(:container_repository) }
 
-              expect(subject).to eq(selected_repository)
-            end
+        if params[:expected_selected_repository] == :none
+          it 'does not select any repository' do
+            expect(subject).to eq(nil)
           end
+        else
+          it 'does select a repository' do
+            selected_repository = expected_selected_repository == :repository ? repository : other_repository
 
-          def update_container_repository(container_repository, cleanup_status, policy_status)
-            container_repository.update_column(:expiration_policy_cleanup_status, "cleanup_#{cleanup_status}")
-
-            policy = container_repository.project.container_expiration_policy
-
-            case policy_status
-            when :disabled
-              policy.update!(enabled: false)
-            when :runnable
-              policy.update!(enabled: true)
-              policy.update_column(:next_run_at, 5.minutes.ago)
-            when :not_runnable
-              policy.update!(enabled: true)
-              policy.update_column(:next_run_at, 5.minutes.from_now)
-            end
+            expect(subject).to eq(selected_repository)
           end
         end
-      end
 
-      context 'with another repository in cleanup unfinished state' do
-        let_it_be(:another_repository) { create(:container_repository, :cleanup_unfinished) }
+        def update_container_repository(container_repository, cleanup_status, policy_status)
+          container_repository.update_column(:expiration_policy_cleanup_status, "cleanup_#{cleanup_status}")
 
-        before do
-          policy.update_column(:next_run_at, 5.minutes.ago)
-        end
+          policy = container_repository.project.container_expiration_policy
 
-        it 'process the cleanup scheduled repository first' do
-          service_response = cleanup_service_response(repository: repository)
-          expect(ContainerExpirationPolicies::CleanupService)
-            .to receive(:new).with(repository).and_return(double(execute: service_response))
-          expect_log_extra_metadata(service_response: service_response)
-
-          subject
+          case policy_status
+          when :disabled
+            policy.update!(enabled: false)
+          when :runnable
+            policy.update!(enabled: true)
+            policy.update_column(:next_run_at, 5.minutes.ago)
+          when :not_runnable
+            policy.update!(enabled: true)
+            policy.update_column(:next_run_at, 5.minutes.from_now)
+          end
         end
       end
     end
 
-    context 'with loopless disabled' do
-      before do
-        stub_feature_flags(container_registry_expiration_policies_loopless: false)
-      end
-
-      context 'with repository in cleanup scheduled state' do
-        it_behaves_like 'handling all repository conditions'
-      end
-
-      context 'with repository in cleanup unfinished state' do
-        before do
-          repository.cleanup_unfinished!
-        end
-
-        it_behaves_like 'handling all repository conditions'
-      end
-
-      context 'with another repository in cleanup unfinished state' do
-        let_it_be(:another_repository) { create(:container_repository, :cleanup_unfinished) }
-
-        it 'process the cleanup scheduled repository first' do
-          service_response = cleanup_service_response(repository: repository)
-          expect(ContainerExpirationPolicies::CleanupService)
-            .to receive(:new).with(repository).and_return(double(execute: service_response))
-          expect_log_extra_metadata(service_response: service_response)
-
-          subject
-        end
-      end
-
-      context 'with multiple repositories in cleanup unfinished state' do
-        let_it_be(:repository2) { create(:container_repository, :cleanup_unfinished, expiration_policy_started_at: 20.minutes.ago) }
-        let_it_be(:repository3) { create(:container_repository, :cleanup_unfinished, expiration_policy_started_at: 10.minutes.ago) }
-
-        before do
-          repository.update!(expiration_policy_cleanup_status: :cleanup_unfinished, expiration_policy_started_at: 30.minutes.ago)
-        end
+    context 'with another repository in cleanup unfinished state' do
+      let_it_be(:another_repository) { create(:container_repository, :cleanup_unfinished) }
 
-        it 'process the repository with the oldest expiration_policy_started_at' do
-          service_response = cleanup_service_response(repository: repository)
-          expect(ContainerExpirationPolicies::CleanupService)
-            .to receive(:new).with(repository).and_return(double(execute: service_response))
-          expect_log_extra_metadata(service_response: service_response)
-
-          subject
-        end
-      end
-
-      context 'with repository in cleanup ongoing state' do
-        before do
-          repository.cleanup_ongoing!
-        end
-
-        it 'does not process it' do
-          expect(Projects::ContainerRepository::CleanupTagsService).not_to receive(:new)
-
-          expect { subject }.not_to change { ContainerRepository.waiting_for_cleanup.count }
-          expect(repository.cleanup_ongoing?).to be_truthy
-        end
-      end
-
-      context 'with no repository in any cleanup state' do
-        before do
-          repository.cleanup_unscheduled!
-        end
-
-        it 'does not process it' do
-          expect(Projects::ContainerRepository::CleanupTagsService).not_to receive(:new)
-
-          expect { subject }.not_to change { ContainerRepository.waiting_for_cleanup.count }
-          expect(repository.cleanup_unscheduled?).to be_truthy
-        end
-      end
-
-      context 'with no container repository waiting' do
-        before do
-          repository.destroy!
-        end
-
-        it 'does not execute the cleanup tags service' do
-          expect(Projects::ContainerRepository::CleanupTagsService).not_to receive(:new)
-
-          expect { subject }.not_to change { ContainerRepository.waiting_for_cleanup.count }
-        end
+      before do
+        policy.update_column(:next_run_at, 5.minutes.ago)
       end
 
-      context 'with feature flag disabled' do
-        before do
-          stub_feature_flags(container_registry_expiration_policies_throttling: false)
-        end
-
-        it 'is a no-op' do
-          expect(Projects::ContainerRepository::CleanupTagsService).not_to receive(:new)
+      it 'process the cleanup scheduled repository first' do
+        service_response = cleanup_service_response(repository: repository)
+        expect(ContainerExpirationPolicies::CleanupService)
+          .to receive(:new).with(repository).and_return(double(execute: service_response))
+        expect_log_extra_metadata(service_response: service_response)
 
-          expect { subject }.not_to change { ContainerRepository.waiting_for_cleanup.count }
-        end
+        subject
       end
     end
 
@@ -509,69 +399,53 @@ RSpec.describe ContainerExpirationPolicies::CleanupContainerRepositoryWorker do
   end
 
   describe '#remaining_work_count' do
-    subject { worker.remaining_work_count }
+    let_it_be(:disabled_repository) { create(:container_repository, :cleanup_scheduled) }
 
-    shared_examples 'handling all conditions' do
-      context 'with container repositories waiting for cleanup' do
-        let_it_be(:unfinished_repositories) { create_list(:container_repository, 2, :cleanup_unfinished) }
+    let(:capacity) { 10 }
 
-        it { is_expected.to eq(3) }
+    subject { worker.remaining_work_count }
 
-        it 'logs the work count' do
-          expect_log_info(
-            cleanup_scheduled_count: 1,
-            cleanup_unfinished_count: 2,
-            cleanup_total_count: 3
-          )
+    before do
+      stub_application_setting(container_registry_expiration_policies_worker_capacity: capacity)
 
-          subject
-        end
-      end
+      ContainerExpirationPolicy.update_all(enabled: true)
+      repository.project.container_expiration_policy.update_column(:next_run_at, 5.minutes.ago)
+      disabled_repository.project.container_expiration_policy.update_column(:enabled, false)
+    end
 
-      context 'with no container repositories waiting for cleanup' do
-        before do
-          repository.cleanup_ongoing!
-          policy.update_column(:next_run_at, 5.minutes.from_now)
-        end
+    context 'with container repositories waiting for cleanup' do
+      let_it_be(:unfinished_repositories) { create_list(:container_repository, 2, :cleanup_unfinished) }
 
-        it { is_expected.to eq(0) }
+      it { is_expected.to eq(3) }
 
-        it 'logs 0 work count' do
-          expect_log_info(
-            cleanup_scheduled_count: 0,
-            cleanup_unfinished_count: 0,
-            cleanup_total_count: 0
-          )
+      it 'logs the work count' do
+        expect_log_info(
+          cleanup_scheduled_count: 1,
+          cleanup_unfinished_count: 2,
+          cleanup_total_count: 3
+        )
 
-          subject
-        end
+        subject
       end
     end
 
-    context 'with loopless enabled' do
-      let_it_be(:disabled_repository) { create(:container_repository, :cleanup_scheduled) }
-
-      let(:capacity) { 10 }
-
+    context 'with no container repositories waiting for cleanup' do
       before do
-        stub_feature_flags(container_registry_expiration_policies_loopless: true)
-        stub_application_setting(container_registry_expiration_policies_worker_capacity: capacity)
-
-        # loopless mode is more accurate that non loopless: policies need to be enabled
-        ContainerExpirationPolicy.update_all(enabled: true)
-        repository.project.container_expiration_policy.update_column(:next_run_at, 5.minutes.ago)
-        disabled_repository.project.container_expiration_policy.update_column(:enabled, false)
+        repository.cleanup_ongoing!
+        policy.update_column(:next_run_at, 5.minutes.from_now)
       end
 
-      it_behaves_like 'handling all conditions'
-    end
+      it { is_expected.to eq(0) }
 
-    context 'with loopless disabled' do
-      before do
-        stub_feature_flags(container_registry_expiration_policies_loopless: false)
-      end
+      it 'logs 0 work count' do
+        expect_log_info(
+          cleanup_scheduled_count: 0,
+          cleanup_unfinished_count: 0,
+          cleanup_total_count: 0
+        )
 
-      it_behaves_like 'handling all conditions'
+        subject
+      end
     end
   end
 
@@ -599,8 +473,4 @@ RSpec.describe ContainerExpirationPolicies::CleanupContainerRepositoryWorker do
     expect(worker.logger)
       .to receive(:info).with(worker.structured_payload(structure))
   end
-
-  def loopless_enabled?
-    Feature.enabled?(:container_registry_expiration_policies_loopless)
-  end
 end
diff --git a/spec/workers/container_expiration_policy_worker_spec.rb b/spec/workers/container_expiration_policy_worker_spec.rb
index e8f9a972f10..8562935b0b5 100644
--- a/spec/workers/container_expiration_policy_worker_spec.rb
+++ b/spec/workers/container_expiration_policy_worker_spec.rb
@@ -34,101 +34,18 @@ RSpec.describe ContainerExpirationPolicyWorker do
       end
     end
 
-    context 'With no container expiration policies' do
-      context 'with loopless disabled' do
-        before do
-          stub_feature_flags(container_registry_expiration_policies_loopless: false)
-        end
-
-        it 'does not execute any policies' do
-          expect(ContainerRepository).not_to receive(:for_project_id)
-
-          expect { subject }.not_to change { ContainerRepository.cleanup_scheduled.count }
-        end
-      end
-    end
-
     context 'with throttling enabled' do
       before do
         stub_feature_flags(container_registry_expiration_policies_throttling: true)
       end
 
-      context 'with loopless disabled' do
-        before do
-          stub_feature_flags(container_registry_expiration_policies_loopless: false)
-        end
-
-        context 'with container expiration policies' do
-          let_it_be(:container_expiration_policy) { create(:container_expiration_policy, :runnable) }
-          let_it_be(:container_repository) { create(:container_repository, project: container_expiration_policy.project) }
-
-          before do
-            expect(worker).to receive(:with_runnable_policy).and_call_original
-          end
-
-          context 'with a valid container expiration policy' do
-            it 'schedules the next run' do
-              expect { subject }.to change { container_expiration_policy.reload.next_run_at }
-            end
-
-            it 'marks the container repository as scheduled for cleanup' do
-              expect { subject }.to change { container_repository.reload.cleanup_scheduled? }.from(false).to(true)
-              expect(ContainerRepository.cleanup_scheduled.count).to eq(1)
-            end
-
-            it 'calls the limited capacity worker' do
-              expect(ContainerExpirationPolicies::CleanupContainerRepositoryWorker).to receive(:perform_with_capacity)
-
-              subject
-            end
-          end
-
-          context 'with a disabled container expiration policy' do
-            before do
-              container_expiration_policy.disable!
-            end
+      it 'calls the limited capacity worker' do
+        expect(ContainerExpirationPolicies::CleanupContainerRepositoryWorker).to receive(:perform_with_capacity)
 
-            it 'does not run the policy' do
-              expect(ContainerRepository).not_to receive(:for_project_id)
-
-              expect { subject }.not_to change { ContainerRepository.cleanup_scheduled.count }
-            end
-          end
-
-          context 'with an invalid container expiration policy' do
-            let(:user) { container_expiration_policy.project.owner }
-
-            before do
-              container_expiration_policy.update_column(:name_regex, '*production')
-            end
-
-            it 'disables the policy and tracks an error' do
-              expect(ContainerRepository).not_to receive(:for_project_id)
-              expect(Gitlab::ErrorTracking).to receive(:log_exception).with(instance_of(described_class::InvalidPolicyError), container_expiration_policy_id: container_expiration_policy.id)
-
-              expect { subject }.to change { container_expiration_policy.reload.enabled }.from(true).to(false)
-              expect(ContainerRepository.cleanup_scheduled).to be_empty
-            end
-          end
-        end
-
-        it_behaves_like 'handling a taken exclusive lease'
+        subject
       end
 
-      context 'with loopless enabled' do
-        before do
-          stub_feature_flags(container_registry_expiration_policies_loopless: true)
-          expect(worker).not_to receive(:with_runnable_policy)
-        end
-
-        it 'calls the limited capacity worker' do
-          expect(ContainerExpirationPolicies::CleanupContainerRepositoryWorker).to receive(:perform_with_capacity)
-
-          subject
-        end
-
-        it_behaves_like 'handling a taken exclusive lease'
-      end
+      it_behaves_like 'handling a taken exclusive lease'
     end
 
     context 'with throttling disabled' do
@@ -193,5 +110,18 @@ RSpec.describe ContainerExpirationPolicyWorker do
         end
       end
     end
+
+    context 'process stale ongoing cleanups' do
+      let_it_be(:stuck_cleanup) { create(:container_repository, :cleanup_ongoing, expiration_policy_started_at: 1.day.ago) }
+      let_it_be(:container_repository) { create(:container_repository, :cleanup_scheduled) }
+      let_it_be(:container_repository) { create(:container_repository, :cleanup_unfinished) }
+
+      it 'set them as unfinished' do
+        expect { subject }
+          .to change { ContainerRepository.cleanup_ongoing.count }.from(1).to(0)
+          .and change { ContainerRepository.cleanup_unfinished.count }.from(1).to(2)
+        expect(stuck_cleanup.reload).to be_cleanup_unfinished
+      end
+    end
   end
 end
diff --git a/spec/workers/deployments/execute_hooks_worker_spec.rb b/spec/workers/deployments/execute_hooks_worker_spec.rb
deleted file mode 100644
index fb1dc8cf290..00000000000
--- a/spec/workers/deployments/execute_hooks_worker_spec.rb
+++ /dev/null
@@ -1,51 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe Deployments::ExecuteHooksWorker do
-  let(:worker) { described_class.new }
-
-  describe '#perform' do
-    before do
-      allow(ProjectServiceWorker).to receive(:perform_async)
-    end
-
-    it 'executes project services for deployment_hooks' do
-      deployment = create(:deployment, :running)
-      project = deployment.project
-      service = create(:service, type: 'SlackService', project: project, deployment_events: true, active: true)
-
-      expect(ProjectServiceWorker).to receive(:perform_async).with(service.id, an_instance_of(Hash))
-
-      worker.perform(deployment.id)
-    end
-
-    it 'does not execute an inactive service' do
-      deployment = create(:deployment, :running)
-      project = deployment.project
-      create(:service, type: 'SlackService', project: project, deployment_events: true, active: false)
-
-      expect(ProjectServiceWorker).not_to receive(:perform_async)
-
-      worker.perform(deployment.id)
-    end
-
-    it 'does not execute if a deployment does not exist' do
-      expect(ProjectServiceWorker).not_to receive(:perform_async)
-
-      worker.perform(non_existing_record_id)
-    end
-
-    it 'execute webhooks' do
-      deployment = create(:deployment, :running)
-      project = deployment.project
-      web_hook = create(:project_hook, deployment_events: true, project: project)
-
-      expect_next_instance_of(WebHookService, web_hook, an_instance_of(Hash), "deployment_hooks") do |service|
-        expect(service).to receive(:async_execute)
-      end
-
-      worker.perform(deployment.id)
-    end
-  end
-end
diff --git a/spec/workers/every_sidekiq_worker_spec.rb b/spec/workers/every_sidekiq_worker_spec.rb
index de848e59d57..34d42addef3 100644
--- a/spec/workers/every_sidekiq_worker_spec.rb
+++ b/spec/workers/every_sidekiq_worker_spec.rb
@@ -130,6 +130,7 @@ RSpec.describe 'Every Sidekiq worker' do
         'AuthorizedProjectUpdate::ProjectGroupLinkCreateWorker' => 3,
         'AuthorizedProjectUpdate::UserRefreshOverUserRangeWorker' => 3,
         'AuthorizedProjectUpdate::UserRefreshWithLowUrgencyWorker' => 3,
+        'AuthorizedProjectUpdate::UserRefreshFromReplicaWorker' => 3,
         'AuthorizedProjectsWorker' => 3,
         'AutoDevops::DisableWorker' => 3,
         'AutoMergeProcessWorker' => 3,
@@ -165,6 +166,7 @@ RSpec.describe 'Every Sidekiq worker' do
         'Ci::ResourceGroups::AssignResourceFromResourceGroupWorker' => 3,
         'Ci::TestFailureHistoryWorker' => 3,
         'Ci::TriggerDownstreamSubscriptionsWorker' => 3,
+        'Ci::SyncReportsToReportApprovalRulesWorker' => 3,
         'CleanupContainerRepositoryWorker' => 3,
         'ClusterConfigureIstioWorker' => 3,
         'ClusterInstallAppWorker' => 3,
@@ -195,7 +197,6 @@ RSpec.describe 'Every Sidekiq worker' do
         'DeleteUserWorker' => 3,
         'Deployments::AutoRollbackWorker' => 3,
         'Deployments::DropOlderDeploymentsWorker' => 3,
-        'Deployments::ExecuteHooksWorker' => 3,
         'Deployments::FinishedWorker' => 3,
         'Deployments::ForwardDeploymentWorker' => 3,
         'Deployments::LinkMergeRequestWorker' => 3,
@@ -212,7 +213,6 @@ RSpec.describe 'Every Sidekiq worker' do
         'ElasticCommitIndexerWorker' => 2,
         'ElasticDeleteProjectWorker' => 2,
         'ElasticFullIndexWorker' => 2,
-        'ElasticIndexerWorker' => 2,
         'ElasticIndexingControlWorker' => 3,
         'ElasticNamespaceIndexerWorker' => 2,
         'ElasticNamespaceRolloutWorker' => 2,
@@ -307,8 +307,6 @@ RSpec.describe 'Every Sidekiq worker' do
         'IncidentManagement::OncallRotations::PersistAllRotationsShiftsJob' => 3,
         'IncidentManagement::OncallRotations::PersistShiftsJob' => 3,
         'IncidentManagement::PagerDuty::ProcessIncidentWorker' => 3,
-        'IncidentManagement::ProcessAlertWorker' => 3,
-        'IncidentManagement::ProcessPrometheusAlertWorker' => 3,
         'InvalidGpgSignatureUpdateWorker' => 3,
         'IrkerWorker' => 3,
         'IssuableExportCsvWorker' => 3,
@@ -374,7 +372,6 @@ RSpec.describe 'Every Sidekiq worker' do
         'PipelineMetricsWorker' => 3,
         'PipelineNotificationWorker' => 3,
         'PipelineProcessWorker' => 3,
-        'PipelineUpdateWorker' => 3,
         'PostReceive' => 3,
         'ProcessCommitWorker' => 3,
         'ProjectCacheWorker' => 3,
@@ -433,7 +430,6 @@ RSpec.describe 'Every Sidekiq worker' do
         'StoreSecurityScansWorker' => 3,
         'SyncSeatLinkRequestWorker' => 20,
         'SyncSeatLinkWorker' => 12,
-        'SyncSecurityReportsToReportApprovalRulesWorker' => 3,
         'SystemHookPushWorker' => 3,
         'TodosDestroyer::ConfidentialEpicWorker' => 3,
         'TodosDestroyer::ConfidentialIssueWorker' => 3,
diff --git a/spec/workers/expire_pipeline_cache_worker_spec.rb b/spec/workers/expire_pipeline_cache_worker_spec.rb
index de42eeeab75..6a1a95b8052 100644
--- a/spec/workers/expire_pipeline_cache_worker_spec.rb
+++ b/spec/workers/expire_pipeline_cache_worker_spec.rb
@@ -42,8 +42,15 @@ RSpec.describe ExpirePipelineCacheWorker do
       subject.perform(617748)
     end
 
-    it_behaves_like 'an idempotent worker' do
-      let(:job_args) { [pipeline.id] }
+    skip "with https://gitlab.com/gitlab-org/gitlab/-/issues/325291 resolved" do
+      it_behaves_like 'an idempotent worker' do
+        let(:job_args) { [pipeline.id] }
+      end
     end
+
+    it_behaves_like 'worker with data consistency',
+                  described_class,
+                  feature_flag: :load_balancing_for_expire_pipeline_cache_worker,
+                  data_consistency: :delayed
   end
 end
diff --git a/spec/workers/git_garbage_collect_worker_spec.rb b/spec/workers/git_garbage_collect_worker_spec.rb
deleted file mode 100644
index 3df64c35166..00000000000
--- a/spec/workers/git_garbage_collect_worker_spec.rb
+++ /dev/null
@@ -1,26 +0,0 @@
-# frozen_string_literal: true
-
-require 'fileutils'
-
-require 'spec_helper'
-
-RSpec.describe GitGarbageCollectWorker do
-  let_it_be(:project) { create(:project, :repository) }
-
-  let(:lease_uuid) { SecureRandom.uuid }
-  let(:lease_key)  { "project_housekeeping:#{project.id}" }
-  let(:task)       { :full_repack }
-  let(:params)     { [project.id, task, lease_key, lease_uuid] }
-
-  subject { described_class.new }
-
-  describe "#perform" do
-    it 'calls the Projects::GitGarbageGitGarbageCollectWorker with the same params' do
-      expect_next_instance_of(Projects::GitGarbageCollectWorker) do |instance|
-        expect(instance).to receive(:perform).with(*params)
-      end
-
-      subject.perform(*params)
-    end
-  end
-end
diff --git a/spec/workers/incident_management/process_alert_worker_spec.rb b/spec/workers/incident_management/process_alert_worker_spec.rb
deleted file mode 100644
index 7db9b191677..00000000000
--- a/spec/workers/incident_management/process_alert_worker_spec.rb
+++ /dev/null
@@ -1,88 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe IncidentManagement::ProcessAlertWorker do
-  let_it_be(:project) { create(:project) }
-  let_it_be(:settings) { create(:project_incident_management_setting, project: project, create_issue: true) }
-
-  describe '#perform' do
-    let_it_be(:started_at) { Time.now.rfc3339 }
-    let_it_be(:payload) { { 'title' => 'title', 'start_time' => started_at } }
-    let_it_be(:alert) { create(:alert_management_alert, project: project, payload: payload, started_at: started_at) }
-
-    let(:created_issue) { Issue.last! }
-
-    subject { described_class.new.perform(nil, nil, alert.id) }
-
-    before do
-      allow(Gitlab::AppLogger).to receive(:warn).and_call_original
-
-      allow(AlertManagement::CreateAlertIssueService)
-        .to receive(:new).with(alert, User.alert_bot)
-        .and_call_original
-    end
-
-    shared_examples 'creates issue successfully' do
-      it 'creates an issue' do
-        expect(AlertManagement::CreateAlertIssueService)
-          .to receive(:new).with(alert, User.alert_bot)
-
-        expect { subject }.to change { Issue.count }.by(1)
-      end
-
-      it 'updates AlertManagement::Alert#issue_id' do
-        subject
-
-        expect(alert.reload.issue_id).to eq(created_issue.id)
-      end
-
-      it 'does not write a warning to log' do
-        subject
-
-        expect(Gitlab::AppLogger).not_to have_received(:warn)
-      end
-    end
-
-    context 'with valid alert' do
-      it_behaves_like 'creates issue successfully'
-
-      context 'when alert cannot be updated' do
-        let_it_be(:alert) { create(:alert_management_alert, :with_validation_errors, project: project, payload: payload) }
-
-        it 'updates AlertManagement::Alert#issue_id' do
-          expect { subject }.not_to change { alert.reload.issue_id }
-        end
-
-        it 'logs a warning' do
-          subject
-
-          expect(Gitlab::AppLogger).to have_received(:warn).with(
-            message: 'Cannot process an Incident',
-            issue_id: created_issue.id,
-            alert_id: alert.id,
-            errors: 'Hosts hosts array is over 255 chars'
-          )
-        end
-      end
-
-      context 'prometheus alert' do
-        let_it_be(:alert) { create(:alert_management_alert, :prometheus, project: project, started_at: started_at) }
-
-        it_behaves_like 'creates issue successfully'
-      end
-    end
-
-    context 'with invalid alert' do
-      let(:invalid_alert_id) { non_existing_record_id }
-
-      subject { described_class.new.perform(nil, nil, invalid_alert_id) }
-
-      it 'does not create issues' do
-        expect(AlertManagement::CreateAlertIssueService).not_to receive(:new)
-
-        expect { subject }.not_to change { Issue.count }
-      end
-    end
-  end
-end
diff --git a/spec/workers/incident_management/process_prometheus_alert_worker_spec.rb b/spec/workers/incident_management/process_prometheus_alert_worker_spec.rb
deleted file mode 100644
index 56f07459a15..00000000000
--- a/spec/workers/incident_management/process_prometheus_alert_worker_spec.rb
+++ /dev/null
@@ -1,28 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe IncidentManagement::ProcessPrometheusAlertWorker do
-  describe '#perform' do
-    let_it_be(:project) { create(:project) }
-    let_it_be(:prometheus_alert) { create(:prometheus_alert, project: project) }
-
-    let(:payload_key) { Gitlab::AlertManagement::Payload::Prometheus.new(project: project, payload: alert_params).gitlab_fingerprint }
-    let!(:prometheus_alert_event) { create(:prometheus_alert_event, prometheus_alert: prometheus_alert, payload_key: payload_key) }
-    let!(:settings) { create(:project_incident_management_setting, project: project, create_issue: true) }
-
-    let(:alert_params) do
-      {
-        startsAt: prometheus_alert.created_at.rfc3339,
-        labels: {
-          gitlab_alert_id: prometheus_alert.prometheus_metric_id
-        }
-      }.with_indifferent_access
-    end
-
-    it 'does nothing' do
-      expect { subject.perform(project.id, alert_params) }
-        .not_to change(Issue, :count)
-    end
-  end
-end
diff --git a/spec/workers/issue_placement_worker_spec.rb b/spec/workers/issue_placement_worker_spec.rb
index e0c17bfadee..780790dbb1b 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, project.id)
+        expect(IssueRebalancingWorker).to receive(:perform_async).with(nil, nil, project.group.id)
 
         run_worker
       end
@@ -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, project.id)
+        expect(IssueRebalancingWorker).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/issue_rebalancing_worker_spec.rb b/spec/workers/issue_rebalancing_worker_spec.rb
index e5c6ac3f854..b6e9429d78e 100644
--- a/spec/workers/issue_rebalancing_worker_spec.rb
+++ b/spec/workers/issue_rebalancing_worker_spec.rb
@@ -20,34 +20,83 @@ RSpec.describe IssueRebalancingWorker do
       end
     end
 
-    it 'runs an instance of IssueRebalancingService' do
-      service = double(execute: nil)
-      expect(IssueRebalancingService).to receive(:new).with(issue).and_return(service)
+    shared_examples 'running the worker' do
+      it 'runs an instance of IssueRebalancingService' do
+        service = double(execute: nil)
+        service_param = arguments.second.present? ? kind_of(Project.id_in([project]).class) : kind_of(group&.all_projects.class)
 
-      described_class.new.perform(nil, issue.project_id)
+        expect(IssueRebalancingService).to receive(:new).with(service_param).and_return(service)
+
+        described_class.new.perform(*arguments)
+      end
+
+      it 'anticipates there being too many issues' 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(IssueRebalancingService::TooManyIssues)
+        expect(IssueRebalancingService).to receive(:new).with(service_param).and_return(service)
+        expect(Gitlab::ErrorTracking).to receive(:log_exception).with(IssueRebalancingService::TooManyIssues, 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(IssueRebalancingService).not_to receive(:new)
+        expect(Gitlab::ErrorTracking).not_to receive(:log_exception)
+
+        described_class.new.perform # all arguments are nil
+      end
     end
 
-    it 'anticipates the inability to find the issue' do
-      expect(Gitlab::ErrorTracking).to receive(:log_exception).with(ActiveRecord::RecordNotFound, include(project_id: -1))
-      expect(IssueRebalancingService).not_to receive(:new)
+    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(IssueRebalancingService).not_to receive(:new)
 
-      described_class.new.perform(nil, -1)
+        described_class.new.perform(*arguments)
+      end
     end
 
-    it 'anticipates there being too many issues' do
-      service = double
-      allow(service).to receive(:execute) { raise IssueRebalancingService::TooManyIssues }
-      expect(IssueRebalancingService).to receive(:new).with(issue).and_return(service)
-      expect(Gitlab::ErrorTracking).to receive(:log_exception).with(IssueRebalancingService::TooManyIssues, include(project_id: issue.project_id))
+    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
 
-      described_class.new.perform(nil, issue.project_id)
+      include_examples 'an idempotent worker' do
+        let(:job_args) { [nil, -1] }
+      end
     end
 
-    it 'takes no action if the value is nil' do
-      expect(IssueRebalancingService).not_to receive(:new)
-      expect(Gitlab::ErrorTracking).not_to receive(:log_exception)
+    context 'with root_namespace param' do
+      it_behaves_like 'running the worker' do
+        let(:arguments) { [nil, nil, group.id] }
+      end
 
-      described_class.new.perform(nil, nil)
+      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/merge_requests/assignees_change_worker_spec.rb b/spec/workers/merge_requests/assignees_change_worker_spec.rb
deleted file mode 100644
index 33478daf8d3..00000000000
--- a/spec/workers/merge_requests/assignees_change_worker_spec.rb
+++ /dev/null
@@ -1,59 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe MergeRequests::AssigneesChangeWorker do
-  include AfterNextHelpers
-
-  let_it_be(:merge_request) { create(:merge_request) }
-  let_it_be(:user) { create(:user) }
-  let_it_be(:old_assignees) { create_list(:user, 3) }
-
-  let(:user_ids) { old_assignees.map(&:id).to_a }
-  let(:worker) { described_class.new }
-
-  it_behaves_like 'an idempotent worker' do
-    let(:job_args) { [merge_request.id, user.id, user_ids] }
-  end
-
-  describe '#perform' do
-    context 'with a non-existing merge request' do
-      it 'does nothing' do
-        expect(::MergeRequests::HandleAssigneesChangeService).not_to receive(:new)
-
-        worker.perform(non_existing_record_id, user.id, user_ids)
-      end
-    end
-
-    context 'with a non-existing user' do
-      it 'does nothing' do
-        expect(::MergeRequests::HandleAssigneesChangeService).not_to receive(:new)
-
-        worker.perform(merge_request.id, non_existing_record_id, user_ids)
-      end
-    end
-
-    context 'when there are no changes' do
-      it 'does nothing' do
-        expect(::MergeRequests::HandleAssigneesChangeService).not_to receive(:new)
-
-        worker.perform(merge_request.id, user.id, merge_request.assignee_ids)
-      end
-    end
-
-    context 'when the old users cannot be found' do
-      it 'does nothing' do
-        expect(::MergeRequests::HandleAssigneesChangeService).not_to receive(:new)
-
-        worker.perform(merge_request.id, user.id, [non_existing_record_id])
-      end
-    end
-
-    it 'gets MergeRequests::UpdateAssigneesService to handle the changes' do
-      expect_next(::MergeRequests::HandleAssigneesChangeService)
-        .to receive(:execute).with(merge_request, match_array(old_assignees), execute_hooks: true)
-
-      worker.perform(merge_request.id, user.id, user_ids)
-    end
-  end
-end
diff --git a/spec/workers/packages/debian/generate_distribution_worker_spec.rb b/spec/workers/packages/debian/generate_distribution_worker_spec.rb
new file mode 100644
index 00000000000..a8751ccceae
--- /dev/null
+++ b/spec/workers/packages/debian/generate_distribution_worker_spec.rb
@@ -0,0 +1,62 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Packages::Debian::GenerateDistributionWorker, type: :worker do
+  describe '#perform' do
+    let(:container_type) { distribution.container_type }
+    let(:distribution_id) { distribution.id }
+
+    subject { described_class.new.perform(container_type, distribution_id) }
+
+    include_context 'with published Debian package'
+
+    [:project, :group].each do |container_type|
+      context "for #{container_type}" do
+        include_context 'with Debian distribution', container_type
+
+        context 'with mocked service' do
+          it 'calls GenerateDistributionService' do
+            expect(Gitlab::ErrorTracking).not_to receive(:log_exception)
+            expect_next_instance_of(::Packages::Debian::GenerateDistributionService) do |service|
+              expect(service).to receive(:execute)
+                .with(no_args)
+            end
+
+            subject
+          end
+        end
+
+        context 'with non existing distribution id' do
+          let(:distribution_id) { non_existing_record_id }
+
+          it 'returns early without error' do
+            expect(Gitlab::ErrorTracking).not_to receive(:log_exception)
+            expect(::Packages::Debian::GenerateDistributionService).not_to receive(:new)
+
+            subject
+          end
+        end
+
+        context 'with nil distribution id' do
+          let(:distribution_id) { nil }
+
+          it 'returns early without error' do
+            expect(Gitlab::ErrorTracking).not_to receive(:log_exception)
+            expect(::Packages::Debian::GenerateDistributionService).not_to receive(:new)
+
+            subject
+          end
+        end
+
+        context 'with valid parameters' do
+          it_behaves_like 'an idempotent worker' do
+            let(:job_args) { [container_type, distribution_id] }
+
+            it_behaves_like 'Generate Debian Distribution and component files'
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/workers/pipeline_hooks_worker_spec.rb b/spec/workers/pipeline_hooks_worker_spec.rb
index 7c75cdc8823..5957b355c8e 100644
--- a/spec/workers/pipeline_hooks_worker_spec.rb
+++ b/spec/workers/pipeline_hooks_worker_spec.rb
@@ -22,4 +22,9 @@ RSpec.describe PipelineHooksWorker do
       end
     end
   end
+
+  it_behaves_like 'worker with data consistency',
+                  described_class,
+                  feature_flag: :load_balancing_for_pipeline_hooks_worker,
+                  data_consistency: :delayed
 end
diff --git a/spec/workers/pipeline_process_worker_spec.rb b/spec/workers/pipeline_process_worker_spec.rb
index 0c1db3ccc5a..f8140d11f2e 100644
--- a/spec/workers/pipeline_process_worker_spec.rb
+++ b/spec/workers/pipeline_process_worker_spec.rb
@@ -3,10 +3,44 @@
 require 'spec_helper'
 
 RSpec.describe PipelineProcessWorker do
+  let_it_be(:pipeline) { create(:ci_pipeline) }
+
+  include_examples 'an idempotent worker' do
+    let(:pipeline) { create(:ci_pipeline, :created) }
+    let(:job_args) { [pipeline.id] }
+
+    before do
+      create(:ci_build, :created, pipeline: pipeline)
+    end
+
+    it 'processes the pipeline' do
+      expect(pipeline.status).to eq('created')
+      expect(pipeline.processables.pluck(:status)).to contain_exactly('created')
+
+      subject
+
+      expect(pipeline.reload.status).to eq('pending')
+      expect(pipeline.processables.pluck(:status)).to contain_exactly('pending')
+
+      subject
+
+      expect(pipeline.reload.status).to eq('pending')
+      expect(pipeline.processables.pluck(:status)).to contain_exactly('pending')
+    end
+  end
+
+  context 'when the FF ci_idempotent_pipeline_process_worker is disabled' do
+    before do
+      stub_feature_flags(ci_idempotent_pipeline_process_worker: false)
+    end
+
+    it 'is not deduplicated' do
+      expect(described_class).not_to be_deduplication_enabled
+    end
+  end
+
   describe '#perform' do
     context 'when pipeline exists' do
-      let(:pipeline) { create(:ci_pipeline) }
-
       it 'processes pipeline' do
         expect_any_instance_of(Ci::ProcessPipelineService).to receive(:execute)
 
@@ -16,14 +50,9 @@ RSpec.describe PipelineProcessWorker do
 
     context 'when pipeline does not exist' do
       it 'does not raise exception' do
-        expect { described_class.new.perform(123) }
+        expect { described_class.new.perform(non_existing_record_id) }
           .not_to raise_error
       end
     end
-
-    it_behaves_like 'worker with data consistency',
-                described_class,
-                feature_flag: :load_balancing_for_pipeline_process_worker,
-                data_consistency: :delayed
   end
 end
diff --git a/spec/workers/post_receive_spec.rb b/spec/workers/post_receive_spec.rb
index a468c8c3482..4d3cc447d9b 100644
--- a/spec/workers/post_receive_spec.rb
+++ b/spec/workers/post_receive_spec.rb
@@ -375,7 +375,7 @@ RSpec.describe PostReceive do
 
     it 'asks the project to trigger all hooks' do
       create(:project_hook, push_events: true, tag_push_events: true, project: project)
-      create(:custom_issue_tracker_service, push_events: true, merge_requests_events: false, project: project)
+      create(:custom_issue_tracker_integration, push_events: true, merge_requests_events: false, project: project)
       allow(Project).to receive(:find_by).and_return(project)
 
       expect(project).to receive(:execute_hooks).twice
diff --git a/spec/workers/process_commit_worker_spec.rb b/spec/workers/process_commit_worker_spec.rb
index 294a05c652b..3df26c774ba 100644
--- a/spec/workers/process_commit_worker_spec.rb
+++ b/spec/workers/process_commit_worker_spec.rb
@@ -138,7 +138,7 @@ RSpec.describe ProcessCommitWorker do
     end
   end
 
-  describe '#update_issue_metrics' do
+  describe '#update_issue_metrics', :clean_gitlab_redis_cache do
     context 'when commit has issue reference' do
       subject(:update_metrics_and_reload) do
         -> {
diff --git a/spec/workers/project_schedule_bulk_repository_shard_moves_worker_spec.rb b/spec/workers/project_schedule_bulk_repository_shard_moves_worker_spec.rb
deleted file mode 100644
index f284e1ab8c6..00000000000
--- a/spec/workers/project_schedule_bulk_repository_shard_moves_worker_spec.rb
+++ /dev/null
@@ -1,12 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe ProjectScheduleBulkRepositoryShardMovesWorker do
-  it_behaves_like 'schedules bulk repository shard moves' do
-    let_it_be_with_reload(:container) { create(:project, :repository).tap { |project| project.track_project_repository } }
-
-    let(:move_service_klass) { Projects::RepositoryStorageMove }
-    let(:worker_klass) { Projects::UpdateRepositoryStorageWorker }
-  end
-end
diff --git a/spec/workers/project_service_worker_spec.rb b/spec/workers/project_service_worker_spec.rb
index 237f501e0ec..9383e7ec5c4 100644
--- a/spec/workers/project_service_worker_spec.rb
+++ b/spec/workers/project_service_worker_spec.rb
@@ -3,7 +3,7 @@ require 'spec_helper'
 
 RSpec.describe ProjectServiceWorker, '#perform' do
   let(:worker) { described_class.new }
-  let(:service) { JiraService.new }
+  let(:service) { Integrations::Jira.new }
 
   before do
     allow(Integration).to receive(:find).and_return(service)
diff --git a/spec/workers/project_update_repository_storage_worker_spec.rb b/spec/workers/project_update_repository_storage_worker_spec.rb
deleted file mode 100644
index 6924e8a93a3..00000000000
--- a/spec/workers/project_update_repository_storage_worker_spec.rb
+++ /dev/null
@@ -1,15 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe ProjectUpdateRepositoryStorageWorker do
-  subject { described_class.new }
-
-  it_behaves_like 'an update storage move worker' do
-    let_it_be_with_refind(:container) { create(:project, :repository) }
-    let_it_be(:repository_storage_move) { create(:project_repository_storage_move) }
-
-    let(:service_klass) { Projects::UpdateRepositoryStorageService }
-    let(:repository_storage_move_klass) { Projects::RepositoryStorageMove }
-  end
-end
diff --git a/spec/workers/projects/post_creation_worker_spec.rb b/spec/workers/projects/post_creation_worker_spec.rb
index c2f42f03299..50c21575878 100644
--- a/spec/workers/projects/post_creation_worker_spec.rb
+++ b/spec/workers/projects/post_creation_worker_spec.rb
@@ -33,7 +33,7 @@ RSpec.describe Projects::PostCreationWorker do
           end
 
           before do
-            create(:clusters_applications_prometheus, :installed, cluster: cluster)
+            create(:clusters_integrations_prometheus, cluster: cluster)
           end
 
           it 'creates PrometheusService record', :aggregate_failures do
@@ -50,7 +50,7 @@ RSpec.describe Projects::PostCreationWorker do
           let(:cluster) { create(:cluster, :instance) }
 
           before do
-            create(:clusters_applications_prometheus, :installed, cluster: cluster)
+            create(:clusters_integrations_prometheus, cluster: cluster)
           end
 
           it 'creates PrometheusService record', :aggregate_failures do
diff --git a/spec/workers/projects/schedule_bulk_repository_shard_moves_worker_spec.rb b/spec/workers/projects/schedule_bulk_repository_shard_moves_worker_spec.rb
index 24957a35b72..7eff8e4dcd7 100644
--- a/spec/workers/projects/schedule_bulk_repository_shard_moves_worker_spec.rb
+++ b/spec/workers/projects/schedule_bulk_repository_shard_moves_worker_spec.rb
@@ -4,7 +4,7 @@ require 'spec_helper'
 
 RSpec.describe Projects::ScheduleBulkRepositoryShardMovesWorker do
   it_behaves_like 'schedules bulk repository shard moves' do
-    let_it_be_with_reload(:container) { create(:project, :repository).tap { |project| project.track_project_repository } }
+    let_it_be_with_reload(:container) { create(:project, :repository) }
 
     let(:move_service_klass) { Projects::RepositoryStorageMove }
     let(:worker_klass) { Projects::UpdateRepositoryStorageWorker }
diff --git a/spec/workers/propagate_integration_inherit_worker_spec.rb b/spec/workers/propagate_integration_inherit_worker_spec.rb
index 39219eaa3b5..2b4f241f755 100644
--- a/spec/workers/propagate_integration_inherit_worker_spec.rb
+++ b/spec/workers/propagate_integration_inherit_worker_spec.rb
@@ -6,7 +6,7 @@ RSpec.describe PropagateIntegrationInheritWorker do
   describe '#perform' do
     let_it_be(:integration) { create(:redmine_service, :instance) }
     let_it_be(:integration1) { create(:redmine_service, inherit_from_id: integration.id) }
-    let_it_be(:integration2) { create(:bugzilla_service, inherit_from_id: integration.id) }
+    let_it_be(:integration2) { create(:bugzilla_integration, inherit_from_id: integration.id) }
     let_it_be(:integration3) { create(:redmine_service) }
 
     it_behaves_like 'an idempotent worker' do
diff --git a/spec/workers/propagate_integration_worker_spec.rb b/spec/workers/propagate_integration_worker_spec.rb
index b8c7f2bebe7..2461b30a2ed 100644
--- a/spec/workers/propagate_integration_worker_spec.rb
+++ b/spec/workers/propagate_integration_worker_spec.rb
@@ -5,7 +5,7 @@ require 'spec_helper'
 RSpec.describe PropagateIntegrationWorker do
   describe '#perform' do
     let(:integration) do
-      PushoverService.create!(
+      Integrations::Pushover.create!(
         template: true,
         active: true,
         device: 'MyDevice',
@@ -21,11 +21,5 @@ RSpec.describe PropagateIntegrationWorker do
 
       subject.perform(integration.id)
     end
-
-    it 'ignores overwrite parameter from previous version' do
-      expect(Admin::PropagateIntegrationService).to receive(:propagate).with(integration)
-
-      subject.perform(integration.id, true)
-    end
   end
 end
diff --git a/spec/workers/propagate_service_template_worker_spec.rb b/spec/workers/propagate_service_template_worker_spec.rb
index 793f0b9b08c..b692ce3d72b 100644
--- a/spec/workers/propagate_service_template_worker_spec.rb
+++ b/spec/workers/propagate_service_template_worker_spec.rb
@@ -7,7 +7,7 @@ RSpec.describe PropagateServiceTemplateWorker do
 
   describe '#perform' do
     it 'calls the propagate service with the template' do
-      template = PushoverService.create!(
+      template = Integrations::Pushover.create!(
         template: true,
         active: true,
         properties: {
diff --git a/spec/workers/prune_web_hook_logs_worker_spec.rb b/spec/workers/prune_web_hook_logs_worker_spec.rb
deleted file mode 100644
index 6cd7a54ac7a..00000000000
--- a/spec/workers/prune_web_hook_logs_worker_spec.rb
+++ /dev/null
@@ -1,26 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe PruneWebHookLogsWorker do
-  describe '#perform' do
-    before do
-      hook = create(:project_hook)
-
-      create(:web_hook_log, web_hook: hook, created_at: 5.months.ago)
-      create(:web_hook_log, web_hook: hook, created_at: 4.months.ago)
-      create(:web_hook_log, web_hook: hook, created_at: 91.days.ago)
-      create(:web_hook_log, web_hook: hook, created_at: 89.days.ago)
-      create(:web_hook_log, web_hook: hook, created_at: 2.months.ago)
-      create(:web_hook_log, web_hook: hook, created_at: 1.month.ago)
-      create(:web_hook_log, web_hook: hook, response_status: '404')
-    end
-
-    it 'removes all web hook logs older than 90 days' do
-      described_class.new.perform
-
-      expect(WebHookLog.count).to eq(4)
-      expect(WebHookLog.last.response_status).to eq('404')
-    end
-  end
-end
diff --git a/spec/workers/remove_expired_group_links_worker_spec.rb b/spec/workers/remove_expired_group_links_worker_spec.rb
index 91031768632..ff5f7b9db27 100644
--- a/spec/workers/remove_expired_group_links_worker_spec.rb
+++ b/spec/workers/remove_expired_group_links_worker_spec.rb
@@ -24,7 +24,7 @@ RSpec.describe RemoveExpiredGroupLinksWorker do
         expect(non_expiring_project_group_link.reload).to be_present
       end
 
-      it 'removes project authorization' do
+      it 'removes project authorization', :sidekiq_inline do
         user = create(:user)
 
         project = expired_project_group_link.project
diff --git a/spec/workers/remove_unreferenced_lfs_objects_worker_spec.rb b/spec/workers/remove_unreferenced_lfs_objects_worker_spec.rb
index 21b9a7b844b..6007d3b34f8 100644
--- a/spec/workers/remove_unreferenced_lfs_objects_worker_spec.rb
+++ b/spec/workers/remove_unreferenced_lfs_objects_worker_spec.rb
@@ -34,14 +34,14 @@ RSpec.describe RemoveUnreferencedLfsObjectsWorker do
     end
 
     it 'removes unreferenced lfs objects' do
-      worker.perform
+      expect(worker.perform).to eq(2)
 
       expect(LfsObject.where(id: unreferenced_lfs_object1.id)).to be_empty
       expect(LfsObject.where(id: unreferenced_lfs_object2.id)).to be_empty
     end
 
     it 'leaves referenced lfs objects' do
-      worker.perform
+      expect(worker.perform).to eq(2)
 
       expect(referenced_lfs_object1.reload).to be_present
       expect(referenced_lfs_object2.reload).to be_present
@@ -50,10 +50,12 @@ RSpec.describe RemoveUnreferencedLfsObjectsWorker do
     it 'removes unreferenced lfs objects after project removal' do
       project1.destroy!
 
-      worker.perform
+      expect(worker.perform).to eq(3)
 
       expect(referenced_lfs_object1.reload).to be_present
       expect(LfsObject.where(id: referenced_lfs_object2.id)).to be_empty
     end
   end
+
+  it_behaves_like 'an idempotent worker'
 end
diff --git a/spec/workers/snippet_schedule_bulk_repository_shard_moves_worker_spec.rb b/spec/workers/snippet_schedule_bulk_repository_shard_moves_worker_spec.rb
deleted file mode 100644
index a5f1c6b7b3d..00000000000
--- a/spec/workers/snippet_schedule_bulk_repository_shard_moves_worker_spec.rb
+++ /dev/null
@@ -1,12 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe SnippetScheduleBulkRepositoryShardMovesWorker do
-  it_behaves_like 'schedules bulk repository shard moves' do
-    let_it_be_with_reload(:container) { create(:snippet, :repository).tap { |snippet| snippet.create_repository } }
-
-    let(:move_service_klass) { Snippets::RepositoryStorageMove }
-    let(:worker_klass) { Snippets::UpdateRepositoryStorageWorker }
-  end
-end
diff --git a/spec/workers/snippet_update_repository_storage_worker_spec.rb b/spec/workers/snippet_update_repository_storage_worker_spec.rb
deleted file mode 100644
index 205cb2e432f..00000000000
--- a/spec/workers/snippet_update_repository_storage_worker_spec.rb
+++ /dev/null
@@ -1,15 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe SnippetUpdateRepositoryStorageWorker do
-  subject { described_class.new }
-
-  it_behaves_like 'an update storage move worker' do
-    let_it_be_with_refind(:container) { create(:snippet, :repository) }
-    let_it_be(:repository_storage_move) { create(:snippet_repository_storage_move) }
-
-    let(:service_klass) { Snippets::UpdateRepositoryStorageService }
-    let(:repository_storage_move_klass) { Snippets::RepositoryStorageMove }
-  end
-end
diff --git a/spec/workers/ssh_keys/expired_notification_worker_spec.rb b/spec/workers/ssh_keys/expired_notification_worker_spec.rb
index 249ee404870..109d24f03ab 100644
--- a/spec/workers/ssh_keys/expired_notification_worker_spec.rb
+++ b/spec/workers/ssh_keys/expired_notification_worker_spec.rb
@@ -15,6 +15,20 @@ RSpec.describe SshKeys::ExpiredNotificationWorker, type: :worker do
   describe '#perform' do
     let_it_be(:user) { create(:user) }
 
+    context 'with a large batch' do
+      before 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) }
+
+      it 'updates all keys regardless of batch size' do
+        worker.perform
+
+        expect(keys.pluck(:expiry_notification_delivered_at)).not_to include(nil)
+      end
+    end
+
     context 'with expiring key today' do
       let_it_be_with_reload(:expired_today) { create(:key, expires_at: Time.current, user: user) }
 
@@ -35,24 +49,24 @@ RSpec.describe SshKeys::ExpiredNotificationWorker, type: :worker do
           perform_multiple(worker: worker)
         end
       end
+    end
+
+    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 }
+      end
 
-      context 'when feature is not enabled' do
+      context 'when key has already been notified of expiration' do
         before do
-          stub_feature_flags(ssh_key_expiration_email_notification: false)
+          expired_past.update!(expiry_notification_delivered_at: 1.day.ago)
         end
 
         it 'does not update notified column' do
-          expect { worker.perform }.not_to change { expired_today.reload.expiry_notification_delivered_at }
+          expect { worker.perform }.not_to change { expired_past.reload.expiry_notification_delivered_at }
         end
       end
     end
-
-    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 not update notified column' do
-        expect { worker.perform }.not_to change { expired_past.reload.expiry_notification_delivered_at }
-      end
-    end
   end
 end
diff --git a/spec/workers/ssh_keys/expiring_soon_notification_worker_spec.rb b/spec/workers/ssh_keys/expiring_soon_notification_worker_spec.rb
index f9276c86cdf..0a1d4a14ad0 100644
--- a/spec/workers/ssh_keys/expiring_soon_notification_worker_spec.rb
+++ b/spec/workers/ssh_keys/expiring_soon_notification_worker_spec.rb
@@ -35,16 +35,6 @@ RSpec.describe SshKeys::ExpiringSoonNotificationWorker, type: :worker do
           perform_multiple(worker: worker)
         end
       end
-
-      context 'when feature is not enabled' do
-        before do
-          stub_feature_flags(ssh_key_expiration_email_notification: false)
-        end
-
-        it 'does not update notified column' do
-          expect { worker.perform }.not_to change { expiring_soon.reload.before_expiry_notification_delivered_at }
-        end
-      end
     end
 
     context 'when key has expired in the past' do
diff --git a/spec/workers/stuck_ci_jobs_worker_spec.rb b/spec/workers/stuck_ci_jobs_worker_spec.rb
index 24d3b6fadf5..84b2d87494e 100644
--- a/spec/workers/stuck_ci_jobs_worker_spec.rb
+++ b/spec/workers/stuck_ci_jobs_worker_spec.rb
@@ -9,12 +9,17 @@ RSpec.describe StuckCiJobsWorker do
   let!(:job) { create :ci_build, runner: runner }
   let(:worker_lease_key) { StuckCiJobsWorker::EXCLUSIVE_LEASE_KEY }
   let(:worker_lease_uuid) { SecureRandom.uuid }
+  let(:created_at) { }
+  let(:updated_at) { }
 
   subject(:worker) { described_class.new }
 
   before do
     stub_exclusive_lease(worker_lease_key, worker_lease_uuid)
-    job.update!(status: status, updated_at: updated_at)
+    job_attributes = { status: status }
+    job_attributes[:created_at] = created_at if created_at
+    job_attributes[:updated_at] = updated_at if updated_at
+    job.update!(job_attributes)
   end
 
   shared_examples 'job is dropped' do
@@ -63,22 +68,70 @@ RSpec.describe StuckCiJobsWorker do
         allow_any_instance_of(Ci::Build).to receive(:stuck?).and_return(false)
       end
 
-      context 'when job was not updated for more than 1 day ago' do
-        let(:updated_at) { 2.days.ago }
+      context 'when job was updated_at more than 1 day ago' do
+        let(:updated_at) { 1.5.days.ago }
 
-        it_behaves_like 'job is dropped'
+        context 'when created_at is the same as updated_at' do
+          let(:created_at) { 1.5.days.ago }
+
+          it_behaves_like 'job is dropped'
+        end
+
+        context 'when created_at is before updated_at' do
+          let(:created_at) { 3.days.ago }
+
+          it_behaves_like 'job is dropped'
+        end
+
+        context 'when created_at is outside lookback window' do
+          let(:created_at) { described_class::BUILD_LOOKBACK - 1.day }
+
+          it_behaves_like 'job is unchanged'
+        end
       end
 
-      context 'when job was updated in less than 1 day ago' do
+      context 'when job was updated less than 1 day ago' do
         let(:updated_at) { 6.hours.ago }
 
-        it_behaves_like 'job is unchanged'
+        context 'when created_at is the same as updated_at' do
+          let(:created_at) { 1.5.days.ago }
+
+          it_behaves_like 'job is unchanged'
+        end
+
+        context 'when created_at is before updated_at' do
+          let(:created_at) { 3.days.ago }
+
+          it_behaves_like 'job is unchanged'
+        end
+
+        context 'when created_at is outside lookback window' do
+          let(:created_at) { described_class::BUILD_LOOKBACK - 1.day }
+
+          it_behaves_like 'job is unchanged'
+        end
       end
 
-      context 'when job was not updated for more than 1 hour ago' do
+      context 'when job was updated more than 1 hour ago' do
         let(:updated_at) { 2.hours.ago }
 
-        it_behaves_like 'job is unchanged'
+        context 'when created_at is the same as updated_at' do
+          let(:created_at) { 2.hours.ago }
+
+          it_behaves_like 'job is unchanged'
+        end
+
+        context 'when created_at is before updated_at' do
+          let(:created_at) { 3.days.ago }
+
+          it_behaves_like 'job is unchanged'
+        end
+
+        context 'when created_at is outside lookback window' do
+          let(:created_at) { described_class::BUILD_LOOKBACK - 1.day }
+
+          it_behaves_like 'job is unchanged'
+        end
       end
     end
 
@@ -87,17 +140,48 @@ RSpec.describe StuckCiJobsWorker do
         allow_any_instance_of(Ci::Build).to receive(:stuck?).and_return(true)
       end
 
-      context 'when job was not updated for more than 1 hour ago' do
-        let(:updated_at) { 2.hours.ago }
+      context 'when job was updated_at more than 1 hour ago' do
+        let(:updated_at) { 1.5.hours.ago }
+
+        context 'when created_at is the same as updated_at' do
+          let(:created_at) { 1.5.hours.ago }
+
+          it_behaves_like 'job is dropped'
+        end
+
+        context 'when created_at is before updated_at' do
+          let(:created_at) { 3.days.ago }
 
-        it_behaves_like 'job is dropped'
+          it_behaves_like 'job is dropped'
+        end
+
+        context 'when created_at is outside lookback window' do
+          let(:created_at) { described_class::BUILD_LOOKBACK - 1.day }
+
+          it_behaves_like 'job is unchanged'
+        end
       end
 
-      context 'when job was updated in less than 1
-       hour ago' do
+      context 'when job was updated in less than 1 hour ago' do
         let(:updated_at) { 30.minutes.ago }
 
-        it_behaves_like 'job is unchanged'
+        context 'when created_at is the same as updated_at' do
+          let(:created_at) { 30.minutes.ago }
+
+          it_behaves_like 'job is unchanged'
+        end
+
+        context 'when created_at is before updated_at' do
+          let(:created_at) { 2.days.ago }
+
+          it_behaves_like 'job is unchanged'
+        end
+
+        context 'when created_at is outside lookback window' do
+          let(:created_at) { described_class::BUILD_LOOKBACK - 1.day }
+
+          it_behaves_like 'job is unchanged'
+        end
       end
     end
   end
@@ -105,7 +189,7 @@ RSpec.describe StuckCiJobsWorker do
   context 'when job is running' do
     let(:status) { 'running' }
 
-    context 'when job was not updated for more than 1 hour ago' do
+    context 'when job was updated_at more than an hour ago' do
       let(:updated_at) { 2.hours.ago }
 
       it_behaves_like 'job is dropped'
@@ -123,7 +207,23 @@ RSpec.describe StuckCiJobsWorker do
       let(:status) { status }
       let(:updated_at) { 2.days.ago }
 
-      it_behaves_like 'job is unchanged'
+      context 'when created_at is the same as updated_at' do
+        let(:created_at) { 2.days.ago }
+
+        it_behaves_like 'job is unchanged'
+      end
+
+      context 'when created_at is before updated_at' do
+        let(:created_at) { 3.days.ago }
+
+        it_behaves_like 'job is unchanged'
+      end
+
+      context 'when created_at is outside lookback window' do
+        let(:created_at) { described_class::BUILD_LOOKBACK - 1.day }
+
+        it_behaves_like 'job is unchanged'
+      end
     end
   end
 
diff --git a/spec/workers/users/update_open_issue_count_worker_spec.rb b/spec/workers/users/update_open_issue_count_worker_spec.rb
deleted file mode 100644
index 700055980d8..00000000000
--- a/spec/workers/users/update_open_issue_count_worker_spec.rb
+++ /dev/null
@@ -1,65 +0,0 @@
-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe Users::UpdateOpenIssueCountWorker do
-  let_it_be(:first_user) { create(:user) }
-  let_it_be(:second_user) { create(:user) }
-
-  describe '#perform' do
-    let(:target_user_ids) { [first_user.id, second_user.id] }
-
-    subject { described_class.new.perform(target_user_ids) }
-
-    context 'when arguments are missing' do
-      context 'when target_user_ids are missing' do
-        context 'when nil' do
-          let(:target_user_ids) { nil }
-
-          it 'raises an error' do
-            expect { subject }.to raise_error(ArgumentError, /No target user ID provided/)
-          end
-        end
-
-        context 'when empty array' do
-          let(:target_user_ids) { [] }
-
-          it 'raises an error' do
-            expect { subject }.to raise_error(ArgumentError, /No target user ID provided/)
-          end
-        end
-
-        context 'when not an ID' do
-          let(:target_user_ids) { "nonsense" }
-
-          it 'raises an error' do
-            expect { subject }.to raise_error(ArgumentError, /No valid target user ID provided/)
-          end
-        end
-      end
-    end
-
-    context 'when successful' do
-      let(:job_args) { [target_user_ids] }
-      let(:fake_service1) { double }
-      let(:fake_service2) { double }
-
-      it 'calls the user update service' do
-        expect(Users::UpdateAssignedOpenIssueCountService).to receive(:new).with(target_user: first_user).and_return(fake_service1)
-        expect(Users::UpdateAssignedOpenIssueCountService).to receive(:new).with(target_user: second_user).and_return(fake_service2)
-        expect(fake_service1).to receive(:execute)
-        expect(fake_service2).to receive(:execute)
-
-        subject
-      end
-
-      it_behaves_like 'an idempotent worker' do
-        it 'recalculates' do
-          subject
-
-          expect(first_user.assigned_open_issues_count).to eq(0)
-        end
-      end
-    end
-  end
-end
diff --git a/spec/workers/web_hook_worker_spec.rb b/spec/workers/web_hook_worker_spec.rb
index becc7461f2a..548cf4c717a 100644
--- a/spec/workers/web_hook_worker_spec.rb
+++ b/spec/workers/web_hook_worker_spec.rb
@@ -10,9 +10,14 @@ RSpec.describe WebHookWorker do
 
   describe '#perform' do
     it 'delegates to WebHookService' do
-      expect_next(WebHookService, project_hook, data.with_indifferent_access, hook_name).to receive(:execute)
+      expect_next(WebHookService, project_hook, data.with_indifferent_access, hook_name, anything).to receive(:execute)
 
       subject.perform(project_hook.id, data, hook_name)
     end
+
+    it_behaves_like 'worker with data consistency',
+                  described_class,
+                  feature_flag: :load_balancing_for_web_hook_worker,
+                  data_consistency: :delayed
   end
 end