Fix memory leak in PriorityQueue waiter management

Add waiter invalidation to prevent abandoned waiters from accumulating
in the @waiting heap when exceptions occur during dequeue operations.

Changes:
- Add Waiter#invalidate! method that nullifies fiber and condition
- Add Waiter#valid? method to check if waiter is still usable
- Add ensure block in dequeue to invalidate waiters on exceptions
- Update all waiter processing to skip invalid waiters (close, enqueue, enqueue_all)
- Add comprehensive tests for waiter invalidation scenarios

This fixes a memory leak where waiters would remain in @waiting indefinitely
if exceptions occurred during wait_for_value, causing both memory growth
and incorrect waiting counts.

The fix uses lazy cleanup - invalid waiters are skipped during normal
operation rather than expensive O(n) deletion, making it safe for
concurrent access patterns.

Author: samuel-williams-shopify

lib/async/priority_queue.rb

@@ -43,6 +43,17 @@ def wait_for_value(mutex)
 				condition.wait(mutex)
 				return self.value
 			end
+			
+			# Invalidate this waiter, making it unusable and detectable as abandoned.
+			def invalidate!
+				self.fiber = nil
+				self.condition = nil
+			end
+			
+			# Check if this waiter has been invalidated.
+			def valid?
+				!fiber.nil? && !condition.nil?
+			end
 		end
 
 		# Create a new priority queue.
@@ -64,12 +75,12 @@ def close
 			@mutex.synchronize do
 				@closed = true
 
-				# Signal all waiting fibers with nil, skipping dead ones:
+				# Signal all waiting fibers with nil, skipping dead/invalid ones:
 				while waiter = @waiting.pop
-					if waiter.fiber.alive?
+					if waiter.valid? && waiter.fiber.alive?
 						waiter.signal(nil)
 					end
-					# Dead waiter discarded, continue to next one.
+					# Dead/invalid waiter discarded, continue to next one.
 				end
 			end
 		end
@@ -105,14 +116,14 @@ def push(item)
 
 				@items << item
 
-				# Wake up the highest priority waiter if any, skipping dead waiters:
+				# Wake up the highest priority waiter if any, skipping dead/invalid waiters:
 				while waiter = @waiting.pop
-					if waiter.fiber.alive?
+					if waiter.valid? && waiter.fiber.alive?
 						value = @items.shift
 						waiter.signal(value)
 						break
 					end
-					# Dead waiter discarded, try next one.
+					# Dead/invalid waiter discarded, try next one.
 				end
 			end
 		end
@@ -133,13 +144,13 @@ def enqueue(*items)
 
 				@items.concat(items)
 
-				# Wake up waiting fibers in priority order, skipping dead waiters:
+				# Wake up waiting fibers in priority order, skipping dead/invalid waiters:
 				while !@items.empty? && (waiter = @waiting.pop)
-					if waiter.fiber.alive?
+					if waiter.valid? && waiter.fiber.alive?
 						value = @items.shift
 						waiter.signal(value)
 					end
-					# Dead waiter discarded, continue to next one.
+					# Dead/invalid waiter discarded, continue to next one.
 				end
 			end
 		end
@@ -175,9 +186,14 @@ def dequeue(priority: 0)
 				waiter = Waiter.new(Fiber.current, priority, sequence, condition, nil)
 				@waiting.push(waiter)
 
-				# Wait for our specific condition variable to be signaled:
-				# The mutex is released during wait, reacquired after:
-				return waiter.wait_for_value(@mutex)
+				begin
+					# Wait for our specific condition variable to be signaled:
+					# The mutex is released during wait, reacquired after:
+					return waiter.wait_for_value(@mutex)
+				ensure
+					# Invalidate waiter if exception occurs - prevents memory leaks
+					waiter.invalidate!
+				end
 			end
 		end
 

test/async/priority_queue.rb

@@ -547,4 +547,91 @@
 			]
 		end
 	end
+	
+	with "waiter invalidation" do
+		it "should invalidate waiters when tasks are stopped to prevent memory leaks" do
+			# Start a task that will wait and then be stopped
+			task = reactor.async do
+				queue.dequeue(priority: 1)
+			end
+			
+			# Give the task time to start waiting
+			sleep(0.01)
+			expect(queue.waiting).to be == 1
+			
+			# Stop the task (simulates exception)
+			task.stop
+			
+			# Give time for cleanup
+			sleep(0.01)
+			
+			# Now enqueue an item - should not try to wake the invalid waiter
+			queue.enqueue("test_item")
+			
+			# The item should still be available for a new waiter
+			result = nil
+			new_task = reactor.async do
+				result = queue.dequeue
+			end
+			
+			new_task.wait
+			expect(result).to be == "test_item"
+		end
+		
+		it "should skip invalid waiters during enqueue" do
+			received_items = []
+			
+			# Start multiple waiters
+			tasks = []
+			3.times do |i|
+				tasks << reactor.async do
+					item = queue.dequeue(priority: i)
+					received_items << [i, item]
+				end
+			end
+			
+			# Give tasks time to start waiting
+			sleep(0.01)
+			expect(queue.waiting).to be == 3
+			
+			# Stop the middle priority task (priority 1)
+			tasks[1].stop
+			
+			# Give time for invalidation
+			sleep(0.01)
+			
+			# Add items to the queue
+			queue.enqueue("item1", "item2")
+			
+			# Give time for processing
+			sleep(0.01)
+			
+			# Should have received items in the valid waiters only
+			# Invalid waiter (priority 1) should be skipped
+			expect(received_items.size).to be == 2
+			
+			# Items should go to highest priority waiters (2, then 0)
+			priorities_served = received_items.map(&:first).sort.reverse
+			expect(priorities_served).to be == [2, 0]
+		end
+	end
+	
+	describe "Waiter" do
+		it "should invalidate correctly" do
+			condition = ConditionVariable.new
+			fiber = Fiber.current
+			waiter = Async::PriorityQueue::Waiter.new(fiber, 1, 1, condition, nil)
+			
+			expect(waiter.valid?).to be == true
+			expect(waiter.fiber).to be == fiber
+			expect(waiter.condition).to be == condition
+			
+			waiter.invalidate!
+			
+			expect(waiter.valid?).to be == false
+			expect(waiter.fiber).to be_nil
+			expect(waiter.condition).to be_nil
+			# value can be whatever - we don't care
+		end
+	end
 end