Geo repository verification gets stuck after 1000 failed repositories

ee/app/finders/geo/project_registry_finder.rb

@@ -172,18 +172,19 @@ module Geo
     # Find all registries that repository or wiki need verification
     # @return [ActiveRecord::Relation<Geo::ProjectRegistry>] list of registries that need verification
     def fdw_find_registries_to_verify(batch_size:)
+      repo_condition =
+        local_registry_table[:repository_verification_checksum_sha].eq(nil)
+          .and(local_registry_table[:last_repository_verification_failure].eq(nil))
+          .and(fdw_repository_state_table[:repository_verification_checksum].not_eq(nil))
+      wiki_condition =
+        local_registry_table[:wiki_verification_checksum_sha].eq(nil)
+          .and(local_registry_table[:last_wiki_verification_failure].eq(nil))
+          .and(fdw_repository_state_table[:wiki_verification_checksum].not_eq(nil))
+
       Geo::ProjectRegistry
         .joins(fdw_inner_join_repository_state)
-        .where(
-          local_registry_table[:repository_verification_checksum_sha].eq(nil).or(
-            local_registry_table[:wiki_verification_checksum_sha].eq(nil)
-          )
-        )
-        .where(
-          fdw_repository_state_table[:repository_verification_checksum].not_eq(nil).or(
-            fdw_repository_state_table[:wiki_verification_checksum].not_eq(nil)
-          )
-        ).limit(batch_size)
+        .where(repo_condition.or(wiki_condition))
+        .limit(batch_size)
     end
 
     # @return [ActiveRecord::Relation<Geo::ProjectRegistry>]
@@ -307,32 +308,41 @@ module Geo
 
     # @return [ActiveRecord::Relation<Geo::ProjectRegistry>] list of registries that need verification
     def legacy_find_registries_to_verify(batch_size:)
-      registries = Geo::ProjectRegistry.where(
-        local_registry_table[:repository_verification_checksum_sha].eq(nil).or(
-          local_registry_table[:wiki_verification_checksum_sha].eq(nil)
-        )
-      ).pluck(:project_id)
+      repo_condition =
+        local_registry_table[:repository_verification_checksum_sha].eq(nil)
+          .and(local_registry_table[:last_repository_verification_failure].eq(nil))
+      wiki_condition =
+        local_registry_table[:wiki_verification_checksum_sha].eq(nil)
+          .and(local_registry_table[:last_wiki_verification_failure].eq(nil))
+
+      registries = Geo::ProjectRegistry
+        .where(repo_condition.or(wiki_condition))
+        .pluck(:project_id, repo_condition.to_sql, wiki_condition.to_sql)
 
       return Geo::ProjectRegistry.none if registries.empty?
 
-      id_and_want_to_sync = registries.map { |project_id| "(#{project_id}, #{quote_value(true)})" }
+      id_and_want_to_sync = registries.map do |project_id, want_to_sync_repo, want_to_sync_wiki|
+        "(#{project_id}, #{quote_value(want_to_sync_repo)}, #{quote_value(want_to_sync_wiki)})"
+      end
+
+      project_registry_sync_table = Arel::Table.new(:project_registry_sync_table)
 
       joined_relation =
-        ProjectRepositoryState.joins(<<~SQL)
+        ProjectRepositoryState.joins(<<~SQL_REPO)
           INNER JOIN
           (VALUES #{id_and_want_to_sync.join(',')})
-          project_registry(project_id, want_to_sync)
-          ON #{legacy_repository_state_table.name}.project_id = project_registry.project_id
-        SQL
+          project_registry_sync_table(project_id, want_to_sync_repo, want_to_sync_wiki)
+          ON #{legacy_repository_state_table.name}.project_id = project_registry_sync_table.project_id
+        SQL_REPO
 
       project_ids = joined_relation
         .where(
-          legacy_repository_state_table[:repository_verification_checksum].not_eq(nil).or(
-            legacy_repository_state_table[:wiki_verification_checksum].not_eq(nil)
-          )
-        ).where(
-          project_registry: { want_to_sync: true }
-        ).limit(batch_size).pluck(:project_id)
+          legacy_repository_state_table[:repository_verification_checksum].not_eq(nil)
+            .and(project_registry_sync_table[:want_to_sync_repo].eq(true))
+          .or(legacy_repository_state_table[:wiki_verification_checksum].not_eq(nil)
+            .and(project_registry_sync_table[:want_to_sync_wiki].eq(true))))
+        .limit(batch_size)
+        .pluck(:project_id)
 
       Geo::ProjectRegistry.where(project_id: project_ids)
     end

