From b99502fa246426a5ecce8ae0abd66792d3818737 Mon Sep 17 00:00:00 2001 From: Sapan Diwakar Date: Mon, 20 Jul 2026 07:07:00 +0200 Subject: [PATCH 1/3] 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 02614847a..91530be50 100644 --- a/README.md +++ b/README.md @@ -404,6 +404,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 eb77df689..2c796d4e1 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 7738391cf..dda3048e0 100644 --- a/lib/solid_queue/processes/runnable.rb +++ b/lib/solid_queue/processes/runnable.rb @@ -8,9 +8,10 @@ def mode=(value) @mode = (value || DEFAULT_MODE).to_s.inquiry end - 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 From cd51ba0f74bbe614467f41dba21d46c55f245c16 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Wed, 29 Jul 2026 11:20:05 +0200 Subject: [PATCH 2/3] Terminate the stalled supervisor gracefully in tests The tests killed the whole process group to clean up, but the supervisor under test is actively replacing its stalled forks, and a group KILL races that forking: a fork spawned concurrently with the kill can escape it, wake up from its stalled boot long after the test has finished, register itself and start polling, holding the test database's write lock and cascading lock timeouts through unrelated tests, as seen on CI with SQLite. Terminating the supervisor gracefully instead lets it clean up its own forks: it stops replacing them first, and a fork stuck in boot is covered by the QUIT escalation, as its signal handlers are installed before the boot callbacks where it stalls. This also lets go of the replacement fork's liveness assertion: the replacement stalls too, so whether it's alive depends on where its own boot timeout is. Co-Authored-By: Claude Opus 4.8 --- test/unit/fork_supervisor_test.rb | 31 ++++++++++++++++--------------- 1 file changed, 16 insertions(+), 15 deletions(-) diff --git a/test/unit/fork_supervisor_test.rb b/test/unit/fork_supervisor_test.rb index 25efb960f..0a31fc2ca 100644 --- a/test/unit/fork_supervisor_test.rb +++ b/test/unit/fork_supervisor_test.rb @@ -236,8 +236,10 @@ def register wait_while_with_timeout(4.seconds) { startup_pids.size < 2 } + # The stalled fork was terminated and replaced by a new one; the healthy + # worker was left alone. The replacement's liveness is not asserted because + # it stalls too and will be replaced in turn. 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 @@ -289,7 +291,6 @@ def standalone? end.new(configured_processes, "fork".inquiry) @stalled_supervisor_pid = fork do - ::Process.setpgrp SolidQueue::ForkSupervisor.new(configuration).start end end @@ -300,21 +301,21 @@ def startup_pids [] end + # Terminate the supervisor gracefully so it cleans up its own forks first, + # including any it's still starting: killing it abruptly instead can leak a + # fork spawned concurrently with the kill, which would keep writing to the + # database long after this test has finished. def terminate_stalled_supervisor - return unless @stalled_supervisor_pid - - begin - ::Process.kill(:KILL, -@stalled_supervisor_pid) + terminate_process(@stalled_supervisor_pid) if @stalled_supervisor_pid && process_exists?(@stalled_supervisor_pid) + rescue Timeout::Error + # The supervisor didn't terminate in time and got killed without a chance + # to clean up its forks: kill any forks it left behind—stalled ones, so + # they can't wake up from their stalled boot and write to the database + # while other tests run, and healthy ones, so they don't keep polling. + orphaned_pids = startup_pids + SolidQueue::Process.where(kind: "Worker").pluck(:pid) + orphaned_pids.uniq.each do |pid| + ::Process.kill(:KILL, 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 From 3504c4e7310d8dc22f00f5541bd7013d2f3797f7 Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Wed, 29 Jul 2026 11:20:21 +0200 Subject: [PATCH 3/3] Track process startups with boot guards the processes themselves manage Move the startup-tracking machinery out of ForkSupervisor into boot guards with a common interface, created by each process when it starts, according to how it runs: a forked process pipes its boot completion to the supervisor through a BootGuards::ForkGuard, created before forking so it survives it, while processes running as threads of their supervisor, or inline, use a BootGuards::NullGuard, which completes on boot without any monitoring. Booting itself completes the guard, so starting a process remains just boot and run. ForkSupervisor checks its forks' boot guards while checking for terminated processes to replace, terminating any fork whose boot has timed out so that it's reaped and replaced as usual, and closing each guard together with the rest of the fork's cleanup, asking the process instances themselves, like the async supervisor does when it checks whether they're alive. The configurable timeout becomes fork_boot_timeout, as it only applies to forked processes: supervised processes running as threads aren't monitored while booting, and the supervisor itself has no parent within Solid Queue that could enforce a timeout on it. The instrumentation event follows suit. Also reuse Timer's monotonic clock instead of redefining it. Co-Authored-By: Claude Opus 4.8 --- README.md | 2 +- lib/solid_queue.rb | 2 +- lib/solid_queue/fork_supervisor.rb | 91 +++++++-------------------- lib/solid_queue/log_subscriber.rb | 4 +- lib/solid_queue/processes/runnable.rb | 90 +++++++++++++++++++++++++- lib/solid_queue/supervisor.rb | 10 +-- lib/solid_queue/timer.rb | 7 +-- test/unit/boot_guards_test.rb | 67 ++++++++++++++++++++ test/unit/fork_supervisor_test.rb | 8 +-- test/unit/log_subscriber_test.rb | 6 +- 10 files changed, 192 insertions(+), 95 deletions(-) create mode 100644 test/unit/boot_guards_test.rb diff --git a/README.md b/README.md index 91530be50..1a7533c1c 100644 --- a/README.md +++ b/README.md @@ -404,7 +404,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. +- `fork_boot_timeout`: how long a forked process can take to finish booting before the supervisor replaces it—defaults to 5 minutes. It only applies in the default `fork` mode. - `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 2c796d4e1..654a6be63 100644 --- a/lib/solid_queue.rb +++ b/lib/solid_queue.rb @@ -29,7 +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 :fork_boot_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 3d7fa2549..63595b00d 100644 --- a/lib/solid_queue/fork_supervisor.rb +++ b/lib/solid_queue/fork_supervisor.rb @@ -2,15 +2,8 @@ module SolidQueue class ForkSupervisor < Supervisor - def initialize(...) - @starting_processes = {} - super - end - private - attr_reader :starting_processes - def perform_graceful_termination term_forks @@ -21,8 +14,6 @@ def perform_graceful_termination def perform_immediate_termination quit_forks - ensure - close_startup_pipes end def term_forks @@ -41,7 +32,20 @@ def check_and_replace_terminated_processes replace_fork(pid, status) end - check_process_startups + check_boot_timeouts + end + + def check_boot_timeouts + process_instances.each do |pid, instance| + terminate_unready_process(pid) if instance.boot_timed_out? + end + end + + def terminate_unready_process(pid) + SolidQueue.instrument(:fork_boot_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 end def reap_terminated_forks @@ -49,12 +53,15 @@ def reap_terminated_forks pid, status = ::Process.waitpid2(-1, ::Process::WNOHANG) break unless pid - if (terminated_fork = process_instances.delete(pid)) && (!status.exited? || status.exitstatus.to_i > 0) - error = Processes::ProcessExitError.new(status) - release_claimed_jobs_by(terminated_fork, with_error: error) + if terminated_fork = process_instances.delete(pid) + 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) + end end - close_startup_pipe(pid) configured_processes.delete(pid) end rescue SystemCallError @@ -64,7 +71,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) + terminated_fork.mark_as_reaped payload[:fork] = terminated_fork error = Processes::ProcessExitError.new(status) release_claimed_jobs_by(terminated_fork, with_error: error) @@ -77,59 +84,5 @@ 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 a9f482631..bb6b67da7 100644 --- a/lib/solid_queue/log_subscriber.rb +++ b/lib/solid_queue/log_subscriber.rb @@ -161,9 +161,9 @@ def replace_fork(event) end end - def fork_startup_timeout(event) + def fork_boot_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)) + warn formatted_event(event, action: "Terminate #{process.kind} that failed to boot in time", **event.payload.slice(:pid).merge(hostname: process.hostname, name: process.name)) end private diff --git a/lib/solid_queue/processes/runnable.rb b/lib/solid_queue/processes/runnable.rb index dda3048e0..cd89a2ea8 100644 --- a/lib/solid_queue/processes/runnable.rb +++ b/lib/solid_queue/processes/runnable.rb @@ -8,10 +8,9 @@ def mode=(value) @mode = (value || DEFAULT_MODE).to_s.inquiry end - def start(on_fork_ready: nil) + def start run_in_mode do boot - on_fork_ready&.call if running_as_fork? run end end @@ -32,6 +31,14 @@ def alive? !running_async? || @thread&.alive? end + def boot_timed_out? + @boot_guard.timed_out? + end + + def mark_as_reaped + @boot_guard.close + end + private DEFAULT_MODE = :async @@ -42,11 +49,14 @@ def mode def run_in_mode(&block) case when running_as_fork? - fork(&block) + @boot_guard = BootGuards::ForkGuard.new + fork(&block).tap { @boot_guard.start } when running_async? + @boot_guard = BootGuards::NullGuard.new @thread = create_thread(&block) @thread.object_id else + @boot_guard = BootGuards::NullGuard.new block.call end end @@ -60,6 +70,8 @@ def boot end end end + + @boot_guard.complete end def shutting_down? @@ -96,4 +108,76 @@ def running_as_fork? mode.fork? end end + + module BootGuards + # Tracks a process that shares memory with its supervisor, whose boot time + # doesn't need monitoring. + class NullGuard + def complete + @completed = true + end + + def start + end + + def completed? + @completed + end + + def timed_out? + false + end + + def close + end + end + + # Tracks a forked process from the moment it's started until its boot + # callbacks finish, over a pipe that survives forking: the forked process + # writes to it when it's done booting, and its supervisor reads from it to + # decide whether the process is taking too long to boot and needs replacing. + class ForkGuard + def initialize + @reader, @writer = IO.pipe + @created_at = SolidQueue::Timer.monotonic_time_now + end + + # Runs in the forked process when it has finished booting + def complete + reader.close + writer.write(".") + rescue Errno::EPIPE + # The supervisor stopped waiting while this process finished booting + ensure + writer.close + end + + # Runs in the parent process right after forking + def start + writer.close + end + + # A byte means boot completed; EOF means the process exited before + # finishing its boot, and will be replaced when it's reaped + def completed? + @completed ||= begin + completed = reader.read_nonblock(1, exception: false) != :wait_readable + reader.close if completed + completed + end + end + + def timed_out? + !completed? && SolidQueue::Timer.monotonic_time_now - created_at >= SolidQueue.fork_boot_timeout + end + + def close + reader.close unless reader.closed? + writer.close unless writer.closed? + end + + private + attr_reader :reader, :writer, :created_at + end + end end diff --git a/lib/solid_queue/supervisor.rb b/lib/solid_queue/supervisor.rb index 4c1984fdb..17b90c544 100644 --- a/lib/solid_queue/supervisor.rb +++ b/lib/solid_queue/supervisor.rb @@ -86,22 +86,16 @@ def supervise shutdown end - def start_process(configured_process, on_fork_ready: nil) + def start_process(configured_process) process_instance = configured_process.instantiate.tap do |instance| instance.supervised_by process instance.mode = mode end - process_id = if on_fork_ready - process_instance.start(on_fork_ready:) - else - process_instance.start - end + process_id = process_instance.start configured_processes[process_id] = configured_process process_instances[process_id] = process_instance - - process_id end def check_and_replace_terminated_processes diff --git a/lib/solid_queue/timer.rb b/lib/solid_queue/timer.rb index 19e691f53..a6c813bf9 100644 --- a/lib/solid_queue/timer.rb +++ b/lib/solid_queue/timer.rb @@ -20,9 +20,8 @@ def wait_until(timeout, condition) end end - private - def monotonic_time_now - ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) - end + def monotonic_time_now + ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) + end end end diff --git a/test/unit/boot_guards_test.rb b/test/unit/boot_guards_test.rb new file mode 100644 index 000000000..198c7e52e --- /dev/null +++ b/test/unit/boot_guards_test.rb @@ -0,0 +1,67 @@ +require "test_helper" + +class BootGuardsTest < ActiveSupport::TestCase + setup do + @previous_fork_boot_timeout = SolidQueue.fork_boot_timeout + SolidQueue.fork_boot_timeout = 0.1.seconds + end + + teardown do + SolidQueue.fork_boot_timeout = @previous_fork_boot_timeout + @guard&.close + end + + test "a fork guard times out when boot doesn't complete within the configured timeout" do + @guard = SolidQueue::Processes::BootGuards::ForkGuard.new + + assert_not @guard.completed? + assert_not @guard.timed_out? + + sleep SolidQueue.fork_boot_timeout + + assert_not @guard.completed? + assert @guard.timed_out? + end + + test "a fork guard completed from the forked process doesn't time out" do + @guard = SolidQueue::Processes::BootGuards::ForkGuard.new + + pid = fork do + @guard.complete + exit!(0) + end + @guard.start + Process.waitpid(pid) + + assert @guard.completed? + + sleep SolidQueue.fork_boot_timeout + + assert @guard.completed? + assert_not @guard.timed_out? + end + + test "a fork guard reads as completed when the forked process exits before finishing its boot" do + @guard = SolidQueue::Processes::BootGuards::ForkGuard.new + + pid = fork { exit!(0) } + @guard.start + Process.waitpid(pid) + + # EOF on the pipe: the process will be replaced when it's reaped + assert @guard.completed? + assert_not @guard.timed_out? + end + + test "a null guard completes on boot and never times out" do + @guard = SolidQueue::Processes::BootGuards::NullGuard.new + + assert_not @guard.completed? + assert_not @guard.timed_out? + + @guard.complete + + assert @guard.completed? + assert_not @guard.timed_out? + end +end diff --git a/test/unit/fork_supervisor_test.rb b/test/unit/fork_supervisor_test.rb index 0a31fc2ca..519e4f881 100644 --- a/test/unit/fork_supervisor_test.rb +++ b/test/unit/fork_supervisor_test.rb @@ -22,7 +22,7 @@ def register setup do @previous_pidfile = SolidQueue.supervisor_pidfile - @previous_process_startup_timeout = SolidQueue.process_startup_timeout + @previous_fork_boot_timeout = SolidQueue.fork_boot_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") @@ -31,7 +31,7 @@ def register teardown do terminate_stalled_supervisor SolidQueue.supervisor_pidfile = @previous_pidfile - SolidQueue.process_startup_timeout = @previous_process_startup_timeout + SolidQueue.fork_boot_timeout = @previous_fork_boot_timeout File.delete(@pidfile) if File.exist?(@pidfile) File.delete(@startup_log) if File.exist?(@startup_log) end @@ -229,7 +229,7 @@ def register end test "replace only the fork that does not finish booting" do - SolidQueue.process_startup_timeout = 0.2.seconds + SolidQueue.fork_boot_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 @@ -244,7 +244,7 @@ def register end test "preserve a fork that finishes booting before the timeout" do - SolidQueue.process_startup_timeout = 0.5.seconds + SolidQueue.fork_boot_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 diff --git a/test/unit/log_subscriber_test.rb b/test/unit/log_subscriber_test.rb index 774cee627..9e2399232 100644 --- a/test/unit/log_subscriber_test.rb +++ b/test/unit/log_subscriber_test.rb @@ -50,13 +50,13 @@ 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 + test "fork boot timeout" do worker = SolidQueue::Worker.new attach_log_subscriber - instrument "fork_startup_timeout.solid_queue", process: worker, pid: 42 + instrument "fork_boot_timeout.solid_queue", process: worker, pid: 42 - assert_match_logged :warn, "Terminate unresponsive Worker during startup", "pid: 42, hostname: \"#{worker.hostname}\", name: \"#{worker.name}\"" + assert_match_logged :warn, "Terminate Worker that failed to boot in time", "pid: 42, hostname: \"#{worker.hostname}\", name: \"#{worker.name}\"" end test "deregister process" do