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