From fcf8cb406ed0b3584ee1ccacc76d5dd82498455e Mon Sep 17 00:00:00 2001
From: Shaokun Wu <shaokun.wu@gmail.com>
Date: Mon, 8 Dec 2025 13:16:33 +0800
Subject: [PATCH] Support passing Async::Semaphore as the parent task for
 Async::Job::Processor::Redis::Server

---
 lib/async/job/processor/redis/server.rb | 26 +++++-----
 test/async/job/processor/server.rb      | 63 +++++++++++++++++++++++++
 2 files changed, 76 insertions(+), 13 deletions(-)

diff --git a/lib/async/job/processor/redis/server.rb b/lib/async/job/processor/redis/server.rb
index d3ead22..30f9691 100644
--- a/lib/async/job/processor/redis/server.rb
+++ b/lib/async/job/processor/redis/server.rb
@@ -53,11 +53,13 @@ def start!
 						
 						@task = true
 						
-						@parent.async(transient: true, annotation: self.class.name) do |task|
+						Async do |task|
 							@task = task
 							
 							while true
-								self.dequeue(task)
+								@parent.async(transient: true, annotation: self.class.name) do
+									self.dequeue
+								end
 							end
 						ensure
 							@task = nil
@@ -120,19 +122,17 @@ def call(job)
 					# If the job fails for any reason, it will be retried.
 					#
 					# If you do not desire this behavior, you should catch exceptions in the delegate.
-					def dequeue(parent)
+					def dequeue
 						_id = @processing_list.fetch
 						
-						parent.async do
-							id = _id; _id = nil
-							
-							job = @coder.load(@job_store.get(id))
-							@delegate.call(job)
-							@processing_list.complete(id)
-						rescue => error
-							Console.error(self, "Job failed with error!", id: id, exception: error)
-							@processing_list.retry(id)
-						end
+						id = _id; _id = nil
+						
+						job = @coder.load(@job_store.get(id))
+						@delegate.call(job)
+						@processing_list.complete(id)
+					rescue => error
+						Console.error(self, "Job failed with error!", id: id, exception: error)
+						@processing_list.retry(id)
 					ensure
 						@processing_list.retry(_id) if _id
 					end
diff --git a/test/async/job/processor/server.rb b/test/async/job/processor/server.rb
index a56075b..b2728b4 100644
--- a/test/async/job/processor/server.rb
+++ b/test/async/job/processor/server.rb
@@ -5,6 +5,7 @@
 
 require "async"
 require "async/redis"
+require "async/semaphore"
 
 require "sus/fixtures/async/reactor_context"
 require "sus/fixtures/console"
@@ -80,8 +81,70 @@
 			expect(server.status_string).to be == "R=0 D=0 P=0/0"
 			
 			server.call(job)
+			sleep 0.1 # Allow some time for the job to be processed.
 			
 			expect(server.status_string).to be == "R=0 D=0 P=0/1"
 		end
 	end
+	
+	with "concurrency limit" do
+		# Delegate that sleeps for 5 seconds to simulate slow job processing
+		let(:slow_delegate) do
+			Class.new do
+				def start
+				end
+				
+				def stop
+				end
+				
+				def call(job)
+					sleep 5
+				end
+			end.new
+		end
+		
+		with "Async::Idler" do
+			let(:idler_server) {subject.new(slow_delegate, prefix:, resolution: 1)}
+			
+			it "can process all jobs concurrently" do
+				idler_server.start
+				
+				# Enqueue 10 jobs
+				10.times do |i|
+					idler_server.call({"data" => "job #{i}"})
+				end
+				
+				# Give time for all jobs to be picked up
+				sleep 0.5
+				
+				# With Async::Idler (unlimited concurrency), all 10 jobs should be in processing status
+				status = idler_server.status_string
+				expect(status).to be =~ /P=(10|[5-9])\//
+				
+				idler_server.stop
+			end
+		end
+		
+		with "Async::Semaphore" do
+			let(:semaphore_server) {subject.new(slow_delegate, prefix:, resolution: 1, parent: Async::Semaphore.new(2))}
+			
+			it "can limit concurrent job processing to 2" do
+				semaphore_server.start
+				
+				# Enqueue 10 jobs
+				10.times do |i|
+					semaphore_server.call({"data" => "job #{i}"})
+				end
+				
+				# Give time for jobs to be picked up
+				sleep 0.5
+				
+				# Only 2 jobs should be in processing status
+				status = semaphore_server.status_string
+				expect(status).to be =~ /P=2\//
+				
+				semaphore_server.stop
+			end
+		end
+	end
 end