From 2c08093d0a0c2e2553dd918b49a31757e2076b40 Mon Sep 17 00:00:00 2001 From: wintan1418 Date: Thu, 30 Jul 2026 15:27:25 +0100 Subject: [PATCH 1/2] Replace terminated forks even if releasing their claimed jobs fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the database restarts under a running fork supervisor, the workers and dispatcher crash and exit. The supervisor reaps them and calls replace_fork, which released the terminated fork's claimed jobs — a database call — before starting the replacement. With the database still down that call raised, the exception crashed the supervise loop, and the terminated forks were never replaced: the scheduler (which survives enqueue errors) kept enqueueing recurring jobs that nothing would ever process, until the whole thing was restarted by hand. Rescue errors from the release, report them through handle_thread_error and start the replacement regardless. The database being unreachable at that point is likely the very reason the fork died, so replacing it can't depend on the database being back. The terminated fork's claimed jobs aren't lost: its stale registration gets pruned once the database is reachable again, which fails them through the regular path so they can be retried. Fixes rails/solid_queue#683 --- lib/solid_queue/fork_supervisor.rb | 12 ++++++++++-- test/unit/fork_supervisor_test.rb | 23 +++++++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/lib/solid_queue/fork_supervisor.rb b/lib/solid_queue/fork_supervisor.rb index 63595b00..c3aed9ed 100644 --- a/lib/solid_queue/fork_supervisor.rb +++ b/lib/solid_queue/fork_supervisor.rb @@ -73,8 +73,16 @@ def replace_fork(pid, status) if terminated_fork = process_instances.delete(pid) terminated_fork.mark_as_reaped payload[:fork] = terminated_fork - error = Processes::ProcessExitError.new(status) - release_claimed_jobs_by(terminated_fork, with_error: error) + + begin + release_claimed_jobs_by(terminated_fork, with_error: Processes::ProcessExitError.new(status)) + rescue StandardError => error + # The database may be unreachable — likely the same reason the fork + # terminated. Starting the replacement can't depend on it: the jobs + # claimed by the terminated fork will be failed when its stale + # registration is pruned once the database is back. + handle_thread_error(error) + end start_process(configured_processes.delete(pid)) end diff --git a/test/unit/fork_supervisor_test.rb b/test/unit/fork_supervisor_test.rb index 519e4f88..9f5a4592 100644 --- a/test/unit/fork_supervisor_test.rb +++ b/test/unit/fork_supervisor_test.rb @@ -129,6 +129,29 @@ def register terminate_process(pid) end + test "replace a terminated fork even if releasing its claimed jobs fails" do + configuration = SolidQueue::Configuration.new(workers: [ { queues: "*", processes: 1 } ], dispatchers: [], skip_recurring: true) + supervisor = SolidQueue::ForkSupervisor.new(configuration) + + configured_process = configuration.configured_processes.first + terminated_fork = stub(kind: "Worker", name: "worker-42", hostname: "localhost", mark_as_reaped: true) + + supervisor.send(:process_instances)[42] = terminated_fork + supervisor.send(:configured_processes)[42] = configured_process + + # The database is unreachable when the supervisor tries to fail the + # terminated fork's claimed jobs, like right after a database restart + # that also crashed the fork + supervisor.expects(:release_claimed_jobs_by).raises(ActiveRecord::ConnectionNotEstablished.new("connection is closed")) + supervisor.expects(:start_process).with(configured_process) + + status = stub(exited?: true, exitstatus: 1, pid: 42, signaled?: false, stopsig: nil, termsig: nil) + + assert_nothing_raised do + supervisor.send(:replace_fork, 42, status) + end + end + test "fail orphaned executions" do 3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) } process = SolidQueue::Process.register(kind: "Worker", pid: 42, name: "worker-123") From a1b7cc61921c7cbf09aa5d912bbab98d6e530bcc Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Thu, 20 Aug 2026 12:57:46 +0200 Subject: [PATCH 2/2] Protect the shutdown reap from release failures too Extract the rescued release into attempt_to_release_claimed_jobs_by and use it from reap_terminated_forks as well: a database that is down when the supervisor shuts down should not abort the reap loop or skip exit hooks, for the same reason it should not stop a replacement. Co-Authored-By: Claude Fable 5 --- lib/solid_queue/fork_supervisor.rb | 23 ++++++++++++----------- test/unit/fork_supervisor_test.rb | 22 ++++++++++++++++++++++ 2 files changed, 34 insertions(+), 11 deletions(-) diff --git a/lib/solid_queue/fork_supervisor.rb b/lib/solid_queue/fork_supervisor.rb index c3aed9ed..00757afe 100644 --- a/lib/solid_queue/fork_supervisor.rb +++ b/lib/solid_queue/fork_supervisor.rb @@ -57,8 +57,7 @@ def reap_terminated_forks terminated_fork.mark_as_reaped if !status.exited? || status.exitstatus.to_i > 0 - error = Processes::ProcessExitError.new(status) - release_claimed_jobs_by(terminated_fork, with_error: error) + attempt_to_release_claimed_jobs_by(terminated_fork, status) end end @@ -74,21 +73,23 @@ def replace_fork(pid, status) terminated_fork.mark_as_reaped payload[:fork] = terminated_fork - begin - release_claimed_jobs_by(terminated_fork, with_error: Processes::ProcessExitError.new(status)) - rescue StandardError => error - # The database may be unreachable — likely the same reason the fork - # terminated. Starting the replacement can't depend on it: the jobs - # claimed by the terminated fork will be failed when its stale - # registration is pruned once the database is back. - handle_thread_error(error) - end + attempt_to_release_claimed_jobs_by(terminated_fork, 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/test/unit/fork_supervisor_test.rb b/test/unit/fork_supervisor_test.rb index 9f5a4592..0d809227 100644 --- a/test/unit/fork_supervisor_test.rb +++ b/test/unit/fork_supervisor_test.rb @@ -152,6 +152,28 @@ def register end end + test "reap a terminated fork on shutdown even if releasing its claimed jobs fails" do + configuration = SolidQueue::Configuration.new(workers: [ { queues: "*", processes: 1 } ], dispatchers: [], skip_recurring: true) + supervisor = SolidQueue::ForkSupervisor.new(configuration) + + terminated_fork = stub(kind: "Worker", name: "worker-42", hostname: "localhost", mark_as_reaped: true) + supervisor.send(:process_instances)[42] = terminated_fork + supervisor.send(:configured_processes)[42] = configuration.configured_processes.first + + # The fork was killed and the database is unreachable when the supervisor + # reaps it during shutdown + status = stub(exited?: false, exitstatus: nil, pid: 42, signaled?: true, stopsig: nil, termsig: 9) + ::Process.stubs(:waitpid2).returns([ 42, status ]).then.returns(nil) + supervisor.expects(:release_claimed_jobs_by).raises(ActiveRecord::ConnectionNotEstablished.new("connection is closed")) + + assert_nothing_raised do + supervisor.send(:reap_terminated_forks) + end + + assert_empty supervisor.send(:process_instances) + assert_empty supervisor.send(:configured_processes) + end + test "fail orphaned executions" do 3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) } process = SolidQueue::Process.register(kind: "Worker", pid: 42, name: "worker-123")