From 3a923d9dc0b5b26bb788e6aa4a6fabd3c8ee8d5e Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Sat, 22 Aug 2026 13:44:39 +0200 Subject: [PATCH] Replace terminated threads even if releasing their claimed jobs fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The async supervisor had the same hole #781 closed for forks: a worker thread that dies when the database is unreachable — likely the very reason it died — took the whole supervise loop down with it when replace_thread tried to fail its claimed jobs before starting the replacement. Hoist the rescued release into the Maintenance concern both supervisors share, and let callers pass the error, since forks and threads terminate with different ones. As with forks, the terminated thread's claimed jobs aren't lost by skipping the release: its stale registration is pruned once the database is reachable again, and pruning fails its claimed executions. Co-Authored-By: Claude Fable 5 --- lib/solid_queue/async_supervisor.rb | 3 +-- lib/solid_queue/fork_supervisor.rb | 14 ++------------ lib/solid_queue/supervisor/maintenance.rb | 10 ++++++++++ test/unit/async_supervisor_test.rb | 20 ++++++++++++++++++++ 4 files changed, 33 insertions(+), 14 deletions(-) diff --git a/lib/solid_queue/async_supervisor.rb b/lib/solid_queue/async_supervisor.rb index 4a7f4aeec..f6ab38342 100644 --- a/lib/solid_queue/async_supervisor.rb +++ b/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 diff --git a/lib/solid_queue/fork_supervisor.rb b/lib/solid_queue/fork_supervisor.rb index 00757afe2..0acc36e7d 100644 --- a/lib/solid_queue/fork_supervisor.rb +++ b/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 diff --git a/lib/solid_queue/supervisor/maintenance.rb b/lib/solid_queue/supervisor/maintenance.rb index d92569d58..23e321f47 100644 --- a/lib/solid_queue/supervisor/maintenance.rb +++ b/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 diff --git a/test/unit/async_supervisor_test.rb b/test/unit/async_supervisor_test.rb index 0ca6a3c2f..f2a70f0ff 100644 --- a/test/unit/async_supervisor_test.rb +++ b/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))