diff --git a/app/models/solid_queue/blocked_execution.rb b/app/models/solid_queue/blocked_execution.rb index 68551a5f..251e4924 100644 --- a/app/models/solid_queue/blocked_execution.rb +++ b/app/models/solid_queue/blocked_execution.rb @@ -62,6 +62,11 @@ def set_expires_at end def acquire_concurrency_lock + # A job whose class no longer resolves can't check its concurrency limits, but it + # can't hold a semaphore either. Release it without one so it fails on execution + # instead of crashing the dispatcher's concurrency maintenance forever. + return true unless job.concurrency_limited? + Semaphore.wait(job) end diff --git a/test/models/solid_queue/blocked_execution_test.rb b/test/models/solid_queue/blocked_execution_test.rb new file mode 100644 index 00000000..82ef5b07 --- /dev/null +++ b/test/models/solid_queue/blocked_execution_test.rb @@ -0,0 +1,40 @@ +# frozen_string_literal: true + +require "test_helper" + +class SolidQueue::BlockedExecutionTest < ActiveSupport::TestCase + self.use_transactional_tests = false + + class NonOverlappingJob < ApplicationJob + limits_concurrency key: ->(job_result, **) { job_result } + + def perform(job_result) + end + end + + setup do + @result = JobResult.create!(queue_name: "default") + end + + test "release a blocked execution whose job class no longer resolves" do + NonOverlappingJob.perform_later(@result) + NonOverlappingJob.perform_later(@result) + + blocked_job = SolidQueue::Job.last + assert blocked_job.blocked? + + # Simulate the job class being renamed or deleted in a later deploy + SolidQueue::Job.where(id: blocked_job.id).update_all(class_name: "NoLongerExistingJob") + semaphore_value = SolidQueue::Semaphore.find_by!(key: blocked_job.concurrency_key).value + + assert SolidQueue::BlockedExecution.release_one(blocked_job.concurrency_key) + + # The execution is promoted to ready, where it will fail on execution and + # be recorded as failed, instead of being retried by the dispatcher forever + assert_not SolidQueue::BlockedExecution.exists?(job_id: blocked_job.id) + assert SolidQueue::ReadyExecution.exists?(job_id: blocked_job.id) + + # Without concurrency limits to check, no semaphore slot is taken + assert_equal semaphore_value, SolidQueue::Semaphore.find_by!(key: blocked_job.concurrency_key).value + end +end