Refactor repositories clean up worker

Does not rely on FDW queries to find
projects to clean up.

ee/app/workers/geo/repositories_clean_up_worker.rb

@@ -14,13 +14,14 @@ module Geo
     # rubocop:disable CodeReuse/ActiveRecord
     def perform(geo_node_id)
       try_obtain_lease do
-        node = Geo::Fdw::GeoNode.find(geo_node_id)
+        node = GeoNode.find(geo_node_id)
         break unless node.selective_sync?
 
-        node.projects_outside_selective_sync.find_in_batches(batch_size: BATCH_SIZE) do |batch|
-          batch.each do |project|
-            clean_up_repositories(project)
-          end
+        Geo::ProjectRegistry.select(:id, :project_id).find_in_batches(batch_size: BATCH_SIZE) do |registries|
+          tracked_project_ids = registries.map(&:project_id)
+          replicable_project_ids = node.projects.id_in(tracked_project_ids).pluck_primary_key
+          unused_tracked_project_ids = tracked_project_ids - replicable_project_ids
+          clean_up_repositories(unused_tracked_project_ids)
         end
       end
     rescue ActiveRecord::RecordNotFound => error
@@ -30,7 +31,15 @@ module Geo
 
     private
 
-    def clean_up_repositories(project)
+    def clean_up_repositories(unused_tracked_project_ids)
+      unused_projects = Project.id_in(unused_tracked_project_ids)
+
+      unused_projects.each do |project|
+        clean_up_repository(project)
+      end
+    end
+
+    def clean_up_repository(project)
       job_id = ::Geo::RepositoryCleanupWorker.perform_async(project.id, project.name, project.disk_path, project.repository.storage)
 
       if job_id

ee/spec/workers/geo/repositories_clean_up_worker_spec.rb

@@ -2,30 +2,25 @@
 
 require 'spec_helper'
 
-RSpec.describe Geo::RepositoriesCleanUpWorker, :geo, :geo_fdw do
+RSpec.describe Geo::RepositoriesCleanUpWorker, :geo do
   include ::EE::GeoHelpers
   include ExclusiveLeaseHelpers
 
   describe '#perform' do
-    let(:secondary) { create(:geo_node) }
-
-    let(:synced_group) { create(:group) }
-    let(:synced_subgroup) { create(:group, parent: synced_group) }
-    let(:unsynced_group) { create(:group) }
-
-    let(:project_1) { create(:project, group: synced_group) }
-    let(:project_2) { create(:project, group: synced_group) }
-    let!(:project_3) { create(:project, :repository, group: unsynced_group) }
-    let(:project_4) { create(:project, :repository, group: unsynced_group) }
-    let(:project_5) { create(:project, group: synced_subgroup) }
-    let(:project_6) { create(:project, group: synced_subgroup) }
-    let(:project_7) { create(:project) }
-    let(:project_8) { create(:project) }
-
-    before do
-      stub_current_geo_node(secondary)
-      stub_exclusive_lease
-
+    let_it_be(:secondary) { create(:geo_node) }
+    let_it_be(:synced_group) { create(:group) }
+    let_it_be(:synced_subgroup) { create(:group, parent: synced_group) }
+    let_it_be(:unsynced_group) { create(:group) }
+    let_it_be(:project_1) { create(:project, group: synced_group) }
+    let_it_be(:project_2) { create(:project, group: synced_group) }
+    let_it_be(:project_3) { create(:project, :repository, group: unsynced_group) }
+    let_it_be(:project_4) { create(:project, :repository, group: unsynced_group) }
+    let_it_be(:project_5) { create(:project, group: synced_subgroup) }
+    let_it_be(:project_6) { create(:project, group: synced_subgroup) }
+    let_it_be(:project_7) { create(:project) }
+    let_it_be(:project_8) { create(:project) }
+
+    before_all do
       create(:geo_project_registry, project: project_1)
       create(:geo_project_registry, project: project_2)
       create(:geo_project_registry, project: project_4)
@@ -35,6 +30,11 @@ RSpec.describe Geo::RepositoriesCleanUpWorker, :geo, :geo_fdw do
       create(:geo_project_registry, project: project_8)
     end
 
+    before do
+      stub_current_geo_node(secondary)
+      stub_exclusive_lease
+    end
+
     it 'does not perform Geo::RepositoryCleanupWorker when cannnot obtain a lease' do
       stub_exclusive_lease_taken
 
@@ -76,10 +76,8 @@ RSpec.describe Geo::RepositoriesCleanUpWorker, :geo, :geo_fdw do
         subject.perform(secondary.id)
       end
 
-      it 'does not leave orphaned entries in the project_registry table' do
-        Sidekiq::Testing.inline! do
-          subject.perform(secondary.id)
-        end
+      it 'does not leave orphaned entries in the project_registry table', :sidekiq_inline do
+        subject.perform(secondary.id)
 
         expect(Geo::ProjectRegistry.where(project_id: [project_3, project_4, project_7, project_8])).to be_empty
       end
@@ -109,10 +107,8 @@ RSpec.describe Geo::RepositoriesCleanUpWorker, :geo, :geo_fdw do
         subject.perform(secondary.id)
       end
 
-      it 'does not leave orphaned entries in the project_registry table' do
-        Sidekiq::Testing.inline! do
-          subject.perform(secondary.id)
-        end
+      it 'does not leave orphaned entries in the project_registry table', :sidekiq_inline do
+        subject.perform(secondary.id)
 
         expect(Geo::ProjectRegistry.where(project_id: [project_1, project_2, project_3, project_4, project_5, project_6])).to be_empty
       end