From 6e11a479eb11082a9669b1c6bbe1f207a09b1ec7 Mon Sep 17 00:00:00 2001 From: wintan1418 Date: Thu, 30 Jul 2026 15:27:25 +0100 Subject: [PATCH] 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")