ee/app/finders/geo/repository_verification_finder.rb

@@ -53,10 +53,12 @@ module Geo
 
     def repository_outdated
       repository_state_table[:repository_verification_checksum].eq(nil)
+        .and(repository_state_table[:last_repository_verification_failure].eq(nil))
     end
 
     def wiki_outdated
       repository_state_table[:wiki_verification_checksum].eq(nil)
+        .and(repository_state_table[:last_wiki_verification_failure].eq(nil))
     end
 
     def repository_never_verified

ee/app/services/geo/repository_updated_service.rb

@@ -34,7 +34,7 @@ module Geo
     def reset_repository_checksum!
       return if repository_state.nil?
 
-      repository_state.update!("#{repository_checksum_column}" => nil)
+      repository_state.update!("#{repository_checksum_column}" => nil, "#{repository_failure_column}" => nil)
     rescue => e
       log_error('Cannot reset repository checksum', e)
       raise Gitlab::Git::Checksum::Failure, "Cannot reset repository checksum: #{e}"
@@ -43,5 +43,9 @@ module Geo
     def repository_checksum_column
       "#{Geo::RepositoryUpdatedEvent.sources.key(source)}_verification_checksum"
     end
+
+    def repository_failure_column
+      "last_#{Geo::RepositoryUpdatedEvent.sources.key(source)}_verification_failure"
+    end
   end
 end

ee/app/workers/geo/repository_verification/secondary/single_worker.rb

@@ -15,7 +15,9 @@ module Geo
         delegate :project, to: :registry
 
         def perform(registry_id)
-          @registry = Geo::ProjectRegistry.find(registry_id)
+          return unless Gitlab::Geo.secondary?
+
+          @registry = Geo::ProjectRegistry.find_by_id(registry_id)
           return if registry.nil? || project.pending_delete?
 
           try_obtain_lease do

ee/db/geo/migrate/20180405074130_add_partial_index_project_repository_verification.rb

+class AddPartialIndexProjectRepositoryVerification < ActiveRecord::Migration
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+  REPO_INDEX_NAME = 'idx_project_registry_on_repo_checksums_and_failure_partial'
+  WIKI_INDEX_NAME = 'idx_project_registry_on_wiki_checksums_and_failure_partial'
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_index(:project_registry,
+      [:project_id],
+      name: REPO_INDEX_NAME,
+      where: 'repository_verification_checksum_sha IS NULL AND last_repository_verification_failure IS NULL'
+    )
+
+    add_concurrent_index(:project_registry,
+    [:project_id],
+      name: WIKI_INDEX_NAME,
+      where: 'wiki_verification_checksum_sha IS NULL AND last_wiki_verification_failure IS NULL'
+    )
+  end
+
+  def down
+    remove_concurrent_index_by_name(:project_registry, REPO_INDEX_NAME)
+    remove_concurrent_index_by_name(:project_registry, WIKI_INDEX_NAME)
+  end
+end

ee/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: 20180331055706) do
+ActiveRecord::Schema.define(version: 20180405074130) do
 
   # These are extensions that must be enabled in order to support this database
   enable_extension "plpgsql"
@@ -73,7 +73,9 @@ ActiveRecord::Schema.define(version: 20180331055706) do
 
   add_index "project_registry", ["last_repository_successful_sync_at"], name: "index_project_registry_on_last_repository_successful_sync_at", using: :btree
   add_index "project_registry", ["last_repository_synced_at"], name: "index_project_registry_on_last_repository_synced_at", using: :btree
+  add_index "project_registry", ["project_id"], name: "idx_project_registry_on_repo_checksums_and_failure_partial", where: "((repository_verification_checksum_sha IS NULL) AND (last_repository_verification_failure IS NULL))", using: :btree
   add_index "project_registry", ["project_id"], name: "idx_project_registry_on_repository_failure_partial", where: "(last_repository_verification_failure IS NOT NULL)", using: :btree
