Merge branch 'fix/sm/atomic-migration' into 'master'ce-40781-os-to-ce

Fix migrate! method (Minimal fix with ExclusiveLock to prevent race conditions)

Closes #4928 and #4980

See merge request gitlab-org/gitlab-ee!4624

7 files changed, 599 insertions, 30 deletions

diff --git a/app/uploaders/object_storage.rb b/app/uploaders/object_storage.rb
index 43d881e5aba..1880cd100dc 100644
--- a/app/uploaders/object_storage.rb
+++ b/app/uploaders/object_storage.rb
@@ -88,7 +88,13 @@ module ObjectStorage
     def changed_mounts
       self.class.uploaders.select do |mount, uploader_class|
         mounted_as = uploader_class.serialization_column(self.class, mount)
-        mount if send(:"#{mounted_as}_changed?") # rubocop:disable GitlabSecurity/PublicSend
+        uploader = send(:"#{mounted_as}") # rubocop:disable GitlabSecurity/PublicSend
+
+        next unless uploader
+        next unless uploader.exists?
+        next unless send(:"#{mounted_as}_changed?") # rubocop:disable GitlabSecurity/PublicSend
+
+        mount
       end.keys
     end
 
@@ -164,7 +170,7 @@ module ObjectStorage
       return unless persist_object_store?
 
       updated = model.update_column(store_serialization_column, object_store)
-      raise ActiveRecordError unless updated
+      raise 'Failed to update object store' unless updated
     end
 
     def use_file
@@ -190,32 +196,12 @@ module ObjectStorage
     #   new_store: Enum (Store::LOCAL, Store::REMOTE)
     #
     def migrate!(new_store)
-      return unless object_store != new_store
-      return unless file
-
-      new_file = nil
-      file_to_delete = file
-      from_object_store = object_store
-      self.object_store = new_store # changes the storage and file
-
-      cache_stored_file! if file_storage?
+      uuid = Gitlab::ExclusiveLease.new(exclusive_lease_key, timeout: 1.hour.to_i).try_obtain
+      raise 'Already running' unless uuid
 
-      with_callbacks(:migrate, file_to_delete) do
-        with_callbacks(:store, file_to_delete) do # for #store_versions!
-          new_file = storage.store!(file)
-          persist_object_store!
-          self.file = new_file
-        end
-      end
-
-      file
-    rescue => e
-      # in case of failure delete new file
-      new_file.delete unless new_file.nil?
-      # revert back to the old file
-      self.object_store = from_object_store
-      self.file = file_to_delete
-      raise e
+      unsafe_migrate!(new_store)
+    ensure
+      Gitlab::ExclusiveLease.cancel(exclusive_lease_key, uuid)
     end
 
     def schedule_background_upload(*args)
@@ -298,5 +284,43 @@ module ObjectStorage
         raise UnknownStoreError
       end
     end
+
+    def exclusive_lease_key
+      "object_storage_migrate:#{model.class}:#{model.id}"
+    end
+
+    #
+    # Move the file to another store
+    #
+    #   new_store: Enum (Store::LOCAL, Store::REMOTE)
+    #
+    def unsafe_migrate!(new_store)
+      return unless object_store != new_store
+      return unless file
+
+      new_file = nil
+      file_to_delete = file
+      from_object_store = object_store
+      self.object_store = new_store # changes the storage and file
+
+      cache_stored_file! if file_storage?
+
+      with_callbacks(:migrate, file_to_delete) do
+        with_callbacks(:store, file_to_delete) do # for #store_versions!
+          new_file = storage.store!(file)
+          persist_object_store!
+          self.file = new_file
+        end
+      end
+
+      file
+    rescue => e
+      # in case of failure delete new file
+      new_file.delete unless new_file.nil?
+      # revert back to the old file
+      self.object_store = from_object_store
+      self.file = file_to_delete
+      raise e
+    end
   end
 end
diff --git a/app/uploaders/records_uploads.rb b/app/uploaders/records_uploads.rb
index 458928bc067..89c74a78835 100644
--- a/app/uploaders/records_uploads.rb
+++ b/app/uploaders/records_uploads.rb
@@ -24,8 +24,7 @@ module RecordsUploads
         uploads.where(path: upload_path).delete_all
         upload.destroy! if upload
 
-        self.upload = build_upload
-        upload.save!
+        self.upload = build_upload.tap(&:save!)
       end
     end
 
