Add latest changes from gitlab-org/gitlab@master

.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'

.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

.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'

app/models/bulk_imports/export_status.rb

@@ -13,11 +13,15 @@ module BulkImports
     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 @@ module BulkImports
 
     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 }

app/workers/bulk_imports/pipeline_worker.rb

@@ -52,7 +52,7 @@ module BulkImports
         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)

spec/models/bulk_imports/export_status_spec.rb

@@ -10,11 +10,9 @@ RSpec.describe BulkImports::ExportStatus do
   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 @@ RSpec.describe BulkImports::ExportStatus do
       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 @@ RSpec.describe BulkImports::ExportStatus do
 
     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

spec/workers/bulk_imports/pipeline_worker_spec.rb

@@ -178,7 +178,7 @@ RSpec.describe BulkImports::PipelineWorker do
         )
 
         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 @@ RSpec.describe BulkImports::PipelineWorker do
     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 @@ RSpec.describe BulkImports::PipelineWorker do
       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