Merge branch 'rodrigo/358743-fix-race-condition-in-bulk-import' into 'master'

Update BulkImports to handle empty relation status responses See merge request gitlab-org/gitlab!88019

Merge branch 'rodrigo/358743-fix-race-condition-in-bulk-import' into 'master'
Update BulkImports to handle empty relation status responses See merge request gitlab-org/gitlab!88019
3c9803ca · Kerri Miller · e6df1842 · a30ed390 · 3c9803ca · 3c9803ca
--- a/.rubocop_todo/rspec/predicate_matcher.yml
+++ b/.rubocop_todo/rspec/predicate_matcher.yml
@@ -503,7 +503,6 @@ RSpec/PredicateMatcher:
    - 'spec/validators/namespace_path_validator_spec.rb'
    - 'spec/validators/project_path_validator_spec.rb'
    - 'spec/workers/bulk_imports/entity_worker_spec.rb'
-    - 'spec/workers/bulk_imports/pipeline_worker_spec.rb'
    - 'spec/workers/ci/delete_objects_worker_spec.rb'
    - 'spec/workers/concerns/worker_attributes_spec.rb'
    - 'spec/workers/container_expiration_policies/cleanup_container_repository_worker_spec.rb'

--- a/.rubocop_todo/rspec/verified_doubles.yml
+++ b/.rubocop_todo/rspec/verified_doubles.yml
@@ -831,7 +831,6 @@ RSpec/VerifiedDoubles:
    - spec/models/application_record_spec.rb
    - spec/models/badge_spec.rb
    - spec/models/badges/project_badge_spec.rb
-    - spec/models/bulk_imports/export_status_spec.rb
    - spec/models/ci/build_spec.rb
    - spec/models/ci/build_trace_chunk_spec.rb
    - spec/models/ci/commit_with_pipeline_spec.rb
@@ -1128,7 +1127,6 @@ RSpec/VerifiedDoubles:
    - spec/views/shared/milestones/_issuables.html.haml_spec.rb
    - spec/views/shared/wikis/_sidebar.html.haml_spec.rb
    - spec/workers/bulk_imports/export_request_worker_spec.rb
-    - spec/workers/bulk_imports/pipeline_worker_spec.rb
    - spec/workers/chat_notification_worker_spec.rb
    - spec/workers/ci/build_prepare_worker_spec.rb
    - spec/workers/ci/create_cross_project_pipeline_worker_spec.rb

--- a/.rubocop_todo/style/if_unless_modifier.yml
+++ b/.rubocop_todo/style/if_unless_modifier.yml
@@ -414,7 +414,6 @@ Style/IfUnlessModifier:
    - 'app/views/projects/merge_requests/index.atom.builder'
    - 'app/workers/authorized_project_update/user_refresh_from_replica_worker.rb'
    - 'app/workers/auto_devops/disable_worker.rb'
-    - 'app/workers/bulk_imports/pipeline_worker.rb'
    - 'app/workers/cleanup_container_repository_worker.rb'
    - 'app/workers/concerns/application_worker.rb'
    - 'app/workers/concerns/packages/cleanup_artifact_worker.rb'

--- a/app/models/bulk_imports/export_status.rb
+++ b/app/models/bulk_imports/export_status.rb
@@ -13,11 +13,15 @@ def initialize(pipeline_tracker, relation)
    end

    def started?
-      export_status['status'] == Export::STARTED
+      !empty? && export_status['status'] == Export::STARTED
    end

    def failed?
-      export_status['status'] == Export::FAILED
+      !empty? && export_status['status'] == Export::FAILED
+    end
+
+    def empty?
+      export_status.nil?
    end

    def error
@@ -30,14 +34,7 @@ def error

    def export_status
      strong_memoize(:export_status) do
-        status = fetch_export_status
-
-        relation_export_status = status&.find { |item| item['relation'] == relation }
-
-        # Consider empty response as failed export
-        raise StandardError, 'Empty relation export status' unless relation_export_status&.present?
-
-        relation_export_status
+        fetch_export_status&.find { |item| item['relation'] == relation }
      end
    rescue StandardError => e
      { 'status' => Export::FAILED, 'error' => e.message }

--- a/app/workers/bulk_imports/pipeline_worker.rb
+++ b/app/workers/bulk_imports/pipeline_worker.rb
@@ -52,7 +52,7 @@ def run(pipeline_tracker)
        raise(Pipeline::ExpiredError, 'Pipeline timeout') if job_timeout?(pipeline_tracker)
        raise(Pipeline::FailedError, export_status.error) if export_status.failed?

-        return reenqueue(pipeline_tracker) if export_status.started?
+        return reenqueue(pipeline_tracker) if export_status.empty? || export_status.started?
      end

      pipeline_tracker.update!(status_event: 'start', jid: jid)

--- a/spec/models/bulk_imports/export_status_spec.rb
+++ b/spec/models/bulk_imports/export_status_spec.rb
@@ -10,11 +10,9 @@
  let_it_be(:tracker) { create(:bulk_import_tracker, entity: entity) }

  let(:response_double) do
