Merge branch '11002-geo-selective-sync-support-for-the-lfs-objects-fdw-queries' into 'master'

Geo - Implement selective sync support for the LFS objects FDW queries

Closes #11002

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

ee/app/finders/geo/expire_uploads_finder.rb

@@ -2,9 +2,11 @@
 
 module Geo
   class ExpireUploadsFinder
+    UPLOAD_TYPE = 'file'
+
     def find_project_uploads(project)
       if Gitlab::Geo::Fdw.enabled?
-        Geo::Fdw::Upload.for_model_with_type(project, 'file')
+        Geo::Fdw::Upload.for_model_with_type(project, UPLOAD_TYPE)
       else
         legacy_find_project_uploads(project)
       end
@@ -12,9 +14,9 @@ module Geo
 
     def find_file_registries_uploads(project)
       if Gitlab::Geo::Fdw.enabled?
-        Gitlab::Geo::Fdw::FileRegistryQueryBuilder.new
+        Gitlab::Geo::Fdw::UploadRegistryQueryBuilder.new
           .for_model(project)
-          .with_type('file')
+          .with_type(UPLOAD_TYPE)
       else
         legacy_find_file_registries_uploads(project)
       end
@@ -28,13 +30,12 @@ module Geo
       return Geo::FileRegistry.none if upload_ids.empty?
 
       values_sql = upload_ids.map { |id| "(#{id})" }.join(',')
