diff --git a/ee/app/finders/geo/legacy_project_registry_retrying_verification_finder.rb b/ee/app/finders/geo/legacy_project_registry_retrying_verification_finder.rb
new file mode 100644
index 0000000000000000000000000000000000000000..6dba064eb75100de748ae3ffe29eee65e03cadf5
--- /dev/null
+++ b/ee/app/finders/geo/legacy_project_registry_retrying_verification_finder.rb
@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+# Finder for retrieving project registries that are retrying verification
+# scoped to a type (repository or wiki) using cross-database joins
+# for selective sync.
+#
+# Basic usage:
+#
+#     Geo::LegacyProjectRegistryRetryingVerificationFinder.new(current_node: Gitlab::Geo.current_node, :repository).execute
+#
+# Valid `type` values are:
+#
+# * `:repository`
+# * `:wiki`
+#
+# Any other value will be ignored.
+module Geo
+  class LegacyProjectRegistryRetryingVerificationFinder < RegistryFinder
+    def initialize(current_node: nil, type:)
+      super(current_node: current_node)
+      @type = type.to_s.to_sym
+    end
+
+    def execute
+      if selective_sync?
+        registries_retrying_verification_for_selective_sync
+      else
+        registries_retrying_verification
+      end
+    end
+
+    private
+
+    attr_reader :type
+
+    def registries_retrying_verification
+      Geo::ProjectRegistry.retrying_verification(type)
+    end
+
+    # rubocop: disable CodeReuse/ActiveRecord
+    def registries_retrying_verification_for_selective_sync
+      legacy_inner_join_registry_ids(
+        registries_retrying_verification,
+        current_node.projects.pluck(:id),
+        Geo::ProjectRegistry,
+        foreign_key: :project_id
+      )
+    end
+    # rubocop: enable CodeReuse/ActiveRecord
+  end
+end
diff --git a/ee/app/finders/geo/project_registry_finder.rb b/ee/app/finders/geo/project_registry_finder.rb
index 18ebc4c9f75796547d1885747c9b02e5a5b9adf0..673a5a0d186a471ed2c169ff1de971a1bd875fab 100644
--- a/ee/app/finders/geo/project_registry_finder.rb
+++ b/ee/app/finders/geo/project_registry_finder.rb
@@ -55,11 +55,11 @@ def count_wikis_checksum_mismatch
     end
 
     def count_repositories_retrying_verification
-      Geo::ProjectRegistry.repositories_retrying_verification.count
+      registries_retrying_verification(:repository).count
     end
 
     def count_wikis_retrying_verification
-      Geo::ProjectRegistry.wikis_retrying_verification.count
+      registries_retrying_verification(:wiki).count
     end
 
     def find_checksum_mismatch_project_registries(type = nil)
@@ -359,5 +359,19 @@ def registries_for_verification_failed_projects(type)
         .new(current_node: current_node, type: type)
         .execute
     end
+
+    def finder_klass_for_registries_retrying_verification
+      if Gitlab::Geo::Fdw.enabled_for_selective_sync?
+        Geo::ProjectRegistryRetryingVerificationFinder
+      else
+        Geo::LegacyProjectRegistryRetryingVerificationFinder
+      end
+    end
+
+    def registries_retrying_verification(type)
+      finder_klass_for_registries_retrying_verification
+        .new(current_node: current_node, type: type)
+        .execute
+    end
   end
 end
diff --git a/ee/app/finders/geo/project_registry_retrying_verification_finder.rb b/ee/app/finders/geo/project_registry_retrying_verification_finder.rb
new file mode 100644
index 0000000000000000000000000000000000000000..883bb7aa5efd0d9ef1ab2e69e001eb583386fe30
--- /dev/null
+++ b/ee/app/finders/geo/project_registry_retrying_verification_finder.rb
@@ -0,0 +1,31 @@
+# frozen_string_literal: true
+
+# Finder for retrieving project registries that are retrying verification
+# scoped to a type (repository or wiki) using FDW queries.
+#
+# Basic usage:
+#
+#     Geo::ProjectRegistryRetryingVerificationFinder.new(current_node: Gitlab::Geo.current_node, :repository).execute
+#
+# Valid `type` values are:
+#
+# * `:repository`
+# * `:wiki`
+#
+# Any other value will be ignored.
+module Geo
+  class ProjectRegistryRetryingVerificationFinder
+    def initialize(current_node:, type:)
+      @current_node = Geo::Fdw::GeoNode.find(current_node.id)
+      @type = type.to_s.to_sym
+    end
+
+    def execute
+      current_node.project_registries.retrying_verification(type)
+    end
+
+    private
+
+    attr_reader :current_node, :type
+  end
+end
diff --git a/ee/app/models/geo/project_registry.rb b/ee/app/models/geo/project_registry.rb
index eb519c6b7024c3b544f2170adf27e0644324b80a..b39ebe8c98037c18a059935953a8f3bc8565cbee 100644
--- a/ee/app/models/geo/project_registry.rb
+++ b/ee/app/models/geo/project_registry.rb
@@ -127,6 +127,17 @@ def self.verification_failed(type)
     end
   end
 
