diff --git a/lib/solid_queue/processes/registrable.rb b/lib/solid_queue/processes/registrable.rb index 08b2750c6..c6faf4c80 100644 --- a/lib/solid_queue/processes/registrable.rb +++ b/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 diff --git a/lib/solid_queue/supervisor/maintenance.rb b/lib/solid_queue/supervisor/maintenance.rb index 23e321f47..ba4b81312 100644 --- a/lib/solid_queue/supervisor/maintenance.rb +++ b/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 diff --git a/test/unit/async_supervisor_test.rb b/test/unit/async_supervisor_test.rb index f2a70f0ff..41754ad81 100644 --- a/test/unit/async_supervisor_test.rb +++ b/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 diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index 77f41edf5..d773d6947 100644 --- a/test/unit/worker_test.rb +++ b/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) }