-      upload_type = 'file'
 
       Geo::FileRegistry.joins(<<~SQL)
         JOIN (VALUES #{values_sql})
           AS uploads (id)
           ON uploads.id = file_registry.file_id
-         AND file_registry.file_type='#{upload_type}'
+         AND file_registry.file_type='#{UPLOAD_TYPE}'
       SQL
     end
     # rubocop:enable CodeReuse/ActiveRecord

ee/app/finders/geo/legacy_lfs_object_registry_finder.rb

+# frozen_string_literal: true
+
+module Geo
+  class LegacyLfsObjectRegistryFinder < RegistryFinder
+    def syncable
+      current_node.lfs_objects.syncable
+    end
+
+    def lfs_objects_synced
+      legacy_inner_join_registry_ids(
+        syncable,
+        Geo::FileRegistry.lfs_objects.synced.pluck_file_key,
+        LfsObject
+      )
+    end
+
+    def lfs_objects_failed
+      legacy_inner_join_registry_ids(
+        syncable,
+        Geo::FileRegistry.lfs_objects.failed.pluck_file_key,
+        LfsObject
+      )
+    end
+
+    def lfs_objects_unsynced(except_file_ids:)
+      registry_file_ids = Geo::FileRegistry.lfs_objects.pluck_file_key | except_file_ids
+
+      legacy_left_outer_join_registry_ids(
+        syncable,
+        registry_file_ids,
+        LfsObject
+      )
+    end
+
+    def lfs_objects_migrated_local(except_file_ids:)
+      legacy_inner_join_registry_ids(
+        current_node.lfs_objects.with_files_stored_remotely,
+        Geo::FileRegistry.lfs_objects.file_id_not_in(except_file_ids).pluck_file_key,
+        LfsObject
+      )
+    end
+
+    def lfs_objects_synced_missing_on_primary
+      legacy_inner_join_registry_ids(
+        syncable,
+        Geo::FileRegistry.lfs_objects.synced.missing_on_primary.pluck_file_key,
+        LfsObject
+      )
+    end
+
+    def registries_for_lfs_objects
+      return Geo::FileRegistry.lfs_objects unless selective_sync?
+
+      legacy_inner_join_registry_ids(
+        Geo::FileRegistry.lfs_objects,
+        current_node.lfs_objects.pluck_primary_key,
+        Geo::FileRegistry,
+        foreign_key: :file_id
+      )
+    end
+  end
+end

ee/app/finders/geo/lfs_object_registry_finder.rb

@@ -7,31 +7,29 @@ module Geo
     end
 
     def count_synced
-      if aggregate_pushdown_supported?
-        find_synced.count
-      else
-        legacy_find_synced.count
-      end
+      lfs_objects_synced.count
     end
 
     def count_failed
-      if aggregate_pushdown_supported?
-        find_failed.count
-      else
-        legacy_find_failed.count
-      end
+      lfs_objects_failed.count
     end
 
     def count_synced_missing_on_primary
-      if aggregate_pushdown_supported? && !use_legacy_queries?
-        fdw_find_synced_missing_on_primary.count
-      else
-        legacy_find_synced_missing_on_primary.count
-      end
+      lfs_objects_synced_missing_on_primary.count
     end
 
     def count_registry
-      Geo::FileRegistry.lfs_objects.count
+      registries_for_lfs_objects.count
+    end
+
+    def syncable
+      if use_legacy_queries_for_selective_sync?
+        legacy_finder.syncable
+      elsif selective_sync?
+        fdw_geo_node.lfs_objects.syncable
+      else
+        LfsObject.syncable
+      end
     end
 
     # Find limited amount of non replicated lfs objects.
@@ -44,206 +42,111 @@ module Geo
     #
     # @param [Integer] batch_size used to limit the results returned
     # @param [Array<Integer>] except_file_ids ids that will be ignored from the query
-    # rubocop: disable CodeReuse/ActiveRecord
+    # rubocop:disable CodeReuse/ActiveRecord
     def find_unsynced(batch_size:, except_file_ids: [])
       relation =
-        if use_legacy_queries?
-          legacy_find_unsynced(except_file_ids: except_file_ids)
+        if use_legacy_queries_for_selective_sync?
+          legacy_finder.lfs_objects_unsynced(except_file_ids: except_file_ids)
         else
-          fdw_find_unsynced(except_file_ids: except_file_ids)
+          lfs_objects_unsynced(except_file_ids: except_file_ids)
         end
 
       relation.limit(batch_size)
     end
-    # rubocop: enable CodeReuse/ActiveRecord
+    # rubocop:enable CodeReuse/ActiveRecord
 
-    # rubocop: disable CodeReuse/ActiveRecord
+    # rubocop:disable CodeReuse/ActiveRecord
     def find_migrated_local(batch_size:, except_file_ids: [])
       relation =
-        if use_legacy_queries?
-          legacy_find_migrated_local(except_file_ids: except_file_ids)
+        if use_legacy_queries_for_selective_sync?
+          legacy_finder.lfs_objects_migrated_local(except_file_ids: except_file_ids)
         else
-          fdw_find_migrated_local(except_file_ids: except_file_ids)
+          lfs_objects_migrated_local(except_file_ids: except_file_ids)
         end
 
       relation.limit(batch_size)
     end
-    # rubocop: enable CodeReuse/ActiveRecord
+    # rubocop:enable CodeReuse/ActiveRecord
 
-    def syncable
-      all.geo_syncable
-    end
-
-    # rubocop: disable CodeReuse/ActiveRecord
+    # rubocop:disable CodeReuse/ActiveRecord
     def find_retryable_failed_registries(batch_size:, except_file_ids: [])
-      find_failed_registries
-        .retry_due
-        .where.not(file_id: except_file_ids)
+      registries_for_lfs_objects
+        .merge(Geo::FileRegistry.failed)
+        .merge(Geo::FileRegistry.retry_due)
+        .file_id_not_in(except_file_ids)
         .limit(batch_size)
     end
-    # rubocop: enable CodeReuse/ActiveRecord
+    # rubocop:enable CodeReuse/ActiveRecord
 
-    # rubocop: disable CodeReuse/ActiveRecord
+    # rubocop:disable CodeReuse/ActiveRecord
     def find_retryable_synced_missing_on_primary_registries(batch_size:, except_file_ids: [])
-      find_synced_missing_on_primary_registries
+      registries_for_lfs_objects
+        .synced
+        .missing_on_primary
         .retry_due
-        .where.not(file_id: except_file_ids)
+        .file_id_not_in(except_file_ids)
         .limit(batch_size)
     end
-    # rubocop: enable CodeReuse/ActiveRecord
+    # rubocop:enable CodeReuse/ActiveRecord
 
     private
 
-    def all
-      if selective_sync?
-        LfsObject.project_id_in(current_node.projects)
-      else
-        LfsObject.all
-      end
+    # rubocop:disable CodeReuse/Finder
+    def legacy_finder
+      @legacy_finder ||= Geo::LegacyLfsObjectRegistryFinder.new(current_node: current_node)
     end
+    # rubocop:enable CodeReuse/Finder
 
-    def find_synced
-      if use_legacy_queries?
-        legacy_find_synced
-      else
-        fdw_find_synced
-      end
+    def fdw_geo_node
+      @fdw_geo_node ||= Geo::Fdw::GeoNode.find(current_node.id)
     end
 
-    def find_failed
-      if use_legacy_queries?
-        legacy_find_failed
+    def registries_for_lfs_objects
+      if use_legacy_queries_for_selective_sync?
+        legacy_finder.registries_for_lfs_objects
       else
-        fdw_find_failed
+        fdw_geo_node.lfs_object_registries
       end
     end
 
-    def find_synced_registries
-      Geo::FileRegistry.lfs_objects.synced
-    end
-
-    def find_failed_registries
-      Geo::FileRegistry.lfs_objects.failed
-    end
-
-    def find_synced_missing_on_primary_registries
-      find_synced_registries.missing_on_primary
+    def lfs_objects_synced
+      if use_legacy_queries_for_selective_sync?
+        legacy_finder.lfs_objects_synced
+      else
+        fdw_geo_node.lfs_objects.synced
+      end
     end
 
-    #
-    # FDW accessors
-    #
-
-    # rubocop: disable CodeReuse/ActiveRecord
-    def fdw_find
-      fdw_all.joins("INNER JOIN file_registry ON file_registry.file_id = #{fdw_table}.id")
-        .geo_syncable
-        .merge(Geo::FileRegistry.lfs_objects)
+    def lfs_objects_failed
+      if use_legacy_queries_for_selective_sync?
+        legacy_finder.lfs_objects_failed
+      else
+        fdw_geo_node.lfs_objects.failed
+      end
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
-    # rubocop: disable CodeReuse/ActiveRecord
-    def fdw_find_unsynced(except_file_ids:)
-      fdw_all.joins("LEFT OUTER JOIN file_registry
-                                          ON file_registry.file_id = #{fdw_table}.id
-                                         AND file_registry.file_type = 'lfs'")
-        .geo_syncable
-        .where(file_registry: { id: nil })
-        .where.not(id: except_file_ids)
+    def lfs_objects_unsynced(except_file_ids:)
+      fdw_geo_node
+        .lfs_objects
+        .syncable
+        .missing_file_registry
+        .id_not_in(except_file_ids)
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
-    # rubocop: disable CodeReuse/ActiveRecord
-    def fdw_find_migrated_local(except_file_ids:)
-      fdw_all.joins("INNER JOIN file_registry ON file_registry.file_id = #{fdw_table}.id")
+    def lfs_objects_migrated_local(except_file_ids:)
+      fdw_geo_node
+        .lfs_objects
+        .inner_join_file_registry
         .with_files_stored_remotely
-        .where.not(id: except_file_ids)
-        .merge(Geo::FileRegistry.lfs_objects)
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    def fdw_find_synced
-      fdw_find.merge(Geo::FileRegistry.synced)
+        .id_not_in(except_file_ids)
     end
 
-    def fdw_find_synced_missing_on_primary
-      fdw_find.merge(Geo::FileRegistry.synced.missing_on_primary)
-    end
-
-    def fdw_find_failed
-      fdw_find.merge(Geo::FileRegistry.failed)
-    end
-
-    # rubocop: disable CodeReuse/ActiveRecord
-    def fdw_all
-      if selective_sync?
-        Geo::Fdw::LfsObject.joins(:project).where(projects: { id: current_node.projects })
+    def lfs_objects_synced_missing_on_primary
+      if use_legacy_queries_for_selective_sync?
+        legacy_finder.lfs_objects_synced_missing_on_primary
       else
-        Geo::Fdw::LfsObject.all
+        fdw_geo_node.lfs_objects.synced.missing_on_primary
       end
     end
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    def fdw_table
-      Geo::Fdw::LfsObject.table_name
-    end
-
-    #
-    # Legacy accessors (non FDW)
-    #
-
-    # rubocop: disable CodeReuse/ActiveRecord
-    def legacy_find_synced
-      legacy_inner_join_registry_ids(
-        syncable,
-        find_synced_registries.pluck(:file_id),
-        LfsObject
-      )
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    # rubocop: disable CodeReuse/ActiveRecord
-    def legacy_find_failed
-      legacy_inner_join_registry_ids(
-        syncable,
-        find_failed_registries.pluck(:file_id),
-        LfsObject
-      )
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    # rubocop: disable CodeReuse/ActiveRecord
-    def legacy_find_unsynced(except_file_ids:)
-      registry_file_ids = Geo::FileRegistry.lfs_objects.pluck(:file_id) | except_file_ids
-
-      legacy_left_outer_join_registry_ids(
-        syncable,
-        registry_file_ids,
-        LfsObject
-      )
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    # rubocop: disable CodeReuse/ActiveRecord
-    def legacy_find_migrated_local(except_file_ids:)
-      registry_file_ids = Geo::FileRegistry.lfs_objects.pluck(:file_id) - except_file_ids
-
-      legacy_inner_join_registry_ids(
-        all.with_files_stored_remotely,
-        registry_file_ids,
-        LfsObject
-      )
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
-
-    # rubocop: disable CodeReuse/ActiveRecord
-    def legacy_find_synced_missing_on_primary
-      legacy_inner_join_registry_ids(
-        syncable,
-        find_synced_missing_on_primary_registries.pluck(:file_id),
-        LfsObject
-      )
-    end
-    # rubocop: enable CodeReuse/ActiveRecord
   end
 end

ee/app/finders/geo/project_registry_finder.rb

@@ -86,14 +86,6 @@ module Geo
 
     private
 
-    def fdw_disabled?
-      !Gitlab::Geo::Fdw.enabled?
-    end
-
-    def use_legacy_queries_for_selective_sync?
-      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

ee/app/finders/geo/registry_finder.rb

@@ -23,7 +23,11 @@ module Geo
     def use_legacy_queries?
       # Selective project replication adds a wrinkle to FDW
       # queries, so we fallback to the legacy version for now.
-      !Gitlab::Geo::Fdw.enabled? || selective_sync?
+      Gitlab::Geo::Fdw.disabled? || selective_sync?
+    end
+
+    def use_legacy_queries_for_selective_sync?
+      use_legacy_queries? && !Gitlab::Geo::Fdw.enabled_for_selective_sync?
     end
 
     # rubocop: disable CodeReuse/ActiveRecord

ee/app/models/ee/lfs_object.rb

@@ -8,12 +8,15 @@ module EE
   module LfsObject
     extend ActiveSupport::Concern
 
+    STORE_COLUMN = :file_store
+
     prepended do
+      include ObjectStorable
+
       after_destroy :log_geo_deleted_event
 
-      scope :geo_syncable, -> { with_files_stored_locally }
       scope :project_id_in, ->(ids) { joins(:projects).merge(::Project.id_in(ids)) }
-      scope :with_files_stored_remotely, -> { where(file_store: LfsObjectUploader::Store::REMOTE) }
+      scope :syncable, -> { with_files_stored_locally }
     end
 
     def log_geo_deleted_event

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

@@ -14,6 +14,19 @@ module Geo
       has_many :geo_node_namespace_links, class_name: 'Geo::Fdw::GeoNodeNamespaceLink'
       has_many :namespaces, class_name: 'Geo::Fdw::Namespace', through: :geo_node_namespace_links
 
+      def lfs_objects
+        return Geo::Fdw::LfsObject.all unless selective_sync?
+
+        Geo::Fdw::LfsObject.project_id_in(projects)
+      end
+
+      def lfs_object_registries
+        return Geo::FileRegistry.lfs_objects unless selective_sync?
+
+        Gitlab::Geo::Fdw::LfsObjectRegistryQueryBuilder.new
+          .for_lfs_objects(lfs_objects)
+      end
+
       def projects
         return Geo::Fdw::Project.all unless selective_sync?
 

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

@@ -9,7 +9,60 @@ module Geo
 
       self.table_name = Gitlab::Geo::Fdw.foreign_table_name('lfs_objects')
 
-      scope :geo_syncable, -> { with_files_stored_locally }
+      has_many :lfs_objects_projects, class_name: 'Geo::Fdw::LfsObjectsProject'
+      has_many :projects, class_name: 'Geo::Fdw::Project', through: :lfs_objects_projects
+
+      scope :project_id_in, ->(ids) { joins(:projects).merge(Geo::Fdw::Project.id_in(ids)) }
+      scope :syncable, -> { with_files_stored_locally }
+
+      class << self
+        def failed
+          inner_join_file_registry
+            .syncable
+            .merge(Geo::FileRegistry.failed)
+        end
+
+        def inner_join_file_registry
+          join_statement =
+            arel_table
+              .join(file_registry_table, Arel::Nodes::InnerJoin)
+              .on(arel_table[:id].eq(file_registry_table[:file_id]))
+
+          joins(join_statement.join_sources)
+            .merge(Geo::FileRegistry.lfs_objects)
+        end
+
+        def missing_file_registry
+          left_outer_join_file_registry
+            .where(file_registry_table[:id].eq(nil))
+        end
+
+        def missing_on_primary
+          inner_join_file_registry
+            .merge(Geo::FileRegistry.synced.missing_on_primary)
+        end
+
+        def synced
+          inner_join_file_registry
+            .syncable
+            .merge(Geo::FileRegistry.synced)
+        end
+
+        private
+
+        def file_registry_table
+          Geo::FileRegistry.arel_table
+        end
+
+        def left_outer_join_file_registry
+          join_statement =
+            arel_table
+              .join(file_registry_table, Arel::Nodes::OuterJoin)
+              .on(arel_table[:id].eq(file_registry_table[:file_id]))
+
+          joins(join_statement.join_sources)
+        end
+      end
     end
   end
 end

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

+# frozen_string_literal: true
+
+module Geo
+  module Fdw
+    class LfsObjectsProject < ::Geo::BaseFdw
+      self.table_name = Gitlab::Geo::Fdw.foreign_table_name('lfs_objects_projects')
+
+      belongs_to :lfs_object, class_name: 'Geo::Fdw::LfsObject', inverse_of: :projects
+      belongs_to :project, class_name: 'Geo::Fdw::Project', inverse_of: :lfs_objects
+    end
+  end
+end

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

@@ -7,6 +7,9 @@ module Geo
 
       self.table_name = Gitlab::Geo::Fdw.foreign_table_name('projects')
 
+      has_many :lfs_objects_projects, class_name: 'Geo::Fdw::LfsObjectsProject'
+      has_many :lfs_objects, class_name: 'Geo::Fdw::LfsObject', through: :lfs_objects_projects
+
       class << self
         def missing_project_registry
           left_outer_join_project_registry

ee/app/models/geo/file_registry.rb

@@ -3,11 +3,11 @@
 class Geo::FileRegistry < Geo::BaseRegistry
   include Geo::Syncable
 
-  scope :lfs_objects, -> { where(file_type: :lfs) }
   scope :attachments, -> { where(file_type: Geo::FileService::DEFAULT_OBJECT_TYPES) }
   scope :failed, -> { where(success: false).where.not(retry_count: nil) }
-  scope :never, -> { where(success: false, retry_count: nil) }
   scope :fresh, -> { order(created_at: :desc) }
+  scope :lfs_objects, -> { where(file_type: :lfs) }
+  scope :never, -> { where(success: false, retry_count: nil) }
   scope :with_file_type, ->(type) { where(file_type: type) }
 
   self.inheritance_column = 'file_type'
@@ -20,6 +20,18 @@ class Geo::FileRegistry < Geo::BaseRegistry
     end
   end
 
+  def self.file_id_in(ids)
+    where(file_id: ids)
+  end
+
+  def self.file_id_not_in(ids)
+    where.not(file_id: ids)
+  end
+
+  def self.pluck_file_key
+    where(nil).pluck(:file_id)
+  end
+
   def self.with_status(status)
     case status
     when 'synced', 'never', 'failed'

ee/app/models/geo_node.rb

@@ -194,6 +194,12 @@ class GeoNode < ApplicationRecord
     end
   end
 
+  def lfs_objects
+    return LfsObject.all unless selective_sync?
+
+    LfsObject.project_id_in(projects)
+  end
+
   def projects
     return Project.all unless selective_sync?
 

ee/changelogs/unreleased/11002-geo-selective-sync-support-for-the-lfs-objects-fdw-queries.yml

+---
+title: Geo - Implement selective sync support for the LFS objects FDW queries
+merge_request: 10757
+author:
+type: changed

ee/lib/gitlab/geo/fdw.rb

@@ -19,6 +19,10 @@ module Gitlab
           value.nil? ? true : value
         end
 
+        def disabled?
+          !enabled?
+        end
+
         def enabled_for_selective_sync?
           enabled? && Feature.enabled?(:use_fdw_queries_for_selective_sync)
         end

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

+# frozen_string_literal: true
+
+# Builder class to create composable queries using FDW to
+# retrieve file registries for LFS objects.
+#
+# Basic usage:
+#
+#     Gitlab::Geo::Fdw::LfsObjectRegistryQueryBuilder
+#       .new
+#       .inner_join_lfs_objects
+#
+module Gitlab
+  module Geo
+    class Fdw
+      class LfsObjectRegistryQueryBuilder < BaseQueryBuilder
+        # rubocop:disable CodeReuse/ActiveRecord
+        def for_lfs_objects(file_ids)
+          query
+            .joins(fdw_inner_join_lfs_objects)
+            .file_id_in(file_ids)
+        end
+        # rubocop:enable CodeReuse/ActiveRecord
+
+        private
+
+        def base
+          ::Geo::FileRegistry
+            .select(file_registry_table[Arel.star])
+            .lfs_objects
+        end
+
+        def file_registry_table
+          ::Geo::FileRegistry.arel_table
+        end
+
+        def fdw_lfs_object_table
+          ::Geo::Fdw::LfsObject.arel_table
+        end
+
+        def fdw_inner_join_lfs_objects
+          file_registry_table
+              .join(fdw_lfs_object_table, Arel::Nodes::InnerJoin)
+              .on(file_registry_table[:file_id].eq(fdw_lfs_object_table[:id]))
+              .join_sources
+        end
+      end
+    end
+  end
+end

ee/lib/gitlab/geo/fdw/file_registry_query_builder.rb → ee/lib/gitlab/geo/fdw/upload_registry_query_builder.rb

@@ -5,14 +5,14 @@
 #
 # Basic usage:
 #
-#     Gitlab::Geo::Fdw::FileRegistryQueryBuilder
+#     Gitlab::Geo::Fdw::UploadRegistryQueryBuilder
 #       .new
 #       .for_project_with_type(project, 'file')
 #
 module Gitlab
   module Geo
     class Fdw
-      class FileRegistryQueryBuilder < BaseQueryBuilder
+      class UploadRegistryQueryBuilder < BaseQueryBuilder
         # rubocop:disable CodeReuse/ActiveRecord
         def for_model(model)
           reflect(

ee/spec/finders/geo/lfs_object_registry_finder_spec.rb

@@ -22,6 +22,7 @@ describe Geo::LfsObjectRegistryFinder, :geo do
   let!(:lfs_object_5) { create(:lfs_object) }
   let(:lfs_object_remote_1) { create(:lfs_object, :object_storage) }
   let(:lfs_object_remote_2) { create(:lfs_object, :object_storage) }
+  let(:lfs_object_remote_3) { create(:lfs_object, :object_storage) }
 
   subject { described_class.new(current_node: secondary) }
 
@@ -30,6 +31,25 @@ describe Geo::LfsObjectRegistryFinder, :geo do
     stub_lfs_object_storage
   end
 
+  it 'responds to file registry finder methods' do
+    file_registry_finder_methods = %i{
+      syncable
+      count_syncable
+      count_synced
+      count_failed
+      count_synced_missing_on_primary
+      count_registry
+      find_unsynced
+      find_migrated_local
+      find_retryable_failed_registries
+      find_retryable_synced_missing_on_primary_registries
+    }
+
+    file_registry_finder_methods.each do |method|
+      expect(subject).to respond_to(method)
+    end
+  end
+
   shared_examples 'counts all the things' do
     describe '#count_syncable' do
       before do
@@ -85,24 +105,62 @@ describe Geo::LfsObjectRegistryFinder, :geo do
       end
     end
 
-    describe '#count_synced' do
-      it 'delegates to #legacy_find_synced' do
-        allow(subject).to receive(:aggregate_pushdown_supported?).and_return(false)
+    describe '#count_registry' do
+      it 'counts file registries for LFS objects' do
+        create(:geo_file_registry, :lfs, file_id: lfs_object_remote_1.id)
+        create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
+        create(:geo_file_registry, :lfs, file_id: lfs_object_3.id)
+        create(:geo_file_registry, :avatar)
 
-        expect(subject).to receive(:legacy_find_synced).and_call_original
+        expect(subject.count_registry).to eq 3
+      end
 
-        subject.count_synced
+      context 'with selective sync by namespace' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
+
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: unsynced_project, lfs_object: lfs_object_3)
+
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+        end
+
+        it 'counts file registries for LFS objects' do
+          create(:geo_file_registry, :lfs, file_id: lfs_object_remote_1.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_3.id)
+          create(:geo_file_registry, :avatar)
+
+          expect(subject.count_registry).to eq 1
+        end
       end
 
-      it 'delegates to #fdw_find_synced for PostgreSQL 10' do
-        allow(subject).to receive(:aggregate_pushdown_supported?).and_return(true)
-        allow(subject).to receive(:use_legacy_queries?).and_return(false)
+      context 'with selective sync by shard' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
+
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: unsynced_project, lfs_object: lfs_object_3)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_4)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_remote_1)
+
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+        end
 