diff --git a/spec/models/lfs_object_spec.rb b/spec/models/lfs_object_spec.rb
new file mode 100644
index 00000000000..87f4daab9be
--- /dev/null
+++ b/spec/models/lfs_object_spec.rb
@@ -0,0 +1,110 @@
+require 'spec_helper'
+
+describe LfsObject do
+  describe '#local_store?' do
+    it 'returns true when file_store is nil' do
+      subject.file_store = nil
+
+      expect(subject.local_store?).to eq true
+    end
+
+    it 'returns true when file_store is equal to LfsObjectUploader::Store::LOCAL' do
+      subject.file_store = LfsObjectUploader::Store::LOCAL
+
+      expect(subject.local_store?).to eq true
+    end
+
+    it 'returns false whe file_store is equal to LfsObjectUploader::Store::REMOTE' do
+      subject.file_store = LfsObjectUploader::Store::REMOTE
+
+      expect(subject.local_store?).to eq false
+    end
+  end
+
+  describe '#destroy' do
+    subject { create(:lfs_object, :with_file) }
+
+    context 'when running in a Geo primary node' do
+      set(:primary) { create(:geo_node, :primary) }
+      set(:secondary) { create(:geo_node) }
+
+      it 'logs an event to the Geo event log' do
+        expect { subject.destroy }.to change(Geo::LfsObjectDeletedEvent, :count).by(1)
+      end
+    end
+  end
+
+  describe '#schedule_background_upload' do
+    before do
+      stub_lfs_setting(enabled: true)
+    end
+
+    subject { create(:lfs_object, :with_file) }
+
+    context 'when object storage is disabled' do
+      before do
+        stub_lfs_object_storage(enabled: false)
+      end
+
+      it 'does not schedule the migration' do
+        expect(ObjectStorage::BackgroundMoveWorker).not_to receive(:perform_async)
+
+        subject
+      end
+    end
+
+    context 'when object storage is enabled' do
+      context 'when background upload is enabled' do
+        context 'when is licensed' do
+          before do
+            stub_lfs_object_storage(background_upload: true)
+          end
+
+          it 'schedules the model for migration' do
+            expect(ObjectStorage::BackgroundMoveWorker)
+              .to receive(:perform_async)
+              .with('LfsObjectUploader', described_class.name, :file, kind_of(Numeric))
+              .once
+
+            subject
+          end
+
+          it 'schedules the model for migration once' do
+            expect(ObjectStorage::BackgroundMoveWorker)
+              .to receive(:perform_async)
+              .with('LfsObjectUploader', described_class.name, :file, kind_of(Numeric))
+              .once
+
+            lfs_object = create(:lfs_object)
+            lfs_object.file = fixture_file_upload(Rails.root + "spec/fixtures/dk.png", "`/png")
+            lfs_object.save!
+          end
+        end
+
+        context 'when is unlicensed' do
+          before do
+            stub_lfs_object_storage(background_upload: true, licensed: false)
+          end
+
+          it 'does not schedule the migration' do
+            expect(ObjectStorage::BackgroundMoveWorker).not_to receive(:perform_async)
+
+            subject
+          end
+        end
+      end
+
+      context 'when background upload is disabled' do
+        before do
+          stub_lfs_object_storage(background_upload: false)
+        end
+
+        it 'schedules the model for migration' do
+          expect(ObjectStorage::BackgroundMoveWorker).not_to receive(:perform_async)
+
+          subject
+        end
+      end
+    end
+  end
+end
diff --git a/spec/requests/lfs_http_spec.rb b/spec/requests/lfs_http_spec.rb
index 0ddbddb112c..f7c04c19903 100644
--- a/spec/requests/lfs_http_spec.rb
+++ b/spec/requests/lfs_http_spec.rb
@@ -997,7 +997,7 @@ describe 'Git LFS API and storage' do
 
           context 'and workhorse requests upload finalize for a new lfs object' do
             before do
-              allow_any_instance_of(LfsObjectUploader).to receive(:exists?) { false }
+              lfs_object.destroy
             end
 
             context 'with object storage disabled' do
diff --git a/spec/support/shared_examples/uploaders/object_storage_shared_examples.rb b/spec/support/shared_examples/uploaders/object_storage_shared_examples.rb
index ba7f2f828dd..cd9974cd6e2 100644
--- a/spec/support/shared_examples/uploaders/object_storage_shared_examples.rb
+++ b/spec/support/shared_examples/uploaders/object_storage_shared_examples.rb
@@ -20,6 +20,19 @@ shared_examples "migrates" do |to_store:, from_store: nil|
     migrate(from)
   end
 
