Merge branch '3691-geo-file-backfill' into 'master'

Prevent failed file syncs from stalling Geo backfill

Closes #3691

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

app/assets/javascripts/geo_nodes.js

@@ -20,7 +20,9 @@ class GeoNodeStatus {
     this.$repositoriesSynced = $('.js-repositories-synced', this.$status);
     this.$repositoriesFailed = $('.js-repositories-failed', this.$status);
     this.$lfsObjectsSynced = $('.js-lfs-objects-synced', this.$status);
+    this.$lfsObjectsFailed = $('.js-lfs-objects-failed', this.$status);
     this.$attachmentsSynced = $('.js-attachments-synced', this.$status);
+    this.$attachmentsFailed = $('.js-attachments-failed', this.$status);
     this.$lastEventSeen = $('.js-last-event-seen', this.$status);
     this.$lastCursorEvent = $('.js-last-cursor-event', this.$status);
     this.$health = $('.js-health', this.$status);
@@ -78,15 +80,21 @@ class GeoNodeStatus {
         status.lfs_objects_count,
         status.lfs_objects_synced_in_percentage);
 
+      const lfsFailedText = gl.text.addDelimiter(status.lfs_objects_failed_count);
+
       const attachmentText = GeoNodeStatus.formatCountAndPercentage(
         status.attachments_synced_count,
         status.attachments_count,
         status.attachments_synced_in_percentage);
 
+      const attachmentFailedText = gl.text.addDelimiter(status.attachments_failed_count);
+
       this.$repositoriesSynced.text(repoText);
       this.$repositoriesFailed.text(repoFailedText);
       this.$lfsObjectsSynced.text(lfsText);
+      this.$lfsObjectsFailed.text(lfsFailedText);
       this.$attachmentsSynced.text(attachmentText);
+      this.$attachmentsFailed.text(attachmentFailedText);
 
       const eventDate = gl.utils.formatDate(new Date(status.last_event_date));
       const cursorDate = gl.utils.formatDate(new Date(status.cursor_last_event_date));

app/models/geo/file_registry.rb

 class Geo::FileRegistry < Geo::BaseRegistry
+  scope :failed, -> { where(success: false) }
+  scope :synced, -> { where(success: true) }
 end

app/models/geo_node_status.rb

@@ -100,7 +100,7 @@ class GeoNodeStatus
 
   def lfs_objects_synced_count
     @lfs_objects_synced_count ||= begin
-      relation = Geo::FileRegistry.where(file_type: :lfs)
+      relation = Geo::FileRegistry.synced.where(file_type: :lfs)
 
       if Gitlab::Geo.current_node.restricted_project_ids
         relation = relation.where(file_id: lfs_objects.pluck(:id))
@@ -114,6 +114,14 @@ class GeoNodeStatus
     @lfs_objects_synced_count = value.to_i
   end
 
+  def lfs_objects_failed_count
+    @lfs_objects_failed_count ||= Geo::FileRegistry.failed.where(file_type: :lfs).count
+  end
+
+  def lfs_objects_failed_count=(value)
+    @lfs_objects_failed_count = value.to_i
+  end
+
   def lfs_objects_synced_in_percentage
     sync_percentage(lfs_objects_count, lfs_objects_synced_count)
   end
@@ -129,7 +137,7 @@ class GeoNodeStatus
   def attachments_synced_count
     @attachments_synced_count ||= begin
       upload_ids = attachments.pluck(:id)
-      synced_ids = Geo::FileRegistry.where(file_type: [:attachment, :avatar, :file]).pluck(:file_id)
+      synced_ids = Geo::FileRegistry.synced.where(file_type: Geo::FileService::DEFAULT_OBJECT_TYPES).pluck(:file_id)
 
       (synced_ids & upload_ids).length
     end
@@ -139,6 +147,14 @@ class GeoNodeStatus
     @attachments_synced_count = value.to_i
   end
 
+  def attachments_failed_count
+    @attachments_failed_count ||= Geo::FileRegistry.failed.where(file_type: Geo::FileService::DEFAULT_OBJECT_TYPES).count
+  end
+
+  def attachments_failed_count=(value)
+    @attachments_failed_count = value.to_i
+  end
+
   def attachments_synced_in_percentage
     sync_percentage(attachments_count, attachments_synced_count)
   end

app/serializers/geo_node_status_entity.rb

@@ -10,6 +10,7 @@ class GeoNodeStatusEntity < Grape::Entity
 
   expose :attachments_count
   expose :attachments_synced_count
+  expose :attachments_failed_count
   expose :attachments_synced_in_percentage do |node|
     number_to_percentage(node.attachments_synced_in_percentage, precision: 2)
   end
@@ -18,6 +19,7 @@ class GeoNodeStatusEntity < Grape::Entity
 
   expose :lfs_objects_count
   expose :lfs_objects_synced_count
+  expose :lfs_objects_failed_count
   expose :lfs_objects_synced_in_percentage do |node|
     number_to_percentage(node.lfs_objects_synced_in_percentage, precision: 2)
   end

app/services/geo/file_download_service.rb

@@ -11,7 +11,7 @@ module Geo
                  success: success,
                  bytes_downloaded: bytes_downloaded,
                  download_time_s: (Time.now - start_time).to_f.round(3))