-        expect(subject).to receive(:fdw_find_synced).and_return(double(count: 1))
+        it 'counts file registries for LFS objects' do
+          create(:geo_file_registry, :lfs, file_id: lfs_object_remote_1.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_4.id)
+          create(:geo_file_registry, :avatar)
 
-        subject.count_synced
+          expect(subject.count_registry).to eq 2
+        end
       end
+    end
 
+    describe '#count_synced' do
       it 'counts LFS objects that has been synced' do
         create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_1.id)
         create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
@@ -119,23 +177,17 @@ describe Geo::LfsObjectRegistryFinder, :geo do
         expect(subject.count_synced).to eq 2
       end
 
-      context 'with selective sync' do
+      context 'with selective sync by namespace' do
         before do
           allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
 
           create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_1)
-          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_2)
           create(:lfs_objects_project, project: unsynced_project, lfs_object: lfs_object_3)
 
           secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
         end
 
-        it 'delegates to #legacy_find_synced' do
-          expect(subject).to receive(:legacy_find_synced).and_call_original
-
-          subject.count_synced
-        end
-
         it 'counts LFS objects that has been synced' do
           create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_1.id)
           create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
@@ -153,25 +205,44 @@ describe Geo::LfsObjectRegistryFinder, :geo do
           expect(subject.count_synced).to eq 1
         end
       end
