Avoid orphaning blob files on secondaries

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

@@ -34,8 +34,6 @@ module Geo
 
     # Called by Gitlab::Geo::Replicator#consume
     def consume_event_deleted(**params)
-      return unless in_replicables_for_geo_node?
-
       replicate_destroy(params)
     end
 
@@ -100,7 +98,7 @@ module Geo
     def replicate_destroy(event_data)
       ::Geo::FileRegistryRemovalService.new(
         replicable_name,
-        model_record.id,
+        model_record_id,
         event_data[:blob_path]
       ).execute
     end

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

@@ -131,27 +131,37 @@ RSpec.shared_examples 'a blob replicator' do
   end
 
   describe '#consume_event_deleted' do
-    context "when the blob's project is in replicables for this geo node" do
-      it 'invokes Geo::FileRegistryRemovalService' do
-        expect(replicator).to receive(:in_replicables_for_geo_node?).and_return(true)
-        service = double(:service)
+    before do
+      # If a delete event was published from the primary, then the model
+      # record generally does not exist anymore.
+      model_record.delete
+    end
 
-        expect(service).to receive(:execute)
-        expect(::Geo::FileRegistryRemovalService)
-          .to receive(:new).with(replicator.replicable_name, replicator.model_record_id, 'blob_path').and_return(service)
+    # The replicator is instantiated by Geo::EventService on the secondary side,
+    # after the model_record no longer exists. This ensures the tests do not
+    # have access to a model_record instance, only model_record_id. Using this
+    # replicator helps avoid a regression of
+    # https://gitlab.com/gitlab-org/gitlab/-/issues/233040
+    let(:secondary_side_replicator) { ::Gitlab::Geo::Replicator.for_replicable_params(replicable_name: replicator.replicable_name, replicable_id: replicator.model_record_id) }
 
-        replicator.consume_event_deleted({ blob_path: 'blob_path' })
-      end
-    end
+    it 'invokes Geo::FileRegistryRemovalService' do
+      service = double(:service)
 
-    context "when the blob's project is not in replicables for this geo node" do
-      it 'does not invoke Geo::FileRegistryRemovalService' do
-        expect(replicator).to receive(:in_replicables_for_geo_node?).and_return(false)
+      expect(service).to receive(:execute)
+      expect(::Geo::FileRegistryRemovalService)
+        .to receive(:new).with(secondary_side_replicator.replicable_name, secondary_side_replicator.model_record_id, 'blob_path').and_return(service)
 
-        expect(::Geo::FileRegistryRemovalService).not_to receive(:new)
+      secondary_side_replicator.consume_event_deleted({ blob_path: 'blob_path' })
+    end
 
-        replicator.consume_event_deleted({ blob_path: '' })
-      end
+    # `in_replicables_for_geo_node?` returns false if the record does not exist,
+    # therefore it is unlikely to be useful during a delete event. This test is
+    # intended to avoid a regression of
+    # https://gitlab.com/gitlab-org/gitlab/-/issues/233040
+    it 'does not invoke in_replicables_for_geo_node?' do
+      expect(secondary_side_replicator).not_to receive(:in_replicables_for_geo_node?)
+
+      secondary_side_replicator.consume_event_deleted({ blob_path: 'blob_path' })
     end
   end