From d903a3649ee6cb1385c2f4f228a94de6100abcaf Mon Sep 17 00:00:00 2001 From: Pissardo Date: Fri, 24 Jul 2026 22:03:40 +0200 Subject: [PATCH] Stop workers when job finalization fails Prevent claimed executions from remaining stuck after transient database errors by terminating the owning worker so shutdown releases the claim. --- app/models/solid_queue/claimed_execution.rb | 25 ++++++++++++ lib/solid_queue/pool.rb | 14 ++++++- lib/solid_queue/worker.rb | 14 ++++++- .../solid_queue/claimed_execution_test.rb | 29 ++++++++++++++ test/unit/worker_test.rb | 39 ++++++++++++++++++- 5 files changed, 117 insertions(+), 4 deletions(-) diff --git a/app/models/solid_queue/claimed_execution.rb b/app/models/solid_queue/claimed_execution.rb index 265692692..f4cc532e9 100644 --- a/app/models/solid_queue/claimed_execution.rb +++ b/app/models/solid_queue/claimed_execution.rb @@ -11,6 +11,17 @@ def success? end end + # Raised when a job has already run (or failed) but we couldn't update its + # claim/finished state because of a transient error. The claim is still held + # by a living worker, so it won't be recovered as orphaned unless the worker + # is stopped and replaced. + class FinalizationError < RuntimeError + def initialize(claimed_execution, cause:) + super("Failed to finalize claimed execution #{claimed_execution.id} (job #{claimed_execution.job_id}): #{cause.class}: #{cause.message}") + set_backtrace(cause.backtrace) if cause.backtrace + end + end + class << self def claiming(job_ids, process_id, &block) job_data = Array(job_ids).collect { |job_id| { job_id: job_id, process_id: process_id } } @@ -70,6 +81,12 @@ def perform failed_with(result.error) raise result.error end + rescue FinalizationError + raise + rescue => error + raise FinalizationError.new(self, cause: error) if still_claimed? + + raise end def release @@ -122,4 +139,12 @@ def unless_already_finalized yield end end + + def still_claimed? + self.class.exists?(id) + rescue + # If we can't check because the DB is unavailable, assume the claim is + # still held so the worker can be stopped and replaced. + true + end end diff --git a/lib/solid_queue/pool.rb b/lib/solid_queue/pool.rb index 9c3d2a298..0d464787e 100644 --- a/lib/solid_queue/pool.rb +++ b/lib/solid_queue/pool.rb @@ -8,9 +8,10 @@ class Pool delegate :shutdown, :shutdown?, :wait_for_termination, to: :executor - def initialize(size, on_idle: nil) + def initialize(size, on_idle: nil, on_unrecoverable_error: nil) @size = size @on_idle = on_idle + @on_unrecoverable_error = on_unrecoverable_error @available_threads = Concurrent::AtomicFixnum.new(size) @mutex = Mutex.new end @@ -26,6 +27,7 @@ def post(execution) mutex.synchronize { on_idle.try(:call) if idle? } end end.on_rejection! do |e| + handle_unrecoverable_error(e) handle_thread_error(e) end end @@ -39,7 +41,7 @@ def idle? end private - attr_reader :available_threads, :on_idle, :mutex + attr_reader :available_threads, :on_idle, :on_unrecoverable_error, :mutex DEFAULT_OPTIONS = { min_threads: 0, @@ -47,6 +49,14 @@ def idle? fallback_policy: :abort } + def handle_unrecoverable_error(error) + return unless error.is_a?(ClaimedExecution::FinalizationError) + + # Only signal shutdown — do not join the worker from this pool thread, + # or wait_for_termination during worker shutdown would deadlock. + on_unrecoverable_error&.call(error) + end + def executor @executor ||= Concurrent::ThreadPoolExecutor.new DEFAULT_OPTIONS.merge(max_threads: size, max_queue: size) end diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index e036a5fd9..1655cec53 100644 --- a/lib/solid_queue/worker.rb +++ b/lib/solid_queue/worker.rb @@ -16,7 +16,11 @@ def initialize(**options) # Ensure that the queues array is deep frozen to prevent accidental modification @queues = Array(options[:queues]).map(&:freeze).freeze - @pool = Pool.new(options[:threads], on_idle: -> { wake_up }) + @pool = Pool.new( + options[:threads], + on_idle: -> { wake_up }, + on_unrecoverable_error: ->(*) { request_termination } + ) super(**options) end @@ -42,6 +46,14 @@ def claim_executions end end + def request_termination + # Signal the poller to shut down without joining from the pool thread. + # Runnable#stop joins when unsupervised, which would deadlock once + # shutdown waits for this pool thread to finish. + @stopped = true + wake_up + end + def shutdown pool.shutdown pool.wait_for_termination(SolidQueue.shutdown_timeout) diff --git a/test/models/solid_queue/claimed_execution_test.rb b/test/models/solid_queue/claimed_execution_test.rb index 72a988f69..01242883c 100644 --- a/test/models/solid_queue/claimed_execution_test.rb +++ b/test/models/solid_queue/claimed_execution_test.rb @@ -17,6 +17,35 @@ class SolidQueue::ClaimedExecutionTest < ActiveSupport::TestCase assert job.reload.finished? end + test "raises FinalizationError when finishing fails while the claim remains" do + claimed_execution = prepare_and_claim_job AddToBufferJob.perform_later(42) + + SolidQueue::ClaimedExecution.any_instance.stubs(:finished).raises(ActiveRecord::StatementInvalid.new("transient DB glitch")) + + error = assert_raises SolidQueue::ClaimedExecution::FinalizationError do + claimed_execution.perform + end + + assert_match(/transient DB glitch/, error.message) + assert_equal ActiveRecord::StatementInvalid, error.cause.class + assert SolidQueue::ClaimedExecution.exists?(claimed_execution.id) + assert_not claimed_execution.job.reload.finished? + end + + test "raises FinalizationError when failing the job fails while the claim remains" do + claimed_execution = prepare_and_claim_job RaisingJob.perform_later(RuntimeError, "A") + + SolidQueue::ClaimedExecution.any_instance.stubs(:failed_with).raises(ActiveRecord::StatementInvalid.new("transient DB glitch")) + + error = assert_raises SolidQueue::ClaimedExecution::FinalizationError do + claimed_execution.perform + end + + assert_match(/transient DB glitch/, error.message) + assert SolidQueue::ClaimedExecution.exists?(claimed_execution.id) + assert_not claimed_execution.job.reload.failed? + end + test "stale performer cannot release a concurrency lock after its claim is pruned" do job_result = JobResult.create!(queue_name: "default", status: "") first_active_job = NonOverlappingUpdateResultJob.perform_later(job_result, name: "A") diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index 3d692404b..ee9c5b5f0 100644 --- a/test/unit/worker_test.rb +++ b/test/unit/worker_test.rb @@ -62,7 +62,9 @@ class WorkerTest < ActiveSupport::TestCase @worker.wake_up assert_equal 1, subscriber.errors.count - assert_equal "everything is broken", subscriber.messages.first + error = subscriber.errors.first.first + assert_instance_of SolidQueue::ClaimedExecution::FinalizationError, error + assert_equal "everything is broken", error.cause.message ensure Rails.error.unsubscribe(subscriber) if Rails.error.respond_to?(:unsubscribe) SolidQueue.on_thread_error = original_on_thread_error @@ -85,6 +87,41 @@ class WorkerTest < ActiveSupport::TestCase Rails.error.unsubscribe(subscriber) if Rails.error.respond_to?(:unsubscribe) end + test "worker stops and releases the claim when finishing a job fails" do + previous_on_thread_error, SolidQueue.on_thread_error = SolidQueue.on_thread_error, ->(*) { } + + SolidQueue::ClaimedExecution.any_instance.stubs(:finished).raises(ActiveRecord::StatementInvalid.new("transient DB glitch")) + + AddToBufferJob.perform_later "hey!" + + @worker.start + + wait_while_with_timeout(2.seconds) { !@worker.pool.shutdown? } + assert @worker.pool.shutdown? + + wait_for_registered_processes(0, timeout: 1.second) + assert_no_registered_processes + + assert_equal 0, SolidQueue::ClaimedExecution.count + assert SolidQueue::Job.last.reload.ready? + ensure + SolidQueue.on_thread_error = previous_on_thread_error + end + + test "worker keeps running after a regular job failure" do + RaisingJob.perform_later(ExpectedTestError, "B") + AddToBufferJob.perform_later "ok" + + @worker.start + + wait_for_jobs_to_finish_for(2.seconds) + @worker.wake_up + + assert_not @worker.pool.shutdown? + assert_equal "ok", JobBuffer.last_value + assert_equal 0, SolidQueue::ClaimedExecution.count + end + test "claim and process more enqueued jobs than the pool size allows to process at once" do 5.times do |i| StoreResultJob.perform_later(:paused, pause: 0.1.second)