-    end
 
-    describe '#count_failed' do
-      it 'delegates to #legacy_find_failed' do
-        allow(subject).to receive(:aggregate_pushdown_supported?).and_return(false)
+      context 'with selective sync by shard' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
 
-        expect(subject).to receive(:legacy_find_failed).and_call_original
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: unsynced_project, lfs_object: lfs_object_3)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_4)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_5)
 
-        subject.count_failed
-      end
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+        end
 
-      it 'delegates to #find_failed' do
-        allow(subject).to receive(:aggregate_pushdown_supported?).and_return(true)
+        it 'counts LFS objects that has been synced' do
+          create(:geo_file_registry, :lfs, file_id: lfs_object_1.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_3.id)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_4.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_5.id)
 
-        expect(subject).to receive(:find_failed).and_call_original
+          expect(subject.count_synced).to eq 1
+        end
+
+        it 'ignores remote LFS objects' do
+          create(:geo_file_registry, :lfs, file_id: lfs_object_1.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_3.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_4.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_5.id)
+          lfs_object_5.update_column(:file_store, ObjectStorage::Store::REMOTE)
 
-        subject.count_failed
+          expect(subject.count_synced).to eq 1
+        end
       end
+    end
 
+    describe '#count_failed' do
       it 'counts LFS objects that sync has failed' do
         create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_1.id)
         create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