+  add_index "project_registry", ["project_id"], name: "idx_project_registry_on_wiki_checksums_and_failure_partial", where: "((wiki_verification_checksum_sha IS NULL) AND (last_wiki_verification_failure IS NULL))", using: :btree
   add_index "project_registry", ["project_id"], name: "idx_project_registry_on_wiki_failure_partial", where: "(last_wiki_verification_failure IS NOT NULL)", using: :btree
   add_index "project_registry", ["project_id"], name: "index_project_registry_on_project_id", unique: true, using: :btree
   add_index "project_registry", ["repository_retry_at"], name: "index_project_registry_on_repository_retry_at", using: :btree

ee/lib/gitlab/geo/log_cursor/daemon.rb

@@ -104,7 +104,8 @@ module Gitlab
 
         def handle_repository_updated_event(event, created_at)
           registry = find_or_initialize_registry(event.project_id,
-            "resync_#{event.source}" => true, "#{event.source}_verification_checksum_sha" => nil)
+            "resync_#{event.source}" => true, "#{event.source}_verification_checksum_sha" => nil,
+            "last_#{event.source}_verification_failure" => nil)
 
           registry.save!
 

ee/spec/finders/geo/project_registry_finder_spec.rb

@@ -385,23 +385,19 @@ describe Geo::ProjectRegistryFinder, :geo do
   end
 
   shared_examples 'find outdated registries for repositories/wikis' do
-    it 'returns registries that verified on primary but not on secondary' do
+    it 'does not return registries that are verified on primary and secondary' do
       project_verified    = create(:repository_state, :repository_verified, :wiki_verified).project
       repository_verified = create(:repository_state, :repository_verified).project
       wiki_verified       = create(:repository_state, :wiki_verified).project
 
       create(:geo_project_registry, :repository_verified, :wiki_verified, project: project_verified)
-      registry_repository_verified = create(:geo_project_registry, :repository_verified, project: repository_verified)
-      registry_wiki_verified       = create(:geo_project_registry, :wiki_verified, project: wiki_verified)
+      create(:geo_project_registry, :repository_verified, project: repository_verified)
+      create(:geo_project_registry, :wiki_verified, project: wiki_verified)
 
-      expect(subject.find_registries_to_verify(batch_size: 100))
-        .to match_array([
-          registry_repository_verified,
-          registry_wiki_verified
-        ])
+      expect(subject.find_registries_to_verify(batch_size: 100)).to be_empty
     end
 
-    it 'does not return registries were unverified/outdated on primary' do
+    it 'does not return registries that were unverified/outdated on primary' do
       project_unverified_primary  = create(:project)
       project_outdated_primary    = create(:repository_state, :repository_outdated, :wiki_outdated).project
       repository_outdated_primary = create(:repository_state, :repository_outdated, :wiki_verified).project
@@ -436,7 +432,15 @@ describe Geo::ProjectRegistryFinder, :geo do
         ])
     end
 
-    it 'does not return registries that both verification failed on primary' do
+    it 'does not return registries that failed on primary' do
+      verification_failed_primary = create(:repository_state, :repository_failed, :wiki_failed).project
+
+      create(:geo_project_registry, project: verification_failed_primary)
+
+      expect(subject.find_registries_to_verify(batch_size: 100)).to be_empty
+    end
+
+    it 'returns registries where one failed and one verified on the primary' do
       verification_failed_primary = create(:repository_state, :repository_failed, :wiki_failed).project
       repository_failed_primary   = create(:repository_state, :repository_failed, :wiki_verified).project
       wiki_failed_primary         = create(:repository_state, :repository_verified, :wiki_failed).project
@@ -452,22 +456,17 @@ describe Geo::ProjectRegistryFinder, :geo do
         ])
     end
 
-    it 'returns registries that verification failed on secondary' do
+    it 'does not return registries where verification failed on secondary' do
       # Verification failed on secondary
       verification_failed_secondary = create(:repository_state, :repository_verified, :wiki_verified).project
       repository_failed_secondary   = create(:repository_state, :repository_verified).project
       wiki_failed_secondary         = create(:repository_state, :wiki_verified).project
 