-    double(parsed_response: [{ 'relation' => 'labels', 'status' => status, 'error' => 'error!' }])
-  end
-
-  let(:invalid_response_double) do
-    double(parsed_response: [{ 'relation' => 'not_a_real_relation', 'status' => status, 'error' => 'error!' }])
+    instance_double(HTTParty::Response,
+      parsed_response: [{ 'relation' => 'labels', 'status' => status, 'error' => 'error!' }]
+    )
  end

  subject { described_class.new(tracker, relation) }
@@ -40,22 +38,34 @@
      it 'returns false' do
        expect(subject.started?).to eq(false)
      end
+    end

-      context 'when returned relation is invalid' do
-        before do
-          allow_next_instance_of(BulkImports::Clients::HTTP) do |client|
-            allow(client).to receive(:get).and_return(invalid_response_double)
-          end
-        end
+    context 'when export status is not present' do
+      let(:response_double) do
+        instance_double(HTTParty::Response, parsed_response: [])
+      end

-        it 'returns false' do
-          expect(subject.started?).to eq(false)
+      it 'returns false' do
+        expect(subject.started?).to eq(false)
+      end
+    end
+
+    context 'when something goes wrong during export status fetch' do
+      before do
+        allow_next_instance_of(BulkImports::Clients::HTTP) do |client|
+          allow(client).to receive(:get).and_raise(
+            BulkImports::NetworkError.new("Unsuccessful response", response: nil)
+          )
        end
      end
+
+      it 'returns false' do
+        expect(subject.started?).to eq(false)
+      end
    end
  end

-  describe '#failed' do
+  describe '#failed?' do
    context 'when export status is failed' do
      let(:status) { BulkImports::Export::FAILED }

@@ -74,12 +84,67 @@

    context 'when export status is not present' do
      let(:response_double) do
-        double(parsed_response: [])
+        instance_double(HTTParty::Response, parsed_response: [])
+      end
+
+      it 'returns false' do
+        expect(subject.started?).to eq(false)
+      end
+    end
+
+    context 'when something goes wrong during export status fetch' do
+      before do
+        allow_next_instance_of(BulkImports::Clients::HTTP) do |client|
+          allow(client).to receive(:get).and_raise(
+            BulkImports::NetworkError.new("Unsuccessful response", response: nil)
+          )
+        end
+      end
+
+      it 'returns false' do
+        expect(subject.started?).to eq(false)
+      end
+    end
+  end
+
+  describe '#empty?' do
+    context 'when export status is present' do
+      let(:status) { 'any status' }
+
+      it { expect(subject.empty?).to eq(false) }
+    end
+
+    context 'when export status is not present' do
+      let(:response_double) do
+        instance_double(HTTParty::Response, parsed_response: [])
      end

      it 'returns true' do
-        expect(subject.failed?).to eq(true)
-        expect(subject.error).to eq('Empty relation export status')
+        expect(subject.empty?).to eq(true)
+      end
+    end
+
+    context 'when export status is empty' do
+      let(:response_double) do
+        instance_double(HTTParty::Response, parsed_response: nil)
+      end
+
+      it 'returns true' do
+        expect(subject.empty?).to eq(true)
+      end
+    end
+
+    context 'when something goes wrong during export status fetch' do
+      before do
+        allow_next_instance_of(BulkImports::Clients::HTTP) do |client|
+          allow(client).to receive(:get).and_raise(
+            BulkImports::NetworkError.new("Unsuccessful response", response: nil)
+          )
+        end
+      end
+
+      it 'returns false' do
+        expect(subject.started?).to eq(false)
      end
    end
  end

--- a/spec/workers/bulk_imports/pipeline_worker_spec.rb
+++ b/spec/workers/bulk_imports/pipeline_worker_spec.rb
@@ -178,7 +178,7 @@ def self.file_extraction_pipeline?
        )

        exception = BulkImports::NetworkError.new(
-          response: double(code: 429, headers: {})
+          response: instance_double(HTTParty::Response, code: 429, headers: {})
        )

        expect_next_instance_of(pipeline_class) do |pipeline|
@@ -260,6 +260,7 @@ def self.relation
    it 'runs the pipeline successfully' do
      allow_next_instance_of(BulkImports::ExportStatus) do |status|
        allow(status).to receive(:started?).and_return(false)
+        allow(status).to receive(:empty?).and_return(false)
        allow(status).to receive(:failed?).and_return(false)
      end

@@ -272,6 +273,28 @@ def self.relation
      it 'reenqueues pipeline worker' do
        allow_next_instance_of(BulkImports::ExportStatus) do |status|
          allow(status).to receive(:started?).and_return(true)
+          allow(status).to receive(:empty?).and_return(false)
+          allow(status).to receive(:failed?).and_return(false)
+        end
+
+        expect(described_class)
+          .to receive(:perform_in)
+          .with(
+            described_class::FILE_EXTRACTION_PIPELINE_PERFORM_DELAY,
+            pipeline_tracker.id,
+            pipeline_tracker.stage,
+            entity.id
+          )
+
+        subject.perform(pipeline_tracker.id, pipeline_tracker.stage, entity.id)
+      end
+    end
+
+    context 'when export status is empty' do
+      it 'reenqueues pipeline worker' do
+        allow_next_instance_of(BulkImports::ExportStatus) do |status|
+          allow(status).to receive(:started?).and_return(false)
+          allow(status).to receive(:empty?).and_return(true)
          allow(status).to receive(:failed?).and_return(false)
        end