Verify project import status again before marking as failed

app/workers/stuck_import_jobs_worker.rb

@@ -22,25 +22,23 @@ class StuckImportJobsWorker
   end
 
   def mark_projects_with_jid_as_failed!
-    completed_jids_count = 0
+    jids_and_ids = enqueued_projects_with_jid.pluck(:import_jid, :id).to_h
 
-    enqueued_projects_with_jid.find_in_batches(batch_size: 500) do |group|
-      jids = group.map(&:import_jid)
+    # Find the jobs that aren't currently running or that exceeded the threshold.
+    completed_jids = Gitlab::SidekiqStatus.completed_jids(jids_and_ids.keys)
+    return unless completed_jids.any?
 
-      # Find the jobs that aren't currently running or that exceeded the threshold.
-      completed_jids = Gitlab::SidekiqStatus.completed_jids(jids).to_set
+    completed_project_ids = jids_and_ids.values_at(*completed_jids)
 
-      if completed_jids.any?
-        completed_jids_count += completed_jids.count
-        group.each do |project|
-          project.mark_import_as_failed(error_message) if completed_jids.include?(project.import_jid)
-        end
+    # We select the projects again, because they may have transitioned from
+    # scheduled/started to finished/failed while we were looking up their Sidekiq status.
+    completed_projects = enqueued_projects_with_jid.where(id: completed_project_ids)
 
-        Rails.logger.info("Marked stuck import jobs as failed. JIDs: #{completed_jids.to_a.join(', ')}")
-      end
-    end
+    Rails.logger.info("Marked stuck import jobs as failed. JIDs: #{completed_projects.map(&:import_jid).join(', ')}")
 
-    completed_jids_count
+    completed_projects.each do |project|
+      project.mark_import_as_failed(error_message)
+    end.count
   end
 
   def enqueued_projects

changelogs/unreleased/dm-stuck-import-jobs-verify.yml

+---
+title: Verify project import status again before marking as failed
+merge_request:
+author:
+type: fixed

spec/workers/stuck_import_jobs_worker_spec.rb

@@ -2,32 +2,46 @@ require 'spec_helper'
 
 describe StuckImportJobsWorker do
   let(:worker) { described_class.new }
-  let(:exclusive_lease_uuid) { SecureRandom.uuid }
-
-  before do
-    allow_any_instance_of(Gitlab::ExclusiveLease).to receive(:try_obtain).and_return(exclusive_lease_uuid)
-  end
 
   shared_examples 'project import job detection' do
-    describe 'long running import' do
-      it 'marks the project as failed' do
-        allow(Gitlab::SidekiqStatus).to receive(:completed_jids).and_return(['123'])
+    context 'when the job has completed' do
+      context 'when the import status was already updated' do
+        before do
+          allow(Gitlab::SidekiqStatus).to receive(:completed_jids) do
+            project.import_start
+            project.import_finish
+
+            [project.import_jid]
+          end
+        end
 
-        expect { worker.perform }.to change { project.reload.import_status }.to('failed')
+        it 'does not mark the project as failed' do
+          worker.perform
+
+          expect(project.reload.import_status).to eq('finished')
+        end
+      end
+
+      context 'when the import status was not updated' do
+        before do
+          allow(Gitlab::SidekiqStatus).to receive(:completed_jids).and_return([project.import_jid])
+        end
+
+        it 'marks the project as failed' do
+          worker.perform
+
+          expect(project.reload.import_status).to eq('failed')
+        end
       end
     end
 
-    describe 'running import' do
-      it 'does not mark the project as failed' do
+    context 'when the job is still in Sidekiq' do
+      before do
         allow(Gitlab::SidekiqStatus).to receive(:completed_jids).and_return([])
-
-        expect { worker.perform }.not_to change { project.reload.import_status }
       end
 
-      describe 'import without import_jid' do
-        it 'marks the project as failed' do
-          expect { worker.perform }.to change { project.reload.import_status }.to('failed')
-        end
+      it 'does not mark the project as failed' do
+        expect { worker.perform }.not_to change { project.reload.import_status }
       end
     end
   end