@@ -188,7 +259,7 @@ describe Geo::LfsObjectRegistryFinder, :geo do
         expect(subject.count_failed).to eq 2
       end
 
-      context 'with selective sync' do
+      context 'with selective sync by namespace' do
         before do
           allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
 
@@ -199,12 +270,6 @@ describe Geo::LfsObjectRegistryFinder, :geo do
           secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
         end
 
-        it 'delegates to #legacy_find_failed' do
-          expect(subject).to receive(:legacy_find_failed).and_call_original
-
-          subject.count_failed
-        end
-
         it 'counts LFS objects that sync has failed' do
           create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_1.id)
           create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
@@ -222,26 +287,44 @@ describe Geo::LfsObjectRegistryFinder, :geo do
           expect(subject.count_failed).to eq 1
         end
       end
-    end
 
-    describe '#count_synced_missing_on_primary' do
-      it 'delegates to #legacy_find_synced_missing_on_primary' do
-        allow(subject).to receive(:aggregate_pushdown_supported?).and_return(false)
+      context 'with selective sync by shard' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
 
-        expect(subject).to receive(:legacy_find_synced_missing_on_primary).and_call_original
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: unsynced_project, lfs_object: lfs_object_3)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_4)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_5)
 
-        subject.count_synced_missing_on_primary
-      end
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+        end
 
