From 25dd72da4ed0dcdf35dddb4627e8417302d674f8 Mon Sep 17 00:00:00 2001 From: Sony Mathew <2040199+sony-mathew@users.noreply.github.com> Date: Thu, 23 Jul 2026 00:50:29 +0530 Subject: [PATCH 1/2] fix(imports): retry deferred stat reconciliation --- .../data_imports/intercom/importer.rb | 20 +++++++++-------- .../data_imports/intercom/importer_spec.rb | 22 +++++++++++++++++++ 2 files changed, 33 insertions(+), 9 deletions(-) diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb index db769e331..99ad7552e 100644 --- a/app/services/data_imports/intercom/importer.rb +++ b/app/services/data_imports/intercom/importer.rb @@ -962,17 +962,19 @@ class DataImports::Intercom::Importer def reconcile_dirty_stats return if @dirty_stat_groups.empty? - @dirty_stat_groups.each_key do |group| - case group - when 'contacts' - reconcile_item_stats('contact') - when 'messages' - reconcile_message_stats - else - raise ArgumentError, "Unsupported Intercom import stat group: #{group}" + with_query_timeout_retry do + @dirty_stat_groups.each_key do |group| + case group + when 'contacts' + reconcile_item_stats('contact') + when 'messages' + reconcile_message_stats + else + raise ArgumentError, "Unsupported Intercom import stat group: #{group}" + end end + persist_stats end - persist_stats @dirty_stat_groups.clear end diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb index e886c75af..aec4880f7 100644 --- a/spec/services/data_imports/intercom/importer_spec.rb +++ b/spec/services/data_imports/intercom/importer_spec.rb @@ -212,6 +212,28 @@ RSpec.describe DataImports::Intercom::Importer do expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) end + it 'retries a query timeout during deferred message stats reconciliation', :aggregate_failures do + described_class.new(data_import: data_import).import_conversations_page + stats = data_import.reload.stats.deep_dup + stats['messages']['imported'] = 0 + data_import.update!(stats: stats) + importer = described_class.new(data_import: data_import) + reconciliation_attempts = 0 + allow(importer).to receive(:sleep) + allow(importer).to receive(:reconcile_message_stats).and_wrap_original do |method| + reconciliation_attempts += 1 + raise ActiveRecord::QueryCanceled, 'statement timeout' if reconciliation_attempts == 1 + + method.call + end + + importer.import_conversations_page + + expect(reconciliation_attempts).to eq(2) + expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) + end + it 'indexes imported messages for advanced search' do allow(ChatwootApp).to receive(:advanced_search_allowed?).and_return(true) allow(ChatwootApp).to receive(:chatwoot_cloud?).and_return(false) From b5d4a5b6af023f3325881c2896fd5f41c83b83e8 Mon Sep 17 00:00:00 2001 From: Sony Mathew <2040199+sony-mathew@users.noreply.github.com> Date: Thu, 23 Jul 2026 00:51:23 +0530 Subject: [PATCH 2/2] fix(imports): preserve skipped stats on retry --- .../data_imports/intercom/importer.rb | 28 ++++++++++--------- .../data_imports/intercom/importer_spec.rb | 24 ++++++++++++++++ 2 files changed, 39 insertions(+), 13 deletions(-) diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb index 99ad7552e..738162ddf 100644 --- a/app/services/data_imports/intercom/importer.rb +++ b/app/services/data_imports/intercom/importer.rb @@ -672,19 +672,21 @@ class DataImports::Intercom::Importer end def skip_already_imported_item(item, mapping, already_handled:) - item.update!( - status: :skipped, - chatwoot_record_type: mapping.chatwoot_record_type, - chatwoot_record_id: mapping.chatwoot_record_id, - last_error_code: ALREADY_IMPORTED_ERROR_CODE, - last_error_message: 'Already imported in a previous import.' - ) - record_already_imported_log( - data_import_item: item, - source_object_type: item.source_object_type, - source_object_id: item.source_object_id, - mapping: mapping - ) + DataImportItem.transaction do + item.update!( + status: :skipped, + chatwoot_record_type: mapping.chatwoot_record_type, + chatwoot_record_id: mapping.chatwoot_record_id, + last_error_code: ALREADY_IMPORTED_ERROR_CODE, + last_error_message: 'Already imported in a previous import.' + ) + record_already_imported_log( + data_import_item: item, + source_object_type: item.source_object_type, + source_object_id: item.source_object_id, + mapping: mapping + ) + end increment_stat(stat_group_for(item.source_object_type), 'skipped') unless already_handled end diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb index aec4880f7..bfb742a41 100644 --- a/spec/services/data_imports/intercom/importer_spec.rb +++ b/spec/services/data_imports/intercom/importer_spec.rb @@ -576,6 +576,30 @@ RSpec.describe DataImports::Intercom::Importer do expect(next_data_import.import_errors.skip_logs.pluck(:details).map { |details| details['reason'] }.uniq).to eq(['already_imported']) end + it 'keeps skipped contact stats when a query timeout is retried after the item update', :aggregate_failures do + described_class.new(data_import: data_import).perform + importer = described_class.new(data_import: next_data_import) + contact_log_attempts = 0 + allow(importer).to receive(:sleep) + allow(importer).to receive(:record_already_imported_log).and_wrap_original do |method, **attributes| + if attributes[:source_object_type] == 'contact' + contact_log_attempts += 1 + raise ActiveRecord::QueryCanceled, 'statement timeout' if contact_log_attempts == 1 + end + + method.call(**attributes) + end + + importer.import_contacts_page + + contact_item = next_data_import.items.find_by!(source_object_type: 'contact', source_object_id: 'contact_1') + expect(contact_log_attempts).to eq(2) + expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once + expect(contact_item).to be_skipped + expect(next_data_import.reload.stats.dig('contacts', 'skipped')).to eq(1) + expect(next_data_import.import_errors.skip_logs.exists?(data_import_item: contact_item)).to be(true) + end + it 'recreates messages when existing message mappings point to deleted records', :aggregate_failures do described_class.new(data_import: data_import).perform conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')