Add LFS object replication using the self service framework

* Default behavior is to use the legacy LFS object replication
* When the geo_lfs_object_replication_ssf feature flag is enabled,
uses the self service framework

https://gitlab.com/gitlab-org/gitlab/-/merge_requests/56286

doc/api/graphql/reference/index.md

@@ -5239,6 +5239,29 @@ The edge type for [`Label`](#label).
 | <a id="labeledgecursor"></a>`cursor` | [`String!`](#string) | A cursor for use in pagination. |
 | <a id="labeledgenode"></a>`node` | [`Label`](#label) | The item at the end of the edge. |
 
+#### `LfsObjectRegistryConnection`
+
+The connection type for [`LfsObjectRegistry`](#lfsobjectregistry).
+
+##### Fields
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="lfsobjectregistryconnectionedges"></a>`edges` | [`[LfsObjectRegistryEdge]`](#lfsobjectregistryedge) | A list of edges. |
+| <a id="lfsobjectregistryconnectionnodes"></a>`nodes` | [`[LfsObjectRegistry]`](#lfsobjectregistry) | A list of nodes. |
+| <a id="lfsobjectregistryconnectionpageinfo"></a>`pageInfo` | [`PageInfo!`](#pageinfo) | Information to aid in pagination. |
+
+#### `LfsObjectRegistryEdge`
+
+The edge type for [`LfsObjectRegistry`](#lfsobjectregistry).
+
+##### Fields
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="lfsobjectregistryedgecursor"></a>`cursor` | [`String!`](#string) | A cursor for use in pagination. |
+| <a id="lfsobjectregistryedgenode"></a>`node` | [`LfsObjectRegistry`](#lfsobjectregistry) | The item at the end of the edge. |
+
 #### `LicenseHistoryEntryConnection`
 
 The connection type for [`LicenseHistoryEntry`](#licensehistoryentry).
@@ -8317,6 +8340,22 @@ four standard [pagination arguments](#connection-pagination-arguments):
 | ---- | ---- | ----------- |
 | <a id="geonodegroupwikirepositoryregistriesids"></a>`ids` | [`[ID!]`](#id) | Filters registries by their ID. |
 
+##### `GeoNode.lfsObjectRegistries`
+
+Find LFS object registries on this Geo node. Available only when feature flag `geo_lfs_object_replication` is enabled.
+
+Returns [`LfsObjectRegistryConnection`](#lfsobjectregistryconnection).
+
+This field returns a [connection](#connections). It accepts the
+four standard [pagination arguments](#connection-pagination-arguments):
+`before: String`, `after: String`, `first: Int`, `last: Int`.
+
+###### Arguments
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="geonodelfsobjectregistriesids"></a>`ids` | [`[ID!]`](#id) | Filters registries by their ID. |
+
 ##### `GeoNode.mergeRequestDiffRegistries`
 
 Find merge request diff registries on this Geo node.
@@ -9327,6 +9366,23 @@ four standard [pagination arguments](#connection-pagination-arguments):
 | <a id="labeltitle"></a>`title` | [`String!`](#string) | Content of the label. |
 | <a id="labelupdatedat"></a>`updatedAt` | [`Time!`](#time) | When this label was last updated. |
 
+### `LfsObjectRegistry`
+
+Represents the Geo sync and verification state of an LFS object.
+
+#### Fields
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="lfsobjectregistrycreatedat"></a>`createdAt` | [`Time`](#time) | Timestamp when the LfsObjectRegistry was created. |
+| <a id="lfsobjectregistryid"></a>`id` | [`ID!`](#id) | ID of the LfsObjectRegistry. |
+| <a id="lfsobjectregistrylastsyncfailure"></a>`lastSyncFailure` | [`String`](#string) | Error message during sync of the LfsObjectRegistry. |
+| <a id="lfsobjectregistrylastsyncedat"></a>`lastSyncedAt` | [`Time`](#time) | Timestamp of the most recent successful sync of the LfsObjectRegistry. |
+| <a id="lfsobjectregistrylfsobjectid"></a>`lfsObjectId` | [`ID!`](#id) | ID of the LFS object. |
+| <a id="lfsobjectregistryretryat"></a>`retryAt` | [`Time`](#time) | Timestamp after which the LfsObjectRegistry should be resynced. |
+| <a id="lfsobjectregistryretrycount"></a>`retryCount` | [`Int`](#int) | Number of consecutive failed sync attempts of the LfsObjectRegistry. |
+| <a id="lfsobjectregistrystate"></a>`state` | [`RegistryState`](#registrystate) | Sync state of the LfsObjectRegistry. |
+
 ### `LicenseHistoryEntry`
 
 Represents an entry from the Cloud License history.

ee/app/finders/geo/lfs_object_legacy_registry_finder.rb

+# frozen_string_literal: true
+
+module Geo
+  class LfsObjectLegacyRegistryFinder < FileRegistryFinder
+    def registry_class
+      Geo::LfsObjectRegistry
+    end
+  end
+end

ee/app/finders/geo/lfs_object_registry_finder.rb

 # frozen_string_literal: true
 
 module Geo
-  class LfsObjectRegistryFinder < FileRegistryFinder
-    def registry_class
-      Geo::LfsObjectRegistry
-    end
+  class LfsObjectRegistryFinder
+    include FrameworkRegistryFinder
   end
 end

ee/app/graphql/resolvers/geo/lfs_object_registries_resolver.rb

+# frozen_string_literal: true
+
+module Resolvers
+  module Geo
+    class LfsObjectRegistriesResolver < BaseResolver
+      type ::Types::Geo::GeoNodeType.connection_type, null: true
+
+      include RegistriesResolver
+    end
+  end
+end

ee/app/graphql/types/geo/geo_node_type.rb

@@ -42,6 +42,11 @@ module Types
             null: true,
             resolver: ::Resolvers::Geo::GroupWikiRepositoryRegistriesResolver,
             description: 'Find group wiki repository registries on this Geo node.'
+      field :lfs_object_registries, ::Types::Geo::LfsObjectRegistryType.connection_type,
+            null: true,
+            resolver: ::Resolvers::Geo::LfsObjectRegistriesResolver,
+            description: 'Find LFS object registries on this Geo node.',
+            feature_flag: :geo_lfs_object_replication
       field :pipeline_artifact_registries, ::Types::Geo::PipelineArtifactRegistryType.connection_type,
             null: true,
             resolver: ::Resolvers::Geo::PipelineArtifactRegistriesResolver,

ee/app/graphql/types/geo/lfs_object_registry_type.rb

+# frozen_string_literal: true
+
+module Types
+  module Geo
+    # rubocop:disable Graphql/AuthorizeTypes because it is included
+    class LfsObjectRegistryType < BaseObject
+      include ::Types::Geo::RegistryType
+
+      graphql_name 'LfsObjectRegistry'
+      description 'Represents the Geo sync and verification state of an LFS object'
+
+      field :lfs_object_id, GraphQL::ID_TYPE, null: false, description: 'ID of the LFS object.'
+    end
+  end
+end

ee/app/helpers/ee/geo_helper.rb

@@ -172,14 +172,6 @@ module EE
           name: 'wiki',
           name_plural: 'wikis'
         },
-        {
-          data_type: 'blob',
-          data_type_title: _('File'),
-          title: _('LFS object'),
-          title_plural: _('LFS objects'),
-          name: 'lfs_object',
-          name_plural: 'lfs_objects'
-        },
         {
           data_type: 'blob',
           data_type_title: _('File'),
@@ -216,6 +208,17 @@ module EE
         }
       ]
 
+      if ::Feature.disabled?(:geo_lfs_object_replication)
+        replicable_types.insert(2, {
+          data_type: 'blob',
+          data_type_title: _('File'),
+          title: _('LFS object'),
+          title_plural: _('LFS objects'),
+          name: 'lfs_object',
+          name_plural: 'lfs_objects'
+        })
+      end
+
       # Adds all the SSF Data Types automatically
       enabled_replicator_classes.each do |replicator_class|
         replicable_types.push(

ee/app/models/concerns/geo/verifiable_replicator.rb

@@ -223,7 +223,9 @@ module Geo
     #
     # @abstract
     def primary_checksum
-      model_record.verification_checksum
+      # If verification is not yet setup, then model_record will not have the verification_checksum
+      # attribute yet. Returning nil is fine here
+      model_record.verification_checksum if model_record.respond_to?(:verification_checksum)
     end
 
     def secondary_checksum

ee/app/models/ee/lfs_object.rb

@@ -11,6 +11,10 @@ module EE
     STORE_COLUMN = :file_store
 
     prepended do
+      include ::Gitlab::Geo::ReplicableModel
+
+      with_replicator Geo::LfsObjectReplicator
+
       after_destroy :log_geo_deleted_event
 
       scope :project_id_in, ->(ids) { joins(:projects).merge(::Project.id_in(ids)) }
@@ -21,11 +25,23 @@ module EE
       # @return [ActiveRecord::Relation<LfsObject>] everything that should be synced to this node, restricted by primary key
       def replicables_for_current_secondary(primary_key_in)
         node = ::Gitlab::Geo.current_node
-        local_storage_only = !node.sync_object_storage
+        node.lfs_objects(primary_key_in: primary_key_in)
+          .merge(selective_sync_scope(node))
+          .merge(object_storage_scope(node))
+      end
+
+      private
+
+      def object_storage_scope(node)
+        return all if node.sync_object_storage?
+
+        with_files_stored_locally
+      end
+
+      def selective_sync_scope(node)
+        return all unless node.selective_sync?
 
-        scope = node.lfs_objects(primary_key_in: primary_key_in)
-        scope = scope.with_files_stored_locally if local_storage_only
-        scope
+        project_id_in(node.projects)
       end
     end
 

ee/app/models/geo/lfs_object_registry.rb

 # frozen_string_literal: true
 
 class Geo::LfsObjectRegistry < Geo::BaseRegistry
-  include ::ShaAttribute
   include ::Geo::Syncable
+  include ::Geo::ReplicableRegistry
+  include ::ShaAttribute
+
+  sha_attribute :sha256
 
   MODEL_CLASS = ::LfsObject
   MODEL_FOREIGN_KEY = :lfs_object_id
 
-  sha_attribute :sha256
-
   belongs_to :lfs_object, class_name: 'LfsObject'
 
+  def self.registry_consistency_worker_enabled?
+    if ::Feature.enabled?(:geo_lfs_object_replication)
+      replicator_class.enabled?
+    else
+      true
+    end
+  end
+
+  def self.failed
+    if Feature.enabled?(:geo_lfs_object_replication)
+      with_state(:failed)
+    else
+      where(success: false).where.not(retry_count: nil)
+    end
+  end
+
+  def self.never_attempted_sync
+    if Feature.enabled?(:geo_lfs_object_replication)
+      pending.where(last_synced_at: nil)
+    else
+      where(success: false, retry_count: nil)
+    end
+  end
+
+  def self.retry_due
+    if Feature.enabled?(:geo_lfs_object_replication)
+      where(arel_table[:retry_at].eq(nil).or(arel_table[:retry_at].lt(Time.current)))
+    else
+      where('retry_at is NULL OR retry_at < ?', Time.current)
+    end
+  end
+
+  def self.synced
+    if Feature.enabled?(:geo_lfs_object_replication)
+      with_state(:synced).or(where(success: true))
+    else
+      where(success: true)
+    end
+  end
+
   # If false, RegistryConsistencyService will frequently check the end of the
   # table to quickly handle new replicables.
   def self.has_create_events?

ee/app/models/geo_node_status.rb

@@ -65,9 +65,6 @@ class GeoNodeStatus < ApplicationRecord
     repositories_synced_count
     repositories_failed_count
     lfs_objects_replication_enabled
-    lfs_objects_count
-    lfs_objects_synced_count
-    lfs_objects_failed_count
     attachments_replication_enabled
     attachments_count
     attachments_synced_count
@@ -84,7 +81,6 @@ class GeoNodeStatus < ApplicationRecord
     wikis_verified_count
     wikis_verification_failed_count
     wikis_verification_total_count
-    lfs_objects_synced_missing_on_primary_count
     job_artifacts_synced_missing_on_primary_count
     attachments_synced_missing_on_primary_count
     repositories_checksummed_count
@@ -401,7 +397,6 @@ class GeoNodeStatus < ApplicationRecord
   attr_in_percentage :wikis_synced,                  :wikis_synced_count,                  :wikis_count
   attr_in_percentage :wikis_checksummed,             :wikis_checksummed_count,             :wikis_count
   attr_in_percentage :wikis_verified,                :wikis_verified_count,                :wikis_count
-  attr_in_percentage :lfs_objects_synced,            :lfs_objects_synced_count,            :lfs_objects_count
   attr_in_percentage :job_artifacts_synced,          :job_artifacts_synced_count,          :job_artifacts_count
   attr_in_percentage :attachments_synced,            :attachments_synced_count,            :attachments_count
   attr_in_percentage :replication_slots_used,        :replication_slots_used_count,        :replication_slots_count
@@ -497,6 +492,7 @@ class GeoNodeStatus < ApplicationRecord
 
   def load_lfs_objects_data
     return unless lfs_objects_replication_enabled
+    return if Feature.enabled?(:geo_lfs_object_replication)
 
     self.lfs_objects_count = lfs_objects_finder.registry_count
     self.lfs_objects_synced_count = lfs_objects_finder.synced_count
@@ -624,7 +620,7 @@ class GeoNodeStatus < ApplicationRecord
   end
 
   def lfs_objects_finder
-    @lfs_objects_finder ||= Geo::LfsObjectRegistryFinder.new
+    @lfs_objects_finder ||= Geo::LfsObjectLegacyRegistryFinder.new
   end
 
   def job_artifacts_finder

ee/app/replicators/geo/lfs_object_replicator.rb

+# frozen_string_literal: true
+
+module Geo
+  class LfsObjectReplicator < Gitlab::Geo::Replicator
+    include ::Geo::BlobReplicatorStrategy
+
+    def carrierwave_uploader
+      model_record.file
+    end
+
+    def self.model
+      ::LfsObject
+    end
+
+    def self.replication_enabled_by_default?
+      false
+    end
+  end
+end

ee/app/workers/geo/file_download_dispatch_worker/lfs_object_job_finder.rb

@@ -8,7 +8,7 @@ module Geo
       FILE_SERVICE_OBJECT_TYPE = :lfs
 
       def registry_finder
-        @registry_finder ||= Geo::LfsObjectRegistryFinder.new
+        @registry_finder ||= Geo::LfsObjectLegacyRegistryFinder.new
       end
     end
   end

ee/changelogs/unreleased/lfs-ssf-behind-feature-flag.yml

+---
+title: Migrate LFS object replication to use self service framework
+merge_request: 56286
+author:
+type: changed

ee/config/feature_flags/development/geo_lfs_object_replication.yml

+---
+name: geo_lfs_object_replication
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/56286
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/276694
+milestone: '13.11'
+type: development
+group: group::geo
+default_enabled: false

ee/lib/ee/api/entities/geo_node_status.rb

@@ -36,7 +36,7 @@ module EE
         expose :db_replication_lag_seconds
 
         expose :attachments_replication_enabled
-        expose :lfs_objects_replication_enabled
+        expose :lfs_objects_replication_enabled, if: -> (*) { ::Feature.enabled?(:geo_lfs_object_replication) }
         expose :job_artifacts_replication_enabled
         expose :container_repositories_replication_enabled
         expose :design_repositories_replication_enabled

ee/lib/gitlab/geo.rb

@@ -20,6 +20,7 @@ module Gitlab
     # solutions can be found at
     # https://gitlab.com/gitlab-org/gitlab/-/issues/227693
     REPLICATOR_CLASSES = [
+      ::Geo::LfsObjectReplicator,
       ::Geo::MergeRequestDiffReplicator,
       ::Geo::PackageFileReplicator,
       ::Geo::TerraformStateVersionReplicator,

ee/lib/gitlab/geo/geo_node_status_check.rb

@@ -256,6 +256,8 @@ module Gitlab
       end
 
       def print_lfs_objects_status
+        return if Feature.enabled?(:geo_lfs_object_replication)
+
         print 'LFS Objects: '.rjust(GEO_STATUS_COLUMN_WIDTH)
         show_failed_value(current_node_status.lfs_objects_failed_count)
         print "#{current_node_status.lfs_objects_synced_count}/#{current_node_status.lfs_objects_count} "

ee/spec/factories/geo/lfs_object_registry.rb

 # frozen_string_literal: true
 
 FactoryBot.define do
-  factory :geo_lfs_object_registry, class: 'Geo::LfsObjectRegistry' do
+  factory :geo_lfs_object_legacy_registry, class: 'Geo::LfsObjectRegistry' do
     sequence(:lfs_object_id)
     success { true }
 
@@ -23,3 +23,28 @@ FactoryBot.define do
     end
   end
 end
+
+FactoryBot.define do
+  factory :geo_lfs_object_registry, class: 'Geo::LfsObjectRegistry' do
+    lfs_object
+    state { Geo::LfsObjectRegistry.state_value(:pending) }
+
+    trait :synced do
+      state { Geo::LfsObjectRegistry.state_value(:synced) }
+      last_synced_at { 5.days.ago }
+    end
+
+    trait :failed do
+      state { Geo::LfsObjectRegistry.state_value(:failed) }
+      last_synced_at { 1.day.ago }
+      retry_count { 2 }
+      last_sync_failure { 'Random error' }
+    end
+
+    trait :started do
+      state { Geo::LfsObjectRegistry.state_value(:started) }
+      last_synced_at { 1.day_ago }
+      retry_count { 0 }
+    end
+  end
+end

ee/spec/factories/geo_node_statuses.rb

@@ -11,10 +11,6 @@ FactoryBot.define do
       attachments_failed_count { 13 }
       attachments_synced_count { 141 }
       attachments_synced_missing_on_primary_count { 89 }
-      lfs_objects_count { 256 }
-      lfs_objects_failed_count { 12 }
-      lfs_objects_synced_count { 123 }
-      lfs_objects_synced_missing_on_primary_count { 90 }
       job_artifacts_count { 580 }
       job_artifacts_failed_count { 3 }
       job_artifacts_synced_count { 577 }
@@ -76,7 +72,6 @@ FactoryBot.define do
 
     trait :replicated_and_verified do
       attachments_failed_count { 0 }
-      lfs_objects_failed_count { 0 }
       job_artifacts_failed_count { 0 }
       container_repositories_failed_count { 0 }
       design_repositories_failed_count { 0 }
@@ -97,7 +92,6 @@ FactoryBot.define do
       wikis_checksum_total_count { 10 }
       wikis_verified_count { 10 }
       wikis_verification_total_count { 10 }
-      lfs_objects_synced_count { 10 }
       job_artifacts_synced_count { 10 }
       attachments_synced_count { 10 }
       replication_slots_used_count { 10 }
@@ -106,7 +100,6 @@ FactoryBot.define do
 
       repositories_count { 10 }
       wikis_count { 10 }
-      lfs_objects_count { 10 }
       job_artifacts_count { 10 }
       attachments_count { 10 }
       replication_slots_count { 10 }

ee/spec/finders/geo/lfs_object_legacy_registry_finder_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Geo::LfsObjectLegacyRegistryFinder, :geo do
+  before do
+    stub_feature_flags(geo_lfs_object_replication: false )
+  end
+
+  it_behaves_like 'a file registry finder' do
+    before do
+      stub_lfs_object_storage
+    end
+
+    let_it_be(:replicable_1) { create(:lfs_object) }
+    let_it_be(:replicable_2) { create(:lfs_object) }
+    let_it_be(:replicable_3) { create(:lfs_object) }
+    let_it_be(:replicable_4) { create(:lfs_object) }
+    let_it_be(:replicable_5) { create(:lfs_object) }
+    let!(:replicable_6) { create(:lfs_object, :object_storage) }
+    let!(:replicable_7) { create(:lfs_object, :object_storage) }
+    let!(:replicable_8) { create(:lfs_object, :object_storage) }
+
+    let_it_be(:registry_1) { create(:geo_lfs_object_legacy_registry, :failed, lfs_object_id: replicable_1.id) }
+    let_it_be(:registry_2) { create(:geo_lfs_object_legacy_registry, lfs_object_id: replicable_2.id, missing_on_primary: true) }
+    let_it_be(:registry_3) { create(:geo_lfs_object_legacy_registry, :never_synced, lfs_object_id: replicable_3.id) }
+    let_it_be(:registry_4) { create(:geo_lfs_object_legacy_registry, :failed, lfs_object_id: replicable_4.id) }
+    let_it_be(:registry_5) { create(:geo_lfs_object_legacy_registry, lfs_object_id: replicable_5.id, missing_on_primary: true, retry_at: 1.day.ago) }
+    let!(:registry_6) { create(:geo_lfs_object_legacy_registry, :failed, lfs_object_id: replicable_6.id) }
+    let!(:registry_7) { create(:geo_lfs_object_legacy_registry, :failed, lfs_object_id: replicable_7.id, missing_on_primary: true) }
+    let!(:registry_8) { create(:geo_lfs_object_legacy_registry, :never_synced, lfs_object_id: replicable_8.id) }
+  end
+end

ee/spec/finders/geo/lfs_object_registry_finder_spec.rb

@@ -2,28 +2,6 @@
 
 require 'spec_helper'
 
-RSpec.describe Geo::LfsObjectRegistryFinder, :geo do
-  it_behaves_like 'a file registry finder' do
-    before do
-      stub_lfs_object_storage
-    end
-
-    let_it_be(:replicable_1) { create(:lfs_object) }
-    let_it_be(:replicable_2) { create(:lfs_object) }
-    let_it_be(:replicable_3) { create(:lfs_object) }
-    let_it_be(:replicable_4) { create(:lfs_object) }
-    let_it_be(:replicable_5) { create(:lfs_object) }
-    let!(:replicable_6) { create(:lfs_object, :object_storage) }
-    let!(:replicable_7) { create(:lfs_object, :object_storage) }
-    let!(:replicable_8) { create(:lfs_object, :object_storage) }
-
-    let_it_be(:registry_1) { create(:geo_lfs_object_registry, :failed, lfs_object_id: replicable_1.id) }
-    let_it_be(:registry_2) { create(:geo_lfs_object_registry, lfs_object_id: replicable_2.id, missing_on_primary: true) }
-    let_it_be(:registry_3) { create(:geo_lfs_object_registry, :never_synced, lfs_object_id: replicable_3.id) }
-    let_it_be(:registry_4) { create(:geo_lfs_object_registry, :failed, lfs_object_id: replicable_4.id) }
-    let_it_be(:registry_5) { create(:geo_lfs_object_registry, lfs_object_id: replicable_5.id, missing_on_primary: true, retry_at: 1.day.ago) }
-    let!(:registry_6) { create(:geo_lfs_object_registry, :failed, lfs_object_id: replicable_6.id) }
-    let!(:registry_7) { create(:geo_lfs_object_registry, :failed, lfs_object_id: replicable_7.id, missing_on_primary: true) }
-    let!(:registry_8) { create(:geo_lfs_object_registry, :never_synced, lfs_object_id: replicable_8.id) }
-  end
+RSpec.describe Geo::LfsObjectRegistryFinder do
+  it_behaves_like 'a framework registry finder', :geo_lfs_object_registry
 end

ee/spec/fixtures/api/schemas/public_api/v4/geo_node_status.json

@@ -15,7 +15,14 @@
     "lfs_objects_count",
     "lfs_objects_failed_count",
     "lfs_objects_synced_count",
-    "lfs_objects_synced_missing_on_primary_count",
+    "lfs_objects_checksum_total_count",
+    "lfs_objects_checksummed_count",
+    "lfs_objects_checksum_failed_count",
+    "lfs_objects_registry_count",
+    "lfs_objects_verification_total_count",
+    "lfs_objects_verified_count",
+    "lfs_objects_verified_in_percentage",
+    "lfs_objects_verification_failed_count",
     "job_artifacts_replication_enabled",
     "job_artifacts_count",
     "job_artifacts_failed_count",
@@ -170,6 +177,14 @@
     "lfs_objects_synced_count": { "type": ["integer", "null"] },
     "lfs_objects_synced_missing_on_primary_count": { "type": ["integer", "null"] },
     "lfs_objects_synced_in_percentage": { "type": "string" },
+    "lfs_objects_checksum_total_count": { "type": ["integer", "null"] },
+    "lfs_objects_checksummed_count": { "type": ["integer", "null"] },
+    "lfs_objects_checksum_failed_count": { "type": ["integer", "null"] },
+    "lfs_objects_registry_count": { "type": ["integer", "null"] },
+    "lfs_objects_verification_total_count": { "type": ["integer", "null"] },
+    "lfs_objects_verified_count": { "type": ["integer", "null"] },
+    "lfs_objects_verified_in_percentage": { "type": "string" },
+    "lfs_objects_verification_failed_count": { "type": ["integer", "null"] },
     "job_artifacts_replication_enabled": { "type": ["boolean", "null"] },
     "job_artifacts_count": { "type": "integer" },
     "job_artifacts_failed_count": { "type": ["integer", "null"] },

ee/spec/graphql/resolvers/geo/lfs_object_registries_resolver_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Resolvers::Geo::LfsObjectRegistriesResolver do
+  it_behaves_like 'a Geo registries resolver', :geo_lfs_object_registry
+end

ee/spec/graphql/types/geo/geo_node_type_spec.rb

@@ -14,6 +14,7 @@ RSpec.describe GitlabSchema.types['GeoNode'] do
       minimum_reverification_interval merge_request_diff_registries
       package_file_registries snippet_repository_registries
       terraform_state_version_registries group_wiki_repository_registries
+      lfs_object_registries
       pipeline_artifact_registries
     ]
 

ee/spec/graphql/types/geo/lfs_object_registry_type_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe GitlabSchema.types['LfsObjectRegistry'] do
+  it_behaves_like 'a Geo registry type'
+
+  it 'has the expected fields (other than those included in RegistryType)' do
+    expected_fields = %i[lfs_object_id]
+
+    expect(described_class).to have_graphql_fields(*expected_fields).at_least
+  end
+end

ee/spec/lib/ee/api/entities/geo_node_status_spec.rb

@@ -5,7 +5,7 @@ require 'spec_helper'
 RSpec.describe EE::API::Entities::GeoNodeStatus do
   include ::EE::GeoHelpers
 
-  let(:geo_node_status) { build(:geo_node_status) }
+  let!(:geo_node_status) { build(:geo_node_status) }
   let(:entity) { described_class.new(geo_node_status, request: double) }
   let(:error) { 'Could not connect to Geo database' }
 
@@ -66,12 +66,18 @@ RSpec.describe EE::API::Entities::GeoNodeStatus do
   end
 
   describe '#lfs_objects_synced_in_percentage' do
-    it 'formats as percentage' do
-      geo_node_status.assign_attributes(lfs_objects_count: 256,
-                                        lfs_objects_failed_count: 12,
-                                        lfs_objects_synced_count: 123)
+    context 'LFS with SSF is disabled' do
+      before do
+        stub_feature_flags(geo_lfs_object_replication: false)
+      end
 
-      expect(subject[:lfs_objects_synced_in_percentage]).to eq '48.05%'
+      it 'formats as percentage' do
+        geo_node_status.assign_attributes(lfs_objects_registry_count: 256,
+                                          lfs_objects_failed_count: 12,
+                                          lfs_objects_synced_count: 123)
+
+        expect(subject[:lfs_objects_synced_in_percentage]).to eq '48.05%'
+      end
     end
   end
 

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

@@ -16,22 +16,28 @@ RSpec.describe Gitlab::Geo::GeoNodeStatusCheck do
       allow(Gitlab.config.geo.registry_replication).to receive(:enabled).and_return(true)
     end
 
-    it 'prints messages for all legacy replication and verification checks' do
-      checks = [
-        /Repositories: /,
-        /Verified Repositories: /,
-        /Wikis: /,
-        /Verified Wikis: /,
-        /LFS Objects: /,
-        /Attachments: /,
-        /CI job artifacts: /,
-        /Container repositories: /,
-        /Design repositories: /,
-        /Repositories Checked: /
-      ]
-
-      checks.each do |text|
-        expect { subject.print_replication_verification_status }.to output(text).to_stdout
+    context 'with legacy replication' do
+      before do
+        stub_feature_flags(geo_lfs_object_replication: false)
+      end
+
+      it 'prints messages for all legacy replication and verification checks' do
+        checks = [
+          /Repositories: /,
+          /Verified Repositories: /,
+          /Wikis: /,
+          /Verified Wikis: /,
+          /LFS Objects: /,
+          /Attachments: /,
+          /CI job artifacts: /,
+          /Container repositories: /,
+          /Design repositories: /,
+          /Repositories Checked: /
+        ]
+
+        checks.each do |text|
+          expect { subject.print_replication_verification_status }.to output(text).to_stdout
+        end
       end
     end
 

ee/spec/models/geo/lfs_object_registry_spec.rb

@@ -6,7 +6,7 @@ RSpec.describe Geo::LfsObjectRegistry, :geo do
   include EE::GeoHelpers
 
   it_behaves_like 'a BulkInsertSafe model', Geo::LfsObjectRegistry do
-    let(:valid_items_for_bulk_insertion) { build_list(:geo_lfs_object_registry, 10) }
+    let(:valid_items_for_bulk_insertion) { build_list(:geo_lfs_object_legacy_registry, 10) }
     let(:invalid_items_for_bulk_insertion) { [] } # class does not have any validations defined
   end
 
@@ -41,9 +41,9 @@ RSpec.describe Geo::LfsObjectRegistry, :geo do
 
     context 'untracked IDs' do
       before do
-        create(:geo_lfs_object_registry, lfs_object_id: lfs_object_1.id)
-        create(:geo_lfs_object_registry, :failed, lfs_object_id: lfs_object_3.id)
-        create(:geo_lfs_object_registry, lfs_object_id: lfs_object_4.id)
+        create(:geo_lfs_object_legacy_registry, lfs_object_id: lfs_object_1.id)
+        create(:geo_lfs_object_legacy_registry, :failed, lfs_object_id: lfs_object_3.id)
+        create(:geo_lfs_object_legacy_registry, lfs_object_id: lfs_object_4.id)
 
         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)
@@ -101,7 +101,7 @@ RSpec.describe Geo::LfsObjectRegistry, :geo do
 
     context 'unused tracked IDs' do
       context 'with an orphaned registry' do
-        let!(:orphaned) { create(:geo_lfs_object_registry, lfs_object_id: non_existing_record_id) }
+        let!(:orphaned) { create(:geo_lfs_object_legacy_registry, lfs_object_id: non_existing_record_id) }
 
         it 'includes tracked IDs that do not exist in the model table' do
           range = non_existing_record_id..non_existing_record_id
@@ -124,7 +124,7 @@ RSpec.describe Geo::LfsObjectRegistry, :geo do
         let(:secondary) { create(:geo_node, selective_sync_type: 'namespaces', namespaces: [synced_group]) }
 
         context 'with a tracked LFS object' do
-          let!(:registry_entry) { create(:geo_lfs_object_registry, lfs_object_id: lfs_object_1.id) }
+          let!(:registry_entry) { create(:geo_lfs_object_legacy_registry, lfs_object_id: lfs_object_1.id) }
           let(:range) { lfs_object_1.id..lfs_object_1.id }
 
           context 'excluded from selective sync' do
@@ -151,7 +151,7 @@ RSpec.describe Geo::LfsObjectRegistry, :geo do
         let(:secondary) { create(:geo_node, selective_sync_type: 'shards', selective_sync_shards: ['broken']) }
 
         context 'with a tracked LFS object' do
-          let!(:registry_entry) { create(:geo_lfs_object_registry, lfs_object_id: lfs_object_1.id) }
+          let!(:registry_entry) { create(:geo_lfs_object_legacy_registry, lfs_object_id: lfs_object_1.id) }
           let(:range) { lfs_object_1.id..lfs_object_1.id }
 
           context 'excluded from selective sync' do
@@ -180,7 +180,7 @@ RSpec.describe Geo::LfsObjectRegistry, :geo do
         context 'with a tracked LFS object' do
           context 'in object storage' do
             it 'includes tracked LFS object IDs that are in object storage' do
-              create(:geo_lfs_object_registry, lfs_object_id: lfs_object_remote_1.id)
+              create(:geo_lfs_object_legacy_registry, lfs_object_id: lfs_object_remote_1.id)
               range = lfs_object_remote_1.id..lfs_object_remote_1.id
 
               _, unused_tracked_ids = described_class.find_registry_differences(range)
@@ -191,7 +191,7 @@ RSpec.describe Geo::LfsObjectRegistry, :geo do
 
           context 'not in object storage' do
             it 'excludes tracked LFS object IDs that are not in object storage' do
-              create(:geo_lfs_object_registry, lfs_object_id: lfs_object_1.id)
+              create(:geo_lfs_object_legacy_registry, lfs_object_id: lfs_object_1.id)
               range = lfs_object_1.id..lfs_object_1.id
 
               _, unused_tracked_ids = described_class.find_registry_differences(range)
@@ -204,3 +204,13 @@ RSpec.describe Geo::LfsObjectRegistry, :geo do
     end
   end
 end
+
+RSpec.describe Geo::LfsObjectRegistry, :geo, type: :model do
+  let_it_be(:registry) { create(:geo_lfs_object_registry) }
+
+  specify 'factory is valid' do
+    expect(registry).to be_valid
+  end
+
+  include_examples 'a Geo framework registry'
+end

ee/spec/models/geo_node_status_spec.rb

@@ -173,7 +173,7 @@ RSpec.describe GeoNodeStatus, :geo do
   describe '#attachments_failed_count' do
     it 'counts failed avatars, attachment, personal snippets and files' do
       # These two should be ignored
-      create(:geo_lfs_object_registry, :with_lfs_object, :failed)
+      create(:geo_lfs_object_legacy_registry, :with_lfs_object, :failed)
       create(:geo_upload_registry, :with_file)
 
       create(:geo_upload_registry, :with_file, :failed, file_type: :personal_file)
@@ -219,60 +219,54 @@ RSpec.describe GeoNodeStatus, :geo do
     end
   end
 
-  describe '#lfs_objects_synced_count' do
-    it 'counts synced LFS objects' do
-      # These four should be ignored
-      create(:geo_upload_registry, :failed)
-      create(:geo_upload_registry, :avatar)
-      create(:geo_upload_registry, file_type: :attachment)
-      create(:geo_lfs_object_registry, :failed)
-
-      create(:geo_lfs_object_registry)
-
-      expect(subject.lfs_objects_synced_count).to eq(1)
+  context 'LFS replication with SSF is disabled' do
+    before do
+      stub_feature_flags(geo_lfs_object_replication: false)
     end
-  end
 
-  describe '#lfs_objects_synced_missing_on_primary_count' do
-    it 'counts LFS objects marked as synced due to file missing on the primary' do
-      # These four should be ignored
-      create(:geo_upload_registry, :failed)
-      create(:geo_upload_registry, :avatar, missing_on_primary: true)
-      create(:geo_upload_registry, file_type: :attachment, missing_on_primary: true)
-      create(:geo_lfs_object_registry, :failed)
+    describe '#lfs_objects_synced_count' do
+      it 'counts synced LFS objects' do
+        # These four should be ignored
+        create(:geo_upload_registry, :failed)
+        create(:geo_upload_registry, :avatar)
+        create(:geo_upload_registry, file_type: :attachment)
+        create(:geo_lfs_object_legacy_registry, :failed)
+
+        create(:geo_lfs_object_legacy_registry)
 
-      create(:geo_lfs_object_registry, missing_on_primary: true)
+        create(:geo_lfs_object_legacy_registry, missing_on_primary: true)
 
-      expect(subject.lfs_objects_synced_missing_on_primary_count).to eq(1)
+        expect(subject.lfs_objects_synced_missing_on_primary_count).to eq(1)
+      end
     end
-  end
 
-  describe '#lfs_objects_failed_count' do
-    it 'counts failed LFS objects' do
-      # These four should be ignored
-      create(:geo_upload_registry, :failed)
-      create(:geo_upload_registry, :avatar, :failed)
-      create(:geo_upload_registry, :failed, file_type: :attachment)
-      create(:geo_lfs_object_registry)
+    describe '#lfs_objects_failed_count' do
+      it 'counts failed LFS objects' do
+        # These four should be ignored
+        create(:geo_upload_registry, :failed)
+        create(:geo_upload_registry, :avatar, :failed)
+        create(:geo_upload_registry, :failed, file_type: :attachment)
+        create(:geo_lfs_object_legacy_registry)
 
-      create(:geo_lfs_object_registry, :failed)
+        create(:geo_lfs_object_legacy_registry, :failed)
 
-      expect(subject.lfs_objects_failed_count).to eq(1)
+        expect(subject.lfs_objects_failed_count).to eq(1)
+      end
     end
-  end
 
-  describe '#lfs_objects_synced_in_percentage' do
-    it 'returns 0 when there are no registries' do
-      expect(subject.lfs_objects_synced_in_percentage).to eq(0)
-    end
+    describe '#lfs_objects_synced_in_percentage' do
+      it 'returns 0 when there are no registries' do
+        expect(subject.lfs_objects_synced_in_percentage).to eq(0)
+      end
 
-    it 'returns the right percentage' do
-      create(:geo_lfs_object_registry)
-      create(:geo_lfs_object_registry, :failed)
-      create(:geo_lfs_object_registry, :never_synced)
-      create(:geo_lfs_object_registry, :never_synced)
+      it 'returns the right percentage' do
+        create(:geo_lfs_object_legacy_registry)
+        create(:geo_lfs_object_legacy_registry, :failed)
+        create(:geo_lfs_object_legacy_registry, :never_synced)
+        create(:geo_lfs_object_legacy_registry, :never_synced)
 
-      expect(subject.lfs_objects_synced_in_percentage).to be_within(0.0001).of(25)
+        expect(subject.lfs_objects_synced_in_percentage).to be_within(0.0001).of(25)
+      end
     end
   end
 
@@ -1180,6 +1174,7 @@ RSpec.describe GeoNodeStatus, :geo do
 
   context 'Replicator stats' do
     where(:replicator, :model_factory, :registry_factory) do
+      Geo::LfsObjectReplicator             | :lfs_object                  | :geo_lfs_object_registry
       Geo::MergeRequestDiffReplicator      | :external_merge_request_diff | :geo_merge_request_diff_registry
       Geo::PackageFileReplicator           | :package_file                | :geo_package_file_registry
       Geo::TerraformStateVersionReplicator | :terraform_state_version     | :geo_terraform_state_version_registry
@@ -1400,7 +1395,7 @@ RSpec.describe GeoNodeStatus, :geo do
       end
 
       it 'does not call LfsObjectRegistryFinder#registry_count' do
-        expect_any_instance_of(Geo::LfsObjectRegistryFinder).not_to receive(:registry_count)
+        expect_any_instance_of(Geo::LfsObjectLegacyRegistryFinder).not_to receive(:registry_count)
 
         subject
       end
@@ -1420,7 +1415,8 @@ RSpec.describe GeoNodeStatus, :geo do
 
     context 'on the secondary' do
       it 'calls LfsObjectRegistryFinder#registry_count' do
-        expect_any_instance_of(Geo::LfsObjectRegistryFinder).to receive(:registry_count).twice
+        stub_feature_flags(geo_lfs_object_replication: false)
+        expect_any_instance_of(Geo::LfsObjectLegacyRegistryFinder).to receive(:registry_count).twice
 
         subject
       end

ee/spec/models/lfs_object_spec.rb

@@ -5,6 +5,10 @@ require 'spec_helper'
 RSpec.describe LfsObject do
   include EE::GeoHelpers
 
+  before do
+    stub_feature_flags(geo_lfs_object_replication: false)
+  end
+
   describe '#destroy' do
     subject { create(:lfs_object, :with_file) }
 

ee/spec/replicators/geo/lfs_object_replicator_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Geo::LfsObjectReplicator do
+  let(:model_record) { build(:lfs_object, :with_file) }
+
+  it_behaves_like 'a blob replicator'
+end

ee/spec/requests/api/graphql/geo/registries_spec.rb

@@ -38,6 +38,13 @@ RSpec.describe 'Gets registries' do
     registry_foreign_key_field_name: 'groupWikiRepositoryId'
   }
 
+  it_behaves_like 'gets registries for', {
+    field_name: 'lfsObjectRegistries',
+    registry_class_name: 'LfsObjectRegistry',
+    registry_factory: :geo_lfs_object_registry,
+    registry_foreign_key_field_name: 'lfsObjectId'
+  }
+
   it_behaves_like 'gets registries for', {
     field_name: 'pipelineArtifactRegistries',
     registry_class_name: 'PipelineArtifactRegistry',

ee/spec/services/geo/file_download_service_spec.rb

@@ -441,6 +441,10 @@ RSpec.describe Geo::FileDownloadService do
     end
 
     context 'LFS object' do
+      before do
+        stub_feature_flags(geo_lfs_object_replication: false)
+      end
+
       it_behaves_like "a service that downloads the file and registers the sync result", 'lfs' do
         let(:file) { create(:lfs_object) }
       end

ee/spec/tasks/geo_rake_spec.rb

@@ -317,6 +317,29 @@ RSpec.describe 'geo rake tasks', :geo do
       let!(:primary_node) { create(:geo_node, :primary) }
       let!(:geo_event_log) { create(:geo_event_log) }
       let!(:geo_node_status) { build(:geo_node_status, :healthy, geo_node: current_node) }
+      let(:checks) do
+        [
+          /Name: /,
+          /URL: /,
+          /GitLab Version: /,
+          /Geo Role: /,
+          /Health Status: /,
+          /Sync Settings: /,
+          /Database replication lag: /,
+          /Repositories: /,
+          /Verified Repositories: /,
+          /Wikis: /,
+          /Verified Wikis: /,
+          /Attachments: /,
+          /CI job artifacts: /,
+          /Container repositories: /,
+          /Design repositories: /,
+          /Repositories Checked: /,
+          /Last event ID seen from primary: /,
+          /Last status report was: /
+        ] + Gitlab::Geo.verification_enabled_replicator_classes.map { |k| /#{k.replicable_title_plural} Verified:/ } +
+          Gitlab::Geo.enabled_replicator_classes.map { |k| /#{k.replicable_title_plural}:/ }
+      end
 
       before do
         stub_licensed_features(geo: true)
@@ -343,32 +366,23 @@ RSpec.describe 'geo rake tasks', :geo do
           expect { run_rake_task('geo:status') }.not_to output(/Health Status Summary/).to_stdout
         end
 
-        it 'prints messages for all the checks' do
-          checks = [
-            /Name: /,
-            /URL: /,
-            /GitLab Version: /,
-            /Geo Role: /,
-            /Health Status: /,
-            /Sync Settings: /,
-            /Database replication lag: /,
-            /Repositories: /,
-            /Verified Repositories: /,
-            /Wikis: /,
-            /Verified Wikis: /,
-            /LFS Objects: /,
-            /Attachments: /,
-            /CI job artifacts: /,
-            /Container repositories: /,
-            /Design repositories: /,
-            /Repositories Checked: /,
-            /Last event ID seen from primary: /,
-            /Last status report was: /
-          ] + Gitlab::Geo.verification_enabled_replicator_classes.map { |k| /#{k.replicable_title_plural} Verified:/ } +
-              Gitlab::Geo.enabled_replicator_classes.map { |k| /#{k.replicable_title_plural}:/ }
-
-          checks.each do |text|
-            expect { run_rake_task('geo:status') }.to output(text).to_stdout
+        context 'with legacy LFS replication enabled' do
+          before do
+            stub_feature_flags(geo_lfs_object_replication: false)
+          end
+
+          it 'prints messages for all the checks' do
+            (checks << /LFS Objects: /).each do |text|
+              expect { run_rake_task('geo:status') }.to output(text).to_stdout
+            end
+          end
+        end
+
+        context 'with SSF LFS replication eneabled' do
+          it 'prints messages for all the checks' do
+            checks.each do |text|
+              expect { run_rake_task('geo:status') }.to output(text).to_stdout
+            end
           end
         end
       end

ee/spec/workers/geo/file_download_dispatch_worker_spec.rb

@@ -12,6 +12,7 @@ RSpec.describe Geo::FileDownloadDispatchWorker, :geo, :use_sql_query_cache_for_t
   before do
     stub_current_geo_node(secondary)
     stub_exclusive_lease(renew: true)
+    stub_feature_flags(geo_lfs_object_replication: false)
     allow_next_instance_of(described_class) do |instance|
       allow(instance).to receive(:over_time?).and_return(false)
     end
@@ -47,8 +48,8 @@ RSpec.describe Geo::FileDownloadDispatchWorker, :geo, :use_sql_query_cache_for_t
   it 'does not schedule duplicated jobs' do
     lfs_object_1 = create(:lfs_object, :with_file)
     lfs_object_2 = create(:lfs_object, :with_file)
-    create(:geo_lfs_object_registry, :never_synced, lfs_object: lfs_object_1)
-    create(:geo_lfs_object_registry, :failed, lfs_object: lfs_object_2)
+    create(:geo_lfs_object_legacy_registry, :never_synced, lfs_object: lfs_object_1)
+    create(:geo_lfs_object_legacy_registry, :failed, lfs_object: lfs_object_2)
 
     stub_const('Geo::Scheduler::SchedulerWorker::DB_RETRIEVE_BATCH_SIZE', 5)
     secondary.update!(files_max_capacity: 4)
@@ -65,9 +66,9 @@ RSpec.describe Geo::FileDownloadDispatchWorker, :geo, :use_sql_query_cache_for_t
     lfs_object_1 = create(:lfs_object, :with_file)
     lfs_object_2 = create(:lfs_object, :with_file)
     lfs_object_3 = create(:lfs_object, :with_file)
-    create(:geo_lfs_object_registry, :never_synced, lfs_object: lfs_object_1)
-    create(:geo_lfs_object_registry, :never_synced, lfs_object: lfs_object_2)
-    create(:geo_lfs_object_registry, :never_synced, lfs_object: lfs_object_3)
+    create(:geo_lfs_object_legacy_registry, :never_synced, lfs_object: lfs_object_1)
+    create(:geo_lfs_object_legacy_registry, :never_synced, lfs_object: lfs_object_2)
+    create(:geo_lfs_object_legacy_registry, :never_synced, lfs_object: lfs_object_3)
 
     stub_const('Geo::Scheduler::SchedulerWorker::DB_RETRIEVE_BATCH_SIZE', 3)
     secondary.update!(files_max_capacity: 6)
@@ -205,8 +206,8 @@ RSpec.describe Geo::FileDownloadDispatchWorker, :geo, :use_sql_query_cache_for_t
 
       context 'with lfs_object_registry entries' do
         before do
-          create(:geo_lfs_object_registry, :never_synced, lfs_object: lfs_object_local_store)
-          create(:geo_lfs_object_registry, :failed, lfs_object: lfs_object_remote_store)
+          create(:geo_lfs_object_legacy_registry, :never_synced, lfs_object: lfs_object_local_store)
+          create(:geo_lfs_object_legacy_registry, :failed, lfs_object: lfs_object_remote_store)
           Geo::LfsObjectRegistry.create!(lfs_object_id: lfs_object_file_missing_on_primary.id, bytes: 1234, success: true, missing_on_primary: true)
         end
 
@@ -372,7 +373,7 @@ RSpec.describe Geo::FileDownloadDispatchWorker, :geo, :use_sql_query_cache_for_t
     result_object = double(:result, success: true, bytes_downloaded: 100, primary_missing_file: false)
     allow_any_instance_of(::Gitlab::Geo::Replication::BaseTransfer).to receive(:download_from_primary).and_return(result_object)
 
-    create_list(:geo_lfs_object_registry, 2, :with_lfs_object, :never_synced)
+    create_list(:geo_lfs_object_legacy_registry, 2, :with_lfs_object, :never_synced)
     create_list(:geo_upload_registry, 2, :avatar, :with_file, :never_synced)
     create_list(:geo_upload_registry, 2, :attachment, :with_file, :never_synced)
     create(:geo_upload_registry, :favicon, :with_file, :never_synced)