From d4c8d891abae60b1b71a4f2b659db791913bb0de Mon Sep 17 00:00:00 2001 From: Pissardo Date: Mon, 27 Jul 2026 22:37:41 +0200 Subject: [PATCH] Keep concurrency locks across non-graceful job release When a claimed concurrency-limited job is released back to ready after a force-kill, its semaphore could expire and maintenance could unblock another job for the same key while the released job was still ready. Workers then claimed both and broke limits_concurrency. Extend the lock on release, skip expiring semaphores still held by ready jobs, run concurrency maintenance synchronously at supervisor boot and dispatcher start, and cover the race with unit tests. --- app/models/solid_queue/claimed_execution.rb | 4 + .../solid_queue/job/concurrency_controls.rb | 6 + .../dispatcher/concurrency_maintenance.rb | 38 ++++++- lib/solid_queue/supervisor.rb | 4 + test/unit/concurrency_maintenance_test.rb | 104 ++++++++++++++++++ 5 files changed, 152 insertions(+), 4 deletions(-) create mode 100644 test/unit/concurrency_maintenance_test.rb diff --git a/app/models/solid_queue/claimed_execution.rb b/app/models/solid_queue/claimed_execution.rb index 265692692..5a0bff263 100644 --- a/app/models/solid_queue/claimed_execution.rb +++ b/app/models/solid_queue/claimed_execution.rb @@ -75,6 +75,10 @@ def perform def release SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id) do unless_already_finalized do + # Keep owning the concurrency slot while the job waits to be claimed + # again so maintenance cannot expire the lock and unblock another job + # for the same key (#735). + job.extend_concurrency_lock job.dispatch_bypassing_concurrency_limits destroy! end diff --git a/app/models/solid_queue/job/concurrency_controls.rb b/app/models/solid_queue/job/concurrency_controls.rb index 30d4399ed..17ff6872f 100644 --- a/app/models/solid_queue/job/concurrency_controls.rb +++ b/app/models/solid_queue/job/concurrency_controls.rb @@ -33,6 +33,12 @@ def blocked? blocked_execution.present? end + def extend_concurrency_lock + return unless concurrency_limited? + + Semaphore.where(key: concurrency_key).update_all(expires_at: concurrency_duration.from_now) + end + private def concurrency_on_conflict job_class.concurrency_on_conflict.to_s.inquiry diff --git a/lib/solid_queue/dispatcher/concurrency_maintenance.rb b/lib/solid_queue/dispatcher/concurrency_maintenance.rb index 81cf770cc..3f22a147b 100644 --- a/lib/solid_queue/dispatcher/concurrency_maintenance.rb +++ b/lib/solid_queue/dispatcher/concurrency_maintenance.rb @@ -6,15 +6,25 @@ class Dispatcher::ConcurrencyMaintenance attr_reader :interval, :batch_size + class << self + def perform(batch_size: SolidQueue::Configuration::DISPATCHER_DEFAULTS[:batch_size]) + new(nil, batch_size).perform + end + end + def initialize(interval, batch_size) @interval = interval @batch_size = batch_size end def start - @concurrency_maintenance_task = Concurrent::TimerTask.new(run_now: true, execution_interval: interval) do - expire_semaphores - unblock_blocked_executions + # Run once inline so stale locks are cleared before workers forked alongside + # this dispatcher can claim jobs (see #735). The timer then handles the + # periodic follow-up without racing that first pass via run_now. + perform + + @concurrency_maintenance_task = Concurrent::TimerTask.new(run_now: false, execution_interval: interval) do + perform end @concurrency_maintenance_task.add_observer do |_, _, error| @@ -28,13 +38,33 @@ def stop @concurrency_maintenance_task&.shutdown end + def perform + expire_semaphores + unblock_blocked_executions + end + private def expire_semaphores wrap_in_app_executor do - Semaphore.expired.in_batches(of: batch_size, &:delete_all) + # Ready concurrency-limited jobs still own their slot after a claimed + # execution is released back to the ready queue. Expiring that + # semaphore would let unblock promote another job for the same key and + # break limits_concurrency (#735). + scope = Semaphore.expired + ready_keys = concurrency_keys_held_by_ready_jobs + scope = scope.where.not(key: ready_keys) if ready_keys.any? + + scope.in_batches(of: batch_size, &:delete_all) end end + def concurrency_keys_held_by_ready_jobs + Job.where(id: ReadyExecution.select(:job_id)) + .where.not(concurrency_key: [ nil, "" ]) + .distinct + .pluck(:concurrency_key) + end + def unblock_blocked_executions wrap_in_app_executor do BlockedExecution.unblock(batch_size) diff --git a/lib/solid_queue/supervisor.rb b/lib/solid_queue/supervisor.rb index 17b90c544..a9291f149 100644 --- a/lib/solid_queue/supervisor.rb +++ b/lib/solid_queue/supervisor.rb @@ -38,6 +38,10 @@ def start boot run_start_hooks + # Clear expired concurrency locks before forking workers so they cannot + # claim jobs that should stay blocked until maintenance runs (#735). + Dispatcher::ConcurrencyMaintenance.perform + start_processes launch_maintenance_task diff --git a/test/unit/concurrency_maintenance_test.rb b/test/unit/concurrency_maintenance_test.rb new file mode 100644 index 000000000..7b5656084 --- /dev/null +++ b/test/unit/concurrency_maintenance_test.rb @@ -0,0 +1,104 @@ +# frozen_string_literal: true + +require "test_helper" + +class ConcurrencyMaintenanceTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + setup do + @result = JobResult.create!(queue_name: "default", status: "") + end + + # Regression for https://github.com/rails/solid_queue/issues/735: + # After a claimed concurrency-limited job is released back to ready (non-graceful + # shutdown), its semaphore may still be expired. Expiring that semaphore and + # unblocking the next job while the released job remains ready leaves two ready + # jobs for the same key — workers can then claim both and violate the limit. + test "expire_semaphores does not drop locks still held by ready jobs" do + claimed_job, blocked_job = enqueue_claimed_and_blocked_pair + + claimed_job.claimed_execution.release + + assert claimed_job.reload.ready? + assert blocked_job.reload.blocked? + + semaphore = SolidQueue::Semaphore.find_by!(key: claimed_job.concurrency_key) + assert_equal 0, semaphore.value + semaphore.update!(expires_at: 1.minute.ago) + + maintenance = SolidQueue::Dispatcher::ConcurrencyMaintenance.new(60, 100) + maintenance.send(:expire_semaphores) + maintenance.send(:unblock_blocked_executions) + + skip_active_record_query_cache do + assert SolidQueue::Semaphore.exists?(key: claimed_job.concurrency_key), + "semaphore held by a ready job must not be expired away" + assert claimed_job.reload.ready? + assert blocked_job.reload.blocked? + assert_equal 1, SolidQueue::ReadyExecution.count + assert_equal 1, SolidQueue::BlockedExecution.count + end + end + + test "perform expires abandoned locks and unblocks waiting jobs" do + first = NonOverlappingUpdateResultJob.perform_later(@result, name: "A") + first_job = wait_for_solid_queue_job(first) + + # Abandon the ready execution without releasing the concurrency lock, + # leaving an expired semaphore with blocked work waiting behind it. + SolidQueue::ReadyExecution.find_by!(job_id: first_job.id).delete + assert_equal 0, SolidQueue::Semaphore.find_by!(key: first_job.concurrency_key).value + + second = NonOverlappingUpdateResultJob.perform_later(@result, name: "B") + second_job = wait_for_solid_queue_job(second) + assert second_job.blocked? + + SolidQueue::Semaphore.where(key: first_job.concurrency_key).update_all(expires_at: 1.minute.ago) + SolidQueue::BlockedExecution.update_all(expires_at: 1.minute.ago) + + SolidQueue::Dispatcher::ConcurrencyMaintenance.perform(batch_size: 100) + + skip_active_record_query_cache do + assert second_job.reload.ready? + assert_equal 0, SolidQueue::BlockedExecution.count + end + end + + test "releasing a claimed job extends its concurrency lock expiry" do + claimed_job, _blocked_job = enqueue_claimed_and_blocked_pair + + semaphore = SolidQueue::Semaphore.find_by!(key: claimed_job.concurrency_key) + semaphore.update!(expires_at: 1.second.from_now) + previous_expiry = semaphore.reload.expires_at + + claimed_job.claimed_execution.release + + skip_active_record_query_cache do + assert claimed_job.reload.ready? + assert_operator SolidQueue::Semaphore.find_by!(key: claimed_job.concurrency_key).expires_at, :>, previous_expiry + end + end + + private + def enqueue_claimed_and_blocked_pair + first = NonOverlappingUpdateResultJob.perform_later(@result, name: "A", pause: 30.seconds) + first_job = wait_for_solid_queue_job(first) + + process = SolidQueue::Process.register(kind: "Worker", pid: Process.pid, name: "maintenance-test-#{SecureRandom.hex(4)}") + claimed = SolidQueue::ReadyExecution.claim("*", 1, process.id) + assert_equal [ first_job.id ], claimed.map(&:job_id) + + second = NonOverlappingUpdateResultJob.perform_later(@result, name: "B") + blocked_job = wait_for_solid_queue_job(second) + assert blocked_job.blocked? + + [ first_job.reload, blocked_job ] + end + + def wait_for_solid_queue_job(active_job) + wait_while_with_timeout!(2.seconds) do + SolidQueue::Job.find_by(active_job_id: active_job.job_id).nil? + end + SolidQueue::Job.find_by!(active_job_id: active_job.job_id) + end +end