-        update_registry(bytes_downloaded) if success
+        update_registry(bytes_downloaded, success: success)
       end
     end
 
@@ -37,13 +37,14 @@ module Geo
       end
     end
 
-    def update_registry(bytes_downloaded)
+    def update_registry(bytes_downloaded, success:)
       transfer = Geo::FileRegistry.find_or_initialize_by(
         file_type: object_type,
         file_id: object_db_id
       )
 
       transfer.bytes = bytes_downloaded
+      transfer.success = success
       transfer.save
     end
 

app/views/admin/geo_nodes/index.html.haml

@@ -62,10 +62,18 @@
                   %span.help-block
                     LFS objects synced:
                     %strong.node-info.js-lfs-objects-synced
+                %p
+                  %span.help-block
+                    LFS objects failed:
+                    %strong.node-info.js-lfs-objects-failed
                 %p
                   %span.help-block
                     Attachments synced:
                     %strong.node-info.js-attachments-synced
+                %p
+                  %span.help-block
+                    Attachments failed:
+                    %strong.node-info.js-attachments-failed
                 %p
                 .advanced-geo-node-status-container
                   .advanced-status.hidden

app/workers/geo/file_download_dispatch_worker.rb

@@ -9,33 +9,47 @@ module Geo
     end
 
     def load_pending_resources
+      unsynced = find_unsynced_objects
+      failed = find_failed_objects
+
+      interleave(unsynced, failed)
+    end
+
+    def find_unsynced_objects
       lfs_object_ids = find_lfs_object_ids
       objects_ids    = find_object_ids
 
       interleave(lfs_object_ids, objects_ids)
     end
 
-    def find_object_ids
-      downloaded_ids = find_downloaded_ids(Geo::FileService::DEFAULT_OBJECT_TYPES)
+    def find_failed_objects
+      Geo::FileRegistry
+        .failed
+        .limit(db_retrieve_batch_size)
+        .pluck(:file_id, :file_type)
+    end
 
-      unsynched_downloads = filter_downloaded_ids(
-        current_node.uploads, downloaded_ids, Upload.table_name)
+    def find_object_ids
+      unsynced_downloads = filter_registry_ids(
+        current_node.uploads,
+        Geo::FileService::DEFAULT_OBJECT_TYPES,
+        Upload.table_name
+      )
 
-      unsynched_downloads
-        .order(created_at: :desc)
+      unsynced_downloads
         .limit(db_retrieve_batch_size)
         .pluck(:id, :uploader)
         .map { |id, uploader| [id, uploader.sub(/Uploader\z/, '').underscore] }
     end
 
     def find_lfs_object_ids
-      downloaded_ids = find_downloaded_ids([:lfs])
+      unsynced_downloads = filter_registry_ids(
+        current_node.lfs_objects,
+        [:lfs],
+        LfsObject.table_name
+      )
 
-      unsynched_downloads = filter_downloaded_ids(
-        current_node.lfs_objects, downloaded_ids, LfsObject.table_name)
-
-      unsynched_downloads
-        .order(created_at: :desc)
+      unsynced_downloads
         .limit(db_retrieve_batch_size)
         .pluck(:id)
         .map { |id| [id, :lfs] }
