Geo: Replicate delete event of Package Files

Implements replication of delete events for Self-Service
framework

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

@@ -9,6 +9,7 @@ module Geo
 
     included do
       event :created
+      event :deleted
     end
 
     def handle_after_create_commit
@@ -26,6 +27,17 @@ module Geo
       download
     end
 
+    def handle_after_destroy
+      publish(:deleted, **deleted_params)
+    end
+
+    # Called by Gitlab::Geo::Replicator#consume
+    def consume_event_deleted(**params)
+      return if excluded_by_selective_sync?
+
+      replicate_destroy(params)
+    end
+
     # Return the carrierwave uploader instance scoped to current model
     #
     # @abstract
@@ -34,6 +46,13 @@ module Geo
       raise NotImplementedError
     end
 
+    # Return the absolute path to locally stored package file
+    #
+    # @return [String] File path
+    def blob_path
+      carrierwave_uploader.class.absolute_path(carrierwave_uploader)
+    end
+
     def calculate_checksum!
       checksum = model_record.calculate_checksum!
       update_verification_state!(checksum: checksum)
@@ -77,6 +96,14 @@ module Geo
       ::Geo::BlobDownloadService.new(replicator: self).execute
     end
 
+    def replicate_destroy(event_data)
+      ::Geo::FileRegistryRemovalService.new(
+        replicable_name,
+        model_record.id,
+        event_data[:blob_path]
+      ).execute
+    end
+
     def schedule_checksum_calculation
       Geo::BlobVerificationPrimaryWorker.perform_async(replicable_name, model_record.id)
     end
@@ -85,6 +112,10 @@ module Geo
       { model_record_id: model_record.id }
     end
 
+    def deleted_params
+      { model_record_id: model_record.id, blob_path: blob_path }
+    end
+
     def needs_checksum?
       return true unless model_record.respond_to?(:needs_checksum?)
 

ee/lib/gitlab/geo/replicable_model.rb

@@ -10,6 +10,7 @@ module Gitlab
       included do
         # If this hook turns out not to apply to all Models, perhaps we should extract a `ReplicableBlobModel`
         after_create_commit -> { replicator.handle_after_create_commit if replicator.respond_to?(:handle_after_create_commit) }
+        after_destroy -> { replicator.handle_after_destroy if replicator.respond_to?(:handle_after_destroy) }
 
         scope :checksummed, -> { where('verification_checksum IS NOT NULL') }
         scope :checksum_failed, -> { where('verification_failure IS NOT NULL') }

ee/lib/gitlab/geo/replicator.rb

@@ -28,8 +28,8 @@ module Gitlab
       #
       # @example Declaring support for :update and :delete events
       #   class MyReplicator < Gitlab::Geo::Replicator
-      #     event :update
-      #     event :delete
+      #     event :updated
+      #     event :deleted
       #   end
       #
       # @param [Symbol] event_name

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

@@ -47,6 +47,17 @@ RSpec.shared_examples 'a blob replicator' do
     end
   end
 
+  describe '#handle_after_destroy' do
+    it 'creates a Geo::Event' do
+      expect do
+        replicator.handle_after_destroy
+      end.to change { ::Geo::Event.count }.by(1)
+
+      expect(::Geo::Event.last.attributes).to include(
+        "replicable_name" => replicator.replicable_name, "event_name" => "deleted", "payload" => { "model_record_id" => replicator.model_record.id, "blob_path" => replicator.blob_path })
+    end
+  end
+
   describe '#calculate_checksum!' do
     it 'calculates the checksum' do
       model_record.save!
@@ -71,7 +82,7 @@ RSpec.shared_examples 'a blob replicator' do
     end
   end
 
-  describe '#consume_created_event' do
+  describe '#consume_event_created' do
     context "when the blob's project is not excluded by selective sync" do
       it 'invokes Geo::BlobDownloadService' do
         expect(replicator).to receive(:excluded_by_selective_sync?).and_return(false)
@@ -95,6 +106,31 @@ RSpec.shared_examples 'a blob replicator' do
     end
   end
 
+  describe '#consume_event_deleted' do
+    context "when the blob's project is not excluded by selective sync" do
+      it 'invokes Geo::FileRegistryRemovalService' do
+        expect(replicator).to receive(:excluded_by_selective_sync?).and_return(false)
+        service = double(:service)
+
+        expect(service).to receive(:execute)
+        expect(::Geo::FileRegistryRemovalService)
+          .to receive(:new).with(replicator.replicable_name, replicator.model_record_id, 'blob_path').and_return(service)
+
+        replicator.consume_event_deleted({ blob_path: 'blob_path' })
+      end
+    end
+
+    context "when the blob's project is excluded by selective sync" do
+      it 'does not invoke Geo::FileRegistryRemovalService' do
+        expect(replicator).to receive(:excluded_by_selective_sync?).and_return(true)
+
+        expect(::Geo::FileRegistryRemovalService).not_to receive(:new)
+
+        replicator.consume_event_deleted({ blob_path: '' })
+      end
+    end
+  end
+
   describe '#carrierwave_uploader' do
     it 'is implemented' do
       expect do