Stop processes whose heartbeats keep failing past the alive threshold

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

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 } ])

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) }