Stop processes and supervisors whose timers stop returning

lib/solid_queue/processes/registrable.rb

@@ -38,6 +38,8 @@ def registered?
       end
 
       def launch_heartbeat
+        @last_heartbeat_returned_at = Concurrent::AtomicReference.new(SolidQueue::Timer.monotonic_time_now)
+
         @heartbeat_task = Concurrent::TimerTask.new(execution_interval: SolidQueue.process_heartbeat_interval) do
           wrap_in_app_executor { heartbeat }
         end
@@ -47,10 +49,40 @@ def launch_heartbeat
         end
 
         @heartbeat_task.execute
+
+        launch_heartbeat_watchdog
+      end
+
+      # A heartbeat that raises is handled below, but one that never returns is not.
+      # Concurrent::TimerTask reschedules a task and notifies its observers only once
+      # the task has completed, so a heartbeat blocked on an unresponsive database --
+      # a half-open socket to a connection pooler that stopped serving, say -- stalls
+      # the heartbeat thread silently and indefinitely, and `presumed_dead?` is never
+      # reached because nothing was raised.
+      #
+      # This watchdog reads nothing but memory, so it cannot block the same way.
+      def launch_heartbeat_watchdog
+        @heartbeat_watchdog_task = Concurrent::TimerTask.new(execution_interval: SolidQueue.process_heartbeat_interval) do
+          stop_to_be_replaced if heartbeat_stalled?
+        end
+
+        @heartbeat_watchdog_task.add_observer do |_, _, error|
+          handle_thread_error(error) if error
+        end
+
+        @heartbeat_watchdog_task.execute
       end
 
       def stop_heartbeat
         @heartbeat_task&.shutdown
+        @heartbeat_watchdog_task&.shutdown
+      end
+
+      # Whether the heartbeat has stopped returning altogether. Note this asks only
+      # whether the call came back, not whether it succeeded: a heartbeat that keeps
+      # raising is `presumed_dead?`'s business.
+      def heartbeat_stalled?
+        SolidQueue::Timer.monotonic_time_now - @last_heartbeat_returned_at.get > SolidQueue.process_alive_threshold
       end
 
       def heartbeat
@@ -64,6 +96,8 @@ def heartbeat
         # registration is still there
         stop_to_be_replaced if presumed_dead?
         raise error
+      ensure
+        @last_heartbeat_returned_at&.set(SolidQueue::Timer.monotonic_time_now)
       end
 
       # Whether this process's registration is prunable: if the last heartbeat that

lib/solid_queue/supervisor/maintenance.rb

@@ -6,8 +6,14 @@ module Supervisor::Maintenance
       after_boot :fail_orphaned_executions
     end
 
+    # How far past its own interval a maintenance run is allowed to go before we
+    # treat it as stalled rather than slow. One full missed cycle of slack.
+    MAINTENANCE_STALL_FACTOR = 2
+
     private
       def launch_maintenance_task
+        @last_maintenance_returned_at = Concurrent::AtomicReference.new(SolidQueue::Timer.monotonic_time_now)
+
         @maintenance_task = Concurrent::TimerTask.new(run_now: true, execution_interval: SolidQueue.process_alive_threshold) do
           prune_dead_processes
         end
@@ -17,14 +23,46 @@ def launch_maintenance_task
         end
 
         @maintenance_task.execute
+
+        launch_maintenance_watchdog
+      end
+
+      # Pruning is how a supervisor notices dead processes, and it runs in a
+      # Concurrent::TimerTask, which reschedules only once its task returns. A
+      # prune blocked on an unresponsive database therefore stops this supervisor
+      # pruning ever again, silently -- the mechanism meant to notice dead
+      # processes is built from the same material as the processes it watches.
+      #
+      # A supervisor cannot replace itself, so stop instead, and leave it to
+      # whatever runs this supervisor to start a new one.
+      def launch_maintenance_watchdog
+        @maintenance_watchdog_task = Concurrent::TimerTask.new(execution_interval: SolidQueue.process_heartbeat_interval) do
+          stop if maintenance_stalled?
+        end
+
+        @maintenance_watchdog_task.add_observer do |_, _, error|
+          handle_thread_error(error) if error
+        end
+
+        @maintenance_watchdog_task.execute
       end
 
       def stop_maintenance_task
         @maintenance_task&.shutdown