-      it 'delegates to #fdw_find_synced_missing_on_primary for PostgreSQL 10' do
-        allow(subject).to receive(:aggregate_pushdown_supported?).and_return(true)
-        allow(subject).to receive(:use_legacy_queries?).and_return(false)
+        it 'counts LFS objects that sync has failed' do
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_1.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_3.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_4.id)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_5.id)
+
+          expect(subject.count_failed).to eq 1
+        end
 
-        expect(subject).to receive(:fdw_find_synced_missing_on_primary).and_return(double(count: 1))
+        it 'ignores remote LFS objects' do
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_1.id)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_2.id)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_3.id)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_4.id)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_5.id)
+          lfs_object_5.update_column(:file_store, ObjectStorage::Store::REMOTE)
 
-        subject.count_synced_missing_on_primary
+          expect(subject.count_failed).to eq 1
+        end
       end
+    end
 
+    describe '#count_synced_missing_on_primary' do
       it 'counts LFS objects that have been synced and are missing on the primary' do
         create(:geo_file_registry, :lfs, file_id: lfs_object_1.id, missing_on_primary: true)
 
@@ -250,14 +333,16 @@ describe Geo::LfsObjectRegistryFinder, :geo do
 
       it 'excludes LFS objects that are not missing on the primary' do
         create(:geo_file_registry, :lfs, file_id: lfs_object_1.id)
+        create(:geo_file_registry, :lfs, file_id: lfs_object_2.id, missing_on_primary: true)
 
-        expect(subject.count_synced_missing_on_primary).to eq 0
+        expect(subject.count_synced_missing_on_primary).to eq 1
       end
 
       it 'excludes LFS objects that are not synced' do
         create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_1.id, missing_on_primary: true)
+        create(:geo_file_registry, :lfs, file_id: lfs_object_2.id, missing_on_primary: true)
 
-        expect(subject.count_synced_missing_on_primary).to eq 0
+        expect(subject.count_synced_missing_on_primary).to eq 1
       end
 
       it 'ignores remote LFS objects' do
@@ -266,7 +351,7 @@ describe Geo::LfsObjectRegistryFinder, :geo do
         expect(subject.count_synced_missing_on_primary).to eq 0
       end
 
-      context 'with selective sync' do
+      context 'with selective sync by namespace' do
         before do
           allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
 
@@ -277,12 +362,6 @@ describe Geo::LfsObjectRegistryFinder, :geo do
           secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
         end
 
-        it 'delegates to #legacy_find_synced_missing_on_primary' do
-          expect(subject).to receive(:legacy_find_synced_missing_on_primary).and_call_original
-
-          subject.count_synced_missing_on_primary
-        end
-
         it 'counts LFS objects that has been synced' do
           create(:geo_file_registry, :lfs, file_id: lfs_object_1.id, missing_on_primary: true)
           create(:geo_file_registry, :lfs, file_id: lfs_object_2.id, missing_on_primary: true)
@@ -293,8 +372,9 @@ describe Geo::LfsObjectRegistryFinder, :geo do
 
         it 'ignores remote LFS objects' do
           create(:geo_file_registry, :lfs, file_id: lfs_object_remote_1.id, missing_on_primary: true)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_2.id, missing_on_primary: true)
 
-          expect(subject.count_synced_missing_on_primary).to eq 0
+          expect(subject.count_synced_missing_on_primary).to eq 1
         end
       end
     end
@@ -302,12 +382,6 @@ describe Geo::LfsObjectRegistryFinder, :geo do
 
   shared_examples 'finds all the things' do
     describe '#find_unsynced' do
-      it 'delegates to the correct method' do
-        expect(subject).to receive("#{method_prefix}_find_unsynced".to_sym).and_call_original
-
-        subject.find_unsynced(batch_size: 10)
-      end
-
       it 'returns LFS objects without an entry on the tracking database' do
         create(:geo_file_registry, :lfs, file_id: lfs_object_1.id)
         create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_3.id)
@@ -325,15 +399,69 @@ describe Geo::LfsObjectRegistryFinder, :geo do
 
         expect(lfs_objects).to match_ids(lfs_object_4, lfs_object_5)
       end
-    end
 
-    describe '#find_migrated_local' do
-      it 'delegates to the correct method' do
-        expect(subject).to receive("#{method_prefix}_find_migrated_local".to_sym).and_call_original
+      context 'with selective sync by namespace' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
+
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_3)
+          create(:lfs_objects_project, project: unsynced_project, lfs_object: lfs_object_4)
 
-        subject.find_migrated_local(batch_size: 10)
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+        end
+
+        it 'returns LFS objects without an entry on the tracking database' do
+          create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
+
+          lfs_objects = subject.find_unsynced(batch_size: 10)
+
+          expect(lfs_objects).to match_ids(lfs_object_1, lfs_object_3)
+        end
+
+        it 'ignores remote LFS objects' do
+          create(:geo_file_registry, :lfs, file_id: lfs_object_2.id)
+          lfs_object_1.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+          lfs_objects = subject.find_unsynced(batch_size: 10)
+
+          expect(lfs_objects).to match_ids(lfs_object_3)
+        end
       end
 