-      registry_verification_failed_secondary = create(:geo_project_registry, :repository_verification_failed, :wiki_verification_failed, project: verification_failed_secondary)
-      registry_repository_failed_secondary   = create(:geo_project_registry, :repository_verification_failed, project: repository_failed_secondary)
-      registry_wiki_failed_secondary         = create(:geo_project_registry, :wiki_verification_failed, project: wiki_failed_secondary)
+      create(:geo_project_registry, :repository_verification_failed, :wiki_verification_failed, project: verification_failed_secondary)
+      create(:geo_project_registry, :repository_verification_failed, project: repository_failed_secondary)
+      create(:geo_project_registry, :wiki_verification_failed, project: wiki_failed_secondary)
 
-      expect(subject.find_registries_to_verify(batch_size: 100))
-        .to match_array([
-          registry_verification_failed_secondary,
-          registry_repository_failed_secondary,
-          registry_wiki_failed_secondary
-        ])
+      expect(subject.find_registries_to_verify(batch_size: 100)).to be_empty
     end
   end
 

ee/spec/finders/geo/repository_verification_finder_spec.rb

@@ -18,6 +18,12 @@ describe Geo::RepositoryVerificationFinder, :postgresql do
         .to match_array(project)
     end
 
+    it 'does not return projects where repository verification failed' do
+      create(:repository_state, :repository_failed, :wiki_verified, project: project)
+
+      expect(subject.find_outdated_projects(batch_size: 10)).to be_empty
+    end
+
     it 'returns projects where wiki verification is outdated' do
       create(:repository_state, :wiki_outdated, project: project)
 
@@ -31,6 +37,12 @@ describe Geo::RepositoryVerificationFinder, :postgresql do
       expect(subject.find_outdated_projects(batch_size: 10))
         .to match_array(project)
     end
+
+    it 'does not return projects where wiki verification failed' do
+      create(:repository_state, :repository_verified, :wiki_failed, project: project)
+
+      expect(subject.find_outdated_projects(batch_size: 10)).to be_empty
+    end
   end
 
   describe '#find_unverified_projects' do

ee/spec/lib/gitlab/geo/log_cursor/daemon_spec.rb

@@ -125,10 +125,11 @@ describe Gitlab::Geo::LogCursor::Daemon, :postgresql, :clean_gitlab_redis_shared
           expect(registry.reload.resync_repository).to be true
         end
 
-        it 'resets the repository verification checksum' do
+        it 'resets the repository verification checksum and failure' do
           daemon.run_once!
 
-          expect(registry.reload).to have_attributes(resync_repository: true, repository_verification_checksum_sha: nil)
+          expect(registry.reload).to have_attributes(
+            resync_repository: true, repository_verification_checksum_sha: nil, last_repository_verification_failure: nil)
         end
       end
 
@@ -145,10 +146,11 @@ describe Gitlab::Geo::LogCursor::Daemon, :postgresql, :clean_gitlab_redis_shared
           expect(registry.reload.resync_wiki).to be true
         end
 
-        it 'resets the wiki verification checksum' do
+        it 'resets the wiki verification checksum and failure' do
           daemon.run_once!
 
-          expect(registry.reload).to have_attributes(resync_wiki: true, wiki_verification_checksum_sha: nil)
+          expect(registry.reload).to have_attributes(
+            resync_wiki: true, wiki_verification_checksum_sha: nil, last_wiki_verification_failure: nil)
         end
       end
 

ee/spec/services/geo/repository_updated_service_spec.rb

@@ -28,6 +28,10 @@ describe Geo::RepositoryUpdatedService do
         it 'does not reset the repository verification checksum' do
           expect { subject.execute }.not_to change(repository_state.reload, "#{method_prefix}_verification_checksum")
         end
+
+        it 'does not reset the repository verification failure' do
+          expect { subject.execute }.not_to change(repository_state.reload, "last_#{method_prefix}_verification_failure")
+        end
       end
 
       context 'when running on a primary node' do
@@ -39,13 +43,18 @@ describe Geo::RepositoryUpdatedService do
           expect { subject.execute }.to change { repository_state.reload.public_send("#{method_prefix}_verification_checksum") }.to(nil)
         end
 
+        it 'resets the repository verification failure' do
+          repository_state.update!("last_#{method_prefix}_verification_failure" => 'xxxx')
+          expect { subject.execute }.to change { repository_state.reload.public_send("last_#{method_prefix}_verification_failure") }.to(nil)
+        end
+
         it 'does not raise an error when project have never been verified' do
           expect { described_class.new(create(:project)) }.not_to raise_error
         end
 
         it 'raises a Gitlab::Git::Checksum error when an error occurs' do
           allow(subject.repository_state).to receive(:update!)