+  it 'returns corresponding file type' do
+    expect(subject).to be_an(CarrierWave::Uploader::Base)
+    expect(subject).to be_a(ObjectStorage::Concern)
+
+    if from == described_class::Store::REMOTE
+      expect(subject.file).to be_a(CarrierWave::Storage::Fog::File)
+    elsif from == described_class::Store::LOCAL
+      expect(subject.file).to be_a(CarrierWave::SanitizedFile)
+    else
+      raise 'Unexpected file type'
+    end
+  end
+
   it 'does nothing when migrating to the current store' do
     expect { migrate(from) }.not_to change { subject.object_store }.from(from)
   end
@@ -38,6 +51,42 @@ shared_examples "migrates" do |to_store:, from_store: nil|
     expect(File.exist?(original_file)).to be_falsey
   end
 
+  it 'can access to the original file during migration' do
+    file = subject.file
+
+    allow(subject).to receive(:delete_migrated_file) { } # Remove as a callback of :migrate
+    allow(subject).to receive(:record_upload) { } # Remove as a callback of :store (:record_upload)
+
+    expect(file.exists?).to be_truthy
+    expect { migrate(to) }.not_to change { file.exists? }
+  end
+
+  context 'when migrate! is not oqqupied by another process' do
+    it 'executes migrate!' do
+      expect(subject).to receive(:object_store=).at_least(1)
+
+      migrate(to)
+    end
+  end
+
+  context 'when migrate! is occupied by another process' do
+    let(:exclusive_lease_key) { "object_storage_migrate:#{subject.model.class}:#{subject.model.id}" }
+
+    before do
+      @uuid = Gitlab::ExclusiveLease.new(exclusive_lease_key, timeout: 1.hour.to_i).try_obtain
+    end
+
+    it 'does not execute migrate!' do
+      expect(subject).not_to receive(:unsafe_migrate!)
+
+      expect { migrate(to) }.to raise_error('Already running')
+    end
+
+    after do
+      Gitlab::ExclusiveLease.cancel(exclusive_lease_key, @uuid)
+    end
+  end
+
   context 'migration is unsuccessful' do
     shared_examples "handles gracefully" do |error:|
       it 'does not update the object_store' do
diff --git a/spec/uploaders/job_artifact_uploader_spec.rb b/spec/uploaders/job_artifact_uploader_spec.rb
index 0bcf28f2c1c..714b2498538 100644
--- a/spec/uploaders/job_artifact_uploader_spec.rb
+++ b/spec/uploaders/job_artifact_uploader_spec.rb
@@ -67,4 +67,14 @@ describe JobArtifactUploader do
     it { is_expected.to include("/#{job_artifact.job_id}/#{job_artifact.id}/") }
     it { is_expected.to end_with("ci_build_artifacts.zip") }
   end
+
+  describe "#migrate!" do
+    before do
+      uploader.store!(fixture_file_upload(Rails.root.join('spec/fixtures/trace/sample_trace')))
+      stub_artifacts_object_storage
+    end
+
+    it_behaves_like "migrates", to_store: described_class::Store::REMOTE
+    it_behaves_like "migrates", from_store: described_class::Store::REMOTE, to_store: described_class::Store::LOCAL
+  end
 end