+      context 'with selective sync by shard' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
+
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: unsynced_project, lfs_object: lfs_object_3)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_4)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_5)
+
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+        end
+
+        it 'counts LFS objects that sync has failed' do
+          create(:geo_file_registry, :lfs, file_id: lfs_object_4.id)
+
+          lfs_objects = subject.find_unsynced(batch_size: 10)
+
+          expect(lfs_objects).to match_ids(lfs_object_5)
+        end
+
+        it 'ignores remote LFS objects' do
+          lfs_object_5.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+          lfs_objects = subject.find_unsynced(batch_size: 10)
+
+          expect(lfs_objects).to match_ids(lfs_object_4)
+        end
+      end
+    end
+
+    describe '#find_migrated_local' do
       it 'returns LFS objects remotely and successfully synced locally' do
         create(:geo_file_registry, :lfs, file_id: lfs_object_remote_1.id)
 
@@ -366,8 +494,201 @@ describe Geo::LfsObjectRegistryFinder, :geo do
 
         expect(lfs_objects).to match_ids(lfs_object_remote_2)
       end
+
+      context 'with selective sync by namespace' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
+
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_remote_1)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_remote_2)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: unsynced_project, lfs_object: lfs_object_remote_3)
+
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+        end
+
+        it 'returns LFS objects remotely and successfully synced locally' do
+          create(:geo_file_registry, :lfs, file_id: lfs_object_remote_2.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_remote_3.id)
+
+          lfs_objects = subject.find_migrated_local(batch_size: 10)
+
+          expect(lfs_objects).to match_ids(lfs_object_remote_2)
+        end
+      end
+
+      context 'with selective sync by shard' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
+
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_remote_1)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_remote_2)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_remote_3)
+
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+        end
+
+        it 'returns LFS objects remotely and successfully synced locally' do
+          create(:geo_file_registry, :lfs, file_id: lfs_object_remote_1.id)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_remote_2.id)
+
+          lfs_objects = subject.find_migrated_local(batch_size: 10)
+
+          expect(lfs_objects).to match_ids(lfs_object_remote_2)
+        end
+      end
+    end
+
+    describe '#find_retryable_failed_registries' do
+      it 'returns registries for LFS objects that have failed to sync' do
+        create(:geo_file_registry, :lfs, file_id: lfs_object_1.id)
+        registry_lfs_object_2 = create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_2.id)
+        registry_lfs_object_3 = create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_3.id)
+        create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_4.id, retry_at: 1.day.from_now)
+
+        registries = subject.find_retryable_failed_registries(batch_size: 10)
+
+        expect(registries).to match_ids(registry_lfs_object_2, registry_lfs_object_3)
+      end
+
+      context 'with selective sync by namespace' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
+
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_3)
+          create(:lfs_objects_project, project: unsynced_project, lfs_object: lfs_object_4)
+
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+        end
+
+        it 'returns registries for LFS objects that have failed to sync' do
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_1.id)
+          registry_lfs_object_2 = create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_2.id)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_3.id, retry_at: 1.day.from_now)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_4.id)
+
+          registries = subject.find_retryable_failed_registries(batch_size: 10)
+
+          expect(registries).to match_ids(registry_lfs_object_2)
+        end
+      end
+
+      context 'with selective sync by shard' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
+
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_3)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_4)
+
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+        end
+
+        it 'returns registries for LFS objects that have failed to sync' do
+          registry_lfs_object_1 = create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_1.id)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_2.id)
+          create(:geo_file_registry, :lfs, :failed, file_id: lfs_object_3.id, retry_at: 1.day.from_now)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_4.id)
+
+          registries = subject.find_retryable_failed_registries(batch_size: 10)
+
+          expect(registries).to match_ids(registry_lfs_object_1)
+        end
+      end
+    end
+
+    describe '#find_retryable_synced_missing_on_primary_registries' do
+      it 'returns registries for LFS objects that have been synced and are missing on the primary' do
+        registry_lfs_object_1 = create(:geo_file_registry, :lfs, file_id: lfs_object_1.id, missing_on_primary: true, retry_at: nil)
+        registry_lfs_object_2 = create(:geo_file_registry, :lfs, file_id: lfs_object_2.id, missing_on_primary: true, retry_at: 1.day.ago)
+        create(:geo_file_registry, :lfs, file_id: lfs_object_3.id, missing_on_primary: true, retry_at: 1.day.from_now)
+        create(:geo_file_registry, :lfs, file_id: lfs_object_4.id, missing_on_primary: false)
+
+        registries = subject.find_retryable_synced_missing_on_primary_registries(batch_size: 10)
+
+        expect(registries).to match_ids(registry_lfs_object_1, registry_lfs_object_2)
+      end
+
+      context 'with selective sync by namespace' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
+
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_3)
+          create(:lfs_objects_project, project: project_broken_storage, lfs_object: lfs_object_4)
+
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+        end
+
+        it 'returns registries for LFS objects that have been synced and are missing on the primary' do
+          create(:geo_file_registry, :lfs, file_id: lfs_object_1.id, missing_on_primary: true, retry_at: nil)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_2.id, missing_on_primary: false)
+          registry_lfs_object_3 = create(:geo_file_registry, :lfs, file_id: lfs_object_3.id, missing_on_primary: true, retry_at: 1.day.ago)
+          registry_lfs_object_4 = create(:geo_file_registry, :lfs, file_id: lfs_object_4.id, missing_on_primary: true, retry_at: nil)
+
+          registries = subject.find_retryable_synced_missing_on_primary_registries(batch_size: 10)
+
+          expect(registries).to match_ids(registry_lfs_object_3, registry_lfs_object_4)
+        end
+      end
+
+      context 'with selective sync by shard' do
+        before do
+          allow_any_instance_of(LfsObjectsProject).to receive(:update_project_statistics).and_return(nil)
+
+          create(:lfs_objects_project, project: synced_project, lfs_object: lfs_object_1)
+          create(:lfs_objects_project, project: synced_project_in_nested_group, lfs_object: lfs_object_2)
+          create(:lfs_objects_project, project: unsynced_project, lfs_object: lfs_object_3)
+
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+        end
+
+        it 'returns registries for LFS objects that have been synced and are missing on the primary' do
+          registry_lfs_object_1 = create(:geo_file_registry, :lfs, file_id: lfs_object_1.id, missing_on_primary: true, retry_at: nil)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_2.id, missing_on_primary: true, retry_at: 1.day.from_now)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_3.id, missing_on_primary: true, retry_at: 1.day.ago)
+          create(:geo_file_registry, :lfs, file_id: lfs_object_4.id, missing_on_primary: false)
+
+          registries = subject.find_retryable_synced_missing_on_primary_registries(batch_size: 10)
+
+          expect(registries).to match_ids(registry_lfs_object_1)
+        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', :geo_fdw, :delete do
+    context 'with use_fdw_queries_for_selective_sync disabled' do
+      before do
+        stub_feature_flags(use_fdw_queries_for_selective_sync: false)
+      end
+
+      include_examples 'counts all the things'
+      include_examples 'finds all the things'
+    end
+
+    context 'with use_fdw_queries_for_selective_sync enabled' do
+      before do
+        stub_feature_flags(use_fdw_queries_for_selective_sync: true)
+      end
+
+      include_examples 'counts all the things'
+      include_examples 'finds all the things'
     end
   end
 
