From f54d113571772f9b5f050d31a150e1f74e020ab5 Mon Sep 17 00:00:00 2001 From: Gabriel Jablonski Date: Thu, 15 Jan 2026 22:16:32 -0300 Subject: [PATCH] fix: whatsapp race condition (#185) * feat: include external created at in whatsapp cloud message * fix: whatsapp providers race condition * refactor: remove redundant comments and streamline message processing lock acquisition * fix: scope message source key by inbox.id to prevent race condition * fix: scope message source key by inbox.id to prevent race condition --- .../whatsapp/baileys_handlers/helpers.rb | 4 +- .../baileys_handlers/messages_upsert.rb | 5 +- .../whatsapp/incoming_message_base_service.rb | 8 +- .../incoming_message_service_helpers.rb | 14 +- .../whatsapp/zapi_handlers/helpers.rb | 4 +- .../zapi_handlers/received_callback.rb | 11 +- .../incoming_message_baileys_service_spec.rb | 13 +- .../whatsapp/incoming_message_service_spec.rb | 126 +++++++++++++++++- .../zapi_handlers/received_callback_spec.rb | 29 +++- 9 files changed, 183 insertions(+), 31 deletions(-) diff --git a/app/services/whatsapp/baileys_handlers/helpers.rb b/app/services/whatsapp/baileys_handlers/helpers.rb index 64faf806c..93f38f66d 100644 --- a/app/services/whatsapp/baileys_handlers/helpers.rb +++ b/app/services/whatsapp/baileys_handlers/helpers.rb @@ -197,9 +197,9 @@ module Whatsapp::BaileysHandlers::Helpers # rubocop:disable Metrics/ModuleLength Redis::Alfred.get(key) end - def cache_message_source_id_in_redis + def acquire_message_processing_lock key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{raw_message_id}") - ::Redis::Alfred.setex(key, true) + Redis::Alfred.set(key, true, nx: true, ex: 1.day) end def clear_message_source_id_from_redis diff --git a/app/services/whatsapp/baileys_handlers/messages_upsert.rb b/app/services/whatsapp/baileys_handlers/messages_upsert.rb index dc508ed88..64c2d4dfa 100644 --- a/app/services/whatsapp/baileys_handlers/messages_upsert.rb +++ b/app/services/whatsapp/baileys_handlers/messages_upsert.rb @@ -24,9 +24,10 @@ module Whatsapp::BaileysHandlers::MessagesUpsert # rubocop:disable Metrics/Modul return unless %w[lid user].include?(jid_type) return unless extract_from_jid(type: 'lid') return if ignore_message? - return if find_message_by_source_id(raw_message_id) || message_under_process? + return if find_message_by_source_id(raw_message_id) + + return unless acquire_message_processing_lock - cache_message_source_id_in_redis set_contact unless @contact diff --git a/app/services/whatsapp/incoming_message_base_service.rb b/app/services/whatsapp/incoming_message_base_service.rb index ccfbb3322..4f60485e7 100644 --- a/app/services/whatsapp/incoming_message_base_service.rb +++ b/app/services/whatsapp/incoming_message_base_service.rb @@ -26,9 +26,10 @@ class Whatsapp::IncomingMessageBaseService # Multiple webhook event can be received against the same message due to misconfigurations in the Meta # business manager account. While we have not found the core reason yet, the following line ensure that # there are no duplicate messages created. - return if find_message_by_source_id(@processed_params[:messages].first[:id]) || message_under_process? + return if find_message_by_source_id(@processed_params[:messages].first[:id]) + + return unless acquire_message_processing_lock - cache_message_source_id_in_redis set_contact return unless @contact @@ -154,7 +155,8 @@ class Whatsapp::IncomingMessageBaseService message_type: :incoming, sender: @contact, source_id: message[:id].to_s, - in_reply_to_external_id: @in_reply_to_external_id + in_reply_to_external_id: @in_reply_to_external_id, + external_created_at: message[:timestamp].to_i ) end diff --git a/app/services/whatsapp/incoming_message_service_helpers.rb b/app/services/whatsapp/incoming_message_service_helpers.rb index 46ad255aa..34d205bd5 100644 --- a/app/services/whatsapp/incoming_message_service_helpers.rb +++ b/app/services/whatsapp/incoming_message_service_helpers.rb @@ -66,23 +66,23 @@ module Whatsapp::IncomingMessageServiceHelpers def find_message_by_source_id(source_id) return unless source_id - @message = Message.find_by(source_id: source_id) + @message = inbox.messages.find_by(source_id: source_id) end def message_under_process? - key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: @processed_params[:messages].first[:id]) + key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{@processed_params[:messages].first[:id]}") Redis::Alfred.get(key) end - def cache_message_source_id_in_redis - return if @processed_params.try(:[], :messages).blank? + def acquire_message_processing_lock + return false if @processed_params.try(:[], :messages).blank? - key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: @processed_params[:messages].first[:id]) - ::Redis::Alfred.setex(key, true) + key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{@processed_params[:messages].first[:id]}") + Redis::Alfred.set(key, true, nx: true, ex: 1.day) end def clear_message_source_id_from_redis - key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: @processed_params[:messages].first[:id]) + key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{@processed_params[:messages].first[:id]}") ::Redis::Alfred.delete(key) end end diff --git a/app/services/whatsapp/zapi_handlers/helpers.rb b/app/services/whatsapp/zapi_handlers/helpers.rb index f26a039d9..ea5ba2439 100644 --- a/app/services/whatsapp/zapi_handlers/helpers.rb +++ b/app/services/whatsapp/zapi_handlers/helpers.rb @@ -11,9 +11,9 @@ module Whatsapp::ZapiHandlers::Helpers !@raw_message[:fromMe] end - def cache_message_source_id_in_redis + def acquire_message_processing_lock key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{inbox.id}_#{raw_message_id}") - Redis::Alfred.setex(key, true) + Redis::Alfred.set(key, true, nx: true, ex: 1.day) end def clear_message_source_id_from_redis diff --git a/app/services/whatsapp/zapi_handlers/received_callback.rb b/app/services/whatsapp/zapi_handlers/received_callback.rb index fb3e4e4ad..4feb1b821 100644 --- a/app/services/whatsapp/zapi_handlers/received_callback.rb +++ b/app/services/whatsapp/zapi_handlers/received_callback.rb @@ -3,16 +3,19 @@ module Whatsapp::ZapiHandlers::ReceivedCallback # rubocop:disable Metrics/Module private - def process_received_callback + def process_received_callback # rubocop:disable Metrics/MethodLength @raw_message = processed_params @message = nil @contact_inbox = nil @contact = nil + @lock_acquired = false return unless should_process_message? - return if find_message_by_source_id(raw_message_id) || message_under_process? + return if find_message_by_source_id(raw_message_id) - cache_message_source_id_in_redis + # Atomically acquire lock to prevent race conditions with concurrent webhook deliveries + @lock_acquired = acquire_message_processing_lock + return unless @lock_acquired return handle_edited_message if @raw_message[:isEdit] @@ -28,7 +31,7 @@ module Whatsapp::ZapiHandlers::ReceivedCallback # rubocop:disable Metrics/Module handle_create_message end ensure - clear_message_source_id_from_redis + clear_message_source_id_from_redis if @lock_acquired end def should_process_message? diff --git a/spec/services/whatsapp/incoming_message_baileys_service_spec.rb b/spec/services/whatsapp/incoming_message_baileys_service_spec.rb index 3f3bb398d..99c5768d8 100644 --- a/spec/services/whatsapp/incoming_message_baileys_service_spec.rb +++ b/spec/services/whatsapp/incoming_message_baileys_service_spec.rb @@ -452,20 +452,25 @@ describe Whatsapp::IncomingMessageBaileysService do end it 'does not create a message if it is already being processed' do - allow(Redis::Alfred).to receive(:get).with(format_message_source_key('msg_123')).and_return(true) + # Simulate lock already acquired by returning false from SETNX + allow(Redis::Alfred).to receive(:set) + .with(format_message_source_key('msg_123'), true, nx: true, ex: 1.day) + .and_return(false) described_class.new(inbox: inbox, params: params).perform expect(inbox.conversations).to be_empty end - it 'caches the message source id in Redis and clears it' do - allow(Redis::Alfred).to receive(:setex).with(format_message_source_key('msg_123'), true) + it 'caches and clears message source id in Redis' do + allow(Redis::Alfred).to receive(:set) + .with(format_message_source_key('msg_123'), true, nx: true, ex: 1.day) + .and_return(true) allow(Redis::Alfred).to receive(:delete).with(format_message_source_key('msg_123')) described_class.new(inbox: inbox, params: params).perform - expect(Redis::Alfred).to have_received(:setex) + expect(Redis::Alfred).to have_received(:set) expect(Redis::Alfred).to have_received(:delete) end end diff --git a/spec/services/whatsapp/incoming_message_service_spec.rb b/spec/services/whatsapp/incoming_message_service_spec.rb index cf58c7581..30fd9a489 100644 --- a/spec/services/whatsapp/incoming_message_service_spec.rb +++ b/spec/services/whatsapp/incoming_message_service_spec.rb @@ -24,6 +24,12 @@ describe Whatsapp::IncomingMessageService do expect(whatsapp_channel.inbox.messages.first.content).to eq('Test') end + it 'stores the external_created_at timestamp from the message' do + described_class.new(inbox: whatsapp_channel.inbox, params: params).perform + message = whatsapp_channel.inbox.messages.first + expect(message.external_created_at).to eq(1_633_034_394) + end + it 'appends to last conversation when if conversation already exists' do contact_inbox = create(:contact_inbox, inbox: whatsapp_channel.inbox, source_id: params[:messages].first[:from]) 2.times.each { create(:conversation, inbox: whatsapp_channel.inbox, contact_inbox: contact_inbox) } @@ -78,6 +84,123 @@ describe Whatsapp::IncomingMessageService do described_class.new(inbox: whatsapp_channel.inbox, params: params).perform expect(whatsapp_channel.inbox.messages.count).to eq(1) end + + it 'will not create duplicate conversations when same message is received for new contact' do + # First call creates contact, conversation and message + described_class.new(inbox: whatsapp_channel.inbox, params: params).perform + expect(whatsapp_channel.inbox.conversations.count).to eq(1) + expect(whatsapp_channel.inbox.messages.count).to eq(1) + + # Second call with same message ID should not create duplicate conversation + described_class.new(inbox: whatsapp_channel.inbox, params: params).perform + expect(whatsapp_channel.inbox.conversations.count).to eq(1) + expect(whatsapp_channel.inbox.messages.count).to eq(1) + end + + it 'will not create duplicate conversations when same message is processed concurrently' do + # Simulate concurrent processing by having Redis key cleared before second call + # but message not yet visible in database due to transaction isolation + service1 = described_class.new(inbox: whatsapp_channel.inbox, params: params) + service2 = described_class.new(inbox: whatsapp_channel.inbox, params: params) + + # Process first message + service1.perform + expect(whatsapp_channel.inbox.conversations.count).to eq(1) + + # Even if we clear the Redis key manually, should still not create duplicates + key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: params[:messages].first[:id]) + Redis::Alfred.delete(key) + + # Second call should check database and find existing message + service2.perform + expect(whatsapp_channel.inbox.conversations.count).to eq(1) + expect(whatsapp_channel.inbox.messages.count).to eq(1) + end + + it 'prevents race condition with atomic lock when two requests arrive simultaneously' do + # This test simulates the exact race condition that was causing duplicate conversations + # Two concurrent webhook deliveries for the same message should result in only one being processed + threads_started = Concurrent::CountDownLatch.new(2) + + thread1 = Thread.new do + threads_started.count_down + threads_started.wait # Wait for both threads to be ready + service = described_class.new(inbox: whatsapp_channel.inbox, params: params) + service.perform + end + + thread2 = Thread.new do + threads_started.count_down + threads_started.wait # Wait for both threads to be ready + service = described_class.new(inbox: whatsapp_channel.inbox, params: params) + service.perform + end + + thread1.join + thread2.join + + # Only one conversation and one message should exist + expect(whatsapp_channel.inbox.conversations.count).to eq(1) + expect(whatsapp_channel.inbox.messages.count).to eq(1) + end + + it 'prevents duplicate when both requests pass the message_under_process? check before cache is set' do + # This test explicitly simulates the race condition timing: + # 1. Request A calls message_under_process? -> returns nil (no lock) + # 2. Request B calls message_under_process? -> returns nil (no lock yet!) + # 3. Request A calls cache_message_source_id_in_redis + # 4. Request B calls cache_message_source_id_in_redis + # 5. Both requests pass find_message_by_source_id (message not yet committed) + # 6. Both create messages -> DUPLICATE! + # + # With atomic lock (SETNX), only one can succeed. + + # Key is scoped by inbox.id to prevent cross-inbox lock collisions + message_source_key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{whatsapp_channel.inbox.id}_#{params[:messages].first[:id]}") + lock_acquired = false + + # Simulate atomic SETNX: only the first call returns true + allow(Redis::Alfred).to receive(:set).with(message_source_key, true, nx: true, ex: 1.day) do + if lock_acquired + false # Second caller fails to acquire lock + else + lock_acquired = true + true # First caller acquires lock + end + end + + allow(Redis::Alfred).to receive(:delete).with(message_source_key) + + # Run two services "simultaneously" (both pass the check before either sets the lock) + service1 = described_class.new(inbox: whatsapp_channel.inbox, params: params) + service2 = described_class.new(inbox: whatsapp_channel.inbox, params: params) + + # Mock find_message_by_source_id on specific instances (simulating uncommitted transaction) + allow(service1).to receive(:find_message_by_source_id).and_return(nil) + allow(service2).to receive(:find_message_by_source_id).and_return(nil) + + service1.perform + service2.perform + + # With atomic lock, only ONE should create a message + expect(whatsapp_channel.inbox.messages.count).to eq(1) + expect(whatsapp_channel.inbox.conversations.count).to eq(1) + end + + it 'creates message in second inbox when same source_id exists in different inbox' do + # Create a message with same source_id in a different inbox + other_whatsapp_channel = create(:channel_whatsapp, sync_templates: false) + other_inbox = other_whatsapp_channel.inbox + other_contact = create(:contact, account: other_inbox.account) + other_contact_inbox = create(:contact_inbox, inbox: other_inbox, contact: other_contact) + other_conversation = create(:conversation, inbox: other_inbox, contact_inbox: other_contact_inbox) + create(:message, conversation: other_conversation, inbox: other_inbox, source_id: params[:messages].first[:id]) + + # Should still create message in our inbox since source_id check should be inbox-scoped + described_class.new(inbox: whatsapp_channel.inbox, params: params).perform + expect(whatsapp_channel.inbox.conversations.count).to eq(1) + expect(whatsapp_channel.inbox.messages.count).to eq(1) + end end context 'when unsupported message types' do @@ -432,7 +555,8 @@ describe Whatsapp::IncomingMessageService do ] }] }.with_indifferent_access expect(Message.find_by(source_id: 'wamid.SDFADSf23sfasdafasdfa')).not_to be_present - key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: 'wamid.SDFADSf23sfasdafasdfa') + # Key is scoped by inbox.id to prevent cross-inbox lock collisions + key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: "#{whatsapp_channel.inbox.id}_wamid.SDFADSf23sfasdafasdfa") Redis::Alfred.setex(key, true) expect(Redis::Alfred.get(key)).to be_truthy diff --git a/spec/services/whatsapp/zapi_handlers/received_callback_spec.rb b/spec/services/whatsapp/zapi_handlers/received_callback_spec.rb index da8e1f12b..1b0a99918 100644 --- a/spec/services/whatsapp/zapi_handlers/received_callback_spec.rb +++ b/spec/services/whatsapp/zapi_handlers/received_callback_spec.rb @@ -825,9 +825,10 @@ describe Whatsapp::ZapiHandlers::ReceivedCallback do end it 'does not create message if it is already being processed' do - allow(Redis::Alfred).to receive(:get) - .with(format_message_source_key('duplicate_123')) - .and_return(true) + # Simulate lock already acquired by returning false from SETNX + allow(Redis::Alfred).to receive(:set) + .with(format_message_source_key('duplicate_123'), true, nx: true, ex: 1.day) + .and_return(false) expect do service.perform @@ -835,16 +836,28 @@ describe Whatsapp::ZapiHandlers::ReceivedCallback do end it 'caches and clears message source id in Redis' do - allow(Redis::Alfred).to receive(:setex) + allow(Redis::Alfred).to receive(:set).and_return(true) allow(Redis::Alfred).to receive(:delete) service.perform - expect(Redis::Alfred).to have_received(:setex) - .with(format_message_source_key('duplicate_123'), true) + expect(Redis::Alfred).to have_received(:set) + .with(format_message_source_key('duplicate_123'), true, nx: true, ex: 1.day) expect(Redis::Alfred).to have_received(:delete) .with(format_message_source_key('duplicate_123')) end + + it 'does not clear lock when acquisition fails' do + allow(Redis::Alfred).to receive(:set) + .with(format_message_source_key('duplicate_123'), true, nx: true, ex: 1.day) + .and_return(false) + allow(Redis::Alfred).to receive(:delete) + + service.perform + + expect(Redis::Alfred).not_to have_received(:delete) + .with(format_message_source_key('duplicate_123')) + end end context 'when attachment download fails' do @@ -1238,6 +1251,10 @@ describe Whatsapp::ZapiHandlers::ReceivedCallback do end it 'waits for the lock if it is already acquired' do + # Stub the message processing lock to always succeed + allow(Redis::Alfred).to receive(:set) + .with(format_message_source_key('msg_123'), true, nx: true, ex: 1.day) + .and_return(true) allow(Redis::Alfred).to receive(:set).with('ZAPI::CONTACT_LOCK::5511987654321', 1, nx: true, ex: 5.seconds).and_return(false, true) allow(Redis::Alfred).to receive(:delete)