Merge branch...

Merge branch '8798-geo-implement-selective-sync-support-for-fdw-queries-to-find-registries-to-verify' into 'master'

Geo - Add selective sync support for the FDW queries to find registries to verify

See merge request gitlab-org/gitlab-ee!10438

ee/app/finders/geo/legacy_project_registry_pending_verification_finder.rb

+# frozen_string_literal: true
+
+# Finder for retrieving project registries that need a repository or
+# wiki verification where projects belong to the specific shard
+# using cross-database joins.
+#
+# Basic usage:
+#
+#     Geo::LegacyProjectRegistryPendingVerificationFinder
+#       .new(current_node: Gitlab::Geo.current_node, shard_name: 'default', batch_size: 1000)
+#       .execute
+module Geo
+  class LegacyProjectRegistryPendingVerificationFinder < RegistryFinder
+    def initialize(current_node: nil, shard_name:, batch_size:)
+      super(current_node: current_node)
+      @shard_name = shard_name
+      @batch_size = batch_size
+    end
+
+    def execute
+      registries = find_registries_pending_verification_on_secondary
+      return Geo::ProjectRegistry.none if registries.empty?
+
+      registries_to_verify = filter_registries_verified_in_primary(registries)
+      return registries_to_verify unless selective_sync?
+
+      legacy_inner_join_registry_ids(
+        registries_to_verify,
+        current_node.projects.pluck_primary_key,
+        Geo::ProjectRegistry,
+        foreign_key: :project_id
+      )
+    end
+
+    private
+
+    attr_reader :batch_size, :shard_name
+
+    # rubocop:disable CodeReuse/ActiveRecord
+    def find_registries_pending_verification_on_secondary
+      Geo::ProjectRegistry
+        .where(Geo::ProjectRegistry.registries_pending_verification)
+        .pluck(
+          :project_id,
+          Geo::ProjectRegistry.repositories_pending_verification.to_sql,
+          Geo::ProjectRegistry.wikis_pending_verification.to_sql
+        )
+    end
+    # rubocop:enable CodeReuse/ActiveRecord
+
+    def filter_registries_verified_in_primary(registries)
+      filtered_project_ids = filter_projects_verified_on_primary(registries)
+      Geo::ProjectRegistry.project_id_in(filtered_project_ids)
+    end
+
+    # rubocop:disable CodeReuse/ActiveRecord
+    def filter_projects_verified_on_primary(registries)
+      inner_join_project_repository_state(registries)
+        .joins(:project)
+        .merge(Project.within_shards(shard_name))
+        .where(
+          legacy_repository_state_table[:repository_verification_checksum].not_eq(nil)
+            .and(project_registry_verify_table[:want_to_verify_repo].eq(true))
+          .or(legacy_repository_state_table[:wiki_verification_checksum].not_eq(nil)
+            .and(project_registry_verify_table[:want_to_verify_wiki].eq(true))))
+        .limit(batch_size)
+        .pluck_project_key
+    end
+    # rubocop:enable CodeReuse/ActiveRecord
+
+    # rubocop:disable CodeReuse/ActiveRecord
+    def inner_join_project_repository_state(registries)
+      id_and_want_to_verify = registries.map do |project_id, want_to_verify_repo, want_to_verify_wiki|
+        "(#{project_id}, #{quote_value(want_to_verify_repo)}, #{quote_value(want_to_verify_wiki)})"
+      end
+
+      ProjectRepositoryState.joins(<<~SQL_REPO)
+        INNER JOIN
+        (VALUES #{id_and_want_to_verify.join(',')})
+        #{project_registry_verify_table.name}(project_id, want_to_verify_repo, want_to_verify_wiki)
+        ON #{legacy_repository_state_table.name}.project_id = #{project_registry_verify_table.name}.project_id
+      SQL_REPO
+    end
+    # rubocop:enable CodeReuse/ActiveRecord
+
+    def legacy_repository_state_table
+      @legacy_repository_state_table ||= ProjectRepositoryState.arel_table
+    end
+
+    def project_registry_verify_table
+      @project_registry_verify_table ||= Arel::Table.new(:project_registry_verify_table)
+    end
+  end
+end

ee/app/finders/geo/project_registry_finder.rb

@@ -66,13 +66,10 @@ module Geo
       registries_retrying_verification(:wiki).count
     end
 
-    # Find all registries that need a repository or wiki verification
     def find_registries_to_verify(shard_name:, batch_size:)
-      if use_legacy_queries?
-        legacy_find_registries_to_verify(shard_name: shard_name, batch_size: batch_size)
-      else
-        fdw_find_registries_to_verify(shard_name: shard_name, batch_size: batch_size)
-      end
+      finder_klass_for_registries_pending_verification
+        .new(current_node: current_node, shard_name: shard_name, batch_size: batch_size)
+        .execute
     end
 
     # rubocop: disable CodeReuse/ActiveRecord
@@ -124,41 +121,6 @@ module Geo
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
-    # Find all registries that repository or wiki need verification
-    # @return [ActiveRecord::Relation<Geo::ProjectRegistry>] list of registries that need verification
-    # rubocop: disable CodeReuse/ActiveRecord
-    def fdw_find_registries_to_verify(shard_name:, batch_size:)
-      repo_condition =
-        local_repo_condition
-          .and(fdw_repository_state_table[:repository_verification_checksum].not_eq(nil))
-
-      wiki_condition =
-        local_wiki_condition
-          .and(fdw_repository_state_table[:wiki_verification_checksum].not_eq(nil))
-
-      Geo::ProjectRegistry
-        .joins(fdw_inner_join_projects)
-        .joins(fdw_inner_join_repository_state)
-        .where(repo_condition.or(wiki_condition))
-        .where(fdw_project_table[:repository_storage].eq(shard_name))
-        .limit(batch_size)
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    def fdw_inner_join_projects
-      local_registry_table
-        .join(fdw_project_table, Arel::Nodes::InnerJoin)
-        .on(local_registry_table[:project_id].eq(fdw_project_table[:id]))
-        .join_sources
-    end
-
-    def fdw_inner_join_repository_state
-      local_registry_table
-        .join(fdw_repository_state_table, Arel::Nodes::InnerJoin)
-        .on(local_registry_table[:project_id].eq(fdw_repository_state_table[:project_id]))
-        .join_sources
-    end
-
     #
     # Legacy accessors (non FDW)
     #
@@ -195,88 +157,16 @@ module Geo
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
-    def quote_value(value)
-      ::Gitlab::SQL::Glob.q(value)
-    end
-
-    # @return [ActiveRecord::Relation<Geo::ProjectRegistry>] list of registries that need verification
-    # rubocop: disable CodeReuse/ActiveRecord
-    def legacy_find_registries_to_verify(shard_name:, batch_size:)
-      registries = Geo::ProjectRegistry
-        .where(local_repo_condition.or(local_wiki_condition))
-        .pluck(:project_id, local_repo_condition.to_sql, local_wiki_condition.to_sql)
-
-      return Geo::ProjectRegistry.none if registries.empty?
-
-      id_and_want_to_sync = registries.map do |project_id, want_to_sync_repo, want_to_sync_wiki|
-        "(#{project_id}, #{quote_value(want_to_sync_repo)}, #{quote_value(want_to_sync_wiki)})"
-      end
-
-      project_registry_sync_table = Arel::Table.new(:project_registry_sync_table)
-
-      joined_relation =
-        ProjectRepositoryState.joins(<<~SQL_REPO)
-          INNER JOIN
-          (VALUES #{id_and_want_to_sync.join(',')})
-          project_registry_sync_table(project_id, want_to_sync_repo, want_to_sync_wiki)
-          ON #{legacy_repository_state_table.name}.project_id = project_registry_sync_table.project_id
-        SQL_REPO
-
-      project_ids = joined_relation
-        .joins(:project)
-        .where(projects: { repository_storage: shard_name })
-        .where(
-          legacy_repository_state_table[:repository_verification_checksum].not_eq(nil)
-            .and(project_registry_sync_table[:want_to_sync_repo].eq(true))
-          .or(legacy_repository_state_table[:wiki_verification_checksum].not_eq(nil)
-            .and(project_registry_sync_table[:want_to_sync_wiki].eq(true))))
-        .limit(batch_size)
-        .pluck(:project_id)
-
-      Geo::ProjectRegistry.where(project_id: project_ids)
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    def legacy_repository_state_table
-      ::ProjectRepositoryState.arel_table
-    end
-
     def fdw_project_table
       Geo::Fdw::Project.arel_table
     end
 
-    def fdw_repository_state_table
-      Geo::Fdw::ProjectRepositoryState.arel_table
-    end
-
-    def local_registry_table
-      Geo::ProjectRegistry.arel_table
-    end
-
-    def local_repo_condition
-      local_registry_table[:repository_verification_checksum_sha].eq(nil)
-        .and(local_registry_table[:last_repository_verification_failure].eq(nil))
-        .and(local_registry_table[:resync_repository].eq(false))
-        .and(repository_missing_on_primary_is_not_true)
-    end
-
-    def local_wiki_condition
-      local_registry_table[:wiki_verification_checksum_sha].eq(nil)
-        .and(local_registry_table[:last_wiki_verification_failure].eq(nil))
-        .and(local_registry_table[:resync_wiki].eq(false))
-        .and(wiki_missing_on_primary_is_not_true)
-    end
-
-    def repository_missing_on_primary_is_not_true
-      Arel::Nodes::SqlLiteral.new("project_registry.repository_missing_on_primary IS NOT TRUE")
-    end
+    private
 
-    def wiki_missing_on_primary_is_not_true
-      Arel::Nodes::SqlLiteral.new("project_registry.wiki_missing_on_primary IS NOT TRUE")
+    def use_legacy_queries_for_selective_sync?
+      selective_sync? && !Gitlab::Geo::Fdw.enabled_for_selective_sync?
     end
 
-    private
-
     def finder_klass_for_synced_registries
       if Gitlab::Geo::Fdw.enabled_for_selective_sync?
         Geo::ProjectRegistrySyncedFinder
@@ -360,5 +250,13 @@ module Geo
         .new(current_node: current_node, type: type)
         .execute
     end
+
+    def finder_klass_for_registries_pending_verification
+      if !Gitlab::Geo::Fdw.enabled? || use_legacy_queries_for_selective_sync?
+        Geo::LegacyProjectRegistryPendingVerificationFinder
+      else
+        Geo::ProjectRegistryPendingVerificationFinder
+      end
+    end
   end
 end

ee/app/finders/geo/project_registry_pending_verification_finder.rb

+# frozen_string_literal: true
+
+# Finder for retrieving project registries that that need a repository or
+# wiki verification where projects belong to the specific shard using
+# FDW queries.
+#
+# Basic usage:
+#
+#     Geo::ProjectRegistryPendingVerificationFinder
+#       .new(current_node: Gitlab::Geo.current_node, shard_name: 'default', batch_size: 1000)
+#       .execute.
+module Geo
+  class ProjectRegistryPendingVerificationFinder
+    def initialize(current_node:, shard_name:, batch_size:)
+      @current_node = Geo::Fdw::GeoNode.find(current_node.id)
+      @shard_name = shard_name
+      @batch_size = batch_size
+    end
+
+    # rubocop:disable CodeReuse/ActiveRecord
+    def execute
+      return Geo::ProjectRegistry.none unless valid_shard?
+
+      Gitlab::Geo::Fdw::ProjectRegistryQueryBuilder
+        .new(current_node.project_registries)
+        .registries_pending_verification
+        .within_shards(shard_name)
+        .limit(batch_size)
+    end
+    # rubocop:enable CodeReuse/ActiveRecord
+
+    private
+
+    attr_reader :current_node, :shard_name, :batch_size
+
+    def valid_shard?
+      return true unless current_node.selective_sync_by_shards?
+
+      current_node.selective_sync_shards.include?(shard_name)
+    end
+  end
+end

ee/app/finders/geo/registry_finder.rb

@@ -55,5 +55,9 @@ module Geo
       joined_relation.where(registry: { registry_present: [nil, false] })
     end
     # rubocop: enable CodeReuse/ActiveRecord
+
+    def quote_value(value)
+      ::Gitlab::SQL::Glob.q(value)
+    end
   end
 end

ee/app/models/ee/project.rb

@@ -87,7 +87,7 @@ module EE
       end
 
       scope :with_wiki_enabled, -> { with_feature_enabled(:wiki) }
-
+      scope :within_shards, -> (shard_names) { where(repository_storage: Array(shard_names)) }
       scope :verification_failed_repos, -> { joins(:repository_state).merge(ProjectRepositoryState.verification_failed_repos) }
       scope :verification_failed_wikis, -> { joins(:repository_state).merge(ProjectRepositoryState.verification_failed_wikis) }
       scope :for_plan_name, -> (name) { joins(namespace: :plan).where(plans: { name: name }) }

ee/app/models/geo/fdw/geo_node.rb

@@ -29,11 +29,8 @@ module Geo
       private
 
       def registries_for_selected_namespaces
-        query = selected_namespaces_and_descendants
-
-        Geo::ProjectRegistry
-          .joins(fdw_inner_join_projects)
-          .where(fdw_projects_table.name => { namespace_id: query.select(:id) })
+        Gitlab::Geo::Fdw::ProjectRegistryQueryBuilder.new
+          .within_namespaces(selected_namespaces_and_descendants.select(:id))
       end
 
       def selected_namespaces_and_descendants
@@ -60,17 +57,8 @@ module Geo
       end
 
       def registries_for_selected_shards
-        Geo::ProjectRegistry
-          .joins(fdw_inner_join_projects)
-          .where(fdw_projects_table.name => { repository_storage: selective_sync_shards })
-      end
-
-      def project_registries_table
-        Geo::ProjectRegistry.arel_table
-      end
-
-      def fdw_projects_table
-        Geo::Fdw::Project.arel_table
+        Gitlab::Geo::Fdw::ProjectRegistryQueryBuilder.new
+          .within_shards(selective_sync_shards)
       end
 
       def fdw_namespaces_table
@@ -80,13 +68,6 @@ module Geo
       def fdw_geo_node_namespace_links_table
         Geo::Fdw::GeoNodeNamespaceLink.arel_table
       end
-
-      def fdw_inner_join_projects
-        project_registries_table
-          .join(fdw_projects_table, Arel::Nodes::InnerJoin)
-          .on(project_registries_table[:project_id].eq(fdw_projects_table[:id]))
-          .join_sources
-      end
     end
   end
 end

ee/app/models/geo/fdw/project.rb

@@ -17,6 +17,14 @@ module Geo
         def search(query)
           fuzzy_search(query, [:path, :name, :description])
         end
+
+        def within_namespaces(namespace_ids)
+          where(arel_table.name => { namespace_id: namespace_ids })
+        end
+
+        def within_shards(shard_names)
+          where(repository_storage: Array(shard_names))
+        end
       end
     end
   end

ee/app/models/geo/project_registry.rb

@@ -31,6 +31,10 @@ class Geo::ProjectRegistry < Geo::BaseRegistry
   scope :wiki_checksum_mismatch, -> { where(wiki_checksum_mismatch: true) }
   scope :with_routes, -> { includes(project: :route).includes(project: { namespace: :route }) }
 
+  def self.project_id_in(ids)
+    where(project_id: ids)
+  end
+
   def self.failed
     repository_sync_failed = arel_table[:repository_retry_count].gt(0)
     wiki_sync_failed = arel_table[:wiki_retry_count].gt(0)
@@ -141,6 +145,30 @@ class Geo::ProjectRegistry < Geo::BaseRegistry
     end
   end
 
+  def self.registries_pending_verification
+    repositories_pending_verification.or(wikis_pending_verification)
+  end
+
+  def self.repositories_pending_verification
+    repository_exists_on_primary =
+      Arel::Nodes::SqlLiteral.new("project_registry.repository_missing_on_primary IS NOT TRUE")
+
+    arel_table[:repository_verification_checksum_sha].eq(nil)
+      .and(arel_table[:last_repository_verification_failure].eq(nil))
+      .and(arel_table[:resync_repository].eq(false))
+      .and(repository_exists_on_primary)
+  end
+
+  def self.wikis_pending_verification
+    wiki_exists_on_primary =
+      Arel::Nodes::SqlLiteral.new("project_registry.wiki_missing_on_primary IS NOT TRUE")
+
+    arel_table[:wiki_verification_checksum_sha].eq(nil)
+      .and(arel_table[:last_wiki_verification_failure].eq(nil))
+      .and(arel_table[:resync_wiki].eq(false))
+      .and(wiki_exists_on_primary)
+  end
+
   def self.flag_repositories_for_resync!
     update_all(
       resync_repository: true,

ee/app/models/project_repository_state.rb

@@ -18,4 +18,8 @@ class ProjectRepositoryState < ApplicationRecord
   scope :verification_failed_wikis, -> { where.not(last_wiki_verification_failure: nil) }
   scope :verified_repos, -> { where.not(repository_verification_checksum: nil).where(last_repository_verification_failure: nil) }
   scope :verified_wikis, -> { where.not(wiki_verification_checksum: nil).where(last_wiki_verification_failure: nil) }
+
+  def self.pluck_project_key
+    where(nil).pluck(:project_id)
+  end
 end

ee/changelogs/unreleased/8798-geo-implement-selective-sync-support-for-fdw-queries-to-find-registries-to-verify.yml

+---
+title: Geo - Add selective sync support for the FDW queries to find registries to
+  verify
+merge_request: 10438
+author:
+type: changed

ee/lib/gitlab/geo/fdw/project_registry_query_builder.rb

+# frozen_string_literal: true
+
+# Builder class to create composable queries using FDW to
+# retrieve project registries.
+#
+# Basic usage:
+#
+#     Gitlab::Geo::Fdw::ProjectRegistryQueryBuilder
+#       .new(Geo::ProjectRegistry.all)
+#       .registries_pending_verification
+#       .within_shards(selective_sync_shards)
+#
+module Gitlab
+  module Geo
+    class Fdw
+      class ProjectRegistryQueryBuilder < SimpleDelegator
+        attr_reader :query
+
+        def initialize(query = nil)
+          @query = query || base_query
+          super(query)
+        end
+
+        # rubocop:disable CodeReuse/ActiveRecord
+        def registries_pending_verification
+          reflect(
+            query
+              .joins(fdw_inner_join_projects)
+              .joins(fdw_inner_join_repository_state)
+              .where(repositories_pending_verification.or(wikis_pending_verification))
+          )
+        end
+        # rubocop:enable CodeReuse/ActiveRecord
+
+        # rubocop:disable CodeReuse/ActiveRecord
+        def within_namespaces(namespace_ids)
+          reflect(
+            query
+              .joins(fdw_inner_join_projects)
+              .merge(projects_within_namespaces(namespace_ids))
+          )
+        end
+        # rubocop:enable CodeReuse/ActiveRecord
+
+        # rubocop:disable CodeReuse/ActiveRecord
+        def within_shards(shard_names)
+          reflect(
+            query
+              .joins(fdw_inner_join_projects)
+              .merge(projects_within_shards(shard_names))
+          )
+        end
+        # rubocop:enable CodeReuse/ActiveRecord
+
+        private
+
+        def base_query
+          ::Geo::ProjectRegistry.select(project_registries_table[Arel.star])
+        end
+
+        def reflect(query)
+          self.class.new(query)
+        end
+
+        def project_registries_table
+          ::Geo::ProjectRegistry.arel_table
+        end
+
+        def fdw_projects_table
+          ::Geo::Fdw::Project.arel_table
+        end
+
+        def fdw_repository_state_table
+          ::Geo::Fdw::ProjectRepositoryState.arel_table
+        end
+
+        def fdw_inner_join_projects
+          project_registries_table
+            .join(fdw_projects_table, Arel::Nodes::InnerJoin)
+            .on(project_registries_table[:project_id].eq(fdw_projects_table[:id]))
+            .join_sources
+        end
+
+        def fdw_inner_join_repository_state
+          project_registries_table
+            .join(fdw_repository_state_table, Arel::Nodes::InnerJoin)
+            .on(project_registries_table[:project_id].eq(fdw_repository_state_table[:project_id]))
+            .join_sources
+        end
+
+        def projects_within_namespaces(namespace_ids)
+          ::Geo::Fdw::Project.within_namespaces(namespace_ids)
+        end
+
+        def projects_within_shards(shard_names)
+          ::Geo::Fdw::Project.within_shards(shard_names)
+        end
+
+        def repositories_pending_verification
+          ::Geo::ProjectRegistry
+            .repositories_pending_verification
+            .and(fdw_repository_state_table[:repository_verification_checksum].not_eq(nil))
+        end
+
+        def wikis_pending_verification
+          ::Geo::ProjectRegistry
+            .wikis_pending_verification
+            .and(fdw_repository_state_table[:wiki_verification_checksum].not_eq(nil))
+        end
+      end
+    end
+  end
+end

ee/spec/finders/geo/legacy_project_registry_pending_verification_finder_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Geo::LegacyProjectRegistryPendingVerificationFinder, :geo do
+  include EE::GeoHelpers
+
+  describe '#execute' do
+    let(:node) { create(:geo_node) }
+
+    subject { described_class.new(current_node: node, shard_name: 'default', batch_size: 100) }
+
+    it 'does not return registries that are verified on primary and secondary' do
+      project_verified    = create(:repository_state, :repository_verified, :wiki_verified).project
+      repository_verified = create(:repository_state, :repository_verified).project
+      wiki_verified       = create(:repository_state, :wiki_verified).project
+
+      create(:geo_project_registry, :repository_verified, :wiki_verified, project: project_verified)
+      create(:geo_project_registry, :repository_verified, project: repository_verified)
+      create(:geo_project_registry, :wiki_verified, project: wiki_verified)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries that were unverified/outdated on primary' do
+      project_unverified_primary  = create(:project)
+      project_outdated_primary    = create(:repository_state, :repository_outdated, :wiki_outdated).project
+      repository_outdated_primary = create(:repository_state, :repository_outdated, :wiki_verified).project
+      wiki_outdated_primary       = create(:repository_state, :repository_verified, :wiki_outdated).project
+
+      create(:geo_project_registry, project: project_unverified_primary)
+      create(:geo_project_registry, :repository_verification_outdated, :wiki_verification_outdated, project: project_outdated_primary)
+      create(:geo_project_registry, :repository_verified, :wiki_verified, project: repository_outdated_primary)
+      create(:geo_project_registry, :repository_verified, :wiki_verified, project: wiki_outdated_primary)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'returns registries that were unverified/outdated on secondary' do
+      project_unverified_secondary  = create(:repository_state, :repository_verified, :wiki_verified).project
+      project_outdated_secondary    = create(:repository_state, :repository_verified, :wiki_verified).project
+      repository_outdated_secondary = create(:repository_state, :repository_verified, :wiki_verified).project
+      wiki_outdated_secondary       = create(:repository_state, :repository_verified, :wiki_verified).project
+
+      registry_unverified_secondary          = create(:geo_project_registry, :synced, project: project_unverified_secondary)
+      registry_outdated_secondary            = create(:geo_project_registry, :synced, :repository_verification_outdated, :wiki_verification_outdated, project: project_outdated_secondary)
+      registry_repository_outdated_secondary = create(:geo_project_registry, :synced, :repository_verification_outdated, :wiki_verified, project: repository_outdated_secondary)
+      registry_wiki_outdated_secondary       = create(:geo_project_registry, :synced, :repository_verified, :wiki_verification_outdated, project: wiki_outdated_secondary)
+
+      expect(subject.execute)
+        .to contain_exactly(
+          registry_unverified_secondary,
+          registry_outdated_secondary,
+          registry_repository_outdated_secondary,
+          registry_wiki_outdated_secondary
+        )
+    end
+
+    it 'does not return registries that failed on primary' do
+      verification_failed_primary = create(:repository_state, :repository_failed, :wiki_failed).project
+
+      create(:geo_project_registry, project: verification_failed_primary)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'returns registries where one failed and one verified on the primary' do
+      verification_failed_primary = create(:repository_state, :repository_failed, :wiki_failed).project
+      repository_failed_primary   = create(:repository_state, :repository_failed, :wiki_verified).project
+      wiki_failed_primary         = create(:repository_state, :repository_verified, :wiki_failed).project
+
+      create(:geo_project_registry, :synced, project: verification_failed_primary)
+      registry_repository_failed_primary = create(:geo_project_registry, :synced, project: repository_failed_primary)
+      registry_wiki_failed_primary       = create(:geo_project_registry, :synced, project: wiki_failed_primary)
+
+      expect(subject.execute)
+        .to contain_exactly(
+          registry_repository_failed_primary,
+          registry_wiki_failed_primary
+        )
+    end
+
+    it 'does not return registries where verification failed on secondary' do
+      verification_failed_secondary = create(:repository_state, :repository_verified, :wiki_verified).project
+      repository_failed_secondary   = create(:repository_state, :repository_verified).project
+      wiki_failed_secondary         = create(:repository_state, :wiki_verified).project
+
+      create(:geo_project_registry, :repository_verification_failed, :wiki_verification_failed, project: verification_failed_secondary)
+      create(:geo_project_registry, :repository_verification_failed, project: repository_failed_secondary)
+      create(:geo_project_registry, :wiki_verification_failed, project: wiki_failed_secondary)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries when the repo needs to be resynced' do
+      project_verified = create(:repository_state, :repository_verified).project
+      create(:geo_project_registry, :repository_sync_failed, project: project_verified)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries when the wiki needs to be resynced' do
+      project_verified = create(:repository_state, :wiki_verified).project
+      create(:geo_project_registry, :wiki_sync_failed, project: project_verified)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries when the repository is missing on primary' do
+      project_verified = create(:repository_state, :repository_verified).project
+      create(:geo_project_registry, :synced, project: project_verified, repository_missing_on_primary: true)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries when the wiki is missing on primary' do
+      project_verified = create(:repository_state, :wiki_verified).project
+      create(:geo_project_registry, :synced, project: project_verified, wiki_missing_on_primary: true)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries where projects belongs to other shards' do
+      project_broken_storage = create(:project, :broken_storage)
+      create(:repository_state, :repository_verified, :wiki_verified, project: project_broken_storage)
+      create(:geo_project_registry, :synced, project: project_broken_storage)
+
+      expect(subject.execute).to be_empty
+    end
+
+    context 'with selective sync by namespace' do
+      it 'returns registries where projects belongs to the namespaces' do
+        group_1 = create(:group)
+        group_2 = create(:group)
+        nested_group_1 = create(:group, parent: group_1)
+        project_1 = create(:project, group: group_1)
+        project_2 = create(:project, group: nested_group_1)
+        project_3 = create(:project, group: group_2)
+
+        create(:repository_state, :repository_verified, :wiki_verified, project: project_1)
+        create(:repository_state, :repository_verified, :wiki_verified, project: project_2)
+        create(:repository_state, :repository_verified, :wiki_verified, project: project_3)
+
+        registry_unverified_secondary = create(:geo_project_registry, :synced, project: project_1)
+        registry_outdated_secondary = create(:geo_project_registry, :synced, :repository_verification_outdated, :wiki_verification_outdated, project: project_2)
+        create(:geo_project_registry, :synced, :repository_verification_outdated, :wiki_verified, project: project_3)
+
+        node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+        expect(subject.execute)
+          .to contain_exactly(
+            registry_unverified_secondary,
+            registry_outdated_secondary
+          )
+      end
+    end
+
+    context 'with selective sync by shard' do
+      let(:project_broken_storage) { create(:project, :broken_storage) }
+      let!(:repository_state_project_broken_storage) { create(:repository_state, :repository_verified, :wiki_verified, project: project_broken_storage) }
+      let!(:registry_repository_broken_shard) { create(:geo_project_registry, :synced, project: project_broken_storage) }
+
+      let(:project) { create(:project) }
+      let!(:project_unverified_secondary) { create(:repository_state, :repository_verified, :wiki_verified, project: project) }
+      let!(:registry_unverified_secondary) { create(:geo_project_registry, :synced, project: project) }
+
+      before do
+        node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+      end
+
+      it 'does not return registries when selected shards to sync does not include the shard_name' do
+        subject = described_class.new(current_node: node, shard_name: 'default', batch_size: 100)
+
+        expect(subject.execute).to be_empty
+      end
+
+      it 'returns registries where projects belongs to the shards' do
+        subject = described_class.new(current_node: node, shard_name: 'broken', batch_size: 100)
+
+        expect(subject.execute).to contain_exactly(registry_repository_broken_shard)
+      end
+    end
+  end
+end

ee/spec/finders/geo/project_registry_finder_spec.rb

 require 'spec_helper'
 
 describe Geo::ProjectRegistryFinder, :geo do
+  using RSpec::Parameterized::TableSyntax
+
   include ::EE::GeoHelpers
 
   # Using let() instead of set() because set() does not work properly
@@ -450,126 +452,6 @@ describe Geo::ProjectRegistryFinder, :geo do
         end
       end
     end
-
-    describe '#find_registries_to_verify' do
-      it 'delegates to the correct method' do
-        expect(subject).to receive("#{method_prefix}_find_registries_to_verify".to_sym).and_call_original
-
-        subject.find_registries_to_verify(shard_name: 'default', batch_size: 10)
-      end
-
-      it 'does not return registries that are verified on primary and secondary' do
-        project_verified    = create(:repository_state, :repository_verified, :wiki_verified).project
-        repository_verified = create(:repository_state, :repository_verified).project
-        wiki_verified       = create(:repository_state, :wiki_verified).project
-
-        create(:geo_project_registry, :repository_verified, :wiki_verified, project: project_verified)
-        create(:geo_project_registry, :repository_verified, project: repository_verified)
-        create(:geo_project_registry, :wiki_verified, project: wiki_verified)
-
-        expect(subject.find_registries_to_verify(shard_name: 'default', batch_size: 100)).to be_empty
-      end
-
-      it 'does not return registries that were unverified/outdated on primary' do
-        project_unverified_primary  = create(:project)
-        project_outdated_primary    = create(:repository_state, :repository_outdated, :wiki_outdated).project
-        repository_outdated_primary = create(:repository_state, :repository_outdated, :wiki_verified).project
-        wiki_outdated_primary       = create(:repository_state, :repository_verified, :wiki_outdated).project
-
-        create(:geo_project_registry, project: project_unverified_primary)
-        create(:geo_project_registry, :repository_verification_outdated, :wiki_verification_outdated, project: project_outdated_primary)
-        create(:geo_project_registry, :repository_verified, :wiki_verified, project: repository_outdated_primary)
-        create(:geo_project_registry, :repository_verified, :wiki_verified, project: wiki_outdated_primary)
-
-        expect(subject.find_registries_to_verify(shard_name: 'default', batch_size: 100)).to be_empty
-      end
-
-      it 'returns registries that were unverified/outdated on secondary' do
-        # Secondary unverified/outdated
-        project_unverified_secondary  = create(:repository_state, :repository_verified, :wiki_verified).project
-        project_outdated_secondary    = create(:repository_state, :repository_verified, :wiki_verified).project
-        repository_outdated_secondary = create(:repository_state, :repository_verified, :wiki_verified).project
-        wiki_outdated_secondary       = create(:repository_state, :repository_verified, :wiki_verified).project
-
-        registry_unverified_secondary          = create(:geo_project_registry, :synced, project: project_unverified_secondary)
-        registry_outdated_secondary            = create(:geo_project_registry, :synced, :repository_verification_outdated, :wiki_verification_outdated, project: project_outdated_secondary)
-        registry_repository_outdated_secondary = create(:geo_project_registry, :synced, :repository_verification_outdated, :wiki_verified, project: repository_outdated_secondary)
-        registry_wiki_outdated_secondary       = create(:geo_project_registry, :synced, :repository_verified, :wiki_verification_outdated, project: wiki_outdated_secondary)
-
-        expect(subject.find_registries_to_verify(shard_name: 'default', batch_size: 100))
-          .to match_array([
-            registry_unverified_secondary,
-            registry_outdated_secondary,
-            registry_repository_outdated_secondary,
-            registry_wiki_outdated_secondary
-          ])
-      end
-
-      it 'does not return registries that failed on primary' do
-        verification_failed_primary = create(:repository_state, :repository_failed, :wiki_failed).project
-
-        create(:geo_project_registry, project: verification_failed_primary)
-
-        expect(subject.find_registries_to_verify(shard_name: 'default', batch_size: 100)).to be_empty
-      end
-
-      it 'returns registries where one failed and one verified on the primary' do
-        verification_failed_primary = create(:repository_state, :repository_failed, :wiki_failed).project
-        repository_failed_primary   = create(:repository_state, :repository_failed, :wiki_verified).project
-        wiki_failed_primary         = create(:repository_state, :repository_verified, :wiki_failed).project
-
-        create(:geo_project_registry, :synced, project: verification_failed_primary)
-        registry_repository_failed_primary = create(:geo_project_registry, :synced, project: repository_failed_primary)
-        registry_wiki_failed_primary       = create(:geo_project_registry, :synced, project: wiki_failed_primary)
-
-        expect(subject.find_registries_to_verify(shard_name: 'default', batch_size: 100))
-          .to match_array([
-            registry_repository_failed_primary,
-            registry_wiki_failed_primary
-          ])
-      end
-
-      it 'does not return registries where verification failed on secondary' do
-        # Verification failed on secondary
-        verification_failed_secondary = create(:repository_state, :repository_verified, :wiki_verified).project
-        repository_failed_secondary   = create(:repository_state, :repository_verified).project
-        wiki_failed_secondary         = create(:repository_state, :wiki_verified).project
-
-        create(:geo_project_registry, :repository_verification_failed, :wiki_verification_failed, project: verification_failed_secondary)
-        create(:geo_project_registry, :repository_verification_failed, project: repository_failed_secondary)
-        create(:geo_project_registry, :wiki_verification_failed, project: wiki_failed_secondary)
-
-        expect(subject.find_registries_to_verify(shard_name: 'default', batch_size: 100)).to be_empty
-      end
-
-      it 'does not return registries when the repo needs to be resynced' do
-        project_verified = create(:repository_state, :repository_verified).project
-        create(:geo_project_registry, :repository_sync_failed, project: project_verified)
-
-        expect(subject.find_registries_to_verify(shard_name: 'default', batch_size: 100)).to be_empty
-      end
-
-      it 'does not return registries when the wiki needs to be resynced' do
-        project_verified = create(:repository_state, :wiki_verified).project
-        create(:geo_project_registry, :wiki_sync_failed, project: project_verified)
-
-        expect(subject.find_registries_to_verify(shard_name: 'default', batch_size: 100)).to be_empty
-      end
-
-      it 'does not return registries when the repository is missing on primary' do
-        project_verified = create(:repository_state, :repository_verified).project
-        create(:geo_project_registry, :synced, project: project_verified, repository_missing_on_primary: true)
-
-        expect(subject.find_registries_to_verify(shard_name: 'default', batch_size: 100)).to be_empty
-      end
-
-      it 'does not return registries when the wiki is missing on primary' do
-        project_verified = create(:repository_state, :wiki_verified).project
-        create(:geo_project_registry, :synced, project: project_verified, wiki_missing_on_primary: true)
-
-        expect(subject.find_registries_to_verify(shard_name: 'default', batch_size: 100)).to be_empty
-      end
-    end
   end
 
   # Disable transactions via :delete method because a foreign table
@@ -606,4 +488,33 @@ describe Geo::ProjectRegistryFinder, :geo do
     include_examples 'counts all the things', 'legacy'
     include_examples 'finds all the things', 'legacy'
   end
+
+  describe '#find_registries_to_verify', :delete do
+    where(:selective_sync, :fdw_enabled, :use_fdw_queries_for_selective_sync, :finder) do
+      false | false | false | Geo::LegacyProjectRegistryPendingVerificationFinder
+      false | false | true  | Geo::LegacyProjectRegistryPendingVerificationFinder
+      false | true  | true  | Geo::ProjectRegistryPendingVerificationFinder
+      false | true  | false | Geo::ProjectRegistryPendingVerificationFinder
+      true  | false | false | Geo::LegacyProjectRegistryPendingVerificationFinder
+      true  | false | true  | Geo::LegacyProjectRegistryPendingVerificationFinder
+      true  | true  | true  | Geo::ProjectRegistryPendingVerificationFinder
+      true  | true  | false | Geo::LegacyProjectRegistryPendingVerificationFinder
+    end
+
+    with_them do
+      before do
+        stub_fdw(fdw_enabled)
+        stub_feature_flags(use_fdw_queries_for_selective_sync: use_fdw_queries_for_selective_sync)
+        stub_selective_sync(secondary, selective_sync)
+      end
+
+      it 'delegates to Geo::ProjectRegistryPendingVerificationFinder' do
+        expect_next_instance_of(finder, current_node: secondary, shard_name: 'default', batch_size: 100) do |finder|
+          expect(finder).to receive(:execute).once
+        end
+
+        subject.find_registries_to_verify(shard_name: 'default', batch_size: 100)
+      end
+    end
+  end
 end

ee/spec/finders/geo/project_registry_pending_verification_finder_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Geo::ProjectRegistryPendingVerificationFinder, :geo do
+  # Disable transactions via :delete method because a foreign table
+  # can't see changes inside a transaction of a different connection.
+  describe '#execute', :delete do
+    let(:node) { create(:geo_node) }
+
+    before do
+      skip('FDW is not configured') unless Gitlab::Geo::Fdw.enabled?
+    end
+
+    subject { described_class.new(current_node: node, shard_name: 'default', batch_size: 100) }
+
+    it 'does not return registries that are verified on primary and secondary' do
+      project_verified    = create(:repository_state, :repository_verified, :wiki_verified).project
+      repository_verified = create(:repository_state, :repository_verified).project
+      wiki_verified       = create(:repository_state, :wiki_verified).project
+
+      create(:geo_project_registry, :repository_verified, :wiki_verified, project: project_verified)
+      create(:geo_project_registry, :repository_verified, project: repository_verified)
+      create(:geo_project_registry, :wiki_verified, project: wiki_verified)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries that were unverified/outdated on primary' do
+      project_unverified_primary  = create(:project)
+      project_outdated_primary    = create(:repository_state, :repository_outdated, :wiki_outdated).project
+      repository_outdated_primary = create(:repository_state, :repository_outdated, :wiki_verified).project
+      wiki_outdated_primary       = create(:repository_state, :repository_verified, :wiki_outdated).project
+
+      create(:geo_project_registry, project: project_unverified_primary)
+      create(:geo_project_registry, :repository_verification_outdated, :wiki_verification_outdated, project: project_outdated_primary)
+      create(:geo_project_registry, :repository_verified, :wiki_verified, project: repository_outdated_primary)
+      create(:geo_project_registry, :repository_verified, :wiki_verified, project: wiki_outdated_primary)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'returns registries that were unverified/outdated on secondary' do
+      project_unverified_secondary  = create(:repository_state, :repository_verified, :wiki_verified).project
+      project_outdated_secondary    = create(:repository_state, :repository_verified, :wiki_verified).project
+      repository_outdated_secondary = create(:repository_state, :repository_verified, :wiki_verified).project
+      wiki_outdated_secondary       = create(:repository_state, :repository_verified, :wiki_verified).project
+
+      registry_unverified_secondary          = create(:geo_project_registry, :synced, project: project_unverified_secondary)
+      registry_outdated_secondary            = create(:geo_project_registry, :synced, :repository_verification_outdated, :wiki_verification_outdated, project: project_outdated_secondary)
+      registry_repository_outdated_secondary = create(:geo_project_registry, :synced, :repository_verification_outdated, :wiki_verified, project: repository_outdated_secondary)
+      registry_wiki_outdated_secondary       = create(:geo_project_registry, :synced, :repository_verified, :wiki_verification_outdated, project: wiki_outdated_secondary)
+
+      expect(subject.execute)
+        .to contain_exactly(
+          registry_unverified_secondary,
+          registry_outdated_secondary,
+          registry_repository_outdated_secondary,
+          registry_wiki_outdated_secondary
+        )
+    end
+
+    it 'does not return registries that failed on primary' do
+      verification_failed_primary = create(:repository_state, :repository_failed, :wiki_failed).project
+
+      create(:geo_project_registry, project: verification_failed_primary)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'returns registries where one failed and one verified on the primary' do
+      verification_failed_primary = create(:repository_state, :repository_failed, :wiki_failed).project
+      repository_failed_primary   = create(:repository_state, :repository_failed, :wiki_verified).project
+      wiki_failed_primary         = create(:repository_state, :repository_verified, :wiki_failed).project
+
+      create(:geo_project_registry, :synced, project: verification_failed_primary)
+      registry_repository_failed_primary = create(:geo_project_registry, :synced, project: repository_failed_primary)
+      registry_wiki_failed_primary       = create(:geo_project_registry, :synced, project: wiki_failed_primary)
+
+      expect(subject.execute)
+        .to contain_exactly(
+          registry_repository_failed_primary,
+          registry_wiki_failed_primary
+        )
+    end
+
+    it 'does not return registries where verification failed on secondary' do
+      verification_failed_secondary = create(:repository_state, :repository_verified, :wiki_verified).project
+      repository_failed_secondary   = create(:repository_state, :repository_verified).project
+      wiki_failed_secondary         = create(:repository_state, :wiki_verified).project
+
+      create(:geo_project_registry, :repository_verification_failed, :wiki_verification_failed, project: verification_failed_secondary)
+      create(:geo_project_registry, :repository_verification_failed, project: repository_failed_secondary)
+      create(:geo_project_registry, :wiki_verification_failed, project: wiki_failed_secondary)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries when the repo needs to be resynced' do
+      project_verified = create(:repository_state, :repository_verified).project
+      create(:geo_project_registry, :repository_sync_failed, project: project_verified)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries when the wiki needs to be resynced' do
+      project_verified = create(:repository_state, :wiki_verified).project
+      create(:geo_project_registry, :wiki_sync_failed, project: project_verified)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries when the repository is missing on primary' do
+      project_verified = create(:repository_state, :repository_verified).project
+      create(:geo_project_registry, :synced, project: project_verified, repository_missing_on_primary: true)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries when the wiki is missing on primary' do
+      project_verified = create(:repository_state, :wiki_verified).project
+      create(:geo_project_registry, :synced, project: project_verified, wiki_missing_on_primary: true)
+
+      expect(subject.execute).to be_empty
+    end
+
+    it 'does not return registries where projects belongs to other shards' do
+      project_broken_storage = create(:project, :broken_storage)
+      create(:repository_state, :repository_verified, :wiki_verified, project: project_broken_storage)
+      create(:geo_project_registry, :synced, project: project_broken_storage)
+
+      expect(subject.execute).to be_empty
+    end
+
+    context 'with selective sync by namespace' do
+      it 'returns registries where projects belongs to the namespaces' do
+        group_1 = create(:group)
+        group_2 = create(:group)
+        nested_group_1 = create(:group, parent: group_1)
+        project_1 = create(:project, group: group_1)
+        project_2 = create(:project, group: nested_group_1)
+        project_3 = create(:project, group: group_2)
+
+        create(:repository_state, :repository_verified, :wiki_verified, project: project_1)
+        create(:repository_state, :repository_verified, :wiki_verified, project: project_2)
+        create(:repository_state, :repository_verified, :wiki_verified, project: project_3)
+
+        registry_unverified_secondary = create(:geo_project_registry, :synced, project: project_1)
+        registry_outdated_secondary = create(:geo_project_registry, :synced, :repository_verification_outdated, :wiki_verification_outdated, project: project_2)
+        create(:geo_project_registry, :synced, :repository_verification_outdated, :wiki_verified, project: project_3)
+
+        node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+        expect(subject.execute)
+          .to contain_exactly(
+            registry_unverified_secondary,
+            registry_outdated_secondary
+          )
+      end
+    end
+
+    context 'with selective sync by shard' do
+      let(:project_broken_storage) { create(:project, :broken_storage) }
+      let!(:repository_state_project_broken_storage) { create(:repository_state, :repository_verified, :wiki_verified, project: project_broken_storage) }
+      let!(:registry_repository_broken_shard) { create(:geo_project_registry, :synced, project: project_broken_storage) }
+
+      let(:project) { create(:project) }
+      let!(:project_unverified_secondary) { create(:repository_state, :repository_verified, :wiki_verified, project: project) }
+      let!(:registry_unverified_secondary) { create(:geo_project_registry, :synced, project: project) }
+
+      before do
+        node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+      end
+
+      it 'does not return registries when selected shards to sync does not include the shard_name' do
+        subject = described_class.new(current_node: node, shard_name: 'default', batch_size: 100)
+
+        expect(subject.execute).to be_empty
+      end
+
+      it 'returns registries where projects belongs to the shards' do
+        subject = described_class.new(current_node: node, shard_name: 'broken', batch_size: 100)
+
+        expect(subject.execute).to contain_exactly(registry_repository_broken_shard)
+      end
+    end
+  end
+end

ee/spec/support/helpers/ee/geo_helpers.rb

@@ -14,8 +14,16 @@ module EE
       allow(::Gitlab::Geo).to receive(:secondary?).and_return(true)
     end
 
+    def stub_fdw(value)
+      allow(::Gitlab::Geo::Fdw).to receive(:enabled?).and_return(value)
+    end
+
     def stub_fdw_disabled
-      allow(::Gitlab::Geo::Fdw).to receive(:enabled?).and_return(false)
+      stub_fdw(false)
+    end
+
+    def stub_selective_sync(node, value)
+      allow(node).to receive(:selective_sync?).and_return(value)
     end
   end
 end