Perform GeoRepositoryDestroyWorker even when repository does not exist

There is a possibility the project does not have a repository or wiki
on disk, but it can have an entry in the project_registry table, e.g.,
when all attempts to sync the have failed. If we don't perform the
destroy worker it can lead to orphaned entries in the project_registry
table.

ee/app/workers/geo/repositories_clean_up_worker.rb

@@ -3,9 +3,10 @@
 module Geo
   class RepositoriesCleanUpWorker
     include ApplicationWorker
+    include ExclusiveLeaseGuard
     include GeoQueue
+    include Gitlab::Geo::LogHelpers
     include Gitlab::ShellAdapter
-    include ExclusiveLeaseGuard
 
     BATCH_SIZE = 250
     LEASE_TIMEOUT = 60.minutes
@@ -23,38 +24,25 @@ module Geo
           end
         end
       end
-    rescue ActiveRecord::RecordNotFound => e
-      log_error('Could not find Geo node, skipping repositories clean up', geo_node_id: geo_node_id, error: e)
+    rescue ActiveRecord::RecordNotFound => error
+      log_error('Could not find Geo node, skipping repositories clean up', error, geo_node_id: geo_node_id)
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
     private
 
-    # rubocop: disable CodeReuse/ActiveRecord
     def clean_up_repositories(project)
-      # There is a possibility project does not have repository or wiki
-      return true unless gitlab_shell.exists?(project.repository_storage, "#{project.disk_path}.git")
-
       job_id = ::GeoRepositoryDestroyWorker.perform_async(project.id, project.name, project.disk_path, project.repository.storage)
 
       if job_id
-        log_info('Repository cleaned up', project_id: project.id, shard: project.repository.storage, disk_path: project.disk_path, job_id: job_id)
+        log_info('Repository clean up scheduled', project_id: project.id, shard: project.repository.storage, disk_path: project.disk_path, job_id: job_id)
       else
         log_error('Could not clean up repository', project_id: project.id, shard: project.repository.storage, disk_path: project.disk_path)
       end
     end
-    # rubocop: enable CodeReuse/ActiveRecord
 
     def lease_timeout
       LEASE_TIMEOUT
     end
-
-    def log_info(message, params = {})
-      Gitlab::Geo::Logger.info({ class: self.class.name, message: message }.merge(params))
-    end
-
-    def log_error(message, params = {})
-      Gitlab::Geo::Logger.error({ class: self.class.name, message: message }.merge(params))
-    end
   end
 end

ee/spec/workers/geo/repositories_clean_up_worker_spec.rb

@@ -10,7 +10,7 @@ describe Geo::RepositoriesCleanUpWorker do
       stub_exclusive_lease
     end
 
-    context 'when node has namespace restrictions' do
+    context 'when node has selective sync enabled' do
       let(:synced_group) { create(:group) }
       let(:geo_node) { create(:geo_node, selective_sync_type: 'namespaces', namespaces: [synced_group]) }
 
@@ -54,16 +54,31 @@ describe Geo::RepositoriesCleanUpWorker do
         end
       end
 
-      it 'does not perform GeoRepositoryDestroyWorker when repository does not exist' do
-        create(:project)
+      context 'when the project repository does not exist on disk' do
+        let(:project) { create(:project) }
 
-        expect(GeoRepositoryDestroyWorker).not_to receive(:perform_async)
+        it 'performs GeoRepositoryDestroyWorker' do
+          expect(GeoRepositoryDestroyWorker).to receive(:perform_async)
+            .with(project.id, anything, anything, anything)
+            .once
+            .and_return(1)
 
           subject.perform(geo_node.id)
         end
+
+        it 'does not leave orphaned entries in the project_registry table' do
+          create(:geo_project_registry, :sync_failed, project: project)
+
+          Sidekiq::Testing.inline! do
+            subject.perform(geo_node.id)
+          end
+
+          expect(Geo::ProjectRegistry.where(project_id: project)).to be_empty
+        end
+      end
     end
 
-    it 'does not perform GeoRepositoryDestroyWorker when does not node have namespace restrictions' do
+    it 'does not perform GeoRepositoryDestroyWorker when node does not selective sync enabled' do
       expect(GeoRepositoryDestroyWorker).not_to receive(:perform_async)
 
       subject.perform(geo_node.id)