-            .with("#{method_prefix}_verification_checksum" => nil)
+            .with("#{method_prefix}_verification_checksum" => nil, "last_#{method_prefix}_verification_failure" => nil)
             .and_raise(ActiveRecord::RecordInvalid.new(repository_state))
 
           expect { subject.execute }.to raise_error Gitlab::Git::Checksum::Failure, /Cannot reset repository checksum/

ee/spec/workers/geo/repository_verification/primary/shard_worker_spec.rb

@@ -3,8 +3,9 @@ require 'spec_helper'
 describe Geo::RepositoryVerification::Primary::ShardWorker, :postgresql, :clean_gitlab_redis_cache do
   include ::EE::GeoHelpers
 
-  let!(:primary) { create(:geo_node, :primary) }
+  let!(:primary)   { create(:geo_node, :primary) }
   let(:shard_name) { Gitlab.config.repositories.storages.keys.first }
+  let(:primary_singleworker) { Geo::RepositoryVerification::Primary::SingleWorker }
 
   before do
     stub_current_geo_node(primary)
@@ -21,8 +22,7 @@ describe Geo::RepositoryVerification::Primary::ShardWorker, :postgresql, :clean_
     it 'performs Geo::RepositoryVerification::Primary::SingleWorker for each project' do
       create_list(:project, 2)
 
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-        .to receive(:perform_async).twice
+      expect(primary_singleworker).to receive(:perform_async).twice
 
       subject.perform(shard_name)
     end
@@ -36,12 +36,9 @@ describe Geo::RepositoryVerification::Primary::ShardWorker, :postgresql, :clean_
       create(:repository_state, :repository_outdated, project: repository_outdated)
       create(:repository_state, :wiki_outdated, project: wiki_outdated)
 
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-        .not_to receive(:perform_async).with(verified_project.id)
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-        .to receive(:perform_async).with(repository_outdated.id)
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-        .to receive(:perform_async).with(wiki_outdated.id)
+      expect(primary_singleworker).not_to receive(:perform_async).with(verified_project.id)
+      expect(primary_singleworker).to receive(:perform_async).with(repository_outdated.id)
+      expect(primary_singleworker).to receive(:perform_async).with(wiki_outdated.id)
 
       subject.perform(shard_name)
     end
@@ -51,8 +48,7 @@ describe Geo::RepositoryVerification::Primary::ShardWorker, :postgresql, :clean_
 
       create(:repository_state, :wiki_verified, project: missing_repository_verification)
 
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-        .to receive(:perform_async).with(missing_repository_verification.id)
+      expect(primary_singleworker).to receive(:perform_async).with(missing_repository_verification.id)
 
       subject.perform(shard_name)
     end
@@ -62,8 +58,7 @@ describe Geo::RepositoryVerification::Primary::ShardWorker, :postgresql, :clean_
 
       create(:repository_state, :repository_verified, project: missing_wiki_verification)
 
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-        .to receive(:perform_async).with(missing_wiki_verification.id)
+      expect(primary_singleworker).to receive(:perform_async).with(missing_wiki_verification.id)
 
       subject.perform(shard_name)
     end
@@ -73,8 +68,7 @@ describe Geo::RepositoryVerification::Primary::ShardWorker, :postgresql, :clean_
 
       Gitlab::Geo::ShardHealthCache.update([])
 
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-        .not_to receive(:perform_async)
+      expect(primary_singleworker).not_to receive(:perform_async)
 
       subject.perform(shard_name)
     end
@@ -82,8 +76,7 @@ describe Geo::RepositoryVerification::Primary::ShardWorker, :postgresql, :clean_
     it 'does not perform Geo::RepositoryVerification::Primary::SingleWorker when not running on a primary' do
       allow(Gitlab::Geo).to receive(:primary?) { false }
 
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-        .not_to receive(:perform_async)
+      expect(primary_singleworker).not_to receive(:perform_async)
 
       subject.perform(shard_name)
     end
