Replace terminated threads even if releasing their claimed jobs fails

lib/solid_queue/async_supervisor.rb

@@ -27,8 +27,7 @@ def replace_thread(thread_id)
           if (instance = process_instances.delete(thread_id))
             payload[:thread] = instance
 
-            error = Processes::ThreadTerminatedError.new(instance.name)
-            release_claimed_jobs_by(instance, with_error: error)
+            attempt_to_release_claimed_jobs_by(instance, with_error: Processes::ThreadTerminatedError.new(instance.name))
 
             start_process(configured_processes.delete(thread_id))
           end

lib/solid_queue/fork_supervisor.rb

@@ -57,7 +57,7 @@ def reap_terminated_forks
           terminated_fork.mark_as_reaped
 
           if !status.exited? || status.exitstatus.to_i > 0
-            attempt_to_release_claimed_jobs_by(terminated_fork, status)
+            attempt_to_release_claimed_jobs_by(terminated_fork, with_error: Processes::ProcessExitError.new(status))
           end
         end
 
@@ -73,23 +73,13 @@ def replace_fork(pid, status)
           terminated_fork.mark_as_reaped
           payload[:fork] = terminated_fork
 
-          attempt_to_release_claimed_jobs_by(terminated_fork, status)
+          attempt_to_release_claimed_jobs_by(terminated_fork, with_error: Processes::ProcessExitError.new(status))
 
           start_process(configured_processes.delete(pid))
         end
       end
     end
 
-    # The database may be unreachable — likely the same reason the fork
-    # terminated. Neither starting a replacement nor shutting down can depend
-    # on it: the jobs claimed by the terminated fork will be failed when its
-    # stale registration is pruned once the database is back.
-    def attempt_to_release_claimed_jobs_by(terminated_fork, status)
-      release_claimed_jobs_by(terminated_fork, with_error: Processes::ProcessExitError.new(status))
-    rescue StandardError => error
-      handle_thread_error(error)
-    end
-
     def all_processes_terminated?
       process_instances.empty?
     end

lib/solid_queue/supervisor/maintenance.rb

@@ -43,5 +43,15 @@ def release_claimed_jobs_by(terminated_process, with_error:)
           end
         end
       end
+
+      # The database may be unreachable — possibly the same reason the
+      # supervised process died. Neither starting a replacement nor shutting
+      # down can depend on it: the jobs claimed by the dead process will be
+      # failed when its stale registration is pruned once the database is back.
+      def attempt_to_release_claimed_jobs_by(terminated_process, with_error:)
+        release_claimed_jobs_by(terminated_process, with_error: with_error)
+      rescue StandardError => error
+        handle_thread_error(error)
+      end
   end
 end

test/unit/async_supervisor_test.rb

@@ -112,6 +112,26 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
     assert_no_match /the database connection pool is/, log.string
   end
 
+  test "replace a terminated thread even if releasing its claimed jobs fails" do
+    configuration = SolidQueue::Configuration.new(workers: [ { queues: "*", processes: 1 } ], dispatchers: [], skip_recurring: true)
+    supervisor = SolidQueue::AsyncSupervisor.new(configuration)
+
+    configured_process = configuration.configured_processes.first
+    terminated_thread = stub(kind: "Worker", name: "worker-42", hostname: "localhost", alive?: false)
+
+    supervisor.send(:process_instances)[42] = terminated_thread
+    supervisor.send(:configured_processes)[42] = configured_process
+
+    # The database is unreachable when the supervisor tries to fail the
+    # terminated thread's claimed jobs, like right after losing its connection
+    supervisor.expects(:release_claimed_jobs_by).raises(ActiveRecord::ConnectionNotEstablished.new("connection is closed"))
+    supervisor.expects(:start_process).with(configured_process)
+
+    assert_nothing_raised do
+      supervisor.send(:check_and_replace_terminated_processes)
+    end
+  end
+
   private
     def run_supervisor_as_thread(**options)
       SolidQueue::Supervisor.start(mode: :async, standalone: false, **options.with_defaults(skip_recurring: true))