diff --git a/app/models/solid_queue/claimed_execution.rb b/app/models/solid_queue/claimed_execution.rb
index 9a606e0b..90b56b42 100644
--- a/app/models/solid_queue/claimed_execution.rb
+++ b/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
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..8b18d5a7 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?(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
diff --git a/lib/solid_queue/processes/unrecoverable_error.rb b/lib/solid_queue/processes/unrecoverable_error.rb
new file mode 100644
index 00000000..bfdd9713
--- /dev/null
+++ b/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
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..3947b05d 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 618a16c6..aff8cde1 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
     polling_query = /SELECT .* FROM .solid_queue_ready_executions. WHERE .solid_queue_ready_executions...queue_name./