@@ -45,12 +59,14 @@ module Geo
     # plucks a list of file IDs from one into the other. This will not scale
     # well with the number of synchronized files--the query will increase
     # linearly in size--so this should be replaced with postgres_fdw ASAP.
-    def filter_downloaded_ids(objects, downloaded_ids, table_name)
-      return objects if downloaded_ids.empty?
+    def filter_registry_ids(objects, file_types, table_name)
+      registry_ids = pluck_registry_ids(Geo::FileRegistry, file_types)
+
+      return objects if registry_ids.empty?
 
       joined_relation = objects.joins(<<~SQL)
         LEFT OUTER JOIN
-        (VALUES #{downloaded_ids.map { |id| "(#{id}, 't')" }.join(',')})
+        (VALUES #{registry_ids.map { |id| "(#{id}, 't')" }.join(',')})
          file_registry(file_id, registry_present)
          ON #{table_name}.id = file_registry.file_id
       SQL
@@ -58,9 +74,9 @@ module Geo
       joined_relation.where(file_registry: { registry_present: [nil, false] })
     end
 
-    def find_downloaded_ids(file_types)
-      downloaded_ids = Geo::FileRegistry.where(file_type: file_types).pluck(:file_id)
-      (downloaded_ids + scheduled_file_ids(file_types)).uniq
+    def pluck_registry_ids(relation, file_types)
+      ids = relation.where(file_type: file_types).pluck(:file_id)
+      (ids + scheduled_file_ids(file_types)).uniq
     end
 
     def scheduled_file_ids(types)

changelogs/unreleased-ee/3691-geo-file-backfill.yml

+---
+title: Prevent failed file syncs from stalling Geo backfill
+merge_request: 3101
+author:
+type: fixed

db/geo/migrate/20171009162208_add_file_registry_success.rb

+class AddFileRegistrySuccess < ActiveRecord::Migration
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+
+  disable_ddl_transaction!
+
+  def up
+    # Ensure existing rows are recorded as successes
+    add_column_with_default :file_registry, :success, :boolean, default: true, allow_null: false
+
+    change_column :file_registry, :success, :boolean, default: false
+  end
+
+  def down
+    # Prevent failures from being converted into successes
+    false_value = Arel::Nodes::False.new.to_sql(Geo::BaseRegistry)
+    connection.execute("DELETE FROM file_registry WHERE success = #{false_value}")
+
+    remove_column :file_registry, :success
+  end
+end

db/geo/migrate/20171009162209_add_file_registry_success_index.rb

+class AddFileRegistrySuccessIndex < ActiveRecord::Migration
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_index :file_registry, :success
+  end
+
+  def down
+    remove_concurrent_index :file_registry, :success
+  end
+end

db/geo/schema.rb

@@ -11,7 +11,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema.define(version: 20171005045404) do
+ActiveRecord::Schema.define(version: 20171009162209) do
 
   # These are extensions that must be enabled in order to support this database
   enable_extension "plpgsql"
@@ -25,10 +25,12 @@ ActiveRecord::Schema.define(version: 20171005045404) do
     t.integer "bytes", limit: 8
     t.string "sha256"
     t.datetime "created_at", null: false
+    t.boolean "success", default: false, null: false
   end
 
   add_index "file_registry", ["file_type", "file_id"], name: "index_file_registry_on_file_type_and_file_id", unique: true, using: :btree
   add_index "file_registry", ["file_type"], name: "index_file_registry_on_file_type", using: :btree
+  add_index "file_registry", ["success"], name: "index_file_registry_on_success", using: :btree
 
   create_table "project_registry", force: :cascade do |t|
     t.integer "project_id", null: false

lib/api/entities.rb

@@ -1018,8 +1018,10 @@ module API
       expose :repositories_failed_count
       expose :lfs_objects_count
       expose :lfs_objects_synced_count
+      expose :lfs_objects_failed_count
       expose :attachments_count
       expose :attachments_synced_count
+      expose :attachments_failed_count
       expose :last_event_id
       expose :last_event_date
       expose :cursor_last_event_id

spec/controllers/admin/geo_nodes_controller_spec.rb

@@ -264,8 +264,10 @@ describe Admin::GeoNodesController, :postgresql do
           id: 1,
           health: nil,
           attachments_count: 329,
+          attachments_failed_count: 13,
           attachments_synced_count: 141,
           lfs_objects_count: 256,
+          lfs_objects_failed_count: 12,
           lfs_objects_synced_count: 123,
           repositories_count: 10,
           repositories_synced_count: 5,

spec/factories/geo/file_registry.rb

@@ -2,6 +2,7 @@ FactoryGirl.define do
   factory :geo_file_registry, class: Geo::FileRegistry do
     sequence(:file_id)
     file_type :file
+    success true
 
     trait :avatar do
       file_type :avatar

spec/fixtures/api/schemas/geo_node_status.json

@@ -5,8 +5,10 @@
     "healthy",
     "health",
     "attachments_count",
+    "attachments_failed_count",
     "attachments_synced_count",
     "lfs_objects_count",
+    "lfs_objects_failed_count",
     "lfs_objects_synced_count",
     "db_replication_lag",
     "repositories_count",
@@ -22,10 +24,12 @@
     "healthy": { "type": "boolean" },
     "health": { "type": "string" },
     "attachments_count": { "type": "integer" },
+    "attachments_failed_count": { "type": "integer" },
     "attachments_synced_count": { "type": "integer" },
     "attachments_synced_in_percentage": { "type": "string" },
     "db_replication_lag": { "type": ["integer", "null"] },
     "lfs_objects_count": { "type": "integer" },
+    "lfs_objects_failed_count": { "type": "integer" },
     "lfs_objects_synced_count": { "type": "integer" },
     "lfs_objects_synced_in_percentage": { "type": "string" },
     "repositories_count": { "type": "integer" },

spec/models/geo/file_registry_spec.rb

+require 'spec_helper'
+
+describe Geo::FileRegistry do
+  set(:failed) { create(:geo_file_registry, success: false) }
+  set(:synced) { create(:geo_file_registry, success: true) }
+
+  describe '.failed' do
+    it 'returns registries in the failed state' do
+      expect(described_class.failed).to contain_exactly(failed)
+    end
+  end
+
+  describe '.synced' do
+    it 'returns registries in the synced state' do
+      expect(described_class.synced).to contain_exactly(synced)
+    end
+  end
+end

spec/models/geo_node_status_spec.rb

@@ -39,6 +39,17 @@ describe GeoNodeStatus do
   end
 
   describe '#attachments_synced_count' do
+    it 'only counts successful syncs' do
+      create_list(:user, 3, avatar: fixture_file_upload(Rails.root + 'spec/fixtures/dk.png', 'image/png'))
+      uploads = Upload.all.pluck(:id)
+
+      create(:geo_file_registry, :avatar, file_id: uploads[0])
+      create(:geo_file_registry, :avatar, file_id: uploads[1])
+      create(:geo_file_registry, :avatar, file_id: uploads[2], success: false)
+
+      expect(subject.attachments_synced_count).to eq(2)
+    end
+
     it 'does not count synced files that were replaced' do
       user = create(:user, avatar: fixture_file_upload(Rails.root + 'spec/fixtures/dk.png', 'image/png'))
 
@@ -68,6 +79,21 @@ describe GeoNodeStatus do
     end
   end
 
+  describe '#attachments_failed_count' do
+    it 'counts failed avatars, attachment, personal snippets and files' do
+      # These two should be ignored
+      create(:geo_file_registry, :lfs, success: false)
+      create(:geo_file_registry)
+
+      create(:geo_file_registry, file_type: :personal_file, success: false)
+      create(:geo_file_registry, file_type: :attachment, success: false)
+      create(:geo_file_registry, :avatar, success: false)
+      create(:geo_file_registry, success: false)
+
+      expect(subject.attachments_failed_count).to eq(4)
+    end
+  end
+
   describe '#attachments_synced_in_percentage' do
     let(:avatar) { fixture_file_upload(Rails.root.join('spec/fixtures/dk.png')) }
     let(:upload_1) { create(:upload, model: group, path: avatar) }
@@ -113,6 +139,20 @@ describe GeoNodeStatus do
     end
   end
 
+  describe '#lfs_objects_failed' do
+    it 'counts failed LFS objects' do
+      # These four should be ignored
+      create(:geo_file_registry, success: false)
+      create(:geo_file_registry, :avatar, success: false)
+      create(:geo_file_registry, file_type: :attachment, success: false)
+      create(:geo_file_registry, :lfs)
+
+      create(:geo_file_registry, :lfs, success: false)
+
+      expect(subject.lfs_objects_failed_count).to eq(1)
+    end
+  end
+
   describe '#lfs_objects_synced_in_percentage' do
     let(:lfs_object_project) { create(:lfs_objects_project, project: project_1) }
 
@@ -218,8 +258,10 @@ describe GeoNodeStatus do
       allow(Gitlab::Geo::HealthCheck).to receive(:db_replication_lag).and_return(nil)
       subject.attachments_count = nil
       subject.attachments_synced_count = nil
+      subject.attachments_failed_count = nil
       subject.lfs_objects_count = nil
       subject.lfs_objects_synced_count = nil
+      subject.lfs_objects_failed_count = nil
       subject.repositories_count = nil
       subject.repositories_synced_count = nil
       subject.repositories_failed_count = nil
@@ -235,9 +277,11 @@ describe GeoNodeStatus do
       expect(subject.repositories_failed_count).to be_zero
       expect(subject.lfs_objects_count).to be_zero
       expect(subject.lfs_objects_synced_count).to be_zero
+      expect(subject.lfs_objects_failed_count).to be_zero
       expect(subject.lfs_objects_synced_in_percentage).to be_zero
       expect(subject.attachments_count).to be_zero
       expect(subject.attachments_synced_count).to be_zero
+      expect(subject.attachments_failed_count).to be_zero
       expect(subject.attachments_synced_in_percentage).to be_zero
       expect(subject.last_event_id).to be_nil
       expect(subject.last_event_date).to be_nil

spec/serializers/geo_node_status_entity_spec.rb

@@ -6,8 +6,10 @@ describe GeoNodeStatusEntity, :postgresql do
       id: 1,
       health: '',
       attachments_count: 329,
+      attachments_failed_count: 25,
       attachments_synced_count: 141,
       lfs_objects_count: 256,
+      lfs_objects_failed_count: 12,
       lfs_objects_synced_count: 123,
       repositories_count: 10,
       repositories_synced_count: 5,
@@ -29,9 +31,11 @@ describe GeoNodeStatusEntity, :postgresql do
   it { is_expected.to have_key(:healthy) }
   it { is_expected.to have_key(:health) }
   it { is_expected.to have_key(:attachments_count) }
+  it { is_expected.to have_key(:attachments_failed_count) }
   it { is_expected.to have_key(:attachments_synced_count) }
   it { is_expected.to have_key(:attachments_synced_in_percentage) }
   it { is_expected.to have_key(:lfs_objects_count) }
+  it { is_expected.to have_key(:lfs_objects_failed_count) }
   it { is_expected.to have_key(:lfs_objects_synced_count) }
   it { is_expected.to have_key(:lfs_objects_synced_in_percentage) }
   it { is_expected.to have_key(:repositories_count) }

spec/services/geo/file_download_service_spec.rb

@@ -8,6 +8,8 @@ describe Geo::FileDownloadService do
 
   before do
     stub_current_geo_node(secondary)
+
+    allow_any_instance_of(Gitlab::ExclusiveLease).to receive(:try_obtain).and_return(true)
   end
 
   describe '#execute' do
@@ -15,15 +17,18 @@ describe Geo::FileDownloadService do
       let(:user) { create(:user, avatar: fixture_file_upload(Rails.root + 'spec/fixtures/dk.png', 'image/png')) }
       let(:upload) { Upload.find_by(model: user, uploader: 'AvatarUploader') }
 
-      subject { described_class.new(:avatar, upload.id) }
+      subject(:execute!) { described_class.new(:avatar, upload.id).execute }
 
       it 'downloads a user avatar' do
-        allow_any_instance_of(Gitlab::ExclusiveLease)
-          .to receive(:try_obtain).and_return(true)
-        allow_any_instance_of(Gitlab::Geo::FileTransfer)
-          .to receive(:download_from_primary).and_return(100)
+        stub_transfer(Gitlab::Geo::FileTransfer, 100)
+
+        expect { execute! }.to change { Geo::FileRegistry.synced.count }.by(1)
+      end
+
+      it 'registers when the download fails' do
+        stub_transfer(Gitlab::Geo::FileTransfer, -1)
 
-        expect { subject.execute }.to change { Geo::FileRegistry.count }.by(1)
+        expect { execute! }.to change { Geo::FileRegistry.failed.count }.by(1)
       end
     end
 
@@ -31,15 +36,18 @@ describe Geo::FileDownloadService do
       let(:group) { create(:group, avatar: fixture_file_upload(Rails.root + 'spec/fixtures/dk.png', 'image/png')) }
       let(:upload) { Upload.find_by(model: group, uploader: 'AvatarUploader') }
 
-      subject { described_class.new(:avatar, upload.id) }
+      subject(:execute!) { described_class.new(:avatar, upload.id).execute }
 
       it 'downloads a group avatar' do
-        allow_any_instance_of(Gitlab::ExclusiveLease)
-          .to receive(:try_obtain).and_return(true)
-        allow_any_instance_of(Gitlab::Geo::FileTransfer)
-          .to receive(:download_from_primary).and_return(100)
+        stub_transfer(Gitlab::Geo::FileTransfer, 100)
+
+        expect { execute! }.to change { Geo::FileRegistry.synced.count }.by(1)
+      end
+
+      it 'registers when the download fails' do
+        stub_transfer(Gitlab::Geo::FileTransfer, -1)
 
-        expect { subject.execute }.to change { Geo::FileRegistry.count }.by(1)
+        expect { execute! }.to change { Geo::FileRegistry.failed.count }.by(1)
       end
     end
 
@@ -47,15 +55,18 @@ describe Geo::FileDownloadService do
       let(:project) { create(:project, avatar: fixture_file_upload(Rails.root + 'spec/fixtures/dk.png', 'image/png')) }
       let(:upload) { Upload.find_by(model: project, uploader: 'AvatarUploader') }
 
-      subject { described_class.new(:avatar, upload.id) }
+      subject(:execute!) { described_class.new(:avatar, upload.id).execute }
 
       it 'downloads a project avatar' do
-        allow_any_instance_of(Gitlab::ExclusiveLease)
-          .to receive(:try_obtain).and_return(true)
-        allow_any_instance_of(Gitlab::Geo::FileTransfer)
-          .to receive(:download_from_primary).and_return(100)
+        stub_transfer(Gitlab::Geo::FileTransfer, 100)
 
-        expect { subject.execute }.to change { Geo::FileRegistry.count }.by(1)
+        expect { execute! }.to change { Geo::FileRegistry.synced.count }.by(1)
+      end
+
+      it 'registers when the download fails' do
+        stub_transfer(Gitlab::Geo::FileTransfer, -1)
+
+        expect { execute! }.to change { Geo::FileRegistry.failed.count }.by(1)
       end
     end
 
@@ -63,30 +74,36 @@ describe Geo::FileDownloadService do
       let(:note) { create(:note, :with_attachment) }
       let(:upload) { Upload.find_by(model: note, uploader: 'AttachmentUploader') }
 
-      subject { described_class.new(:attachment, upload.id) }
+      subject(:execute!) { described_class.new(:attachment, upload.id).execute }
 
       it 'downloads the attachment' do
-        allow_any_instance_of(Gitlab::ExclusiveLease)
-          .to receive(:try_obtain).and_return(true)
-        allow_any_instance_of(Gitlab::Geo::FileTransfer)
-          .to receive(:download_from_primary).and_return(100)
+        stub_transfer(Gitlab::Geo::FileTransfer, 100)
 
-        expect { subject.execute }.to change { Geo::FileRegistry.count }.by(1)
+        expect { execute! }.to change { Geo::FileRegistry.synced.count }.by(1)
+      end
+
+      it 'registers when the download fails' do
+        stub_transfer(Gitlab::Geo::FileTransfer, -1)
+
+        expect { execute! }.to change { Geo::FileRegistry.failed.count }.by(1)
       end
     end
 
     context 'with a snippet' do
       let(:upload) { create(:upload, :personal_snippet) }
 
-      subject { described_class.new(:personal_file, upload.id) }
+      subject(:execute!) { described_class.new(:personal_file, upload.id).execute }
 
       it 'downloads the file' do
-        allow_any_instance_of(Gitlab::ExclusiveLease)
-          .to receive(:try_obtain).and_return(true)
-        allow_any_instance_of(Gitlab::Geo::FileTransfer)
-          .to receive(:download_from_primary).and_return(100)
+        stub_transfer(Gitlab::Geo::FileTransfer, 100)
 
-        expect { subject.execute }.to change { Geo::FileRegistry.count }.by(1)
+        expect { execute! }.to change { Geo::FileRegistry.synced.count }.by(1)
+      end
+
+      it 'registers when the download fails' do
+        stub_transfer(Gitlab::Geo::FileTransfer, -1)
+
+        expect { execute! }.to change { Geo::FileRegistry.failed.count }.by(1)
       end
     end
 
@@ -101,12 +118,15 @@ describe Geo::FileDownloadService do
       end
 
       it 'downloads the file' do
-        allow_any_instance_of(Gitlab::ExclusiveLease)
-          .to receive(:try_obtain).and_return(true)
-        allow_any_instance_of(Gitlab::Geo::FileTransfer)
-          .to receive(:download_from_primary).and_return(100)
+        stub_transfer(Gitlab::Geo::FileTransfer, 100)
 
-        expect { subject.execute }.to change { Geo::FileRegistry.count }.by(1)
+        expect { subject.execute }.to change { Geo::FileRegistry.synced.count }.by(1)
+      end
+
+      it 'registers when the download fails' do
+        stub_transfer(Gitlab::Geo::FileTransfer, -1)
+
+        expect { subject.execute }.to change { Geo::FileRegistry.failed.count }.by(1)
       end
     end
 
@@ -115,18 +135,21 @@ describe Geo::FileDownloadService do
 
       subject { described_class.new(:lfs, lfs_object.id) }
 
-      before do
-        allow_any_instance_of(Gitlab::ExclusiveLease)
-          .to receive(:try_obtain).and_return(true)
-        allow_any_instance_of(Gitlab::Geo::LfsTransfer)
-          .to receive(:download_from_primary).and_return(100)
+      it 'downloads an LFS object' do
+        stub_transfer(Gitlab::Geo::LfsTransfer, 100)
+
+        expect { subject.execute }.to change { Geo::FileRegistry.synced.count }.by(1)
       end
 
-      it 'downloads an LFS object' do
-        expect { subject.execute }.to change { Geo::FileRegistry.count }.by(1)
+      it 'registers when the download fails' do
+        stub_transfer(Gitlab::Geo::LfsTransfer, -1)
+
+        expect { subject.execute }.to change { Geo::FileRegistry.failed.count }.by(1)
       end
 
       it 'logs a message' do
+        stub_transfer(Gitlab::Geo::LfsTransfer, 100)
+
         expect(Gitlab::Geo::Logger).to receive(:info).with(hash_including(:message, :download_time_s, success: true, bytes_downloaded: 100)).and_call_original
 
         subject.execute
@@ -138,5 +161,10 @@ describe Geo::FileDownloadService do
         expect { described_class.new(:bad, 1).execute }.to raise_error(NameError)
       end
     end
+
+    def stub_transfer(kls, result)
+      instance = double("(instance of #{kls})", download_from_primary: result)
+      allow(kls).to receive(:new).and_return(instance)
+    end
   end
 end

spec/workers/geo/file_download_dispatch_worker_spec.rb

@@ -84,6 +84,27 @@ describe Geo::FileDownloadDispatchWorker, :postgresql do
       end
     end
 
+    context 'with a failed file' do
+      let!(:failed_registry) { create(:geo_file_registry, :lfs, file_id: 999, success: false) }
+
+      it 'does not stall backfill' do
+        unsynced = create(:lfs_object, :with_file)
+
+        stub_const('Geo::BaseSchedulerWorker::DB_RETRIEVE_BATCH_SIZE', 1)
+
+        expect(GeoFileDownloadWorker).not_to receive(:perform_async).with(:lfs, failed_registry.file_id)
+        expect(GeoFileDownloadWorker).to receive(:perform_async).with(:lfs, unsynced.id)
+
+        subject.perform
+      end
+
+      it 'retries failed files' do
+        expect(GeoFileDownloadWorker).to receive(:perform_async).with('lfs', failed_registry.file_id)
+
+        subject.perform
+      end
+    end
+
     context 'when node has namespace restrictions' do
       let(:synced_group) { create(:group) }
       let!(:project_in_synced_group) { create(:project, group: synced_group) }