+  def self.retrying_verification(type)
+    case type
+    when :repository
+      repositories_retrying_verification
+    when :wiki
+      wikis_retrying_verification
+    else
+      none
+    end
+  end
+
   def self.flag_repositories_for_resync!
     update_all(
       resync_repository: true,
diff --git a/ee/changelogs/unreleased/8798-geo-implement-selective-sync-support-for-fdw-queries-to-find-registries-retrying-verification.yml b/ee/changelogs/unreleased/8798-geo-implement-selective-sync-support-for-fdw-queries-to-find-registries-retrying-verification.yml
new file mode 100644
index 0000000000000000000000000000000000000000..886d44dec9c9115011049b5987a153677f70e0b1
--- /dev/null
+++ b/ee/changelogs/unreleased/8798-geo-implement-selective-sync-support-for-fdw-queries-to-find-registries-retrying-verification.yml
@@ -0,0 +1,5 @@
+---
+title: Geo - Add selective sync support for queries to find registries retrying verification
+merge_request: 10436
+author:
+type: changed
diff --git a/ee/spec/factories/geo/project_registry.rb b/ee/spec/factories/geo/project_registry.rb
index cb0cb4f382d883b318779f14f58968d5d173ae22..2bb290391f60ee2bba0704bfff42686c3c35b9f1 100644
--- a/ee/spec/factories/geo/project_registry.rb
+++ b/ee/spec/factories/geo/project_registry.rb
@@ -95,6 +95,11 @@
       last_repository_verification_failure nil
     end
 
+    trait :repository_retrying_verification do
+      repository_verification_retry_count 1
+      resync_repository true
+    end
+
     trait :wiki_verified do
       wiki_verification_checksum_sha 'e079a831cab27bcda7d81cd9b48296d0c3dd92ef'
       last_wiki_verification_failure nil
@@ -114,5 +119,10 @@
       wiki_verification_checksum_sha nil
       last_wiki_verification_failure nil
     end
+
+    trait :wiki_retrying_verification do
+      wiki_verification_retry_count 1
+      resync_wiki true
+    end
   end
 end
diff --git a/ee/spec/finders/geo/legacy_project_registry_retrying_verification_finder_spec.rb b/ee/spec/finders/geo/legacy_project_registry_retrying_verification_finder_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..9af383116e345f9993d107acbd8d1076a4ecfcc6
--- /dev/null
+++ b/ee/spec/finders/geo/legacy_project_registry_retrying_verification_finder_spec.rb
@@ -0,0 +1,119 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Geo::LegacyProjectRegistryRetryingVerificationFinder, :geo do
+  include EE::GeoHelpers
+
+  describe '#execute' do
+    let(:node) { create(:geo_node) }
+    let(:group_1) { create(:group) }
+    let(:group_2) { create(:group) }
+    let(:nested_group_1) { create(:group, parent: group_1) }
+    let(:project_1) { create(:project, group: group_1) }
+    let(:project_2) { create(:project, group: nested_group_1) }
+    let(:project_3) { create(:project, group: nested_group_1) }
+    let(:project_4) { create(:project, :broken_storage, group: group_2) }
+    let(:project_5) { create(:project, :broken_storage, group: group_2) }
+    let!(:retrying_verification) { create(:geo_project_registry, :repository_retrying_verification, :wiki_retrying_verification, project: project_1) }
+    let!(:repository_retrying_verification) { create(:geo_project_registry, :repository_retrying_verification, :wiki_verified, project: project_2) }
+    let!(:wiki_retrying_verification) { create(:geo_project_registry, :repository_verified, :wiki_retrying_verification, project: project_3) }
+    let!(:wiki_retrying_verification_broken_shard) { create(:geo_project_registry, :repository_verified, :wiki_retrying_verification, project: project_4) }
+    let!(:repository_retrying_verification_broken_shard) { create(:geo_project_registry, :repository_retrying_verification, :wiki_verified, project: project_5) }
+    let!(:verified) { create(:geo_project_registry, :repository_verified, :wiki_verified) }
+
+    shared_examples 'finds registries retrying verification' do
+      context 'with repository type' do
+        subject { described_class.new(current_node: node, type: :repository) }
+
+        context 'without selective sync' do
+          it 'returns all registries retrying verification' do
+            expect(subject.execute).to contain_exactly(retrying_verification, repository_retrying_verification, repository_retrying_verification_broken_shard)
+          end
+        end
+
+        context 'with selective sync by namespace' do
+          it 'returns registries retrying verification where projects belongs to the namespaces' do
+            node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+            expect(subject.execute).to contain_exactly(retrying_verification, repository_retrying_verification)
+          end
+        end
+
+        context 'with selective sync by shard' do
+          it 'returns registries retrying verification where projects belongs to the shards' do
+            node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+            expect(subject.execute).to contain_exactly(repository_retrying_verification_broken_shard)
+          end
+        end
+      end
+
+      context 'with wiki type' do
+        subject { described_class.new(current_node: node, type: :wiki) }
+
+        context 'without selective sync' do
+          it 'returns all registries retrying verification' do
+            expect(subject.execute).to contain_exactly(retrying_verification, wiki_retrying_verification, wiki_retrying_verification_broken_shard)
+          end
+        end
+
+        context 'with selective sync by namespace' do
+          it 'returns registries retrying verification where projects belongs to the namespaces' do
+            node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+            expect(subject.execute).to contain_exactly(retrying_verification, wiki_retrying_verification)
+          end
+        end
+
+        context 'with selective sync by shard' do
+          it 'returns registries retrying verification where projects belongs to the shards' do
+            node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+            expect(subject.execute).to contain_exactly(wiki_retrying_verification_broken_shard)
+          end
+        end
+      end
+
+      context 'with invalid type' do
+        subject { described_class.new(current_node: node, type: :invalid) }
+
+        context 'without selective sync' do
+          it 'returns nothing' do
+            expect(subject.execute).to be_empty
+          end
+        end
+
+        context 'with selective sync by namespace' do
+          it 'returns nothing' do
+            expect(subject.execute).to be_empty
+          end
+        end
+
+        context 'with selective sync by shard' do
+          it 'returns nothing' do
+            expect(subject.execute).to be_empty
+          end
+        end
+      end
+    end
+
+    # Disable transactions via :delete method because a foreign table
+    # can't see changes inside a transaction of a different connection.
+    context 'FDW', :delete do
+      before do
+        skip('FDW is not configured') unless Gitlab::Geo::Fdw.enabled?
+      end
+
+      include_examples 'finds registries retrying verification'
+    end
+
+    context 'Legacy' do
+      before do
+        stub_fdw_disabled
+      end
+
+      include_examples 'finds registries retrying verification'
+    end
+  end
+end
diff --git a/ee/spec/finders/geo/project_registry_finder_spec.rb b/ee/spec/finders/geo/project_registry_finder_spec.rb
index 1d92c8c4ff2d1fd5d76d35aeb520e12163766457..f8eaf6cc8049e6f0c73c51df7290b90851cf9a8b 100644
--- a/ee/spec/finders/geo/project_registry_finder_spec.rb
+++ b/ee/spec/finders/geo/project_registry_finder_spec.rb
@@ -195,6 +195,56 @@
         expect(subject.count_verification_failed_wikis).to eq 1
       end
     end
+
+    describe '#count_repositories_retrying_verification' do
+      before do
+        project_1_in_synced_group = create(:project, group: synced_group)
+        project_2_in_synced_group = create(:project, group: synced_group)
+
+        create(:geo_project_registry, :repository_retrying_verification, :wiki_retrying_verification, project: project_synced)
+        create(:geo_project_registry, :repository_retrying_verification, project: project_1_in_synced_group)
+        create(:geo_project_registry, :wiki_retrying_verification, project: project_2_in_synced_group)
+      end
+
+      it 'counts registries that repository retrying verification' do
+        expect(subject.count_repositories_retrying_verification).to eq 2
+      end
+
+      context 'with selective sync' do
+        before do
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+        end
+
+        it 'counts registries that repository retrying verification' do
+          expect(subject.count_repositories_retrying_verification).to eq 1
+        end
+      end
+    end
+
+    describe '#count_wikis_retrying_verification' do
+      before do
+        project_1_in_synced_group = create(:project, group: synced_group)
+        project_2_in_synced_group = create(:project, group: synced_group)
+
+        create(:geo_project_registry, :repository_retrying_verification, :wiki_retrying_verification, project: project_synced)
+        create(:geo_project_registry, :repository_retrying_verification, project: project_1_in_synced_group)
+        create(:geo_project_registry, :wiki_retrying_verification, project: project_2_in_synced_group)
+      end
+
+      it 'counts registries that wiki retrying verification' do
+        expect(subject.count_wikis_retrying_verification).to eq 2
+      end
+
+      context 'with selective sync' do
+        before do
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+        end
+
+        it 'counts registries that wiki retrying verification' do
+          expect(subject.count_wikis_retrying_verification).to eq 1
+        end
+      end
+    end
   end
 
   describe '#find_checksum_mismatch_project_registries' do
diff --git a/ee/spec/finders/geo/project_registry_retrying_verification_finder_spec.rb b/ee/spec/finders/geo/project_registry_retrying_verification_finder_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..ad9bf10a2bca7fe96697fa6234a958840e11a1a3
--- /dev/null
+++ b/ee/spec/finders/geo/project_registry_retrying_verification_finder_spec.rb
@@ -0,0 +1,103 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Geo::ProjectRegistryRetryingVerificationFinder, :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) }
+    let(:group_1) { create(:group) }
+    let(:group_2) { create(:group) }
+    let(:nested_group_1) { create(:group, parent: group_1) }
+    let(:project_1) { create(:project, group: group_1) }
+    let(:project_2) { create(:project, group: nested_group_1) }
+    let(:project_3) { create(:project, group: nested_group_1) }
+    let(:project_4) { create(:project, :broken_storage, group: group_2) }
+    let(:project_5) { create(:project, :broken_storage, group: group_2) }
+    let!(:retrying_verification) { create(:geo_project_registry, :repository_retrying_verification, :wiki_retrying_verification, project: project_1) }
+    let!(:repository_retrying_verification) { create(:geo_project_registry, :repository_retrying_verification, :wiki_verified, project: project_2) }
+    let!(:wiki_retrying_verification) { create(:geo_project_registry, :repository_verified, :wiki_retrying_verification, project: project_3) }
+    let!(:wiki_retrying_verification_broken_shard) { create(:geo_project_registry, :repository_verified, :wiki_retrying_verification, project: project_4) }
+    let!(:repository_retrying_verification_broken_shard) { create(:geo_project_registry, :repository_retrying_verification, :wiki_verified, project: project_5) }
+    let!(:verified) { create(:geo_project_registry, :repository_verified, :wiki_verified) }
+
+    before do
+      skip('FDW is not configured') unless Gitlab::Geo::Fdw.enabled?
+    end
+
+    context 'with repository type' do
+      subject { described_class.new(current_node: node, type: :repository) }
+
+      context 'without selective sync' do
+        it 'returns all registries retrying verification' do
+          expect(subject.execute).to contain_exactly(retrying_verification, repository_retrying_verification, repository_retrying_verification_broken_shard)
+        end
+      end
+
+      context 'with selective sync by namespace' do
+        it 'returns registries retrying verification where projects belongs to the namespaces' do
+          node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+          expect(subject.execute).to contain_exactly(retrying_verification, repository_retrying_verification)
+        end
+      end
+
+      context 'with selective sync by shard' do
+        it 'returns registries retrying verification where projects belongs to the shards' do
+          node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+          expect(subject.execute).to contain_exactly(repository_retrying_verification_broken_shard)
+        end
+      end
+    end
+
+    context 'with wiki type' do
+      subject { described_class.new(current_node: node, type: :wiki) }
+
+      context 'without selective sync' do
+        it 'returns all registries retrying verification' do
+          expect(subject.execute).to contain_exactly(retrying_verification, wiki_retrying_verification, wiki_retrying_verification_broken_shard)
+        end
+      end
+
+      context 'with selective sync by namespace' do
+        it 'returns registries retrying verification where projects belongs to the namespaces' do
+          node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+          expect(subject.execute).to contain_exactly(retrying_verification, wiki_retrying_verification)
+        end
+      end
+
+      context 'with selective sync by shard' do
+        it 'returns registries retrying verification where projects belongs to the shards' do
+          node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+          expect(subject.execute).to contain_exactly(wiki_retrying_verification_broken_shard)
+        end
+      end
+    end
+
+    context 'with invalid type' do
+      subject { described_class.new(current_node: node, type: :invalid) }
+
+      context 'without selective sync' do
+        it 'returns nothing' do
+          expect(subject.execute).to be_empty
+        end
+      end
+
+      context 'with selective sync by namespace' do
+        it 'returns nothing' do
+          expect(subject.execute).to be_empty
+        end
+      end
+
+      context 'with selective sync by shard' do
+        it 'returns nothing' do
+          expect(subject.execute).to be_empty
+        end
+      end
+    end
+  end
+end