Apply retry at time to needs_verification scope

Which is used in `needs_verification_count`. This scope must be kept in
sync with changes to `verification_failed_batch` and
`verification_pending_batch`.

Because `VerificationBatchWorker` (via `LimitedCapacity::Worker`) uses
`needs_verification_count` to determine whether it should reenqueue
itself, and how many jobs should be enqueued by `perform_with_capacity`.

ee/lib/gitlab/geo/verification_state.rb

@@ -36,7 +36,7 @@ module Gitlab
         scope :not_checksummed, -> { where(verification_checksum: nil) }
         scope :verification_timed_out, -> { verification_started.where("verification_started_at < ?", VERIFICATION_TIMEOUT.ago) }
         scope :retry_due, -> { where(arel_table[:verification_retry_at].eq(nil).or(arel_table[:verification_retry_at].lt(Time.current))) }
-        scope :needs_verification, -> { with_verification_state(:verification_pending, :verification_failed) }
+        scope :needs_verification, -> { with_verification_state(:verification_pending).or(with_verification_state(:verification_failed).retry_due) }
         # rubocop:enable CodeReuse/ActiveRecord
 
         state_machine :verification_state, initial: :verification_pending do

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

@@ -145,6 +145,30 @@ RSpec.describe Gitlab::Geo::VerificationState do
     end
   end
 
+  describe '.needs_verification' do
+    it 'includes verification_pending' do
+      subject.save!
+
+      expect(subject.class.needs_verification).to include(subject)
+    end
+
+    it 'includes verification_failed and retry_due' do
+      subject.verification_started
+      subject.verification_failed_with_message!('foo')
+      subject.update!(verification_retry_at: 1.minute.ago)
+
+      expect(subject.class.needs_verification).to include(subject)
+    end
+
+    it 'excludes verification_failed with future verification_retry_at' do
+      subject.verification_started
+      subject.verification_failed_with_message!('foo')
+      subject.update!(verification_retry_at: 1.minute.from_now)
+
+      expect(subject.class.needs_verification).not_to include(subject)
+    end
+  end
+
   describe '.fail_verification_timeouts' do
     before do
       subject.verification_started!

ee/spec/support/shared_examples/models/geo_verifiable_registry_shared_examples.rb

@@ -86,12 +86,22 @@ RSpec.shared_examples 'a Geo verifiable registry' do
       subject.save!
     end
 
-    it 'returns the number of rows which are synced and (pending or failed) verification' do
-      create(registry_class_factory, :synced, verification_state: verification_state_value(:verification_failed), verification_failure: 'foo') # rubocop:disable Rails/SaveBang
+    it 'returns the number of rows which are synced and pending verification' do
+      expect(described_class.needs_verification_count(limit: 3)).to eq(1)
+    end
+
+    it 'includes rows which are synced and failed verification and are due for retry' do
+      create(registry_class_factory, :synced, verification_state: verification_state_value(:verification_failed), verification_failure: 'foo', verification_retry_at: 1.minute.ago) # rubocop:disable Rails/SaveBang
 
       expect(described_class.needs_verification_count(limit: 3)).to eq(2)
     end
 
+    it 'excludes rows which are synced and failed verification and have a future retry time' do
+      create(registry_class_factory, :synced, verification_state: verification_state_value(:verification_failed), verification_failure: 'foo', verification_retry_at: 1.minute.from_now) # rubocop:disable Rails/SaveBang
+
+      expect(described_class.needs_verification_count(limit: 3)).to eq(1)
+    end
+
     it 'excludes rows which are not synced or are not (pending or failed) verification' do
       # rubocop:disable Rails/SaveBang
       create(registry_class_factory, verification_state: verification_state_value(:verification_pending))