Import/Fork/Mirror jobs now skip starting transition if the worker was killed...

Import/Fork/Mirror jobs now skip starting transition if the worker was killed and reset without proper cleanup.

app/workers/concerns/project_start_import.rb

+module ProjectStartImport
+  def start(project)
+    if project.import_started? && project.import_jid == self.jid
+      return true
+    end
+
+    project.import_start
+  end
+end

app/workers/repository_fork_worker.rb

@@ -4,6 +4,7 @@ class RepositoryForkWorker
   include Sidekiq::Worker
   include Gitlab::ShellAdapter
   include DedicatedSidekiqQueue
+  include ProjectStartImport
 
   sidekiq_options status_expiration: StuckImportJobsWorker::IMPORT_JOBS_EXPIRATION
 
@@ -37,7 +38,7 @@ class RepositoryForkWorker
   private
 
   def start_fork(project)
-    return true if project.import_start
+    return true if start(project)
 
     Rails.logger.info("Project #{project.full_path} was in inconsistent state (#{project.import_status}) while forking.")
     false

app/workers/repository_import_worker.rb

@@ -4,6 +4,7 @@ class RepositoryImportWorker
   include Sidekiq::Worker
   include DedicatedSidekiqQueue
   include ExceptionBacktrace
+  include ProjectStartImport
 
   sidekiq_options status_expiration: StuckImportJobsWorker::IMPORT_JOBS_EXPIRATION
 
@@ -38,7 +39,7 @@ class RepositoryImportWorker
   private
 
   def start_import(project)
-    return true if project.import_start
+    return true if start(project)
 
     Rails.logger.info("Project #{project.full_path} was in inconsistent state (#{project.import_status}) while importing.")
     false

app/workers/repository_update_mirror_worker.rb

@@ -4,6 +4,7 @@ class RepositoryUpdateMirrorWorker
   include Sidekiq::Worker
   include Gitlab::ShellAdapter
   include DedicatedSidekiqQueue
+  include ProjectStartImport
 
   LEASE_KEY = 'repository_update_mirror_worker_start_scheduler'.freeze
   LEASE_TIMEOUT = 2.seconds
@@ -45,7 +46,7 @@ class RepositoryUpdateMirrorWorker
   end
 
   def start_mirror(project)
-    if project.import_start
+    if start(project)
       Rails.logger.info("Mirror update for #{project.full_path} started. Waiting duration: #{project.mirror_waiting_duration}")
       Gitlab::Metrics.add_event_with_values(
         :mirrors_running,

spec/workers/repository_fork_worker_spec.rb

@@ -12,6 +12,28 @@ describe RepositoryForkWorker do
   end
 
   describe "#perform" do
+    describe 'when a worker was reset without cleanup' do
+      let(:jid) { '12345678' }
+      let(:started_project) { create(:project, :repository, :import_started) }
+
+      it 'creates a new repository from a fork' do
+        allow(subject).to receive(:jid).and_return(jid)
+
+        expect(shell).to receive(:fork_repository).with(
+          '/test/path',
+          project.full_path,
+          project.repository_storage_path,
+          fork_project.namespace.full_path
+        ).and_return(true)
+
+        subject.perform(
+          project.id,
+          '/test/path',
+          project.full_path,
+          fork_project.namespace.full_path)
+      end
+    end
+
     it "creates a new repository from a fork" do
       expect(shell).to receive(:fork_repository).with(
         '/test/path',

spec/workers/repository_import_worker_spec.rb

@@ -6,6 +6,23 @@ describe RepositoryImportWorker do
   subject { described_class.new }
 
   describe '#perform' do
+    context 'when worker was reset without cleanup' do
+      let(:jid) { '12345678' }
+      let(:started_project) { create(:project, :import_started, import_jid: jid) }
+
+      it 'imports the project successfully' do
+        allow(subject).to receive(:jid).and_return(jid)
+
+        expect_any_instance_of(Projects::ImportService).to receive(:execute)
+          .and_return({ status: :ok })
+
+        expect_any_instance_of(Repository).to receive(:expire_emptiness_caches)
+        expect_any_instance_of(Project).to receive(:import_finish)
+
+        subject.perform(project.id)
+      end
+    end
+
     context 'when the import was successful' do
       it 'imports a project' do
         expect_any_instance_of(Projects::ImportService).to receive(:execute)

spec/workers/repository_update_mirror_worker_spec.rb

@@ -2,10 +2,12 @@ require 'rails_helper'
 
 describe RepositoryUpdateMirrorWorker do
   describe '#perform' do
+    let(:jid) { '12345678' }
     let!(:project) { create(:project, :mirror, :import_scheduled) }
 
     before do
       allow_any_instance_of(Gitlab::ExclusiveLease).to receive(:try_obtain).and_return(true)
+      allow(subject).to receive(:jid).and_return(jid)
     end
 
     it 'sets status as finished when update mirror service executes successfully' do
@@ -36,16 +38,22 @@ describe RepositoryUpdateMirrorWorker do
       expect(project.reload.import_status).to eq('failed')
     end
 
+    context 'when worker was reset without cleanup' do
+      let(:started_project) { create(:project, :mirror, :import_started, import_jid: jid) }
+
+      it 'sets status as finished when update mirror service executes successfully' do
+        expect_any_instance_of(Projects::UpdateMirrorService).to receive(:execute).and_return(status: :success)
+
+        expect { subject.perform(started_project.id) }.to change { started_project.reload.import_status }.to('finished')
+      end
+    end
+
     context 'reschedule mirrors' do
       before do
         allow_any_instance_of(Projects::UpdateMirrorService).to receive(:execute).and_return(status: :success)
       end
 
       context 'when we obtain the lease' do
-        before do
-          allow_any_instance_of(Gitlab::ExclusiveLease).to receive(:try_obtain).and_return(true)
-        end
-
         it 'performs UpdateAllMirrorsWorker when reschedule_immediately? returns true' do
           allow(Gitlab::Mirror).to receive(:reschedule_immediately?).and_return(true)