Stop Redis server lifecycle tasks

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

@@ -34,6 +34,7 @@ def initialize(client, key)
 
 						@add = @client.script(:load, ADD)
 						@move = @client.script(:load, MOVE)
+						@task = nil
 					end
 
 					# @returns [Integer] The number of jobs currently in the delayed queue.
@@ -45,9 +46,16 @@ def size
 					# @parameter ready_list [ReadyList] The ready list to move jobs to.
 					# @parameter resolution [Integer] The check interval in seconds.
 					# @parameter parent [Async::Task] The parent task to run the background loop in.
-					# @returns [Async::Task] The background processing task.
+					# @returns [Async::Task | false] The background processing task, or false if already started.
 					def start(ready_list, resolution: 10, parent: Async::Task.current)
-						parent.async do
+						return false if @task
+
+						# Reserve ownership before spawning because Async tasks may begin eagerly.
+						@task = true
+
+						task = parent.async do |task|
+							@task = task
+
 							while true
 								count = move(destination: ready_list.key)
 
@@ -57,7 +65,26 @@ def start(ready_list, resolution: 10, parent: Async::Task.current)
 
 								sleep(resolution)
 							end
+						ensure
+							@task = nil if @task.equal?(task)
 						end
+
+						# A non-greedy parent may return before the child assigns its handle.
+						@task = task if @task == true
+						return task
+					rescue
+						@task = nil
+						raise
+					end
+
+					# Stop the owned delayed job promotion task.
+					def stop
+						task = @task
+						@task = nil
+						return unless task.respond_to?(:stop)
+
+						task.stop
+						task.wait if task.alive?
 					end
 
 					# @attribute [String] The Redis key for this delayed jobs queue.

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

@@ -75,6 +75,7 @@ def initialize(client, key, id, ready_list, job_store)
 						@complete = @client.script(:load, COMPLETE)
 
 						@complete_count = 0
+						@task = nil
 					end
 
 					# @attribute [String] The base Redis key for this processing list.
@@ -133,11 +134,17 @@ def requeue(start_time, delay, factor)
 					# @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.
+					# @returns [Async::Task | false] The background processing task, or false if already started.
 					def start(delay: 5, factor: 2, parent: Async::Task.current)
+						return false if @task
+
+						# Reserve ownership before spawning because Async tasks may begin eagerly.
+						@task = true
 						start_time = Time.now.to_f
 
-						parent.async do |task|
+						task = parent.async do |task|
+							@task = task
+
 							while true
 								task.defer_stop do
 									count = self.requeue(start_time, delay, factor)
@@ -149,7 +156,26 @@ def start(delay: 5, factor: 2, parent: Async::Task.current)
 
 								sleep(delay)
 							end
+						ensure
+							@task = nil if @task.equal?(task)
 						end
+
+						# A non-greedy parent may return before the child assigns its handle.
+						@task = task if @task == true
+						return task
+					rescue
+						@task = nil
+						raise
+					end
+
+					# Stop the owned heartbeat and abandoned job recovery task.
+					def stop
+						task = @task
+						@task = nil
+						return unless task.respond_to?(:stop)
+
+						task.stop
+						task.wait if task.alive?
 					end
 				end
 			end

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

@@ -80,7 +80,11 @@ def start
 
 					# Stop the server and all background processing tasks.
 					def stop
+						# Stop workers before their heartbeat protection. Pending jobs remain in Redis
+						# and become recoverable when this server's heartbeat expires naturally.
 						@task&.stop
+						@processing_list.stop
+						@delayed_jobs.stop
 
 						super
 					end

test/async/job/processor/delayed_jobs.rb

@@ -125,7 +125,7 @@
 			remaining_score = client.zscore(delayed_jobs.key, job_id)
 			expect(remaining_score).to be_nil
 		ensure
-			task&.stop
+			delayed_jobs.stop
 		end
 
 		it "logs debug messages when moving jobs" do
@@ -147,6 +147,23 @@
 				severity: be == :debug,
 				message: be(:include?, "Moved 1 delayed jobs to ready list")
 			)
+		ensure
+			delayed_jobs.stop
+		end
+
+		it "owns a single background task" do
+			task = delayed_jobs.start(ready_list, resolution: 1)
+
+			expect(task.alive?).to be == true
+			expect(delayed_jobs.start(ready_list, resolution: 1)).to be == false
+
+			delayed_jobs.stop
+			expect(task.finished?).to be == true
+
+			restarted_task = delayed_jobs.start(ready_list, resolution: 1)
+			expect(restarted_task).not.to be == false
+		ensure
+			delayed_jobs.stop
 		end
 	end
 end

test/async/job/processor/processing_list.rb

@@ -154,7 +154,22 @@
 			fetched_job = processing_list.fetch
 			expect(fetched_job).to be == abandoned_job_id
 		ensure
-			task&.stop
+			processing_list.stop
+		end
+
+		it "owns a single background task" do
+			task = processing_list.start(delay: 1)
+
+			expect(task.alive?).to be == true
+			expect(processing_list.start(delay: 1)).to be == false
+
+			processing_list.stop
+			expect(task.finished?).to be == true
+
+			restarted_task = processing_list.start(delay: 1)
+			expect(restarted_task).not.to be == false
+		ensure
+			processing_list.stop
 		end
 	end
 end

test/async/job/processor/server.rb

@@ -84,4 +84,31 @@
 			expect(server.status_string).to be == "R=0 D=0 P=0/1"
 		end
 	end
+
+	with "#stop" do
+		it "stops every server-owned task" do
+			dispatcher_task = server.instance_variable_get(:@task)
+			processing_list = server.instance_variable_get(:@processing_list)
+			processing_task = processing_list.instance_variable_get(:@task)
+			delayed_jobs = server.instance_variable_get(:@delayed_jobs)
+			delayed_task = delayed_jobs.instance_variable_get(:@task)
+
+			server.stop
+
+			expect(dispatcher_task.finished?).to be == true
+			expect(processing_task.finished?).to be == true
+			expect(delayed_task.finished?).to be == true
+
+			# Shutdown remains safe when both the owner and its parent call it.
+			server.stop
+		end
+
+		it "can restart after stopping all server-owned tasks" do
+			server.stop
+			server.start
+			server.call(job)
+
+			expect(buffer.pop).to be == job
+		end
+	end
 end