+        @maintenance_watchdog_task&.shutdown
+      end
+
+      # Whether maintenance has stopped returning altogether, as opposed to
+      # failing: a prune that raises is reported by the task's observer.
+      def maintenance_stalled?
+        SolidQueue::Timer.monotonic_time_now - @last_maintenance_returned_at.get >
+          MAINTENANCE_STALL_FACTOR * SolidQueue.process_alive_threshold
       end
 
       def prune_dead_processes
         wrap_in_app_executor { SolidQueue::Process.prune(excluding: process) }
+      ensure
+        @last_maintenance_returned_at&.set(SolidQueue::Timer.monotonic_time_now)
       end
 
       def fail_orphaned_executions

test/unit/async_supervisor_test.rb

@@ -15,6 +15,31 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
     assert_no_registered_processes
   end
 
+  test "stop when maintenance stops returning for longer than the stall threshold" do
+    old_alive_threshold, SolidQueue.process_alive_threshold = SolidQueue.process_alive_threshold, 0.2.seconds
+    old_heartbeat_interval, SolidQueue.process_heartbeat_interval = SolidQueue.process_heartbeat_interval, 0.2.seconds
+
+    # A prune that never returns, so the maintenance TimerTask is never
+    # rescheduled and this supervisor would silently stop pruning forever
+    SolidQueue::Supervisor::Maintenance.module_eval do
+      alias_method :prune_dead_processes_without_blocking, :prune_dead_processes
+      define_method(:prune_dead_processes) { sleep 30 }
+    end
+
+    supervisor = run_supervisor_as_thread
+    wait_while_with_timeout(5) { !supervisor.send(:stopped?) }
+
+    assert supervisor.send(:stopped?)
+  ensure
+    SolidQueue::Supervisor::Maintenance.module_eval do
+      remove_method :prune_dead_processes
+      alias_method :prune_dead_processes, :prune_dead_processes_without_blocking
+      remove_method :prune_dead_processes_without_blocking
+    end
+    SolidQueue.process_alive_threshold = old_alive_threshold
+    SolidQueue.process_heartbeat_interval = old_heartbeat_interval
+  end
+
   test "start standalone" do
     pid = run_supervisor_as_fork(mode: :async)
     wait_for_registered_processes(4, timeout: 5.seconds) # supervisor + dispatcher + 2 workers

test/unit/worker_test.rb

@@ -271,6 +271,34 @@ class WorkerTest < ActiveSupport::TestCase
     SolidQueue.process_alive_threshold = old_alive_threshold
   end
 
+  test "terminate when heartbeats stop returning for longer than the alive threshold" do
+    old_heartbeat_interval, SolidQueue.process_heartbeat_interval = SolidQueue.process_heartbeat_interval, 0.2.seconds
+    old_alive_threshold, SolidQueue.process_alive_threshold = SolidQueue.process_alive_threshold, 0.5.seconds
+
+    # Unlike a heartbeat that raises, this one never comes back at all, so the
+    # heartbeat TimerTask is never rescheduled and never notifies its observers
+    SolidQueue::Process.class_eval do
+      alias_method :heartbeat_without_blocking, :heartbeat
+      define_method(:heartbeat) { sleep 30 }
+    end
+
+    @worker.start
+    wait_for_registered_processes(1, timeout: 1.second)
+
+    assert_not @worker.pool.shutdown?
+
+    wait_while_with_timeout(3) { !@worker.pool.shutdown? }
+    assert @worker.pool.shutdown?
+  ensure
+    SolidQueue::Process.class_eval do
+      remove_method :heartbeat
+      alias_method :heartbeat, :heartbeat_without_blocking
+      remove_method :heartbeat_without_blocking
+    end
+    SolidQueue.process_heartbeat_interval = old_heartbeat_interval
+    SolidQueue.process_alive_threshold = old_alive_threshold
+  end
+
   test "sleeps `10.minutes` if at capacity" do
     3.times { |i| StoreResultJob.perform_later(i, pause: 5.seconds) }