@@ -106,14 +99,79 @@ describe Geo::RepositoryVerification::Primary::ShardWorker, :postgresql, :clean_
 
       create(:repository_state, :repository_outdated, project: missing_outdated)
 
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-        .to receive(:perform_async).with(healthy_unverified.id)
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-        .not_to receive(:perform_async).with(missing_not_verified.id)
-      expect(Geo::RepositoryVerification::Primary::SingleWorker)
-      .not_to receive(:perform_async).with(missing_outdated.id)
+      expect(primary_singleworker).to receive(:perform_async).with(healthy_unverified.id)
+      expect(primary_singleworker).not_to receive(:perform_async).with(missing_not_verified.id)
+      expect(primary_singleworker).not_to receive(:perform_async).with(missing_outdated.id)
 
       Sidekiq::Testing.inline! { subject.perform(shard_name) }
     end
+
+    # test that jobs are always moving forward and we're not querying the same things
+    # over and over
+    describe 'resource loading' do
+      before do
+        allow(subject).to receive(:db_retrieve_batch_size) { 1 }
+      end
+
+      let(:project_repo_verified)   { create(:repository_state, :repository_verified).project }
+      let(:project_repo_failed)     { create(:repository_state, :repository_failed).project }
+      let(:project_wiki_verified)   { create(:repository_state, :wiki_verified).project }
+      let(:project_wiki_failed)     { create(:repository_state, :wiki_failed).project }
+      let(:project_both_verified)   { create(:repository_state, :repository_verified, :wiki_verified).project }
+      let(:project_both_failed)     { create(:repository_state, :repository_failed, :wiki_failed).project }
+      let(:project_repo_unverified) { create(:repository_state).project }
+      let(:project_wiki_unverified) { create(:repository_state).project }
+
+      it 'handles multiple batches of projects needing verification' do
+        project1 = project_repo_unverified
+        project2 = project_wiki_unverified
+
+        expect(primary_singleworker).to receive(:perform_async).with(project1.id).once
+
+        subject.perform(shard_name)
+
+        project1.repository_state.update_attributes!(
+          repository_verification_checksum: 'f079a831cab27bcda7d81cd9b48296d0c3dd92ee',
+          wiki_verification_checksum: 'e079a831cab27bcda7d81cd9b48296d0c3dd92ef')
+
+        expect(primary_singleworker).to receive(:perform_async).with(project2.id).once
+
+        subject.perform(shard_name)
+      end
+
+      it 'handles multiple batches of projects needing verification, skipping failed repos' do
+        project1 = project_repo_unverified
+        project2 = project_wiki_unverified
+        project3 = project_both_failed      # rubocop:disable Lint/UselessAssignment
+        project4 = project_repo_verified
+        project5 = project_wiki_verified
+
+        expect(primary_singleworker).to receive(:perform_async).with(project1.id).once
+
+        subject.perform(shard_name)
+
+        project1.repository_state.update_attributes!(
+          repository_verification_checksum: 'f079a831cab27bcda7d81cd9b48296d0c3dd92ee',
+          wiki_verification_checksum: 'e079a831cab27bcda7d81cd9b48296d0c3dd92ef')
+
+        expect(primary_singleworker).to receive(:perform_async).with(project2.id).once
+
+        subject.perform(shard_name)
+
+        project2.repository_state.update_attributes!(
+          repository_verification_checksum: 'f079a831cab27bcda7d81cd9b48296d0c3dd92ee',
+          wiki_verification_checksum: 'e079a831cab27bcda7d81cd9b48296d0c3dd92ef')
+
+        expect(primary_singleworker).to receive(:perform_async).with(project4.id).once
+
+        subject.perform(shard_name)
+
+        project4.repository_state.update_attributes!(last_wiki_verification_failure: 'Failed!')
+
+        expect(primary_singleworker).to receive(:perform_async).with(project5.id).once
+
+        subject.perform(shard_name)
+      end
+    end
   end
 end

ee/spec/workers/geo/repository_verification/secondary/shard_worker_spec.rb

@@ -5,11 +5,13 @@ describe Geo::RepositoryVerification::Secondary::ShardWorker, :postgresql, :clea
 
   let!(:secondary) { create(:geo_node) }
   let(:shard_name) { Gitlab.config.repositories.storages.keys.first }
+  let(:secondary_singleworker) { Geo::RepositoryVerification::Secondary::SingleWorker }
 
   set(:project) { create(:project) }
 
   before do
     stub_current_geo_node(secondary)
