Stop workers when job finalization fails

app/models/solid_queue/claimed_execution.rb

@@ -1,126 +1,158 @@
 # frozen_string_literal: true
 
-class SolidQueue::ClaimedExecution < SolidQueue::Execution
-  belongs_to :process
+module SolidQueue
+  class ClaimedExecution < Execution
+    belongs_to :process
 
-  scope :orphaned, -> { where.missing(:process) }
+    scope :orphaned, -> { where.missing(:process) }
 
-  class Result < Struct.new(:success, :error)
-    def success?
-      success
+    class Result < Struct.new(:success, :error)
+      def success?
+        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 < Processes::UnrecoverableError
+      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
-  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 } }
+    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 } }
 
-      SolidQueue.instrument(:claim, process_id: process_id, job_ids: job_ids) do |payload|
-        insert_all!(job_data)
-        where(job_id: job_ids, process_id: process_id).load.tap do |claimed|
-          block.call(claimed)
+        SolidQueue.instrument(:claim, process_id: process_id, job_ids: job_ids) do |payload|
+          insert_all!(job_data)
+          where(job_id: job_ids, process_id: process_id).load.tap do |claimed|
+            block.call(claimed)
 
-          payload[:size] = claimed.size
-          payload[:claimed_job_ids] = claimed.map(&:job_id)
+            payload[:size] = claimed.size
+            payload[:claimed_job_ids] = claimed.map(&:job_id)
+          end
         end
       end
-    end
 
-    def release_all
-      SolidQueue.instrument(:release_many_claimed) do |payload|
-        includes(:job).tap do |executions|
-          executions.each(&:release)
+      def release_all
+        SolidQueue.instrument(:release_many_claimed) do |payload|
+          includes(:job).tap do |executions|
+            executions.each(&:release)
 
-          payload[:size] = executions.size
+            payload[:size] = executions.size
+          end
         end
       end
-    end
 
-    def fail_all_with(error)
-      includes(:job).tap do |executions|
-        return if executions.empty?
+      def fail_all_with(error)
+        includes(:job).tap do |executions|
+          return if executions.empty?
+
+          SolidQueue.instrument(:fail_many_claimed) do |payload|
+            executions.each do |execution|
+              execution.failed_with(error)
+            end
 
-        SolidQueue.instrument(:fail_many_claimed) do |payload|
-          executions.each do |execution|
-            execution.failed_with(error)
+            payload[:process_ids] = executions.map(&:process_id).uniq
+            payload[:job_ids] = executions.map(&:job_id).uniq
+            payload[:size] = executions.size
+            payload[:error] = error
           end
-
-          payload[:process_ids] = executions.map(&:process_id).uniq
-          payload[:job_ids] = executions.map(&:job_id).uniq
-          payload[:size] = executions.size
-          payload[:error] = error
         end
       end
-    end
 
-    def discard_all_in_batches(*)
-      raise UndiscardableError, "Can't discard jobs in progress"
-    end
+      def discard_all_in_batches(*)
+        raise UndiscardableError, "Can't discard jobs in progress"
+      end
 
-    def discard_all_from_jobs(*)
-      raise UndiscardableError, "Can't discard jobs in progress"
+      def discard_all_from_jobs(*)
+        raise UndiscardableError, "Can't discard jobs in progress"
+      end
     end
-  end
 
-  def perform
-    result = execute
+    def perform
+      result = execute
 
-    if result.success?
-      finished
-    else
-      failed_with(result.error)
-      raise result.error
+      if result.success?
+        finalizing { finished }
+      else
+        finalizing { failed_with(result.error) }
+        raise result.error
+      end
     end
-  end
 
-  def release
-    SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id) do
-      unless_already_finalized do
-        job.dispatch_bypassing_concurrency_limits
-        destroy!
+    def release
+      SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id) do
+        unless_already_finalized do
+          job.dispatch_bypassing_concurrency_limits
+          destroy!
+        end
       end
     end
-  end
 
-  def discard
-    raise UndiscardableError, "Can't discard a job in progress"
-  end
-
-  def failed_with(error)
-    finalize { job.failed_with(error) }
-  end
-
-  private
-    def execute
-      ActiveJob::Base.execute(job.arguments.merge("provider_job_id" => job.id))
-      Result.new(true, nil)
-    rescue Exception => e
-      Result.new(false, e)
+    def discard
+      raise UndiscardableError, "Can't discard a job in progress"
     end
 
-    def finished
-      finalize { job.finished! }
+    def failed_with(error)
+      finalize { job.failed_with(error) }
     end
 
-    def finalize
-      finalized = unless_already_finalized do
+    private
+      # A failure here means the job already ran but we couldn't record the
+      # outcome, and the claim is still held by this living worker, where no
+      # recovery can reach it: it's a process problem, not a job problem
+      def finalizing
         yield
-        destroy!
-        true
+      rescue => error
+        raise FinalizationError.new(self, cause: error) if still_claimed?
+
+        raise
       end
 
-      # Unblock the next job outside the finalize transaction so a failure while
-      # releasing the concurrency lock or dispatching the next job can't roll back
-      # a job that already finished or failed. Only the actor that owned and
-      # finalized the claim gets here, so the lock is released exactly once.
-      job.unblock_next_blocked_job if finalized
-    end
+      def execute
+        ActiveJob::Base.execute(job.arguments.merge("provider_job_id" => job.id))
+        Result.new(true, nil)
+      rescue Exception => e
+        Result.new(false, e)
+      end
 
-    def unless_already_finalized
-      transaction do
-        return false unless self.class.unscoped.lock.find_by(id: id)
+      def finished
+        finalize { job.finished! }
+      end
 
-        yield
+      def finalize
+        finalized = unless_already_finalized do
+          yield
+          destroy!
+          true
+        end
+
+        # Unblock the next job outside the finalize transaction so a failure while
+        # releasing the concurrency lock or dispatching the next job can't roll back
+        # a job that already finished or failed. Only the actor that owned and
+        # finalized the claim gets here, so the lock is released exactly once.
+        job.unblock_next_blocked_job if finalized
       end
-    end
+
+      def unless_already_finalized
+        transaction do
+          return false unless self.class.unscoped.lock.find_by(id: id)
+
+          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
 end

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

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?(Processes::UnrecoverableError)
+
+        # 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
+      end
+
       def reserve_capacity!
         mutex.synchronize do
           raise RuntimeError, "Execution pool is at capacity" if @available_capacity <= 0

lib/solid_queue/processes/unrecoverable_error.rb

@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  module Processes
+    # Errors that mean the process raising them can no longer account for the
+    # work it holds, so it should stop and let its supervisor replace it
+    class UnrecoverableError < RuntimeError; end
+  end
+end

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

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)

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")