Refactor methods to count failed registries

These changes refactor the finder to make it easier to remove
the legacy queries in the future.

ee/app/finders/geo/legacy_project_registry_sync_failed_finder.rb

+# frozen_string_literal: true
+
+# Finder for retrieving project registries that synchronization have
+# failed scoped to a type (repository or wiki) using cross-database
+# joins for selective sync.
+#
+# Basic usage:
+#
+#     Geo::LegacyProjectRegistrySyncFailedFinder.new(current_node: Gitlab::Geo.current_node, :repository).execute
+#
+# Valid `type` values are:
+#
+# * `:repository`
+# * `:wiki`
+#
+# Any other value will be ignored.
+module Geo
+  class LegacyProjectRegistrySyncFailedFinder < RegistryFinder
+    def initialize(current_node: nil, type:)
+      super(current_node: current_node)
+      @type = type.to_s.to_sym
+    end
+
+    def execute
+      if selective_sync?
+        failed_registries_for_selective_sync
+      else
+        failed_registries
+      end
+    end
+
+    private
+
+    attr_reader :type
+
+    def failed_registries
+      Geo::ProjectRegistry.sync_failed(type)
+    end
+
+    # rubocop: disable CodeReuse/ActiveRecord
+    def failed_registries_for_selective_sync
+      legacy_inner_join_registry_ids(
+        failed_registries,
+        current_node.projects.pluck(:id),
+        Geo::ProjectRegistry,
+        foreign_key: :project_id
+      )
+    end
+    # rubocop: enable CodeReuse/ActiveRecord
+  end
+end

ee/app/finders/geo/project_registry_finder.rb

@@ -15,19 +15,15 @@ module Geo
     end
 
     def count_failed_repositories
-      find_failed_project_registries('repository').count
+      registries_for_failed_projects(:repository).count
     end
 
     def count_failed_wikis
-      find_failed_project_registries('wiki').count
+      registries_for_failed_projects(:wiki).count
     end
 
     def find_failed_project_registries(type = nil)
-      if selective_sync?
-        legacy_find_filtered_failed_projects(type)
-      else
-        find_filtered_failed_project_registries(type)
-      end
+      registries_for_failed_projects(type)
     end
 
     def count_verified_repositories
@@ -143,21 +139,24 @@ module Geo
         .execute
     end
 
-    def find_verified_repositories
-      Geo::ProjectRegistry.verified_repos
-    end
-
-    def find_filtered_failed_project_registries(type = nil)
-      case type
-      when 'repository'
-        Geo::ProjectRegistry.failed_repos
-      when 'wiki'
-        Geo::ProjectRegistry.failed_wikis
+    def finder_klass_for_failed_registries
+      if Gitlab::Geo::Fdw.enabled_for_selective_sync?
+        Geo::ProjectRegistrySyncFailedFinder
       else
-        Geo::ProjectRegistry.failed
+        Geo::LegacyProjectRegistrySyncFailedFinder
       end
     end
 
+    def registries_for_failed_projects(type)
+      finder_klass_for_failed_registries
+        .new(current_node: current_node, type: type)
+        .execute
+    end
+
+    def find_verified_repositories
+      Geo::ProjectRegistry.verified_repos
+    end
+
     def find_filtered_verification_failed_project_registries(type = nil)
       case type
       when 'repository'
@@ -302,18 +301,6 @@ module Geo
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
-    # @return [ActiveRecord::Relation<Geo::ProjectRegistry>] list of projects that sync has failed
-    # rubocop: disable CodeReuse/ActiveRecord
-    def legacy_find_filtered_failed_projects(type = nil)
-      legacy_inner_join_registry_ids(
-        find_filtered_failed_project_registries(type),
-        current_node.projects.pluck(:id),
-        Geo::ProjectRegistry,
-        foreign_key: :project_id
-      )
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
     # @return [ActiveRecord::Relation<Geo::ProjectRegistry>] list of projects that verification has failed
     # rubocop: disable CodeReuse/ActiveRecord
     def legacy_find_filtered_verification_failed_projects(type = nil)

ee/app/finders/geo/project_registry_sync_failed_finder.rb