+    allow(Gitlab::Geo::Fdw).to receive(:enabled?).and_return(false)
   end
 
   describe '#perform' do
@@ -20,46 +22,103 @@ describe Geo::RepositoryVerification::Secondary::ShardWorker, :postgresql, :clea
       Gitlab::Geo::ShardHealthCache.update([shard_name])
     end
 
-    it 'schedule job for each project' do
+    it 'schedules job for each project' do
       other_project = create(:project)
       create(:repository_state, :repository_verified, project: project)
       create(:repository_state, :repository_verified, project: other_project)
       create(:geo_project_registry, :repository_verification_outdated, project: project)
       create(:geo_project_registry, :repository_verification_outdated, project: other_project)
 
-      expect(Geo::RepositoryVerification::Secondary::SingleWorker)
-        .to receive(:perform_async).twice
+      expect(secondary_singleworker).to receive(:perform_async).twice
 
       subject.perform(shard_name)
     end
 
-    it 'schedule job for projects missing repository verification' do
-      create(:repository_state, :wiki_verified, project: project)
+    it 'schedules job for projects missing repository verification' do
+      create(:repository_state, :repository_verified, :wiki_verified, project: project)
       missing_repository_verification = create(:geo_project_registry, :wiki_verified, project: project)
 
-      expect(Geo::RepositoryVerification::Secondary::SingleWorker)
-        .to receive(:perform_async).with(missing_repository_verification.id)
+      expect(secondary_singleworker).to receive(:perform_async).with(missing_repository_verification.id)
 
       subject.perform(shard_name)
     end
 
-    it 'schedule job for projects missing wiki verification' do
-      create(:repository_state, :repository_verified, project: project)
+    it 'schedules job for projects missing wiki verification' do
+      create(:repository_state, :repository_verified, :wiki_verified, project: project)
       missing_wiki_verification = create(:geo_project_registry, :repository_verified, project: project)
 
-      expect(Geo::RepositoryVerification::Secondary::SingleWorker)
-        .to receive(:perform_async).with(missing_wiki_verification.id)
+      expect(secondary_singleworker).to receive(:perform_async).with(missing_wiki_verification.id)
 
       subject.perform(shard_name)
     end
 
+    # test that when jobs are always moving forward and we're not querying the same things
+    # over and over
+    describe 'resource loading' do
+      before do
+        allow(subject).to receive(:db_retrieve_batch_size) { 1 }
+      end
+
+      let(:project1_repo_verified) { create(:repository_state, :repository_verified).project }
+      let(:project2_repo_verified) { create(:repository_state, :repository_verified).project }
+      let(:project3_repo_failed)   { create(:repository_state, :repository_failed).project }
+      let(:project4_wiki_verified) { create(:repository_state, :wiki_verified).project }
+      let(:project5_both_verified) { create(:repository_state, :repository_verified, :wiki_verified).project }
+      let(:project6_both_verified) { create(:repository_state, :repository_verified, :wiki_verified).project }
+
+      it 'handles multiple batches of projects needing verification' do
+        reg1 = create(:geo_project_registry, :repository_verification_outdated, project: project1_repo_verified)
+        reg2 = create(:geo_project_registry, :repository_verification_outdated, project: project2_repo_verified)
+
+        expect(secondary_singleworker).to receive(:perform_async).with(reg1.id).once
+
+        subject.perform(shard_name)
+
+        reg1.update_attributes!(repository_verification_checksum_sha: project1_repo_verified.repository_state.repository_verification_checksum)
+
+        expect(secondary_singleworker).to receive(:perform_async).with(reg2.id).once
+
+        subject.perform(shard_name)
+      end
+
+      it 'handles multiple batches of projects needing verification, skipping failed repos' do
+        reg1 = create(:geo_project_registry, :repository_verification_outdated, project: project1_repo_verified)
+        reg2 = create(:geo_project_registry, :repository_verification_outdated, project: project2_repo_verified)
+        create(:geo_project_registry, :repository_verification_outdated, project: project3_repo_failed)
+        reg4 = create(:geo_project_registry, :wiki_verification_outdated, project: project4_wiki_verified)
+        create(:geo_project_registry, :repository_verification_failed, :wiki_verification_failed, project: project5_both_verified)
+        reg6 = create(:geo_project_registry, project: project6_both_verified)
+
+        expect(secondary_singleworker).to receive(:perform_async).with(reg1.id).once
+
+        subject.perform(shard_name)
+
+        reg1.update_attributes!(repository_verification_checksum_sha: project1_repo_verified.repository_state.repository_verification_checksum)
+
+        expect(secondary_singleworker).to receive(:perform_async).with(reg2.id).once
+
+        subject.perform(shard_name)
+
+        reg2.update_attributes!(repository_verification_checksum_sha: project2_repo_verified.repository_state.repository_verification_checksum)
+
+        expect(secondary_singleworker).to receive(:perform_async).with(reg4.id).once
+
+        subject.perform(shard_name)
+
+        reg4.update_attributes!(last_wiki_verification_failure: 'Failed!')
+
+        expect(secondary_singleworker).to receive(:perform_async).with(reg6.id).once
+
+        subject.perform(shard_name)
+      end
+    end
+
     it 'does not schedule jobs when shard becomes unhealthy' do
       create(:repository_state, project: project)
 
       Gitlab::Geo::ShardHealthCache.update([])
 
