diff --git a/app/models/solid_queue/job/concurrency_controls.rb b/app/models/solid_queue/job/concurrency_controls.rb index b59c464ad..33610ca1b 100644 --- a/app/models/solid_queue/job/concurrency_controls.rb +++ b/app/models/solid_queue/job/concurrency_controls.rb @@ -8,7 +8,7 @@ module ConcurrencyControls included do has_one :blocked_execution - delegate :concurrency_limit, :concurrency_duration, to: :job_class + delegate :concurrency_limit, :concurrency_duration, to: :job_class, allow_nil: true before_destroy :unblock_next_blocked_job, if: -> { concurrency_limited? && ready? } end @@ -48,8 +48,10 @@ def acquire_concurrency_lock Semaphore.wait(self) end + # A job whose class can't be loaded where it runs still holds the lock its + # key took when it was dispatched, so release by key alone. def release_concurrency_lock - return false unless concurrency_limited? + return false unless concurrency_key.present? Semaphore.signal(self) end diff --git a/app/models/solid_queue/semaphore.rb b/app/models/solid_queue/semaphore.rb index 7d111ac55..1e0f26462 100644 --- a/app/models/solid_queue/semaphore.rb +++ b/app/models/solid_queue/semaphore.rb @@ -85,7 +85,7 @@ def key end def expires_at - job.concurrency_duration.from_now + (job.concurrency_duration || SolidQueue.default_concurrency_control_period).from_now end def limit diff --git a/test/models/solid_queue/claimed_execution_concurrency_test.rb b/test/models/solid_queue/claimed_execution_concurrency_test.rb index 1847d49f3..404416dba 100644 --- a/test/models/solid_queue/claimed_execution_concurrency_test.rb +++ b/test/models/solid_queue/claimed_execution_concurrency_test.rb @@ -158,6 +158,23 @@ class SolidQueue::ClaimedExecutionConcurrencyTest < ActiveSupport::TestCase end end + test "a job whose class is gone by the time it runs releases its concurrency lock" do + first_job, claimed_execution, concurrency_key = enqueue_blocked_group + assert_equal 0, SolidQueue::Semaphore.find_by!(key: concurrency_key).value + + # A worker on a release that no longer defines the class claims the job. + first_job.update_columns(class_name: "RemovedJobClass") + claimed_execution.reload + + assert_raises SolidQueue::Job::ClassMissingError do + claimed_execution.perform + end + + assert first_job.reload.failed? + assert SolidQueue::ReadyExecution.where(job_id: SolidQueue::Job.where(concurrency_key: concurrency_key).select(:id)).exists?, + "the failed job should release the next blocked job of its key" + end + private # Enqueues three concurrency-limited jobs over the same key. The first is # claimed; the other two are blocked. Returns the first job, its claimed