From ad264b6ab81ce68c4cca9fc4c42dab2dc9a1db4f Mon Sep 17 00:00:00 2001 From: wintan1418 Date: Thu, 30 Jul 2026 13:47:20 +0100 Subject: [PATCH] Stop processes whose heartbeats keep failing past the alive threshold MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A process that finds its registration gone already stops itself so the supervisor can replace it (RecordNotFound on heartbeat). But heartbeats can also fail without confirming anything about the registration — a dropped database connection after the host sleeps and resumes, SQLite busy errors — and those errors were only reported, leaving the process running unregistered indefinitely: a zombie that polls or schedules nothing anyone can see, while the supervisor never replaces it because it never exits. Once heartbeats have been failing for longer than the alive threshold, every other process will have considered this one dead and pruned its registration anyway, so treat it the same as finding the registration gone: stop, and let the supervisor start a replacement that can register afresh. Also add a regression test covering the existing replacement path for a scheduler whose registration is pruned while it's still alive. Related to rails/solid_queue#763 --- lib/solid_queue/processes/registrable.rb | 19 +++++++++++++++ test/unit/process_recovery_test.rb | 30 ++++++++++++++++++++++++ test/unit/worker_test.rb | 20 ++++++++++++++++ 3 files changed, 69 insertions(+) diff --git a/lib/solid_queue/processes/registrable.rb b/lib/solid_queue/processes/registrable.rb index 35b4e01bd..add149626 100644 --- a/lib/solid_queue/processes/registrable.rb +++ b/lib/solid_queue/processes/registrable.rb @@ -55,9 +55,28 @@ def stop_heartbeat def heartbeat process&.heartbeat + @heartbeats_failing_since = nil rescue ActiveRecord::RecordNotFound self.process = nil wake_up + rescue => error + # Errors other than a missing registration (e.g. the DB connection dropping) + # can prevent the heartbeat from going through, and even from finding out + # whether the registration is still there. If this persists past the alive + # threshold, other processes will have considered this one dead and pruned + # its registration, so stop and let the supervisor replace it with a process + # that can register afresh, rather than running unregistered indefinitely. + @heartbeats_failing_since ||= Time.current + if heartbeats_failing_for_too_long? + self.process = nil + wake_up + end + + raise error + end + + def heartbeats_failing_for_too_long? + @heartbeats_failing_since && Time.current - @heartbeats_failing_since > SolidQueue.process_alive_threshold end def reload_metadata diff --git a/test/unit/process_recovery_test.rb b/test/unit/process_recovery_test.rb index 296d6b958..4b40e6fad 100644 --- a/test/unit/process_recovery_test.rb +++ b/test/unit/process_recovery_test.rb @@ -15,6 +15,36 @@ class ProcessRecoveryTest < ActiveSupport::TestCase JobResult.delete_all end + test "alive scheduler whose registration is pruned is torn down and replaced" do + old_heartbeat_interval, SolidQueue.process_heartbeat_interval = SolidQueue.process_heartbeat_interval, 1.second + + @pid = run_supervisor_as_fork(skip_recurring: false) + wait_for_registered_processes(5, timeout: 3.seconds) # supervisor + 2 workers + dispatcher + scheduler + + scheduler_process = SolidQueue::Process.find_by(kind: "Scheduler") + assert scheduler_process.present? + + # Simulate another process's prune sweep removing the scheduler's registration + # while the scheduler itself is still alive + scheduler_process.delete + + # The scheduler should notice its registration is gone on its next heartbeat, + # terminate, and be replaced by the supervisor with a fresh registration + wait_while_with_timeout(10.seconds) do + skip_active_record_query_cache do + SolidQueue::Process.where(kind: "Scheduler").where.not(id: scheduler_process.id).none? + end + end + + skip_active_record_query_cache do + new_scheduler_process = SolidQueue::Process.where(kind: "Scheduler").last + assert new_scheduler_process.present? + assert_not_equal scheduler_process.id, new_scheduler_process.id + end + ensure + SolidQueue.process_heartbeat_interval = old_heartbeat_interval + end + test "supervisor handles missing process record and fails claimed executions properly" do # Start a supervisor with one worker @pid = run_supervisor_as_fork(workers: [ { queues: "*", polling_interval: 0.1, processes: 1 } ]) diff --git a/test/unit/worker_test.rb b/test/unit/worker_test.rb index c21cd5741..7dec98c44 100644 --- a/test/unit/worker_test.rb +++ b/test/unit/worker_test.rb @@ -213,6 +213,26 @@ class WorkerTest < ActiveSupport::TestCase SolidQueue.process_heartbeat_interval = old_heartbeat_interval end + test "terminate when heartbeats have been failing 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 + + SolidQueue::Process.any_instance.stubs(:heartbeat).raises(ActiveRecord::StatementInvalid.new("connection lost")) + + @worker.start + wait_for_registered_processes(1, timeout: 1.second) + + assert_not @worker.pool.shutdown? + + # Heartbeats keep failing without the registration being confirmed gone, so + # the worker should give up once the failures outlast the alive threshold + wait_while_with_timeout(3) { !@worker.pool.shutdown? } + assert @worker.pool.shutdown? + ensure + 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) }