-      expect(Geo::RepositoryVerification::Secondary::SingleWorker)
-        .not_to receive(:perform_async)
+      expect(secondary_singleworker).not_to receive(:perform_async)
 
       subject.perform(shard_name)
     end
@@ -67,8 +126,7 @@ describe Geo::RepositoryVerification::Secondary::ShardWorker, :postgresql, :clea
     it 'does not schedule jobs when no geo database is configured' do
       allow(Gitlab::Geo).to receive(:geo_database_configured?) { false }
 
-      expect(Geo::RepositoryVerification::Secondary::SingleWorker)
-        .not_to receive(:perform_async)
+      expect(secondary_singleworker).not_to receive(:perform_async)
 
       subject.perform(shard_name)
 
@@ -80,8 +138,7 @@ describe Geo::RepositoryVerification::Secondary::ShardWorker, :postgresql, :clea
     it 'does not schedule jobs when not running on a secondary' do
       allow(Gitlab::Geo).to receive(:primary?) { false }
 
-      expect(Geo::RepositoryVerification::Secondary::SingleWorker)
-        .not_to receive(:perform_async)
+      expect(secondary_singleworker).not_to receive(:perform_async)
 
       subject.perform(shard_name)
     end

ee/spec/workers/geo/repository_verification/secondary/single_worker_spec.rb

+require 'spec_helper'
+
+describe Geo::RepositoryVerification::Secondary::SingleWorker, :postgresql, :clean_gitlab_redis_cache do
+  include ::EE::GeoHelpers
+
+  let!(:secondary) { create(:geo_node) }
+  let(:project)    { create(:project, :repository, :wiki_repo) }
+  let(:registry)   { create(:geo_project_registry, :synced, project: project) }
+
+  before do
+    stub_current_geo_node(secondary)
+  end
+
+  describe '#perform' do
+    before do
+      allow_any_instance_of(Gitlab::ExclusiveLease).to receive(:try_obtain) { true }
+    end
+
+    it 'does not calculate the checksum when not running on a secondary' do
+      allow(Gitlab::Geo).to receive(:secondary?) { false }
+
+      expect_any_instance_of(Geo::RepositoryVerifySecondaryService).not_to receive(:execute)
+
+      subject.perform(registry.id)
+    end
+
+    it 'does not calculate the checksum when project is pending deletion' do
+      registry.project.update!(pending_delete: true)
+
+      expect_any_instance_of(Geo::RepositoryVerifySecondaryService).not_to receive(:execute)
+
+      subject.perform(registry.id)
+    end
+
+    it 'does not raise an error when registry could not be found' do
+      expect { subject.perform(-1) }.not_to raise_error
+    end
+
+    it 'runs verification for both repository and wiki' do
+      create(:repository_state, project: project, repository_verification_checksum: 'my_checksum', wiki_verification_checksum: 'my_checksum')
+
+      expect(Geo::RepositoryVerifySecondaryService).to receive(:new).with(registry, :repository).and_call_original
+      expect(Geo::RepositoryVerifySecondaryService).to receive(:new).with(registry, :wiki).and_call_original
+
+      subject.perform(registry.id)
+    end
+  end
+end