diff --git a/spec/uploaders/object_storage_spec.rb b/spec/uploaders/object_storage_spec.rb
new file mode 100644
index 00000000000..64b59acb286
--- /dev/null
+++ b/spec/uploaders/object_storage_spec.rb
@@ -0,0 +1,377 @@
+require 'rails_helper'
+require 'carrierwave/storage/fog'
+
+class Implementation < GitlabUploader
+  include ObjectStorage::Concern
+  include ::RecordsUploads::Concern
+  prepend ::ObjectStorage::Extension::RecordsUploads
+
+  storage_options Gitlab.config.uploads
+
+  private
+
+  # user/:id
+  def dynamic_segment
+    File.join(model.class.to_s.underscore, model.id.to_s)
+  end
+end
+
+describe ObjectStorage do
+  let(:uploader_class) { Implementation }
+  let(:object) { build_stubbed(:user) }
+  let(:uploader) { uploader_class.new(object, :file) }
+
+  before do
+    allow(uploader_class).to receive(:object_store_enabled?).and_return(true)
+  end
+
+  describe '#object_store=' do
+    it "reload the local storage" do
+      uploader.object_store = described_class::Store::LOCAL
+      expect(uploader.file_storage?).to be_truthy
+    end
+
+    it "reload the REMOTE storage" do
+      uploader.object_store = described_class::Store::REMOTE
+      expect(uploader.file_storage?).to be_falsey
+    end
+  end
+
+  context 'object_store is Store::LOCAL' do
+    before do
+      uploader.object_store = described_class::Store::LOCAL
+    end
+
+    describe '#store_dir' do
+      it 'is the composition of (base_dir, dynamic_segment)' do
+        expect(uploader.store_dir).to start_with("uploads/-/system/user/")
+      end
+    end
+  end
+
+  context 'object_store is Store::REMOTE' do
+    before do
+      uploader.object_store = described_class::Store::REMOTE
+    end
+
+    describe '#store_dir' do
+      it 'is the composition of (dynamic_segment)' do
+        expect(uploader.store_dir).to start_with("user/")
+      end
+    end
+  end
+
+  describe '#object_store' do
+    it "delegates to <mount>_store on model" do
+      expect(object).to receive(:file_store)
+
+      uploader.object_store
+    end
+
+    context 'when store is null' do
+      before do
+        expect(object).to receive(:file_store).and_return(nil)
+      end
+
+      it "returns Store::LOCAL" do
+        expect(uploader.object_store).to eq(described_class::Store::LOCAL)
+      end
+    end
+
+    context 'when value is set' do
+      before do
+        expect(object).to receive(:file_store).and_return(described_class::Store::REMOTE)
+      end
+
+      it "returns the given value" do
+        expect(uploader.object_store).to eq(described_class::Store::REMOTE)
+      end
+    end
+  end
+
+  describe '#file_cache_storage?' do
+    context 'when file storage is used' do
+      before do
+        uploader_class.cache_storage(:file)
+      end
+
+      it { expect(uploader).to be_file_cache_storage }
+    end
+
+    context 'when is remote storage' do
+      before do
+        uploader_class.cache_storage(:fog)
+      end
+
+      it { expect(uploader).not_to be_file_cache_storage }
+    end
+  end
+
+  # this means the model shall include
+  #   include RecordsUpload::Concern
+  #   prepend ObjectStorage::Extension::RecordsUploads
+  # the object_store persistence is delegated to the `Upload` model.
+  #
+  context 'when persist_object_store? is false' do
+    let(:object) { create(:project, :with_avatar) }
+    let(:uploader) { object.avatar }
+
+    it { expect(object).to be_a(Avatarable) }
+    it { expect(uploader.persist_object_store?).to be_falsey }
+
+    describe 'delegates the object_store logic to the `Upload` model' do
+      it 'sets @upload to the found `upload`' do
+        expect(uploader.upload).to eq(uploader.upload)
+      end
+
+      it 'sets @object_store to the `Upload` value' do
+        expect(uploader.object_store).to eq(uploader.upload.store)
+      end
+    end
+
+    describe '#migrate!' do
+      let(:new_store) { ObjectStorage::Store::REMOTE }
+
+      before do
+        stub_uploads_object_storage(uploader: AvatarUploader)
+      end
+
+      subject { uploader.migrate!(new_store) }
+
+      it 'persist @object_store to the recorded upload' do
+        subject
+
+        expect(uploader.upload.store).to eq(new_store)
+      end
+
+      describe 'fails' do
+        it 'is handled gracefully' do
+          store = uploader.object_store
+          expect_any_instance_of(Upload).to receive(:save!).and_raise("An error")
+
+          expect { subject }.to raise_error("An error")
+          expect(uploader.exists?).to be_truthy
+          expect(uploader.upload.store).to eq(store)
+        end
+      end
+    end
+  end
+
+  # this means the model holds an <mounted_as>_store attribute directly
+  # and do not delegate the object_store persistence to the `Upload` model.
+  #
+  context 'persist_object_store? is true' do
+    context 'when using JobArtifactsUploader' do
+      let(:store) { described_class::Store::LOCAL }
+      let(:object) { create(:ci_job_artifact, :archive, file_store: store) }
+      let(:uploader) { object.file }
+
+      context 'checking described_class' do
+        it "uploader include described_class::Concern" do
+          expect(uploader).to be_a(described_class::Concern)
+        end
+      end
+
+      describe '#use_file' do
+        context 'when file is stored locally' do
+          it "calls a regular path" do
+            expect { |b| uploader.use_file(&b) }.not_to yield_with_args(%r[tmp/cache])
+          end
+        end
+
+        context 'when file is stored remotely' do
+          let(:store) { described_class::Store::REMOTE }
+
+          before do
+            stub_artifacts_object_storage
+          end
+
+          it "calls a cache path" do
+            expect { |b| uploader.use_file(&b) }.to yield_with_args(%r[tmp/cache])
+          end
+        end
+      end
+
+      describe '#migrate!' do
+        subject { uploader.migrate!(new_store) }
+
+        shared_examples "updates the underlying <mounted>_store" do
+          it do
+            subject
+
+            expect(object.file_store).to eq(new_store)
+          end
+        end
+
+        context 'when using the same storage' do
+          let(:new_store) { store }
+
+          it "to not migrate the storage" do
+            subject
+
+            expect(uploader).not_to receive(:store!)
+            expect(uploader.object_store).to eq(store)
+          end
+        end
+
+        context 'when migrating to local storage' do
+          let(:store) { described_class::Store::REMOTE }
+          let(:new_store) { described_class::Store::LOCAL }
+
+          before do
+            stub_artifacts_object_storage
+          end
+
+          include_examples "updates the underlying <mounted>_store"
+
+          it "local file does not exist" do
+            expect(File.exist?(uploader.path)).to eq(false)
+          end
+
+          it "remote file exist" do
+            expect(uploader.file.exists?).to be_truthy
+          end
+
+          it "does migrate the file" do
+            subject
+
+            expect(uploader.object_store).to eq(new_store)
+            expect(File.exist?(uploader.path)).to eq(true)
+          end
+        end
+
+        context 'when migrating to remote storage' do
+          let(:new_store) { described_class::Store::REMOTE }
+          let!(:current_path) { uploader.path }
+
+          it "file does exist" do
+            expect(File.exist?(current_path)).to eq(true)
+          end
+
+          context 'when storage is disabled' do
+            before do
+              stub_artifacts_object_storage(enabled: false)
+            end
+
+            it "to raise an error" do
+              expect { subject }.to raise_error(/Object Storage is not enabled/)
+            end
+          end
+
+          context 'when storage is unlicensed' do
+            before do
+              stub_artifacts_object_storage(licensed: false)
+            end
+
+            it "raises an error" do
+              expect { subject }.to raise_error(/Object Storage feature is missing/)
+            end
+          end
+
+          context 'when credentials are set' do
+            before do
+              stub_artifacts_object_storage
+            end
+
+            include_examples "updates the underlying <mounted>_store"
+
+            it "does migrate the file" do
+              subject
+
+              expect(uploader.object_store).to eq(new_store)
+            end
+
+            it "does delete original file" do
+              subject
+
+              expect(File.exist?(current_path)).to eq(false)
+            end
+
+            context 'when subject save fails' do
+              before do
+                expect(uploader).to receive(:persist_object_store!).and_raise(RuntimeError, "exception")
+              end
+
+              it "original file is not removed" do
+                expect { subject }.to raise_error(/exception/)
+
+                expect(File.exist?(current_path)).to eq(true)
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+
+  describe '#fog_directory' do
+    let(:remote_directory) { 'directory' }
+
+    before do
+      uploader_class.storage_options double(object_store: double(remote_directory: remote_directory))
+    end
+
+    subject { uploader.fog_directory }
+
+    it { is_expected.to eq(remote_directory) }
+  end
+
+  describe '#fog_credentials' do
+    let(:connection) { Settingslogic.new("provider" => "AWS") }
+
+    before do
+      uploader_class.storage_options double(object_store: double(connection: connection))
+    end
+
+    subject { uploader.fog_credentials }
+
+    it { is_expected.to eq(provider: 'AWS') }
+  end
+
+  describe '#fog_public' do
+    subject { uploader.fog_public }
+
+    it { is_expected.to eq(false) }
+  end
+
+  describe '#verify_license!' do
+    subject { uploader.verify_license!(nil) }
+
+    context 'when using local storage' do
+      before do
+        expect(object).to receive(:file_store) { described_class::Store::LOCAL }
+      end
+
+      it "does not raise an error" do
+        expect { subject }.not_to raise_error
+      end
+    end
+
+    context 'when using remote storage' do
+      before do
+        uploader_class.storage_options double(object_store: double(enabled: true))
+        expect(object).to receive(:file_store) { described_class::Store::REMOTE }
+      end
+
+      context 'feature is not available' do
+        before do
+          expect(License).to receive(:feature_available?).with(:object_storage).and_return(false)
+        end
+
+        it "does raise an error" do
+          expect { subject }.to raise_error(/Object Storage feature is missing/)
+        end
+      end
+
+      context 'feature is available' do
+        before do
+          expect(License).to receive(:feature_available?).with(:object_storage).and_return(true)
+        end
+
+        it "does not raise an error" do
+          expect { subject }.not_to raise_error
+        end
+      end
+    end
+  end
+end