Replace forked processes that fail to boot

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.

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
 

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

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)}"

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

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

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

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