Merge branch 'codex/cw-7519-intercom-shorter-jobs-heartbeats' into codex/cw-7615-intercom-query-timeout-retries

app/services/data_imports/intercom/importer.rb

@@ -62,10 +62,12 @@ class DataImports::Intercom::Importer
   end
 
   def fail!(error)
-    return if @data_import.reload.abandoned?
+    @data_import.with_lock do
+      next if @data_import.abandoned? || stale_import_run?
 
-    record_run_error(error)
-    @data_import.update!(status: :failed, last_error_at: Time.current)
+      record_run_error(error)
+      @data_import.update!(status: :failed, last_error_at: Time.current)
+    end
   end
 
   def import_contacts_page(starting_after: cursor_for('contacts'))

spec/services/data_imports/intercom/importer_spec.rb

@@ -501,6 +501,26 @@ RSpec.describe DataImports::Intercom::Importer do
       expect(data_import.last_error_at).to be_nil
       expect(data_import.import_errors.exists?).to be(false)
     end
+
+    it 'does not fail an import after a newer run takes over', :aggregate_failures do
+      run_id = 'intercom-run-1'
+      data_import.update!(
+        status: :processing,
+        source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id }
+      )
+      importer = described_class.new(data_import: data_import, run_id: run_id)
+
+      DataImport.find(data_import.id).update!(
+        status: :pending,
+        source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'intercom-run-2' }
+      )
+
+      importer.fail!(StandardError.new('boom'))
+
+      expect(data_import.reload).to be_pending
+      expect(data_import.last_error_at).to be_nil
+      expect(data_import.import_errors.exists?).to be(false)
+    end
   end
 
   context 'when the Intercom records were imported by an earlier run' do