diff --git a/app/models/solid_queue/claimed_execution.rb b/app/models/solid_queue/claimed_execution.rb index 26569269..214a0161 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 } } @@ -62,14 +73,7 @@ def discard_all_from_jobs(*) end def perform - result = execute - - if result.success? - finished - else - failed_with(result.error) - raise result.error - end + finalize_result(execute) end def release @@ -90,6 +94,21 @@ def failed_with(error) end private + def finalize_result(result) + if result.success? + finished + else + 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 execute ActiveJob::Base.execute(job.arguments.merge("provider_job_id" => job.id)) Result.new(true, nil) @@ -122,4 +141,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/fiber_pool.rb b/lib/solid_queue/fiber_pool.rb index 0a29795f..98b68a88 100644 --- a/lib/solid_queue/fiber_pool.rb +++ b/lib/solid_queue/fiber_pool.rb @@ -2,7 +2,7 @@ module SolidQueue class FiberPool < Pool - def initialize(size, on_idle: nil) + def initialize(size, on_idle: nil, on_unrecoverable_error: nil) super @state_mutex = Mutex.new @@ -108,6 +108,7 @@ def perform_execution(execution) handle_thread_error(error) register_fatal_error(error) rescue Exception => error + handle_unrecoverable_error(error) handle_thread_error(error) ensure restore_capacity diff --git a/lib/solid_queue/pool.rb b/lib/solid_queue/pool.rb index f58ac8fc..563f2c58 100644 --- a/lib/solid_queue/pool.rb +++ b/lib/solid_queue/pool.rb @@ -4,15 +4,20 @@ module SolidQueue class Pool include AppExecutor - def self.build(type:, size:, on_idle: nil) - SolidQueue.const_get("#{type.to_s.camelize}Pool").new(size, on_idle: on_idle) + def self.build(type:, size:, on_idle: nil, on_unrecoverable_error: nil) + SolidQueue.const_get("#{type.to_s.camelize}Pool").new( + size, + on_idle: on_idle, + on_unrecoverable_error: on_unrecoverable_error + ) end attr_reader :size - 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_capacity = size @mutex = Mutex.new end @@ -41,7 +46,7 @@ def idle? end private - attr_reader :mutex, :on_idle + attr_reader :mutex, :on_idle, :on_unrecoverable_error def schedule(execution) raise NotImplementedError @@ -50,11 +55,20 @@ def schedule(execution) def perform_execution(execution) wrap_in_app_executor { execution.perform } rescue Exception => error + handle_unrecoverable_error(error) handle_thread_error(error) ensure restore_capacity end + 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 reserve_capacity! mutex.synchronize do raise RuntimeError, "Execution pool is at capacity" if @available_capacity <= 0 diff --git a/lib/solid_queue/thread_pool.rb b/lib/solid_queue/thread_pool.rb index b1485264..85098783 100644 --- a/lib/solid_queue/thread_pool.rb +++ b/lib/solid_queue/thread_pool.rb @@ -17,6 +17,7 @@ def schedule(execution) end.on_rejection! do |error| # Backstop for errors raised outside perform_execution's own rescue, # such as when restoring capacity or waking up the worker + handle_unrecoverable_error(error) handle_thread_error(error) end end diff --git a/lib/solid_queue/worker.rb b/lib/solid_queue/worker.rb index 9ba65bf3..78870a44 100644 --- a/lib/solid_queue/worker.rb +++ b/lib/solid_queue/worker.rb @@ -22,7 +22,8 @@ def initialize(**options) @pool = Pool.build \ type: execution_pool_type, size: execution_pool_size, - on_idle: -> { wake_up } + on_idle: -> { wake_up }, + on_unrecoverable_error: ->(*) { request_termination } super(**options) end @@ -48,6 +49,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 72a988f6..01242883 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/fiber_pool_test.rb b/test/unit/fiber_pool_test.rb index f82694ba..51c44405 100644 --- a/test/unit/fiber_pool_test.rb +++ b/test/unit/fiber_pool_test.rb @@ -19,7 +19,7 @@ def perform def test_builds_a_fiber_pool pool = mock - SolidQueue::FiberPool.expects(:new).with(5, on_idle: nil).returns(pool) + SolidQueue::FiberPool.expects(:new).with(5, on_idle: nil, on_unrecoverable_error: nil).returns(pool) assert_equal pool, SolidQueue::Pool.build(type: :fiber, size: 5) end diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index c21cd574..8f853faa 100644 --- a/test/unit/worker_test.rb +++ b/test/unit/worker_test.rb @@ -111,7 +111,7 @@ class WorkerTest < ActiveSupport::TestCase subscriber = ErrorBuffer.new Rails.error.subscribe(subscriber) - SolidQueue::ClaimedExecution::Result.expects(:new).raises(ExpectedTestError.new("everything is broken")).at_least_once + SolidQueue::ClaimedExecution.any_instance.stubs(:finished).raises(ActiveRecord::StatementInvalid.new("transient DB glitch")) AddToBufferJob.perform_later "hey!" @@ -120,8 +120,9 @@ class WorkerTest < ActiveSupport::TestCase wait_for_jobs_to_finish_for(1.second) @worker.wake_up - assert_equal 1, subscriber.errors.count - assert_equal "everything is broken", subscriber.messages.first + finalization_errors = subscriber.errors.map(&:first).grep(SolidQueue::ClaimedExecution::FinalizationError) + assert_equal 1, finalization_errors.count + assert_match(/transient DB glitch/, finalization_errors.first.message) ensure Rails.error.unsubscribe(subscriber) if Rails.error.respond_to?(:unsubscribe) SolidQueue.on_thread_error = original_on_thread_error @@ -144,6 +145,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 "polling queries are logged" do log = StringIO.new with_active_record_logger(ActiveSupport::Logger.new(log)) do