Handle concurrent VerificationWorker

And also prevent backfill from working the same record.

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

@@ -47,12 +47,26 @@ module Geo
     end
 
     def verify_async
+      # Marking started prevents backfill (VerificationBatchWorker) from picking
+      # this up too.
+      # Also, if another verification job is running, this will make that job
+      # set state to pending after it finishes, since the calculated checksum
+      # is already invalidated.
+      model_record.verification_started!
+
       Geo::VerificationWorker.perform_async(replicable_name, model_record.id)
     end
 
+    # Calculates checksum and asks the model/registry to update verification
+    # state.
     def verify
+      # Deduplicate verification job
+      return unless model_record.verification_started?
+
+      calculation_started_at = Time.current
       checksum = model_record.calculate_checksum
-      model_record.verification_succeeded_with_checksum!(checksum)
+
+      model_record.verification_succeeded_with_checksum!(checksum, calculation_started_at)
     rescue => e
       model_record.verification_failed_with_message!('Error calculating the checksum', e)
     end

ee/lib/gitlab/geo/verification_state.rb

@@ -51,7 +51,6 @@ module Gitlab
 
           before_transition any => :verification_started do |instance, _|
             instance.verification_started_at = Time.current
-            instance.verification_failure = nil
           end
 
           before_transition any => :verification_pending do |instance, _|
@@ -76,7 +75,7 @@ module Gitlab
           end
 
           event :verification_started do
-            transition [:verification_pending, :verification_succeeded, :verification_failed] => :verification_started
+            transition [:verification_pending, :verification_started, :verification_succeeded, :verification_failed] => :verification_started
           end
 
           event :verification_succeeded do
@@ -102,10 +101,21 @@ module Gitlab
       # Convenience method to update checksum and transition to success state.
       #
       # @param [String] checksum value generated by the checksum routine
-      def verification_succeeded_with_checksum!(checksum)
+      # @param [DateTime] calculation_started_at the moment just before the
+      #                   checksum routine was called
+      def verification_succeeded_with_checksum!(checksum, calculation_started_at)
         self.verification_checksum = checksum
 
         self.verification_succeeded!
+
+        if resource_updated_during_checksum?(calculation_started_at)
+          # just let backfill pick it up
+          self.verification_pending!
+        end
+      end
+
+      def resource_updated_during_checksum?(calculation_started_at)
+        self.reset.verification_started_at > calculation_started_at
       end
 
       # Convenience method to update failure message and transition to failed

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

@@ -27,11 +27,26 @@ RSpec.describe Gitlab::Geo::VerificationState do
   subject { DummyModel.new }
 
   describe '#verification_succeeded_with_checksum!' do
-    it 'saves the checksum' do
-      subject.verification_succeeded_with_checksum!('abc123')
+    context 'when the resource was updated during checksum calculation' do
+      let(:calculation_started_at) { subject.verification_started_at - 1.second }
 
-      expect(subject.reload.verification_checksum).to eq('abc123')
-      expect(subject.verified_at).not_to be_nil
+      it 'sets state to pending' do
+        subject.verification_succeeded_with_checksum!('abc123', calculation_started_at)
+
+        expect(subject.reload.verification_pending?).to be_truthy
+      end
+    end
+
+    context 'when the resource was not updated during checksum calculation' do
+      let(:calculation_started_at) { subject.verification_started_at + 1.second }
+
+      it 'saves the checksum' do
+        subject.verification_succeeded_with_checksum!('abc123', calculation_started_at)
+
+        expect(subject.reload.verification_succeeded?).to be_truthy
+        expect(subject.reload.verification_checksum).to eq('abc123')
+        expect(subject.verified_at).not_to be_nil
+      end
     end
   end
 
@@ -41,6 +56,7 @@ RSpec.describe Gitlab::Geo::VerificationState do
 
       subject.verification_failed_with_message!('Failure to calculate checksum', error)
 
+      expect(subject.reload.verification_failed?).to be_truthy
       expect(subject.reload.verification_failure).to eq 'Failure to calculate checksum: An error message'
       expect(subject.verification_retry_count).to be 1
     end

ee/spec/support/shared_examples/models/concerns/verifiable_replicator_shared_examples.rb

@@ -122,12 +122,26 @@ RSpec.shared_examples 'a verifiable replicator' do
     end
   end
 
-  describe '#verify' do
+  describe '#verify_async' do
     before do
-      model_record.verification_started
       model_record.save!
     end
 
+    context 'on a Geo primary' do
+      before do
+        stub_primary_node
+      end
+
+      it 'calls verification_started! and enqueues VerificationWorker' do
+        expect(model_record).to receive(:verification_started!)
+        expect(Geo::VerificationWorker).to receive(:perform_async).with(replicator.replicable_name, model_record.id)
+
+        replicator.verify_async
+      end
+    end
+  end
+
+  describe '#verify' do
     context 'on a Geo primary' do
       before do
         stub_primary_node
@@ -141,7 +155,7 @@ RSpec.shared_examples 'a verifiable replicator' do
         context 'when the checksum succeeds' do
           it 'delegates checksum calculation and the state change to model_record' do
             expect(model_record).to receive(:calculate_checksum).and_return('abc123')
-            expect(model_record).to receive(:verification_succeeded_with_checksum!).with('abc123')
+            expect(model_record).to receive(:verification_succeeded_with_checksum!).with('abc123', kind_of(Time))
 
             replicator.verify
           end
@@ -160,6 +174,14 @@ RSpec.shared_examples 'a verifiable replicator' do
           end
         end
       end
+
+      context 'when verification was not started' do
+        it 'does not call calculate_checksum!' do
+          expect(model_record).not_to receive(:calculate_checksum)
+
+          replicator.verify
+        end
+      end
     end
   end
 end