Replace forked processes that fail to boot

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.
+- `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.

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 :fork_boot_timeout, default: 5.minutes
 
   mattr_accessor :shutdown_timeout, default: 5.seconds
 

lib/solid_queue/fork_supervisor.rb

@@ -31,16 +31,35 @@ def check_and_replace_terminated_processes
 
         replace_fork(pid, status)
       end
+
+      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
       loop do
         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
 
         configured_processes.delete(pid)
@@ -52,6 +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)
+          terminated_fork.mark_as_reaped
           payload[:fork] = terminated_fork
           error = Processes::ProcessExitError.new(status)
           release_claimed_jobs_by(terminated_fork, with_error: error)

lib/solid_queue/log_subscriber.rb

@@ -161,6 +161,11 @@ def replace_fork(event)
     end
   end
 
+  def fork_boot_timeout(event)
+    process = event.payload[:process]
+    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
     def formatted_event(event, action:, **attributes)
       "SolidQueue-#{SolidQueue::VERSION} #{action} (#{event.duration.round(1)}ms)  #{formatted_attributes(**attributes)}"

lib/solid_queue/processes/runnable.rb

@@ -31,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
 
@@ -41,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
@@ -59,6 +70,8 @@ def boot
             end
           end
         end
+
+        @boot_guard.complete
       end
 
       def shutting_down?
@@ -95,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

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

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

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_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")
   end
 
   teardown do
+    terminate_stalled_supervisor
     SolidQueue.supervisor_pidfile = @previous_pidfile
+    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
 
   test "start" do
@@ -206,6 +228,34 @@ class ForkSupervisorTest < ActiveSupport::TestCase
     assert_equal "RuntimeError", failed.exception_class
   end
 
+  test "replace only the fork that does not finish booting" do
+    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
+
+    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_equal healthy_pid, find_processes_registered_as("Worker").sole.pid
+  end
+
+  test "preserve a fork that finishes booting before the timeout" do
+    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
+
+    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 +273,49 @@ 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
+        SolidQueue::ForkSupervisor.new(configuration).start
+      end
+    end
+
+    def startup_pids
+      File.readlines(@startup_log).map(&:to_i)
+    rescue Errno::ENOENT
+      []
+    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
+      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
+      end
+    end
 end