Use FDW to find unsynced projects when feature flag is enabled

ee/app/finders/geo/project_registry_finder.rb

@@ -72,18 +72,11 @@ module Geo
         .execute
     end
 
-    # rubocop: disable CodeReuse/ActiveRecord
-    def find_unsynced_projects(batch_size:)
-      relation =
-        if use_legacy_queries?
-          legacy_find_unsynced_projects
-        else
-          fdw_find_unsynced_projects
-        end
-
-      relation.limit(batch_size)
+    def find_unsynced_projects(shard_name:, batch_size:)
+      finder_klass_for_unsynced_projects
+        .new(current_node: current_node, shard_name: shard_name, batch_size: batch_size)
+        .execute
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
     # rubocop: disable CodeReuse/ActiveRecord
     def find_projects_updated_recently(batch_size:)
@@ -100,18 +93,6 @@ module Geo
 
     protected
 
-    #
-    # FDW accessors
-    #
-
-    # @return [ActiveRecord::Relation<Geo::Fdw::Project>]
-    # rubocop: disable CodeReuse/ActiveRecord
-    def fdw_find_unsynced_projects
-      Geo::Fdw::Project.joins("LEFT OUTER JOIN project_registry ON project_registry.project_id = #{fdw_project_table.name}.id")
-        .where(project_registry: { project_id: nil })
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
     # @return [ActiveRecord::Relation<Geo::Fdw::Project>]
     # rubocop: disable CodeReuse/ActiveRecord
     def fdw_find_projects_updated_recently
@@ -121,21 +102,6 @@ module Geo
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
-    #
-    # Legacy accessors (non FDW)
-    #
-
-    # @return [ActiveRecord::Relation<Project>] list of unsynced projects
-    # rubocop: disable CodeReuse/ActiveRecord
-    def legacy_find_unsynced_projects
-      legacy_left_outer_join_registry_ids(
-        current_node.projects,
-        Geo::ProjectRegistry.pluck(:project_id),
-        Project
-      )
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
     # @return [ActiveRecord::Relation<Project>] list of projects updated recently
     # rubocop: disable CodeReuse/ActiveRecord
     def legacy_find_projects_updated_recently
@@ -171,6 +137,14 @@ module Geo
       fdw_disabled? || selective_sync? && !Gitlab::Geo::Fdw.enabled_for_selective_sync?
     end
 
+    def finder_klass_for_unsynced_projects
+      if use_legacy_queries_for_selective_sync?
+        Geo::LegacyProjectUnsyncedFinder
+      else
+        Geo::ProjectUnsyncedFinder
+      end
+    end
+
     def finder_klass_for_synced_registries
       if use_legacy_queries_for_selective_sync?
         Geo::LegacyProjectRegistrySyncedFinder

ee/app/workers/geo/repository_shard_sync_worker.rb

@@ -77,19 +77,19 @@ module Geo
 
     # rubocop: disable CodeReuse/ActiveRecord
     def find_project_ids_not_synced(batch_size:)
-      shard_restriction(finder.find_unsynced_projects(batch_size: batch_size))
-        .where.not(id: scheduled_project_ids)
+      finder.find_unsynced_projects(shard_name: shard_name, batch_size: batch_size)
+        .id_not_in(scheduled_project_ids)
         .reorder(last_repository_updated_at: :desc)
-        .pluck(:id)
+        .pluck_primary_key
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
     # rubocop: disable CodeReuse/ActiveRecord
     def find_project_ids_updated_recently(batch_size:)
       shard_restriction(finder.find_projects_updated_recently(batch_size: batch_size))
-        .where.not(id: scheduled_project_ids)
+        .id_not_in(scheduled_project_ids)
         .order('project_registry.last_repository_synced_at ASC NULLS FIRST, projects.last_repository_updated_at ASC')
-        .pluck(:id)
+        .pluck_primary_key
     end
     # rubocop: enable CodeReuse/ActiveRecord
 

ee/spec/finders/geo/project_registry_finder_spec.rb

@@ -380,43 +380,6 @@ describe Geo::ProjectRegistryFinder, :geo do
   end
 
   shared_examples 'finds all the things' do |method_prefix|
-    describe '#find_unsynced_projects' do
-      it 'delegates to the correct method' do
-        expect(subject).to receive("#{method_prefix}_find_unsynced_projects".to_sym).and_call_original
-
-        subject.find_unsynced_projects(batch_size: 10)
-      end
-
-      it 'returns projects without an entry on the tracking database' do
-        project_not_synced = create(:project)
-        create(:geo_project_registry, :synced, :repository_dirty, project: project_1_in_synced_group)
-
-        projects = subject.find_unsynced_projects(batch_size: 10)
-
-        expect(projects).to match_ids(project_not_synced)
-      end
-
-      context 'with selective sync' do
-        before do
-          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
-        end
-
-        it 'delegates to #legacy_find_unsynced_projects' do
-          expect(subject).to receive(:legacy_find_unsynced_projects).and_call_original
-
-          subject.find_unsynced_projects(batch_size: 10)
-        end
-
-        it 'returns untracked projects in the synced group' do
-          create(:geo_project_registry, :sync_failed, project: project_1_in_synced_group)
-
-          projects = subject.find_unsynced_projects(batch_size: 10)
-
-          expect(projects).to match_ids(project_2_in_synced_group)
-        end
-      end
-    end
-
     describe '#find_projects_updated_recently' do
       it 'delegates to the correct method' do
         expect(subject).to receive("#{method_prefix}_find_projects_updated_recently".to_sym).and_call_original
@@ -522,6 +485,13 @@ describe Geo::ProjectRegistryFinder, :geo do
     include_examples 'finds all the things', 'legacy'
   end
 
+  describe '#find_unsynced_projects', :delete do
+    include_examples 'delegates to the proper finder',
+      Geo::LegacyProjectUnsyncedFinder,
+      Geo::ProjectUnsyncedFinder,
+      :find_unsynced_projects, [shard_name: 'default', batch_size: 100]
+  end
+
   describe '#find_failed_project_registries', :delete do
     include_examples 'delegates to the proper finder',
       Geo::LegacyProjectRegistrySyncFailedFinder,