diff --git a/.env.example b/.env.example index bc7380a29..8a4f0bb5d 100644 --- a/.env.example +++ b/.env.example @@ -98,6 +98,8 @@ SMTP_OPENSSL_VERIFY_MODE=peer # Mail Incoming # This is the domain set for the reply emails when conversation continuity is enabled MAILER_INBOUND_EMAIL_DOMAIN= +# Maximum time in seconds to process a single IMAP email +# EMAIL_PROCESSING_TIMEOUT_SECONDS=60 # Set this to the appropriate ingress channel with regards to incoming emails # Possible values are : # relay for Exim, Postfix, Qmail diff --git a/app/jobs/inboxes/fetch_imap_emails_job.rb b/app/jobs/inboxes/fetch_imap_emails_job.rb index e98edf409..e2c48488b 100644 --- a/app/jobs/inboxes/fetch_imap_emails_job.rb +++ b/app/jobs/inboxes/fetch_imap_emails_job.rb @@ -36,7 +36,8 @@ class Inboxes::FetchImapEmailsJob < MutexApplicationJob else Imap::FetchEmailService.new(channel: channel, interval: interval).perform end - inbound_emails.map do |inbound_mail| + + inbound_emails.each do |inbound_mail| process_mail(inbound_mail, channel) end rescue OAuth2::Error => e @@ -44,11 +45,38 @@ class Inboxes::FetchImapEmailsJob < MutexApplicationJob channel.authorization_error! end + def should_skip_email?(message_id) + failure_count = Rails.cache.read("email_failures:#{message_id}") || 0 + failure_count >= 3 + end + + def mark_email_as_failed(message_id) + failure_count = Rails.cache.read("email_failures:#{message_id}") || 0 + Rails.cache.write("email_failures:#{message_id}", failure_count + 1, expires_in: 6.hours) + end + def process_mail(inbound_mail, channel) - Imap::ImapMailbox.new.process(inbound_mail, channel) - rescue StandardError => e - ChatwootExceptionTracker.new(e, account: channel.account).capture_exception - Rails.logger.error(" - #{channel.provider} Email dropped: #{inbound_mail.from} and message_source_id: #{inbound_mail.message_id}") + # Skip if this email has failed multiple times recently + if should_skip_email?(inbound_mail.message_id) + Rails.logger.warn "[IMAP] Skipping problematic email: #{inbound_mail.message_id}" + return + end + + begin + Timeout.timeout(email_processing_timeout) do + Imap::ImapMailbox.new.process(inbound_mail, channel) + end + rescue Timeout::Error + mark_email_as_failed(inbound_mail.message_id) + Rails.logger.error "[IMAP] Email processing timeout (#{email_processing_timeout}s): #{inbound_mail.message_id}" + rescue StandardError => e + mark_email_as_failed(inbound_mail.message_id) + Rails.logger.error "[IMAP] Failed to process email #{inbound_mail.message_id}: #{e.message}" + ChatwootExceptionTracker.new(e, account: channel.account).capture_exception + end + end + + def email_processing_timeout + GlobalConfigService.load('EMAIL_PROCESSING_TIMEOUT_SECONDS', 60).to_i end end diff --git a/spec/jobs/inboxes/fetch_imap_emails_job_spec.rb b/spec/jobs/inboxes/fetch_imap_emails_job_spec.rb index da4b95b15..04336c619 100644 --- a/spec/jobs/inboxes/fetch_imap_emails_job_spec.rb +++ b/spec/jobs/inboxes/fetch_imap_emails_job_spec.rb @@ -88,7 +88,10 @@ RSpec.describe Inboxes::FetchImapEmailsJob do end context 'when the fetch service returns the email objects' do - let(:inbound_mail) { create_inbound_email_from_fixture('welcome.eml').mail } + let(:inbound_mail) { instance_double(Mail::Message, message_id: 'message-id') } + let(:failure_cache_key) { "email_failures:#{inbound_mail.message_id}" } + let(:second_inbound_mail) { instance_double(Mail::Message, message_id: 'second-message-id') } + let(:second_failure_cache_key) { "email_failures:#{second_inbound_mail.message_id}" } let(:mailbox) { double } let(:exception_tracker) { double } let(:fetch_service) { double } @@ -101,6 +104,11 @@ RSpec.describe Inboxes::FetchImapEmailsJob do allow(fetch_service).to receive(:perform).and_return([inbound_mail]) end + after do + Rails.cache.delete(failure_cache_key) + Rails.cache.delete(second_failure_cache_key) + end + it 'calls the mailbox to create emails' do allow(mailbox).to receive(:process) @@ -111,6 +119,36 @@ RSpec.describe Inboxes::FetchImapEmailsJob do described_class.perform_now(imap_email_channel) end + it 'marks the email as failed when processing times out' do + allow(Timeout).to receive(:timeout).and_raise(Timeout::Error) + allow(Rails.cache).to receive(:read).and_call_original + allow(Rails.cache).to receive(:read).with(failure_cache_key).and_return(nil) + + expect(Rails.cache).to receive(:write).with(failure_cache_key, 1, expires_in: 6.hours) + + described_class.perform_now(imap_email_channel) + end + + it 'continues processing remaining emails when one email fails' do + allow(fetch_service).to receive(:perform).and_return([inbound_mail, second_inbound_mail]) + allow(mailbox).to receive(:process).with(inbound_mail, imap_email_channel).and_raise(StandardError) + allow(mailbox).to receive(:process).with(second_inbound_mail, imap_email_channel) + allow(exception_tracker).to receive(:capture_exception) + + described_class.perform_now(imap_email_channel) + + expect(mailbox).to have_received(:process).with(second_inbound_mail, imap_email_channel) + end + + it 'skips emails that have failed multiple times recently' do + allow(Rails.cache).to receive(:read).and_call_original + allow(Rails.cache).to receive(:read).with(failure_cache_key).and_return(3) + + expect(mailbox).not_to receive(:process) + + described_class.perform_now(imap_email_channel) + end + it 'logs errors if mailbox returns errors' do allow(mailbox).to receive(:process).and_raise(StandardError)