Enable BulkImports::PipelineWorker to re-run started trackers

app/models/bulk_imports/tracker.rb

@@ -50,6 +50,8 @@ def pipeline_class
 
     event :start do
       transition created: :started
+      # To avoid errors when re-starting a pipeline in case of network errors
+      transition started: :started
     end
 
     event :finish do

app/workers/bulk_imports/pipeline_worker.rb

@@ -16,7 +16,7 @@ class PipelineWorker # rubocop:disable Scalability/IdempotentWorker
 
     def perform(pipeline_tracker_id, stage, entity_id)
       pipeline_tracker = ::BulkImports::Tracker
-        .with_status(:created)
+        .with_status(:created, :started)
         .find_by_id(pipeline_tracker_id)
 
       if pipeline_tracker.present?

spec/workers/bulk_imports/pipeline_worker_spec.rb

@@ -27,42 +27,59 @@ def self.ndjson_pipeline?
       .and_return([[0, pipeline_class]])
   end
 
-  it 'runs the given pipeline successfully' do
-    pipeline_tracker = create(
-      :bulk_import_tracker,
-      entity: entity,
-      pipeline_name: 'FakePipeline'
-    )
-
-    expect_next_instance_of(Gitlab::Import::Logger) do |logger|
-      expect(logger)
-        .to receive(:info)
-        .with(
-          worker: described_class.name,
-          pipeline_name: 'FakePipeline',
-          entity_id: entity.id
-        )
-    end
+  shared_examples 'successfully runs the pipeline' do
+    it 'runs the given pipeline successfully' do
+      expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+        expect(logger)
+          .to receive(:info)
+          .with(
+            worker: described_class.name,
+            pipeline_name: 'FakePipeline',
+            entity_id: entity.id
+          )
+      end
 
-    expect(BulkImports::EntityWorker)
-      .to receive(:perform_async)
-      .with(entity.id, pipeline_tracker.stage)
+      expect(BulkImports::EntityWorker)
+        .to receive(:perform_async)
+        .with(entity.id, pipeline_tracker.stage)
 
-    expect(subject).to receive(:jid).and_return('jid')
+      expect(subject).to receive(:jid).and_return('jid')
 
-    subject.perform(pipeline_tracker.id, pipeline_tracker.stage, entity.id)
+      subject.perform(pipeline_tracker.id, pipeline_tracker.stage, entity.id)
 
-    pipeline_tracker.reload
+      pipeline_tracker.reload
 
-    expect(pipeline_tracker.status_name).to eq(:finished)
-    expect(pipeline_tracker.jid).to eq('jid')
+      expect(pipeline_tracker.status_name).to eq(:finished)
+      expect(pipeline_tracker.jid).to eq('jid')
+    end
+  end
+
+  it_behaves_like 'successfully runs the pipeline' do
+    let(:pipeline_tracker) do
+      create(
+        :bulk_import_tracker,
+        entity: entity,
+        pipeline_name: 'FakePipeline'
+      )
+    end
+  end
+
+  it_behaves_like 'successfully runs the pipeline' do
+    let(:pipeline_tracker) do
+      create(
+        :bulk_import_tracker,
+        :started,
+        entity: entity,
+        pipeline_name: 'FakePipeline'
+      )
+    end
   end
 
   context 'when the pipeline cannot be found' do
     it 'logs the error' do
       pipeline_tracker = create(
         :bulk_import_tracker,
-        :started,
+        :finished,
         entity: entity,
         pipeline_name: 'FakePipeline'
       )