Fix: [Geo]Replicating objects in object storage schedules removal

and re-syncs object again

We just skip MigratedLocalFilesCleanUpWorker when sync_object_storage
for node is enabled

changelogs/unreleased/204856-replicating-objects-in-object-storage-schedules-removal-and-re-syn.yml

+---
+title: 'Geo: Don''t clean up files in object storage when Geo is responsible of syncing
+  them'
+merge_request: 24901
+author:
+type: fixed

ee/app/workers/geo/migrated_local_files_clean_up_worker.rb

@@ -7,10 +7,10 @@ module Geo
     MAX_CAPACITY = 1000
 
     def perform
+      # No need to run when objects stored in Object Storage should be synced too
+      return if sync_object_storage_enabled?
       # No need to run when nothing is configured to be in Object Storage
-      return unless attachments_object_store_enabled? ||
-          lfs_objects_object_store_enabled? ||
-          job_artifacts_object_store_enabled?
+      return unless object_store_enabled?
 
       super
     end
@@ -93,6 +93,16 @@ module Geo
       JobArtifactUploader.object_store_enabled?
     end
 
+    def object_store_enabled?
+      attachments_object_store_enabled? ||
+        lfs_objects_object_store_enabled? ||
+        job_artifacts_object_store_enabled?
+    end
+
+    def sync_object_storage_enabled?
+      current_node.sync_object_storage
+    end
+
     def attachments_finder
       @attachments_finder ||= AttachmentRegistryFinder.new(current_node_id: current_node.id)
     end

ee/spec/workers/geo/migrated_local_files_clean_up_worker_spec.rb

@@ -7,7 +7,7 @@ describe Geo::MigratedLocalFilesCleanUpWorker, :geo, :geo_fdw do
   include ExclusiveLeaseHelpers
 
   let(:primary)   { create(:geo_node, :primary, host: 'primary-geo-node') }
-  let(:secondary) { create(:geo_node) }
+  let(:secondary) { create(:geo_node, :local_storage_only) }
 
   before do
     stub_current_geo_node(secondary)
@@ -22,6 +22,14 @@ describe Geo::MigratedLocalFilesCleanUpWorker, :geo, :geo_fdw do
     subject.perform
   end
 
+  it 'does not run when sync_object_storage is enabled' do
+    secondary.update_column(:sync_object_storage, true)
+
+    expect(subject).not_to receive(:try_obtain_lease)
+
+    subject.perform
+  end
+
   context 'with LFS objects' do
     let(:lfs_object_local) { create(:lfs_object) }
     let(:lfs_object_remote) { create(:lfs_object, :object_storage) }