When a primary Package is checksummed, reset verification on secondaries

We create "checksum_succeeded" event on primary and propagate it
all the way to secondary.

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

@@ -9,6 +9,10 @@ module Geo
     DEFAULT_VERIFICATION_BATCH_SIZE = 10
     DEFAULT_REVERIFICATION_BATCH_SIZE = 1000
 
+    included do
+      event :checksum_succeeded
+    end
+
     class_methods do
       extend Gitlab::Utils::Override
 
@@ -152,6 +156,20 @@ module Geo
       end
     end
 
+    def handle_after_checksum_succeeded
+      return false unless Gitlab::Geo.primary?
+      return unless self.class.verification_enabled?
+
+      publish(:checksum_succeeded, **event_params)
+    end
+
+    # Called by Gitlab::Geo::Replicator#consume
+    def consume_event_checksum_succeeded(**params)
+      return unless Gitlab::Geo.secondary?
+
+      registry.verification_pending!
+    end
+
     # Schedules a verification job after a model record is created/updated
     def after_verifiable_update
       verify_async if should_primary_verify?

ee/lib/gitlab/geo/verification_state.rb

@@ -85,7 +85,7 @@ module Gitlab
           end
 
           event :verification_pending do
-            transition [:verification_started, :verification_succeeded, :verification_failed] => :verification_pending
+            transition [:verification_pending, :verification_started, :verification_succeeded, :verification_failed] => :verification_pending
           end
         end
 
@@ -313,6 +313,8 @@ module Gitlab
         if resource_updated_during_checksum?(calculation_started_at)
           # just let backfill pick it up
           self.verification_pending!
+        elsif Gitlab::Geo.primary?
+          self.replicator.handle_after_checksum_succeeded
         end
       end
 

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

@@ -359,6 +359,26 @@ RSpec.describe Gitlab::Geo::VerificationState do
         expect(subject.verified_at).not_to be_nil
       end
     end
+
+    context 'primary node' do
+      it 'calls replicator.handle_after_checksum_succeeded' do
+        stub_current_geo_node(primary_node)
+
+        expect(subject.replicator).to receive(:handle_after_checksum_succeeded)
+
+        subject.verification_succeeded_with_checksum!('abc123', Time.current)
+      end
+    end
+
+    context 'secondary node' do
+      it 'does not call replicator.handle_after_checksum_succeeded' do
+        stub_current_geo_node(secondary_node)
+
+        expect(subject.replicator).not_to receive(:handle_after_checksum_succeeded)
+
+        subject.verification_succeeded_with_checksum!('abc123', Time.current)
+      end
+    end
   end
 
   describe '#verification_failed_with_message!' do

ee/spec/support/helpers/ee/geo_helpers.rb

@@ -67,6 +67,10 @@ module EE
           true
         end
 
+        def handle_after_checksum_succeeded
+          true
+        end
+
         protected
 
         def consume_event_test(user:, other:)

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

@@ -12,6 +12,12 @@
 RSpec.shared_examples 'a verifiable replicator' do
   include EE::GeoHelpers
 
+  describe 'events' do
+    it 'has checksum_succeeded event' do
+      expect(described_class.supported_events).to include(:checksum_succeeded)
+    end
+  end
+
   describe '.verification_enabled?' do
     context 'when replication is enabled' do
       before do
@@ -402,6 +408,88 @@ RSpec.shared_examples 'a verifiable replicator' do
     end
   end
 
+  describe '#handle_after_checksum_succeeded' do
+    context 'on a Geo primary' do
+      before do
+        stub_primary_node
+      end
+
+      it 'creates checksum_succeeded event' do
+        expect { replicator.handle_after_checksum_succeeded }.to change { ::Geo::Event.count }.by(1)
+        expect(::Geo::Event.last.event_name).to eq 'checksum_succeeded'
+      end
+
+      it 'is called on verification success' do
+        model_record.verification_started
+
+        expect { model_record.verification_succeeded_with_checksum!('abc123', Time.current) }.to change { ::Geo::Event.count }.by(1)
+        expect(::Geo::Event.last.event_name).to eq 'checksum_succeeded'
+      end
+    end
+
+    context 'on a Geo secondary' do
+      before do
+        stub_secondary_node
+      end
+
+      it 'does not create an event' do
+        expect { replicator.handle_after_checksum_succeeded }.not_to change { ::Geo::Event.count }
+      end
+    end
+  end
+
+  describe '#consume_event_checksum_succeeded' do
+    context 'with a persisted model_record' do
+      before do
+        model_record.save!
+      end
+
+      context 'on a Geo primary' do
+        before do
+          stub_primary_node
+        end
+
+        it 'does nothing' do
+          expect(replicator).not_to receive(:registry)
+
+          replicator.consume_event_checksum_succeeded
+        end
+      end
+
+      context 'on a Geo secondary' do
+        let(:registry) { replicator.registry }
+
+        before do
+          stub_secondary_node
+        end
+
+        context 'with a registry which is verified' do
+          it 'sets state to verification_pending' do
+            registry.verification_started
+            registry.verification_succeeded_with_checksum!('foo', Time.current)
+
+            expect do
+              replicator.consume_event_checksum_succeeded
+            end.to change { registry.reload.verification_state }
+              .from(verification_state_value(:verification_succeeded))
+              .to(verification_state_value(:verification_pending))
+          end
+        end
+
+        context 'with a registry which is pending verification' do
+          it 'does not change state from verification_pending' do
+            registry.save!
+
+            expect do
+              replicator.consume_event_checksum_succeeded
+            end.not_to change { registry.reload.verification_state }
+              .from(verification_state_value(:verification_pending))
+          end
+        end
+      end
+    end
+  end
+
   context 'integration tests' do
     before do
       model_record.save!
@@ -458,4 +546,8 @@ RSpec.shared_examples 'a verifiable replicator' do
       end
     end
   end
+
+  def verification_state_value(state_name)
+    model_record.class.verification_state_value(state_name)
+  end
 end