From 16b66e19a38fb43351bc5b58ba420fdc7851c998 Mon Sep 17 00:00:00 2001 From: Julik Tarkhanov Date: Thu, 24 Sep 2026 15:33:28 +0200 Subject: [PATCH 1/2] Stop processes whose heartbeats stop returning #778 stops a process whose heartbeats keep failing, but both of its paths hang off a raise. A heartbeat can also never return at all: Process#heartbeat does a real round-trip, and on a half-open socket to a database that stopped answering, the thread blocks in the driver indefinitely. Nothing is raised, so presumed_dead? is never reached. Concurrent::TimerTask does not help either, because it reschedules the task and notifies its observers only after the task returns - so the heartbeat thread goes quiet permanently, having neither succeeded nor failed. Its timeout_interval is a no-op that warns it was never implementable. So record when the heartbeat last returned, in an ensure so that success and failure both count, and have a second timer stop the process once that goes older than the alive threshold. The watchdog reads nothing but memory, so it cannot block the way the heartbeat it watches can. This only covers the case where the run loop is still able to act on being unregistered. A process whose run loop is itself blocked needs something harsher, which I have deliberately left out of this change. --- lib/solid_queue/processes/registrable.rb | 34 ++++++++++++++++++++++++ test/unit/worker_test.rb | 28 +++++++++++++++++++ 2 files changed, 62 insertions(+) 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/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) } From 09b7ef515dfbc91e37ef268ae9840799e1820df6 Mon Sep 17 00:00:00 2001 From: Julik Tarkhanov Date: Thu, 24 Sep 2026 15:41:53 +0200 Subject: [PATCH 2/2] Stop a supervisor whose maintenance stops returning Same defect as the heartbeat watchdog, one level up. Pruning is how supervisors notice dead processes, and launch_maintenance_task runs it in a Concurrent::TimerTask, so a prune blocked on an unresponsive database stops this supervisor pruning ever again - the mechanism meant to notice dead processes is built from the same material as the processes it watches, and fails at exactly the moment they do. Track when maintenance last returned and stop the supervisor once that goes past MAINTENANCE_STALL_FACTOR times the alive threshold, which leaves a full missed cycle of slack since the task's own interval is the alive threshold. Stopping rather than arranging replacement, because a supervisor cannot replace itself - its run loop breaks on stopped?, so whatever runs it gets to start a new one. Supervisors need this separately from the heartbeat watchdog in Registrable: stop_to_be_replaced only unregisters and wakes the loop, and unlike Runnable#shutting_down?, Supervisor#supervise never checks registered?. --- lib/solid_queue/supervisor/maintenance.rb | 38 +++++++++++++++++++++++ test/unit/async_supervisor_test.rb | 25 +++++++++++++++ 2 files changed, 63 insertions(+) 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