+# frozen_string_literal: true
+
+# Finder for retrieving project registries that synchronization have
+# failed scoped to a type (repository or wiki) using FDW queries.
+#
+# Basic usage:
+#
+#     Geo::ProjectRegistrySyncFailedFinder.new(current_node: Gitlab::Geo.current_node, :repository).execute
+#
+# Valid `type` values are:
+#
+# * `:repository`
+# * `:wiki`
+#
+# Any other value will be ignored.
+module Geo
+  class ProjectRegistrySyncFailedFinder
+    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.sync_failed(type)
+    end
+
+    private
+
+    attr_reader :current_node, :type
+  end
+end

ee/app/models/geo/project_registry.rb

@@ -99,6 +99,17 @@ class Geo::ProjectRegistry < Geo::BaseRegistry
     end
   end
 
+  def self.sync_failed(type)
+    case type
+    when :repository
+      failed_repos
+    when :wiki
+      failed_wikis
+    else
+      failed
+    end
+  end
+
   def self.flag_repositories_for_resync!
     update_all(
       resync_repository: true,

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

+---
+title: Geo - Add selective sync support for the FDW queries to count failed registries
+merge_request: 9527
+author:
+type: changed

ee/spec/finders/geo/legacy_project_registry_sync_failed_finder_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Geo::LegacyProjectRegistrySyncFailedFinder, :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!(:registry_failed) { create(:geo_project_registry, :sync_failed, project: project_1) }
+    let!(:registry_repository_failed) { create(:geo_project_registry, :synced, :repository_sync_failed, project: project_2) }
+    let!(:registry_wiki_failed) { create(:geo_project_registry, :synced, :wiki_sync_failed, project: project_3) }
+    let!(:registry_wiki_failed_broken_shard) { create(:geo_project_registry, :synced, :wiki_sync_failed, project: project_4) }
+    let!(:registry_repository_failed_broken_shard) { create(:geo_project_registry, :synced, :repository_sync_failed, project: project_5) }
+    let!(:registry_synced) { create(:geo_project_registry, :synced) }
+
+    shared_examples 'finds failed registries' do
+      context 'with repository type' do
+        subject { described_class.new(current_node: node, type: :repository) }
+
+        context 'without selective sync' do
+          it 'returns all failed registries' do
+            expect(subject.execute).to match_array([registry_failed, registry_repository_failed, registry_repository_failed_broken_shard])
+          end
+        end
+
+        context 'with selective sync by namespace' do
+          it 'returns failed registries where projects belongs to the namespaces' do
+            node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+            expect(subject.execute).to match_array([registry_failed, registry_repository_failed])
+          end
+        end
+
+        context 'with selective sync by shard' do
+          it 'returns failed registries where projects belongs to the shards' do
+            node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+            expect(subject.execute).to match_array([registry_repository_failed_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 failed registries' do
+            expect(subject.execute).to match_array([registry_failed, registry_wiki_failed, registry_wiki_failed_broken_shard])
+          end
+        end
+
+        context 'with selective sync by namespace' do
+          it 'returns failed registries where projects belongs to the namespaces' do
+            node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+            expect(subject.execute).to match_array([registry_failed, registry_wiki_failed])
+          end
+        end
+
+        context 'with selective sync by shard' do
+          it 'returns failed registries where projects belongs to the shards' do
+            node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+            expect(subject.execute).to match_array([registry_wiki_failed_broken_shard])
+          end
+        end
+      end
+
+      context 'with no type' do
+        subject { described_class.new(current_node: node, type: :invalid) }
+
+        context 'without selective sync' do
+          it 'returns all failed registries' do
+            expect(subject.execute).to match_array([registry_failed, registry_repository_failed, registry_wiki_failed, registry_repository_failed_broken_shard, registry_wiki_failed_broken_shard])
+          end
+        end
+
+        context 'with selective sync by namespace' do
+          it 'returns all failed registries where projects belongs to the namespaces' do
+            node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+            expect(subject.execute).to match_array([registry_failed, registry_repository_failed, registry_wiki_failed])
+          end
+        end
+
+        context 'with selective sync by shard' do
+          it 'returns all failed registries where projects belongs to the shards' do
+            node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+            expect(subject.execute).to match_array([registry_repository_failed_broken_shard, registry_wiki_failed_broken_shard])
+          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 failed registries'
+    end
+
+    context 'Legacy' do
+      before do
+        stub_fdw_disabled
+      end
+
+      include_examples 'finds failed registries'
+    end
+  end
+end

ee/spec/finders/geo/project_registry_finder_spec.rb

@@ -24,7 +24,7 @@ describe Geo::ProjectRegistryFinder, :geo do
     stub_current_geo_node(secondary)
   end
 
-  shared_examples 'counts all the things' do
+  shared_examples 'counts all the things' do |method_prefix|
     describe '#count_synced_repositories' do
       it 'counts repositories that have been synced' do
         create(:geo_project_registry, :sync_failed)
@@ -90,12 +90,6 @@ describe Geo::ProjectRegistryFinder, :geo do
     end
 
     describe '#count_failed_repositories' do
-      it 'delegates to #find_failed_project_registries' do
-        expect(subject).to receive(:find_failed_project_registries).with('repository').and_call_original
-
-        subject.count_failed_repositories
-      end
-
       it 'counts projects that sync has failed' do
         create(:geo_project_registry, :synced)
         create(:geo_project_registry, :sync_failed, project: project_synced)
@@ -110,12 +104,6 @@ describe Geo::ProjectRegistryFinder, :geo do
           secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
         end
 
-        it 'delegates to #find_failed_repositories' do
-          expect(subject).to receive(:find_failed_project_registries).with('repository').and_call_original
-
-          subject.count_failed_repositories
-        end
-
         it 'counts projects that sync has failed' do
           project_1_in_synced_group = create(:project, group: synced_group)
           project_2_in_synced_group = create(:project, group: synced_group)
@@ -130,12 +118,6 @@ describe Geo::ProjectRegistryFinder, :geo do
     end
 
     describe '#count_failed_wikis' do
-      it 'delegates to #find_failed_project_registries' do
-        expect(subject).to receive(:find_failed_project_registries).with('wiki').and_call_original
-
-        subject.count_failed_wikis
-      end
-
       it 'counts projects that sync has failed' do
         create(:geo_project_registry, :synced)
         create(:geo_project_registry, :sync_failed, project: project_synced)
@@ -150,12 +132,6 @@ describe Geo::ProjectRegistryFinder, :geo do
           secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
         end
 
-        it 'delegates to #find_failed_wikis' do
-          expect(subject).to receive(:find_failed_project_registries).with('wiki').and_call_original
-
-          subject.count_failed_wikis
-        end
-
         it 'counts projects that sync has failed' do
           project_1_in_synced_group = create(:project, group: synced_group)
           project_2_in_synced_group = create(:project, group: synced_group)
@@ -343,7 +319,7 @@ describe Geo::ProjectRegistryFinder, :geo do
     end
   end
 
-  shared_examples 'finds all the things' do
+  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
@@ -436,12 +412,6 @@ describe Geo::ProjectRegistryFinder, :geo do
       let!(:repository_sync_failed) { create(:geo_project_registry, :repository_sync_failed, project: project_1_in_synced_group) }
       let!(:wiki_sync_failed) { create(:geo_project_registry, :wiki_sync_failed, project: project_2_in_synced_group) }
 
-      it 'delegates to #find_failed_project_registries' do
-        expect(subject).to receive(:find_failed_project_registries).with('repository').and_call_original
-
-        subject.count_failed_repositories
-      end
-
       it 'returns only project registries that repository sync has failed' do
         expect(subject.find_failed_project_registries('repository')).to match_array([sync_failed, repository_sync_failed])
       end
@@ -455,12 +425,6 @@ describe Geo::ProjectRegistryFinder, :geo do
           secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
         end
 
-        it 'delegates to #legacy_find_filtered_failed_projects' do
-          expect(subject).to receive(:legacy_find_filtered_failed_projects).and_call_original
-
-          subject.find_failed_project_registries
-        end
-
         it 'returns project registries that sync has failed' do
           expect(subject.find_failed_project_registries).to match_array([repository_sync_failed, wiki_sync_failed])
         end
@@ -612,7 +576,8 @@ describe Geo::ProjectRegistryFinder, :geo do
         stub_feature_flags(use_fdw_queries_for_selective_sync: false)
       end
 
-      include_examples 'counts all the things'
+      include_examples 'counts all the things', 'fdw'
+      include_examples 'finds all the things', 'fdw'
     end
 
     context 'with use_fdw_queries_for_selective_sync enabled' do
@@ -620,11 +585,8 @@ describe Geo::ProjectRegistryFinder, :geo do
         stub_feature_flags(use_fdw_queries_for_selective_sync: true)
       end
 
-      include_examples 'counts all the things'
-    end
-
-    include_examples 'finds all the things' do
-      let(:method_prefix) { 'fdw' }
+      include_examples 'counts all the things', 'fdw'
+      include_examples 'finds all the things', 'fdw'
     end
   end
 
@@ -633,10 +595,7 @@ describe Geo::ProjectRegistryFinder, :geo do
       stub_fdw_disabled
     end
 
-    include_examples 'counts all the things'
-
-    include_examples 'finds all the things' do
-      let(:method_prefix) { 'legacy' }
-    end
+    include_examples 'counts all the things', 'legacy'
+    include_examples 'finds all the things', 'legacy'
   end
 end

ee/spec/finders/geo/project_registry_sync_failed_finder_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Geo::ProjectRegistrySyncFailedFinder, :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!(:registry_failed) { create(:geo_project_registry, :sync_failed, project: project_1) }
+    let!(:registry_repository_failed) { create(:geo_project_registry, :synced, :repository_sync_failed, project: project_2) }
+    let!(:registry_wiki_failed) { create(:geo_project_registry, :synced, :wiki_sync_failed, project: project_3) }
+    let!(:registry_wiki_failed_broken_shard) { create(:geo_project_registry, :synced, :wiki_sync_failed, project: project_4) }
+    let!(:registry_repository_failed_broken_shard) { create(:geo_project_registry, :synced, :repository_sync_failed, project: project_5) }
+    let!(:registry_synced) { create(:geo_project_registry, :synced) }
+
+    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 failed registries' do
+          expect(subject.execute).to match_array([registry_failed, registry_repository_failed, registry_repository_failed_broken_shard])
+        end
+      end
+
+      context 'with selective sync by namespace' do
+        it 'returns failed registries where projects belongs to the namespaces' do
+          node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+          expect(subject.execute).to match_array([registry_failed, registry_repository_failed])
+        end
+      end
+
+      context 'with selective sync by shard' do
+        it 'returns failed registries where projects belongs to the shards' do
+          node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+          expect(subject.execute).to match_array([registry_repository_failed_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 failed registries' do
+          expect(subject.execute).to match_array([registry_failed, registry_wiki_failed, registry_wiki_failed_broken_shard])
+        end
+      end
+
+      context 'with selective sync by namespace' do
+        it 'returns failed registries where projects belongs to the namespaces' do
+          node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+          expect(subject.execute).to match_array([registry_failed, registry_wiki_failed])
+        end
+      end
+
+      context 'with selective sync by shard' do
+        it 'returns failed registries where projects belongs to the shards' do
+          node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+          expect(subject.execute).to match_array([registry_wiki_failed_broken_shard])
+        end
+      end
+    end
+
+    context 'with no type' do
+      subject { described_class.new(current_node: node, type: :invalid) }
+
+      context 'without selective sync' do
+        it 'returns all failed registries' do
+          expect(subject.execute).to match_array([registry_failed, registry_repository_failed, registry_wiki_failed, registry_repository_failed_broken_shard, registry_wiki_failed_broken_shard])
+        end
+      end
+
+      context 'with selective sync by namespace' do
+        it 'returns all failed registries where projects belongs to the namespaces' do
+          node.update!(selective_sync_type: 'namespaces', namespaces: [group_1, nested_group_1])
+
+          expect(subject.execute).to match_array([registry_failed, registry_repository_failed, registry_wiki_failed])
+        end
+      end
+
+      context 'with selective sync by shard' do
+        it 'returns all failed registries where projects belongs to the shards' do
+          node.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+          expect(subject.execute).to match_array([registry_repository_failed_broken_shard, registry_wiki_failed_broken_shard])
+        end
+      end
+    end
+  end
+end