From 73ba0b26e53d1ce94431f13f45c913ca77148edf Mon Sep 17 00:00:00 2001 From: Shivam Mishra Date: Fri, 5 Jun 2026 14:06:45 +0530 Subject: [PATCH] fix(instagram): process webhook events after mutex retry exhaustion (#14647) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Instagram webhooks use a Redis mutex to protect the first-message path for a contact/account pair. When multiple webhook events for the same Instagram contact arrive at nearly the same time, they can all enter the “find or create” flow together. There are two pieces involved: `ContactInbox` maps the external Instagram scoped user id to a Chatwoot contact inside an inbox. That side already has a unique `(inbox_id, source_id)` index and retry handling, so duplicate contact-inbox creation is mostly protected at the database layer. `Conversation` creation is more fragile. The message builder looks for an existing active conversation for the contact/inbox and creates one when none is found. If two first-message jobs run concurrently, both can see “no active conversation yet” and both can create a conversation. The mutex exists to bump those jobs apart long enough for the first one to create the conversation, so the next one appends to it instead of creating a duplicate. In production, the old behavior could turn that race dampener into a drop path. Once `Webhooks::InstagramEventsJob` exhausted its lock retries, ActiveJob stopped retrying with `MutexApplicationJob::LockAcquisitionError`. Since ActiveJob retries are not FIFO, later jobs could still win the lock while older jobs kept getting deferred until exhaustion. ## Mutex Application Job Changes This adds `retry_on_lock_conflict` as a small wrapper around the existing `retry_on LockAcquisitionError` pattern. The new API keeps the default behavior intact, but lets jobs explicitly define what should happen after lock retry exhaustion: ```ruby retry_on_lock_conflict wait: 1.second, attempts: 3, on_exhaustion: :process_without_lock ``` The fallback receives the original job arguments, so job classes do not need to reach into ActiveJob internals like `arguments.first`. ## Instagram Fallback Instagram now processes the webhook payload even after the mutex retry window is exhausted. This matches what the mutex was meant to do in the first place: bump concurrent events apart long enough to avoid duplicate conversation creation, not permanently block or drop webhook events. If a job cannot acquire the lock after a few retries, it continues through the normal Instagram processing path without the mutex. ## Retry And Lock Tuning The Instagram lock retry window now uses deterministic backoff: ```ruby retry_on_lock_conflict wait: ->(executions) { executions.seconds }, attempts: 3, on_exhaustion: :process_without_lock ``` The lock TTL is also set explicitly: ```ruby with_lock(key, 3.seconds) ``` This matters because Rails treats `attempts` as total executions, not retries after the first execution. With a fixed `wait: 1.second` and `attempts: 3`, the exhaustion handler can run roughly two seconds after the first lock conflict. That is earlier than a 3-second lock TTL, so the fallback could process without the lock while the original mutex is still valid. That would reopen the duplicate-conversation race the lock is meant to dampen. Using a proc wait makes the timing predictable and avoids Rails jitter for this retry path. The first conflict waits about 1 second, the second waits about 2 seconds, and the final attempt happens around the 3-second mark. That lines the retry window up with the Redis lock TTL before falling back to `process_without_lock`. The protected race window is intentionally small. `ContactInbox` creation already has database uniqueness protection, while conversation creation is the softer `find active conversation || create` path. We only need to give the first job enough time to create the conversation state that later jobs should reuse. The mutex still does not preserve message order, and ActiveJob retries are not FIFO. A longer retry window would mostly add latency for hot Instagram contacts without making ordering more correct. After the lock TTL has elapsed, processing without the lock is the better tradeoff than dead-lettering customer messages. --- app/jobs/mutex_application_job.rb | 13 ++++++ app/jobs/webhooks/instagram_events_job.rb | 20 ++++++++- spec/jobs/mutex_application_job_spec.rb | 53 +++++++++++++++++++++++ 3 files changed, 84 insertions(+), 2 deletions(-) diff --git a/app/jobs/mutex_application_job.rb b/app/jobs/mutex_application_job.rb index 58c7cbf36..4143eab2a 100644 --- a/app/jobs/mutex_application_job.rb +++ b/app/jobs/mutex_application_job.rb @@ -14,6 +14,19 @@ class MutexApplicationJob < ApplicationJob class LockAcquisitionError < StandardError; end + def self.retry_on_lock_conflict(wait:, attempts:, on_exhaustion: :raise) + retry_on LockAcquisitionError, wait: wait, attempts: attempts do |job, error| + raise error if on_exhaustion == :raise + + job.public_send(on_exhaustion, *job.arguments) + end + end + + # Redis::LockManager#unlock is not owner-checked. If a job runs past the TTL, + # Redis can expire the key, a newer job can acquire it, and the older job can + # then delete the newer job's lock on unlock. Current mutex users treat locks as + # short race dampeners, so this is acceptable for now. Future iterations should + # move Redis::LockManager to token-checked unlocks. def with_lock(lock_key, timeout = Redis::LockManager::LOCK_TIMEOUT) lock_manager = Redis::LockManager.new diff --git a/app/jobs/webhooks/instagram_events_job.rb b/app/jobs/webhooks/instagram_events_job.rb index 6383daff8..c27363127 100644 --- a/app/jobs/webhooks/instagram_events_job.rb +++ b/app/jobs/webhooks/instagram_events_job.rb @@ -1,6 +1,14 @@ class Webhooks::InstagramEventsJob < MutexApplicationJob queue_as :default - retry_on LockAcquisitionError, wait: 1.second, attempts: 8 + # This lock is only a short race dampener for first-message conversation creation. + # ContactInbox creation is already protected by a unique index, but conversation + # lookup is `find active conversation || create`, so concurrent first messages from + # the same IG contact can create duplicate conversations. + # + # ActiveJob retries are not FIFO, so a longer retry window does not preserve message + # order. Use deterministic backoff so the final attempt happens after the 3s lock TTL, + # then process without the lock instead of dropping the webhook. + retry_on_lock_conflict wait: ->(executions) { executions.seconds }, attempts: 3, on_exhaustion: :process_without_lock # @return [Array] We will support further events like reaction or seen in future SUPPORTED_EVENTS = [:message, :read].freeze @@ -9,11 +17,19 @@ class Webhooks::InstagramEventsJob < MutexApplicationJob @entries = entries key = format(::Redis::Alfred::IG_MESSAGE_MUTEX, sender_id: contact_instagram_id, ig_account_id: ig_account_id) - with_lock(key) do + # Keep the lock TTL just long enough for the first job to fetch profile data and + # create the contact/conversation. A longer TTL would add user-visible latency for + # hot contacts without giving us ordering guarantees. + with_lock(key, 3.seconds) do process_entries(entries) end end + def process_without_lock(entries) + Rails.logger.warn("[#{self.class.name}] Processing without lock after lock retry exhaustion") + process_entries(entries) + end + # https://developers.facebook.com/docs/messenger-platform/instagram/features/webhook def process_entries(entries) entries.each do |entry| diff --git a/spec/jobs/mutex_application_job_spec.rb b/spec/jobs/mutex_application_job_spec.rb index 91a56407d..4c8fa3394 100644 --- a/spec/jobs/mutex_application_job_spec.rb +++ b/spec/jobs/mutex_application_job_spec.rb @@ -55,4 +55,57 @@ RSpec.describe MutexApplicationJob do end.to raise_error(StandardError) end end + + describe '.retry_on_lock_conflict' do + let(:job_class) do + Class.new(described_class) do + retry_on_lock_conflict wait: 1.second, attempts: 1, on_exhaustion: :process_without_lock + + attr_reader :fallback_args + + def perform(lock_key, _payload) + with_lock(lock_key) { raise 'lock should not be acquired' } + end + + def process_without_lock(lock_key, payload) + @fallback_args = [lock_key, payload] + end + end + end + + let(:payload) { { 'message' => 'hello' } } + + before do + stub_const('LockConflictTestJob', job_class) + end + + it 'runs the configured handler with the original job arguments when lock retries are exhausted' do + allow(lock_manager).to receive(:lock).with(lock_key, Redis::LockManager::LOCK_TIMEOUT).and_return(false) + + job = job_class.new(lock_key, payload) + + expect { job.perform_now }.not_to raise_error + expect(job.fallback_args).to eq([lock_key, payload]) + end + + context 'without an exhaustion handler' do + let(:job_class) do + Class.new(described_class) do + retry_on_lock_conflict wait: 1.second, attempts: 1 + + def perform(lock_key) + with_lock(lock_key) { raise 'lock should not be acquired' } + end + end + end + + it 'raises the lock acquisition error when retries are exhausted' do + allow(lock_manager).to receive(:lock).with(lock_key, Redis::LockManager::LOCK_TIMEOUT).and_return(false) + + expect do + job_class.perform_now(lock_key) + end.to raise_error(StandardError) { |error| expect(error.class.name).to eq('MutexApplicationJob::LockAcquisitionError') } + end + end + end end