From 580cbf91c5ea059224a79e275865b706e3bbef3a Mon Sep 17 00:00:00 2001 From: Sony Mathew Date: Thu, 23 Jul 2026 22:52:06 +0530 Subject: [PATCH] fix: preserve Intercom bulk failure isolation (#15116) This preserves per-message failure isolation when an Intercom bulk message chunk cannot be committed. Query timeouts are retried once after the failed transaction rolls back, while other database failures move directly to an entry-by-entry fallback. The fallback reacquires the import lock and refreshes each entry before writing, so a competing worker cannot create duplicate messages. Each individual write uses a savepoint, and search indexing happens only after the import lock transaction commits. This is Phase 2, Task 6 of CW-7615 and is stacked on #15111. The review delta is one commit across the importer and its focused spec. ## Closes - [CW-7615](https://linear.app/chatwoot/issue/CW-7615/optimize-intercom-import-reliability-and-bulk-message-ingestion) ## How to test 1. Run an Intercom import where a bulk mapping write times out once and confirm the chunk succeeds on retry. 2. Force a persistent timeout or database write failure and confirm valid messages are imported through the fallback. 3. Force one fallback record to fail and confirm only that message is recorded as an import error. 4. Simulate another worker repairing a message before fallback and confirm no duplicate is created. --- .../data_imports/intercom/importer.rb | 129 ++++++++++--- .../data_imports/intercom/importer_spec.rb | 170 +++++++++++++++++- 2 files changed, 267 insertions(+), 32 deletions(-) diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb index 88ca3df17..385df025b 100644 --- a/app/services/data_imports/intercom/importer.rb +++ b/app/services/data_imports/intercom/importer.rb @@ -1,5 +1,7 @@ # rubocop:disable Metrics/ClassLength, Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity, Metrics/MethodLength, Rails/SkipsModelValidations class DataImports::Intercom::Importer + class InvalidMessagePayloadError < StandardError; end + PageResult = Struct.new(:next_cursor, keyword_init: true) do def done? next_cursor.blank? @@ -27,6 +29,7 @@ class DataImports::Intercom::Importer :current_entries, :previous_entries, :messages, + :failed_entries, keyword_init: true ) @@ -450,19 +453,43 @@ class DataImports::Intercom::Importer result.skipped_entries.each do |entry| reconcile_bulk_message_entry(conversation, entry) { record_bulk_skipped_message(conversation, entry) } end + Array(result.failed_entries).each { |entry, error| fail_message(conversation, entry.source_id, entry.part, error) } increment_stat('messages', 'imported', result.imported_entries.size) result.messages.each { |message| reindex_message_for_search(message) } end def bulk_message_batch_result(conversation, contact, batch_builder, entries) - bulk_write_message_entries(conversation, contact, batch_builder, entries) - rescue StandardError + with_query_timeout_retry do + bulk_write_message_entries(conversation, contact, batch_builder, entries) + end + rescue ActiveRecord::ActiveRecordError + fallback_message_entries(conversation, contact, batch_builder, entries) unless @import_stopped + nil + end + + def fallback_message_entries(conversation, contact, batch_builder, entries) entries.each do |entry| break unless continue_import_with_heartbeat? - import_message(conversation, contact, entry) + message = fallback_message_entry(conversation, contact, batch_builder, entry) + reindex_message_for_search(message) if message.is_a?(Message) end - nil + end + + def fallback_message_entry(conversation, contact, batch_builder, entry) + with_query_timeout_retry do + @data_import.with_lock do + if inactive_import_run? + @import_stopped = true + next + end + + refreshed_entry = batch_builder.refresh([entry]).entries.first + import_message(conversation, contact, refreshed_entry, reindex: false) + end + end + rescue ActiveRecord::ActiveRecordError => e + fail_message(conversation, entry.source_id, entry.part, e) end def bulk_write_message_entries(conversation, contact, batch_builder, entries) @@ -487,9 +514,23 @@ class DataImports::Intercom::Importer writable_entries = entries.select do |entry| %i[repairable_stale_mapping existing_message new_message].include?(entry.classification) end - content_by_source_id = writable_entries.to_h { |entry| [entry.source_id, content_for(entry.part)] } + content_by_source_id = {} + attributes_by_source_id = {} + failed_entries = [] + writable_entries.select! do |entry| + validate_message_payload!(entry.part) + content = content_for(entry.part) + content_by_source_id[entry.source_id] = content + if content.present? && entry.message.blank? + attributes_by_source_id[entry.source_id] = message_attributes(conversation, contact, entry.part, entry.source_id, content) + end + true + rescue InvalidMessagePayloadError => e + failed_entries << [entry, e] + false + end skipped_entries, imported_entries = writable_entries.partition { |entry| content_by_source_id[entry.source_id].blank? } - messages = insert_messages(conversation, contact, imported_entries, content_by_source_id) + messages = insert_messages(imported_entries, attributes_by_source_id) upsert_message_mappings(conversation, imported_entries, skipped_entries, messages) MessageBatchResult.new( @@ -497,16 +538,15 @@ class DataImports::Intercom::Importer skipped_entries: skipped_entries, current_entries: grouped_entries.fetch(:current_import, []), previous_entries: grouped_entries.fetch(:previous_import, []), - messages: messages + messages: messages, + failed_entries: failed_entries ) end - def insert_messages(conversation, contact, entries, content_by_source_id) + def insert_messages(entries, attributes_by_source_id) new_entries = entries.reject(&:message) if new_entries.present? - attributes = new_entries.map do |entry| - message_attributes(conversation, contact, entry.part, entry.source_id, content_by_source_id.fetch(entry.source_id)) - end + attributes = new_entries.map { |entry| attributes_by_source_id.fetch(entry.source_id) } result = Message.insert_all!(attributes, returning: %w[id source_id]) inserted_messages = Message.where(id: result.pluck('id')).index_by(&:source_id) end @@ -516,6 +556,36 @@ class DataImports::Intercom::Importer end end + def validate_message_payload!(part) + raise InvalidMessagePayloadError, 'Intercom message payload must be an object' unless part.is_a?(Hash) + + %w[author assigned_to event_details].each do |field| + value = part[field] + next if value.nil? || value.is_a?(Hash) + + raise InvalidMessagePayloadError, "Intercom message #{field} must be an object" + end + + participant = part.dig('event_details', 'participant') + unless participant.nil? || participant.is_a?(Hash) + raise InvalidMessagePayloadError, 'Intercom message event_details.participant must be an object' + end + + %w[created_at updated_at].each do |field| + value = part[field] + valid_timestamp = value.nil? || value.is_a?(Integer) || value.is_a?(Float) || + (value.is_a?(String) && (value.blank? || value.match?(/\A-?\d+(?:\.\d+)?\z/))) + raise InvalidMessagePayloadError, "Intercom message #{field} must be a Unix timestamp" unless valid_timestamp + end + + %w[body subject].each do |field| + value = part[field] + next unless value.is_a?(String) && !value.valid_encoding? + + raise InvalidMessagePayloadError, "Intercom message #{field} must use valid encoding" + end + end + def upsert_message_mappings(conversation, imported_entries, skipped_entries, messages) now = Time.current mapping_attributes = imported_entries.zip(messages).map do |entry, message| @@ -584,19 +654,23 @@ class DataImports::Intercom::Importer fail_message(conversation, source_entry[:source_id], source_entry[:part], e) end - def import_message(conversation, contact, entry) - with_query_timeout_retry do - case entry.classification - when :current_import - reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part) - when :previous_import - skip_existing_message_mapping(conversation, entry.mapping, entry.part) - when :repairable_stale_mapping, :existing_message, :new_message - create_message(conversation, contact, entry) - else - raise ArgumentError, "Unsupported Intercom message classification: #{entry.classification}" + def import_message(conversation, contact, entry, reindex: true) + message = with_query_timeout_retry do + Message.transaction(requires_new: true) do + case entry.classification + when :current_import + reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part) + when :previous_import + skip_existing_message_mapping(conversation, entry.mapping, entry.part) + when :repairable_stale_mapping, :existing_message, :new_message + create_message(conversation, contact, entry) + else + raise ArgumentError, "Unsupported Intercom message classification: #{entry.classification}" + end end end + reindex_message_for_search(message) if reindex && message.is_a?(Message) + message rescue StandardError => e fail_message(conversation, entry.source_id, entry.part, e) end @@ -607,15 +681,12 @@ class DataImports::Intercom::Importer attrs = message_attributes(conversation, contact, entry.part, entry.source_id, content) message = entry.message - Message.transaction do - unless message - result = Message.insert_all!([attrs], returning: %w[id]) - message = Message.find(result.rows.first.first) - end - record_message_mapping(entry, message) + unless message + result = Message.insert_all!([attrs], returning: %w[id]) + message = Message.find(result.rows.first.first) end + record_message_mapping(entry, message) increment_stat('messages', 'imported') - reindex_message_for_search(message) message end diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb index 205215812..212d90887 100644 --- a/spec/services/data_imports/intercom/importer_spec.rb +++ b/spec/services/data_imports/intercom/importer_spec.rb @@ -196,6 +196,169 @@ RSpec.describe DataImports::Intercom::Importer do expect(data_import.reload.stats.dig('messages', 'imported')).to eq(206) end + context 'when a bulk message chunk fails' do + it 'retries a query timeout once and keeps the successful bulk result', :aggregate_failures do + importer = described_class.new(data_import: data_import) + mapping_attempts = 0 + allow(importer).to receive(:sleep) + allow(DataImportMapping).to receive(:upsert_all).and_wrap_original do |method, records, **kwargs| + mapping_attempts += 1 + raise ActiveRecord::QueryCanceled, 'statement timeout' if mapping_attempts == 1 + + method.call(records, **kwargs) + end + expect(importer).not_to receive(:fallback_message_entries) + + importer.perform + + expect(mapping_attempts).to eq(2) + expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once + expect(account.messages.count).to eq(3) + expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3) + expect(data_import.import_errors).to be_empty + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) + end + + it 'falls back individually after the query timeout retry is exhausted', :aggregate_failures do + importer = described_class.new(data_import: data_import) + mapping_attempts = 0 + reindex_transaction_depths = [] + transaction_depth_before_import = Message.connection.open_transactions + allow(importer).to receive(:sleep) + allow(importer).to receive(:fallback_message_entries).and_call_original + allow(importer).to receive(:create_message).and_call_original + allow(importer).to receive(:reindex_message_for_search).and_wrap_original do |method, message| + reindex_transaction_depths << Message.connection.open_transactions + method.call(message) + end + allow(DataImportMapping).to receive(:upsert_all) do + mapping_attempts += 1 + raise ActiveRecord::QueryCanceled, 'statement timeout' + end + + importer.perform + + expect(mapping_attempts).to eq(2) + expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once + expect(importer).to have_received(:fallback_message_entries).once + expect(importer).to have_received(:create_message).exactly(3).times + expect(account.messages.count).to eq(3) + expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3) + expect(data_import.import_errors).to be_empty + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) + expect(reindex_transaction_depths).to all(eq(transaction_depth_before_import)) + end + + it 'falls back immediately for a non-timeout database error', :aggregate_failures do + importer = described_class.new(data_import: data_import) + mapping_attempts = 0 + allow(importer).to receive(:sleep) + allow(importer).to receive(:fallback_message_entries).and_call_original + allow(DataImportMapping).to receive(:upsert_all) do + mapping_attempts += 1 + raise ActiveRecord::StatementInvalid, 'bulk mapping failed' + end + + importer.perform + + expect(mapping_attempts).to eq(1) + expect(importer).not_to have_received(:sleep) + expect(importer).to have_received(:fallback_message_entries).once + expect(account.messages.count).to eq(3) + expect(data_import.import_errors).to be_empty + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) + end + + it 'refreshes fallback entries when another worker repairs a message', :aggregate_failures do + importer = described_class.new(data_import: data_import) + allow(importer).to receive(:bulk_write_message_entries).and_raise(ActiveRecord::StatementInvalid, 'bulk failed') + allow(importer).to receive(:fallback_message_entries).and_wrap_original do |method, conversation, contact, batch_builder, entries| + source_entry = entries.first + message = create( + :message, + account: account, + inbox: conversation.inbox, + conversation: conversation, + source_id: "intercom:#{source_entry.source_id}", + created_at: Time.zone.at(source_entry.part['created_at']), + updated_at: Time.zone.at(source_entry.part['created_at']) + ) + DataImportMapping.create!( + account: account, + data_import: data_import, + source_provider: 'intercom', + source_object_type: 'message', + source_object_id: source_entry.source_id, + chatwoot_record_type: 'Message', + chatwoot_record_id: message.id, + metadata: {} + ) + method.call(conversation, contact, batch_builder, entries) + end + + importer.perform + + source_id = 'intercom:conversation:conversation_1:source:source_1' + expect(account.messages.where(source_id: source_id).count).to eq(1) + expect(account.messages.count).to eq(3) + expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3) + expect(data_import.import_errors).to be_empty + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) + end + + it 'retries a fallback refresh timeout after the lock transaction rolls back', :aggregate_failures do + importer = described_class.new(data_import: data_import) + refresh_attempts = 0 + allow(importer).to receive(:sleep) + allow(importer).to receive(:bulk_write_message_entries).and_raise(ActiveRecord::StatementInvalid, 'bulk failed') + allow(DataImports::Intercom::MessageBatchBuilder).to receive(:new).and_wrap_original do |method, **kwargs| + method.call(**kwargs).tap do |batch_builder| + allow(batch_builder).to receive(:refresh).and_wrap_original do |refresh, entries| + refresh_attempts += 1 + raise ActiveRecord::QueryCanceled, 'statement timeout' if refresh_attempts == 1 + + refresh.call(entries) + end + end + end + + importer.perform + + expect(refresh_attempts).to eq(4) + expect(importer).to have_received(:sleep).with(be_between(0.2, 0.5)).once + expect(account.messages.count).to eq(3) + expect(data_import.mappings.where(source_object_type: 'message').count).to eq(3) + expect(data_import.import_errors).to be_empty + expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3) + end + end + + it 'isolates a malformed message payload while importing valid messages', :aggregate_failures do + malformed_conversation = conversation_payload.deep_dup + malformed_conversation['source']['subject'] = nil + malformed_conversation['source']['body'] = nil + malformed_conversation.dig('conversation_parts', 'conversation_parts').first['created_at'] = { 'unexpected' => true } + allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(malformed_conversation) + importer = described_class.new(data_import: data_import) + expect(importer).not_to receive(:fallback_message_entries) + + importer.perform + + expect(account.messages.pluck(:source_id)).to eq(['intercom:conversation:conversation_1:part:part_2']) + error = data_import.import_errors.find_by!( + source_object_type: 'message', + source_object_id: 'conversation:conversation_1:part:part_1' + ) + expect(error).to have_attributes( + error_code: described_class::InvalidMessagePayloadError.name, + message: 'Intercom message created_at must be a Unix timestamp' + ) + expect(data_import.import_errors.where(source_object_type: 'message').count).to eq(1) + expect(data_import.reload).to be_completed_with_errors + expect(data_import.stats.dig('messages', 'imported')).to eq(1) + expect(data_import.stats.dig('errors', 'count')).to eq(1) + end + it 'imports historical records without dispatching record events or outbound side effects', :aggregate_failures do dispatched_events = [] allow(Rails.configuration.dispatcher).to receive(:dispatch) do |event_name, *_args| @@ -493,7 +656,7 @@ RSpec.describe DataImports::Intercom::Importer do data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id }) importer = described_class.new(data_import: data_import, run_id: run_id) allow(importer).to receive(:bulk_write_message_entries).and_raise(ActiveRecord::StatementInvalid, 'bulk failed') - allow(importer).to receive(:import_message).and_wrap_original do |method, *args| + allow(importer).to receive(:fallback_message_entry).and_wrap_original do |method, *args| method.call(*args).tap do DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' }) end @@ -501,7 +664,7 @@ RSpec.describe DataImports::Intercom::Importer do importer.import_conversations_page - expect(importer).to have_received(:import_message).once + expect(importer).to have_received(:fallback_message_entry).once expect(account.messages.count).to eq(1) expect(data_import.reload.stats.dig('conversations', 'imported')).to eq(0) end @@ -567,7 +730,7 @@ RSpec.describe DataImports::Intercom::Importer do it 'rolls back a newly inserted message when mapping persistence fails', :aggregate_failures do importer = described_class.new(data_import: data_import) - allow(DataImportMapping).to receive(:upsert_all).and_raise(StandardError, 'bulk mapping failed') + allow(DataImportMapping).to receive(:upsert_all).and_raise(ActiveRecord::StatementInvalid, 'bulk mapping failed') allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message| raise StandardError, 'mapping failed' if entry.source_id == 'conversation:conversation_1:source:source_1' @@ -1430,6 +1593,7 @@ RSpec.describe DataImports::Intercom::Importer do expect(data_import.reload).to be_completed_with_errors expect(data_import.stats.dig('errors', 'count')).to eq(1) expect(insert_attempts.size).to eq(2) + expect(data_import.import_errors.where(source_object_type: 'message').count).to eq(1) expect(account.messages.find_by(source_id: 'intercom:conversation:conversation_1:source:source_1')).to be_present end end