-  it_behaves_like 'a file registry finder'
+  context 'Legacy' do
+    before do
+      stub_fdw_disabled
+    end
+
+    include_examples 'counts all the things'
+    include_examples 'finds all the things'
+  end
 end

ee/spec/lib/gitlab/geo/fdw/file_registry_query_builder_spec.rb → ee/spec/lib/gitlab/geo/fdw/upload_registry_query_builder_spec.rb

@@ -4,7 +4,7 @@ require 'spec_helper'
 
 # Disable transactions via :delete method because a foreign table
 # can't see changes inside a transaction of a different connection.
-describe Gitlab::Geo::Fdw::FileRegistryQueryBuilder, :geo, :delete do
+describe Gitlab::Geo::Fdw::UploadRegistryQueryBuilder, :geo, :delete do
   let(:project) { create(:project) }
   let(:upload_1) { create(:upload, :issuable_upload, model: project) }
   let(:upload_2) { create(:upload, :issuable_upload, model: project) }

ee/spec/lib/gitlab/geo/fdw_spec.rb

@@ -44,6 +44,49 @@ describe Gitlab::Geo::Fdw, :geo do
     end
   end
 
+  describe '.enabled?' do
+    it 'returns true when foreign server does not exist' do
+      drop_foreign_server
+
+      expect(described_class.disabled?).to eq true
+    end
+
+    it 'returns true when foreign server exists but foreign schema does not exist' do
+      drop_foreign_schema
+
+      expect(described_class.disabled?).to eq true
+    end
+
+    it 'returns true when foreign server and schema exists but foreign tables are empty' do
+      drop_foreign_schema
+      create_foreign_schema
+
+      expect(described_class.disabled?).to eq true
+    end
+
+    it 'returns true when fdw is disabled in `config/database_geo.yml`' do
+      allow(Rails.configuration).to receive(:geo_database).and_return('fdw' => false)
+
+      expect(described_class.disabled?).to eq true
+    end
+
+    it 'returns false when fdw is set in `config/database_geo.yml`' do
+      allow(Rails.configuration).to receive(:geo_database).and_return('fdw' => true)
+
+      expect(described_class.disabled?).to eq false
+    end
+
+    it 'returns false when fdw is nil in `config/database_geo.yml`' do
+      allow(Rails.configuration).to receive(:geo_database).and_return('fdw' => nil)
+
+      expect(described_class.disabled?).to eq false
+    end
+
+    it 'returns false with a functional fdw environment' do
+      expect(described_class.disabled?).to eq false
+    end
+  end
+
   describe '.enabled_for_selective_sync?' do
     context 'when the feature flag is enabled' do
       before do

ee/spec/models/geo/fdw/lfs_object_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Geo::Fdw::LfsObject, :geo, type: :model do
+  context 'relationships' do
+    it { is_expected.to have_many(:lfs_objects_projects).class_name('Geo::Fdw::LfsObjectsProject') }
+    it { is_expected.to have_many(:projects).class_name('Geo::Fdw::Project').through(:lfs_objects_projects) }
+  end
+end

ee/spec/models/geo/fdw/lfs_objects_project_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Geo::Fdw::LfsObjectsProject, :geo, type: :model do
+  context 'relationships' do
+    it { is_expected.to belong_to(:lfs_object).class_name('Geo::Fdw::LfsObject').inverse_of(:projects) }
+    it { is_expected.to belong_to(:project).class_name('Geo::Fdw::Project').inverse_of(:lfs_objects) }
+  end
+end

ee/spec/models/geo/fdw/project_spec.rb

@@ -3,6 +3,11 @@
 require 'spec_helper'
 
 RSpec.describe Geo::Fdw::Project, :geo, type: :model do
+  context 'relationships' do
+    it { is_expected.to have_many(:lfs_objects_projects).class_name('Geo::Fdw::LfsObjectsProject') }
+    it { is_expected.to have_many(:lfs_objects).class_name('Geo::Fdw::LfsObject').through(:lfs_objects_projects) }
+  end
+
   describe '.search' do
     set(:test_project) { create(:project, description: 'kitten mittens') }
     set(:project) { described_class.find(test_project.id) }