Geo: Apply selective sync to project/wiki events

For example, if a project is outside selective sync, it should not be
synced, nor should it create a registry record (since that will impact
counts and also lead to syncing).

ee/changelogs/unreleased/mk-fix-project-wiki-events-for-selective-sync.yml

+---
+title: 'Geo: Skip project/wiki repo create/update events based on selective sync'
+merge_request: 39488
+author:
+type: fixed

ee/lib/gitlab/geo/log_cursor/events/base_event.rb

@@ -37,6 +37,20 @@ module Gitlab
             yield if healthy_shard_for?(event)
           end
 
+          def replicable_project?
+            strong_memoize(:replicable_project) do
+              # If a registry exists, then it *should* be replicated. The
+              # registry will be removed by the delete event or
+              # RegistryConsistencyWorker if it should no longer be replicated.
+              #
+              # This early exit helps keep event processing efficient especially
+              # for repository updates which are a large proportion of events.
+              next true if registry.persisted?
+
+              Gitlab::Geo.current_node.projects_include?(event.project_id)
+            end
+          end
+
           def log_event(message, params = {})
             logger.event_info(
               created_at,

ee/lib/gitlab/geo/log_cursor/events/repository_created_event.rb

@@ -8,12 +8,14 @@ module Gitlab
           include BaseEvent
 
           def process
-            registry.repository_created!(event)
+            if replicable_project?
+              registry.repository_created!(event)
 
-            job_id = nil
+              job_id = nil
 
-            enqueue_job_if_shard_healthy(event) do
-              job_id = ::Geo::ProjectSyncWorker.perform_async(event.project_id, sync_repository: true, sync_wiki: true)
+              enqueue_job_if_shard_healthy(event) do
+                job_id = ::Geo::ProjectSyncWorker.perform_async(event.project_id, sync_repository: true, sync_wiki: true)
+              end
             end
 
             log_event(job_id)
@@ -29,6 +31,7 @@ module Gitlab
               wiki_path: event.wiki_path,
               resync_repository: registry.resync_repository,
               resync_wiki: registry.resync_wiki,
+              replicable_project: replicable_project?,
               job_id: job_id)
           end
         end

ee/lib/gitlab/geo/log_cursor/events/repository_updated_event.rb

@@ -8,14 +8,16 @@ module Gitlab
           include BaseEvent
 
           def process
-            registry.repository_updated!(event.source, scheduled_at)
+            if replicable_project?
+              registry.repository_updated!(event.source, scheduled_at)
 
-            job_id = enqueue_job_if_shard_healthy(event) do
-              ::Geo::ProjectSyncWorker.perform_async(
-                event.project_id,
-                sync_repository: event.repository?,
-                sync_wiki: event.wiki?
-              )
+              job_id = enqueue_job_if_shard_healthy(event) do
+                ::Geo::ProjectSyncWorker.perform_async(
+                  event.project_id,
+                  sync_repository: event.repository?,
+                  sync_wiki: event.wiki?
+                )
+              end
             end
 
             log_event(job_id)
@@ -31,6 +33,7 @@ module Gitlab
               resync_repository: registry.resync_repository,
               resync_wiki: registry.resync_wiki,
               scheduled_at: scheduled_at,
+              replicable_project: replicable_project?,
               job_id: job_id)
           end
 

ee/spec/lib/gitlab/geo/log_cursor/events/repository_created_event_spec.rb

@@ -3,6 +3,9 @@
 require 'spec_helper'
 
 RSpec.describe Gitlab::Geo::LogCursor::Events::RepositoryCreatedEvent, :clean_gitlab_redis_shared_state do
+  include ::EE::GeoHelpers
+
+  let_it_be(:secondary) { create(:geo_node) }
   let(:logger) { Gitlab::Geo::LogCursor::Logger.new(described_class, Logger::INFO) }
   let(:project) { create(:project) }
   let(:repository_created_event) { create(:geo_repository_created_event, project: project) }
@@ -15,6 +18,10 @@ RSpec.describe Gitlab::Geo::LogCursor::Events::RepositoryCreatedEvent, :clean_gi
     Sidekiq::Testing.fake! { example.run }
   end
 
+  before do
+    stub_current_geo_node(secondary)
+  end
+
   RSpec.shared_examples 'RepositoryCreatedEvent' do
     it 'creates a new project registry' do
       expect { subject.process }.to change(Geo::ProjectRegistry, :count).by(1)
@@ -35,6 +42,18 @@ RSpec.describe Gitlab::Geo::LogCursor::Events::RepositoryCreatedEvent, :clean_gi
       registry = Geo::ProjectRegistry.last
       expect(registry).to have_attributes(project_id: project.id, resync_repository: true, resync_wiki: false)
     end
+
+    context 'when outside selective sync' do
+      before do
+        selective_sync_secondary = create(:geo_node, selective_sync_type: 'shards', selective_sync_shards: ['non-existent'])
+
+        stub_current_geo_node(selective_sync_secondary)
+      end
+
+      it 'does not create a new project registry' do
+        expect { subject.process }.not_to change(Geo::ProjectRegistry, :count)
+      end
+    end
   end
 
   describe '#process' do

ee/spec/lib/gitlab/geo/log_cursor/events/repository_updated_event_spec.rb

@@ -28,6 +28,18 @@ RSpec.describe Gitlab::Geo::LogCursor::Events::RepositoryUpdatedEvent, :clean_gi
       expect { subject.process }.to change(Geo::ProjectRegistry, :count).by(1)
     end
 
+    context 'when outside selective sync' do
+      before do
+        selective_sync_secondary = create(:geo_node, selective_sync_type: 'shards', selective_sync_shards: ['non-existent'])
+
+        stub_current_geo_node(selective_sync_secondary)
+      end
+
+      it 'does not create a new project registry' do
+        expect { subject.process }.not_to change(Geo::ProjectRegistry, :count)
+      end
+    end
+
     context 'when we have an event source' do
       before do
         repository_updated_event.update!(source: event_source)