Fix artifacts object storage geo replication

This fixes the geo replication bug wherein the default behavior of
Carrierwave is skipped when job artifact is uploaded to final location.

Changelog: fixed

app/uploaders/object_storage.rb

@@ -25,7 +25,29 @@ def message
   end
 
   class DirectUploadStorage < ::CarrierWave::Storage::Fog
+    extend ::Gitlab::Utils::Override
+
+    # This override only applies to object storage uploaders (e.g JobArtifactUploader).
+    # - The DirectUploadStorage is only used when object storage is enabled. See `#storage_for`
+    # - This method is called in two possible ways:
+    #   - When a model (e.g. JobArtifact) is saved
+    #   - When uploader.replace_file_without_saving! is called directly
+    #     - For example, see `Gitlab::Geo::Replication::BlobDownloader#download_file`
+    # - We need this override to add the special behavior that bypasses
+    #   CarrierWave's default storing mechanism, which copies a tempfile
+    #   to its final location. In the case of files that are directly uploaded
+    #   by Workhorse to the final location (determined by presence of `<mounted_as>_final_path`) in
+    #   the object storage, the extra copy/delete step of CarrierWave
+    #   is unnecessary.
+    # - We also need to ensure to only bypass the default store behavior if the file given
+    #   is a `CarrierWave::Storage::Fog::File` (uploaded to object storage) and with `<mounted_as>_final_path`
+    #   defined. For everything else, we want to still use the default CarrierWave storage behavior.
+    #   - For example, during Geo replication of job artifacts, `replace_file_without_saving!` is
+    #     called with a sanitized Tempfile. In this case, we want to use the default behavior of
+    #     moving the tempfile to its final location and let CarrierWave upload the file to object storage.
+    override :store!
     def store!(file)
+      return super unless file.is_a?(::CarrierWave::Storage::Fog::File)
       return super unless @uploader.direct_upload_final_path.present?
 
       # The direct_upload_final_path is defined which means

spec/uploaders/object_storage_spec.rb

@@ -1210,6 +1210,86 @@ def escape_path(path)
     end
   end
 
+  describe '#replace_file_without_saving!' do
+    context 'when object storage and direct upload is enabled' do
+      let(:upload_path) { 'some/path/123' }
+
+      let!(:fog_connection) do
+        stub_uploads_object_storage(uploader_class, enabled: true, direct_upload: true)
+      end
+
+      let!(:fog_file) do
+        fog_connection.directories.new(key: 'uploads').files.create( # rubocop:disable Rails/SaveBang
+          key: upload_path,
+          body: 'old content'
+        )
+      end
+
+      before do
+        uploader.object_store = described_class::Store::REMOTE
+      end
+
+      # This scenario can happen when replicating object storage uploads.
+      # See Geo::Replication::BlobDownloader#download_file
+      # A SanitizedFile from a Tempfile will be passed to replace_file_without_saving!
+      context 'and given file is not a CarrierWave::Storage::Fog::File' do
+        let(:temp_file) { Tempfile.new("test") }
+        let(:new_file) { CarrierWave::SanitizedFile.new(temp_file) }
+
+        before do
+          temp_file.write('new content')
+          temp_file.close
+          FileUtils.touch(temp_file)
+          allow(object).to receive(:file_final_path).and_return(file_final_path)
+        end
+
+        after do
+          FileUtils.rm_f(temp_file)
+        end
+
+        shared_examples 'skipping triggers for local file' do
+          it 'allows file to be replaced without triggering any callbacks' do
+            expect(uploader).not_to receive(:with_callbacks)
+
+            uploader.replace_file_without_saving!(new_file)
+          end
+
+          it 'does not trigger pending upload checks' do
+            expect(ObjectStorage::PendingDirectUpload).not_to receive(:complete)
+
+            uploader.replace_file_without_saving!(new_file)
+          end
+        end
+
+        context 'and uploader model has the file_final_path' do
+          let(:file_final_path) { upload_path }
+
+          it_behaves_like 'skipping triggers for local file'
+
+          it 'uses default CarrierWave behavior and uploads the file to object storage using the final path' do
+            uploader.replace_file_without_saving!(new_file)
+
+            updated_content = fog_connection.directories.get('uploads').files.get(upload_path).body
+            expect(updated_content).to eq('new content')
+          end
+        end
+
+        context 'and uploader model has no file_final_path' do
+          let(:file_final_path) { nil }
+
+          it_behaves_like 'skipping triggers for local file'
+
+          it 'uses default CarrierWave behavior and uploads the file to object storage using the uploader store path' do
+            uploader.replace_file_without_saving!(new_file)
+
+            content = fog_connection.directories.get('uploads').files.get(uploader.store_path).body
+            expect(content).to eq('new content')
+          end
+        end
+      end
+    end
+  end
+
   describe '.generate_final_store_path' do
     let(:root_id) { 12345 }
     let(:expected_root_hashed_path) { Gitlab::HashedPath.new(root_hash: root_id) }