Move processing heartbeats off the reactor

lib/async/job/processor/redis/processing_list.rb

@@ -3,6 +3,8 @@
 # Released under the MIT License.
 # Copyright, 2024-2025, by Samuel Williams.
 
+require "kernel/sync"
+
 module Async
 	module Job
 		module Processor
@@ -75,6 +77,10 @@ def initialize(client, key, id, ready_list, job_store)
 						@complete = @client.script(:load, COMPLETE)
 
 						@complete_count = 0
+						@task = nil
+						@mutex = Mutex.new
+						@condition = ConditionVariable.new
+						@stop_requested = false
 					end
 
 					# @attribute [String] The base Redis key for this processing list.
@@ -129,26 +135,83 @@ def requeue(start_time, delay, factor)
 						return count
 					end
 
-					# Start the background heartbeat and abandoned job recovery task.
+					# Start the background heartbeat and abandoned job recovery thread.
 					# @parameter delay [Integer] The heartbeat update interval in seconds.
 					# @parameter factor [Integer] The heartbeat expiration factor.
-					# @parameter parent [Async::Task] The parent task to run the background loop in.
-					# @returns [Async::Task] The background processing task.
-					def start(delay: 5, factor: 2, parent: Async::Task.current)
+					# @returns [Thread | false] The background processing thread, or `false` if already started.
+					def start(delay: 5, factor: 2)
+						@mutex.synchronize do
+							return false if @task
+
+							@stop_requested = false
+							@task = Thread.new do
+								Thread.current.report_on_exception = false
+								Thread.current.name = "#{self.class.name}:#{@id}" if Thread.current.respond_to?(:name=)
+
+								run_heartbeat_loop(delay: delay, factor: factor)
+							rescue => error
+								Console.error(self, "Heartbeat loop failed!", exception: error)
+								raise
+							ensure
+								@mutex.synchronize do
+									@task = nil if @task.equal?(Thread.current)
+								end
+							end
+						end
+					end
+
+					# Stop the background heartbeat and abandoned job recovery thread.
+					def stop
+						task = @mutex.synchronize do
+							@stop_requested = true
+							@condition.broadcast
+
+							@task
+						end
+
+						task&.join unless task == Thread.current
+					end
+
+					private
+
+					def run_heartbeat_loop(delay:, factor:)
 						start_time = Time.now.to_f
+						client = Async::Redis::Client.new(@client.endpoint)
+						requeue = Sync do
+							client.script(:load, REQUEUE)
+						end
 
-						parent.async do |task|
-							while true
-								task.defer_stop do
-									count = self.requeue(start_time, delay, factor)
-
-									if count > 0
-										Console.warn(self, "Requeued #{count} abandoned jobs.")
-									end
-								end
-
-								sleep(delay)
+						loop do
+							count = Sync do
+								requeue_with(client, requeue, start_time, delay, factor)
 							end
+
+							if count > 0
+								Console.warn(self, "Requeued #{count} abandoned jobs.")
+							end
+
+							break if wait_for_stop(delay)
+						end
+					ensure
+						Sync do
+							client&.close
+						end
+					end
+
+					def requeue_with(client, requeue, start_time, delay, factor)
+						uptime = (Time.now.to_f - start_time).round(2)
+						expiry = (delay*factor).ceil
+						client.set(@heartbeat_key, JSON.dump(uptime: uptime), seconds: expiry)
+						client.evalsha(requeue, 2, @key, @ready_list.key)
+					end
+
+					def wait_for_stop(delay)
+						@mutex.synchronize do
+							return true if @stop_requested
+
+							@condition.wait(@mutex, delay)
+
+							@stop_requested
 						end
 					end
 				end

lib/async/job/processor/redis/server.rb

@@ -28,15 +28,19 @@ class Server < Generic
 					# @parameter prefix [String] The Redis key prefix for job data.
 					# @parameter coder [Async::Job::Coder] The job serialization codec.
 					# @parameter resolution [Integer] The resolution in seconds for delayed job processing.
+					# @parameter heartbeat_interval [Integer] The interval in seconds between processing heartbeats.
+					# @parameter heartbeat_factor [Integer] The factor used to calculate heartbeat expiry.
 					# @parameter parent [Async::Task] The parent task for background processing.
-					def initialize(delegate, client, prefix: "async-job", coder: Coder::DEFAULT, resolution: 10, parent: nil)
+					def initialize(delegate, client, prefix: "async-job", coder: Coder::DEFAULT, resolution: 10, heartbeat_interval: 5, heartbeat_factor: 2, parent: nil)
 						super(delegate)
 
 						@id = SecureRandom.uuid
 						@client = client
 						@prefix = prefix
 						@coder = coder
 						@resolution = resolution
+						@heartbeat_interval = heartbeat_interval
+						@heartbeat_factor = heartbeat_factor
 
 						@job_store = JobStore.new(@client, "#{@prefix}:jobs")
 						@delayed_jobs = DelayedJobs.new(@client, "#{@prefix}:delayed")
@@ -73,14 +77,15 @@ def start
 						@delayed_jobs.start(@ready_list, resolution: @resolution)
 
 						# Start the processing processor, which will move jobs to the ready processor when they are abandoned:
-						@processing_list.start
+						@processing_list.start(delay: @heartbeat_interval, factor: @heartbeat_factor)
 
 						self.start!
 					end
 
 					# Stop the server and all background processing tasks.
 					def stop
 						@task&.stop
+						@processing_list.stop
 
 						super
 					end

test/async/job/processor/processing_list.rb

@@ -150,11 +150,12 @@
 			client.del("#{prefix}:processing:#{dead_server_id}")
 
 			task = processing_list.start(delay: 0.1, factor: 2)
+			expect(task).to be_a(Thread)
 
 			fetched_job = processing_list.fetch
 			expect(fetched_job).to be == abandoned_job_id
 		ensure
-			task&.stop
+			processing_list.stop
 		end
 	end
 end

test/async/job/processor/server.rb

@@ -80,6 +80,12 @@
 			expect(server.status_string).to be == "R=0 D=0 P=0/0"
 
 			server.call(job)
+			buffer.pop
+
+			deadline = Time.now + 0.5
+			while server.status_string != "R=0 D=0 P=0/1" && Time.now < deadline
+				sleep(0.01)
+			end
 
 			expect(server.status_string).to be == "R=0 D=0 P=0/1"
 		end