From b5aee2b5db4a31d70bb875205471cd37a7cabd84 Mon Sep 17 00:00:00 2001 From: Sapan Diwakar Date: Mon, 20 Jul 2026 07:07:00 +0200 Subject: [PATCH] supervisor: replace forks that fail to boot --- README.md | 1 + lib/solid_queue.rb | 1 + lib/solid_queue/fork_supervisor.rb | 67 +++++++++++++++++++ lib/solid_queue/log_subscriber.rb | 5 ++ lib/solid_queue/processes/runnable.rb | 3 +- lib/solid_queue/supervisor.rb | 10 ++- test/unit/fork_supervisor_test.rb | 94 +++++++++++++++++++++++++++ test/unit/log_subscriber_test.rb | 9 +++ 8 files changed, 187 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 30e4bf35c..28f3ccd10 100644 --- a/README.md +++ b/README.md @@ -402,6 +402,7 @@ There are several settings that control how Solid Queue works that you can set a - `use_skip_locked`: whether to use `FOR UPDATE SKIP LOCKED` when performing locking reads. This will be automatically detected in the future, and for now, you only need to set this to `false` if your database doesn't support it. For MySQL, that'd be versions < 8; for MariaDB, versions < 10.6; and for PostgreSQL, versions < 9.5. If you use SQLite, this has no effect, as writes are sequential. - `process_heartbeat_interval`: the heartbeat interval that all processes will follow—defaults to 60 seconds. - `process_alive_threshold`: how long to wait until a process is considered dead after its last heartbeat—defaults to 5 minutes. +- `process_startup_timeout`: how long a forked process can take to finish booting before the supervisor replaces it—defaults to 5 minutes. - `shutdown_timeout`: time the supervisor will wait since it sent the `TERM` signal to its supervised processes before sending a `QUIT` version to them requesting immediate termination—defaults to 5 seconds. - `silence_polling`: whether to silence Active Record logs emitted when polling for both workers and dispatchers—defaults to `true`. - `supervisor_pidfile`: path to a pidfile that the supervisor will create when booting to prevent running more than one supervisor in the same host, or in case you want to use it for a health check. It's `nil` by default. diff --git a/lib/solid_queue.rb b/lib/solid_queue.rb index bd2269e5f..4f3bcbe65 100644 --- a/lib/solid_queue.rb +++ b/lib/solid_queue.rb @@ -29,6 +29,7 @@ module SolidQueue mattr_accessor :process_heartbeat_interval, default: 60.seconds mattr_accessor :process_alive_threshold, default: 5.minutes + mattr_accessor :process_startup_timeout, default: 5.minutes mattr_accessor :shutdown_timeout, default: 5.seconds diff --git a/lib/solid_queue/fork_supervisor.rb b/lib/solid_queue/fork_supervisor.rb index b8bb7f3ff..3d7fa2549 100644 --- a/lib/solid_queue/fork_supervisor.rb +++ b/lib/solid_queue/fork_supervisor.rb @@ -2,8 +2,15 @@ module SolidQueue class ForkSupervisor < Supervisor + def initialize(...) + @starting_processes = {} + super + end + private + attr_reader :starting_processes + def perform_graceful_termination term_forks @@ -14,6 +21,8 @@ def perform_graceful_termination def perform_immediate_termination quit_forks + ensure + close_startup_pipes end def term_forks @@ -31,6 +40,8 @@ def check_and_replace_terminated_processes replace_fork(pid, status) end + + check_process_startups end def reap_terminated_forks @@ -43,6 +54,7 @@ def reap_terminated_forks release_claimed_jobs_by(terminated_fork, with_error: error) end + close_startup_pipe(pid) configured_processes.delete(pid) end rescue SystemCallError @@ -52,6 +64,7 @@ def reap_terminated_forks def replace_fork(pid, status) SolidQueue.instrument(:replace_fork, supervisor_pid: ::Process.pid, pid: pid, status: status) do |payload| if terminated_fork = process_instances.delete(pid) + close_startup_pipe(pid) payload[:fork] = terminated_fork error = Processes::ProcessExitError.new(status) release_claimed_jobs_by(terminated_fork, with_error: error) @@ -64,5 +77,59 @@ def replace_fork(pid, status) def all_processes_terminated? process_instances.empty? end + + def start_process(configured_process) + reader, writer = IO.pipe + inherited_readers = starting_processes.values.map { |startup| startup[:reader] } + on_fork_ready = proc do + inherited_readers.each(&:close) + reader.close + writer.write(".") + rescue Errno::EPIPE + # The supervisor stopped waiting while this process finished booting. + ensure + writer.close + end + process_id = super(configured_process, on_fork_ready:) + writer.close + starting_processes[process_id] = { reader:, started_at: monotonic_time_now } + process_id + rescue Exception + reader&.close unless reader&.closed? + writer&.close unless writer&.closed? + raise + end + + def check_process_startups + starting_processes.delete_if do |pid, startup| + reader = startup[:reader] + + # A byte means boot completed; EOF means the child exited and waitpid will replace it. + if reader.read_nonblock(1, exception: false) != :wait_readable + reader.close + true + elsif monotonic_time_now - startup[:started_at] >= SolidQueue.process_startup_timeout + SolidQueue.instrument(:fork_startup_timeout, process: process_instances[pid], pid: pid) do + # A child stuck in boot cannot reach its run loop to stop gracefully. + signal_process(pid, :KILL) + end + reader.close + true + end + end + end + + def close_startup_pipe(pid) + starting_processes.delete(pid)&.fetch(:reader)&.close + end + + def close_startup_pipes + starting_processes.each_value { |startup| startup[:reader].close } + starting_processes.clear + end + + def monotonic_time_now + ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) + end end end diff --git a/lib/solid_queue/log_subscriber.rb b/lib/solid_queue/log_subscriber.rb index 96fb19bf5..a9f482631 100644 --- a/lib/solid_queue/log_subscriber.rb +++ b/lib/solid_queue/log_subscriber.rb @@ -161,6 +161,11 @@ def replace_fork(event) end end + def fork_startup_timeout(event) + process = event.payload[:process] + warn formatted_event(event, action: "Terminate unresponsive #{process.kind} during startup", **event.payload.slice(:pid).merge(hostname: process.hostname, name: process.name)) + end + private def formatted_event(event, action:, **attributes) "SolidQueue-#{SolidQueue::VERSION} #{action} (#{event.duration.round(1)}ms) #{formatted_attributes(**attributes)}" diff --git a/lib/solid_queue/processes/runnable.rb b/lib/solid_queue/processes/runnable.rb index c6e002e4f..964eadb5e 100644 --- a/lib/solid_queue/processes/runnable.rb +++ b/lib/solid_queue/processes/runnable.rb @@ -6,9 +6,10 @@ module Runnable attr_writer :mode - def start + def start(on_fork_ready: nil) run_in_mode do boot + on_fork_ready&.call if running_as_fork? run end end diff --git a/lib/solid_queue/supervisor.rb b/lib/solid_queue/supervisor.rb index 17b90c544..4c1984fdb 100644 --- a/lib/solid_queue/supervisor.rb +++ b/lib/solid_queue/supervisor.rb @@ -86,16 +86,22 @@ def supervise shutdown end - def start_process(configured_process) + def start_process(configured_process, on_fork_ready: nil) process_instance = configured_process.instantiate.tap do |instance| instance.supervised_by process instance.mode = mode end - process_id = process_instance.start + process_id = if on_fork_ready + process_instance.start(on_fork_ready:) + else + process_instance.start + end configured_processes[process_id] = configured_process process_instances[process_id] = process_instance + + process_id end def check_and_replace_terminated_processes diff --git a/test/unit/fork_supervisor_test.rb b/test/unit/fork_supervisor_test.rb index 9ec81b510..25efb960f 100644 --- a/test/unit/fork_supervisor_test.rb +++ b/test/unit/fork_supervisor_test.rb @@ -1,17 +1,39 @@ require "test_helper" class ForkSupervisorTest < ActiveSupport::TestCase + class StalledWorker < SolidQueue::Worker + def initialize(startup_log:, startup_delay:, **options) + @startup_log = startup_log + @startup_delay = startup_delay + super(**options) + end + + private + attr_reader :startup_log, :startup_delay + + def register + File.open(startup_log, "a") { |file| file.puts(::Process.pid) } + sleep startup_delay + super + end + end + self.use_transactional_tests = false setup do @previous_pidfile = SolidQueue.supervisor_pidfile + @previous_process_startup_timeout = SolidQueue.process_startup_timeout @pidfile = Rails.application.root.join("tmp/pids/pidfile_#{SecureRandom.hex}.pid") SolidQueue.supervisor_pidfile = @pidfile + @startup_log = Rails.application.root.join("tmp/pids/startup_#{SecureRandom.hex}.log") end teardown do + terminate_stalled_supervisor SolidQueue.supervisor_pidfile = @previous_pidfile + SolidQueue.process_startup_timeout = @previous_process_startup_timeout File.delete(@pidfile) if File.exist?(@pidfile) + File.delete(@startup_log) if File.exist?(@startup_log) end test "start" do @@ -206,6 +228,32 @@ class ForkSupervisorTest < ActiveSupport::TestCase assert_equal "RuntimeError", failed.exception_class end + test "replace only the fork that does not finish booting" do + SolidQueue.process_startup_timeout = 0.2.seconds + run_stalled_supervisor(startup_delay: 60.seconds, with_healthy_worker: true) + wait_for_registered_processes(2, timeout: 2.seconds) + healthy_pid = find_processes_registered_as("Worker").sole.pid + + wait_while_with_timeout(4.seconds) { startup_pids.size < 2 } + + assert_not process_exists?(startup_pids.first) + assert process_exists?(startup_pids.second) + assert_equal healthy_pid, find_processes_registered_as("Worker").sole.pid + end + + test "preserve a fork that finishes booting before the timeout" do + SolidQueue.process_startup_timeout = 0.5.seconds + run_stalled_supervisor(startup_delay: 0.1.seconds) + wait_for_registered_processes(2, timeout: 2.seconds) + worker_pid = find_processes_registered_as("StalledWorker").sole.pid + + sleep 1.1.seconds + + assert_equal [ worker_pid ], startup_pids + assert process_exists?(worker_pid) + assert_equal worker_pid, find_processes_registered_as("StalledWorker").sole.pid + end + private def assert_registered_workers(supervisor_pid: nil, count: 1) assert_registered_processes(kind: "Worker", count: count, supervisor_pid: supervisor_pid) @@ -223,4 +271,50 @@ def assert_registered_supervisor(pid) assert_equal pid, processes.first.pid end end + + def run_stalled_supervisor(startup_delay:, with_healthy_worker: false) + configured_process_class = Struct.new(:process_class, :attributes) do + def instantiate + process_class.new(**attributes) + end + end + configured_processes = [ + configured_process_class.new(StalledWorker, { startup_log: @startup_log.to_s, startup_delay: }) + ] + configured_processes << configured_process_class.new(SolidQueue::Worker, {}) if with_healthy_worker + configuration = Struct.new(:configured_processes, :mode) do + def standalone? + true + end + end.new(configured_processes, "fork".inquiry) + + @stalled_supervisor_pid = fork do + ::Process.setpgrp + SolidQueue::ForkSupervisor.new(configuration).start + end + end + + def startup_pids + File.readlines(@startup_log).map(&:to_i) + rescue Errno::ENOENT + [] + end + + def terminate_stalled_supervisor + return unless @stalled_supervisor_pid + + begin + ::Process.kill(:KILL, -@stalled_supervisor_pid) + rescue Errno::ESRCH + begin + ::Process.kill(:KILL, @stalled_supervisor_pid) + rescue Errno::ESRCH + end + end + + begin + ::Process.waitpid(@stalled_supervisor_pid) + rescue Errno::ECHILD + end + end end diff --git a/test/unit/log_subscriber_test.rb b/test/unit/log_subscriber_test.rb index aabdd15a0..774cee627 100644 --- a/test/unit/log_subscriber_test.rb +++ b/test/unit/log_subscriber_test.rb @@ -50,6 +50,15 @@ def set_logger(logger) assert_match_logged :error, "Error enqueuing recurring task", "task: :example_task, enqueue_error: \"Everything is broken\", at: \"#{time.iso8601}\"" end + test "fork startup timeout" do + worker = SolidQueue::Worker.new + + attach_log_subscriber + instrument "fork_startup_timeout.solid_queue", process: worker, pid: 42 + + assert_match_logged :warn, "Terminate unresponsive Worker during startup", "pid: 42, hostname: \"#{worker.hostname}\", name: \"#{worker.name}\"" + end + test "deregister process" do process = SolidQueue::Process.register(kind: "Worker", pid: 42, hostname: "localhost", name: "worker-123") last_heartbeat_at = process.last_heartbeat_at.iso8601