WorkerPool loses wakeup when the waiting fiber re-enters the pool

ext/io/event/worker_pool_test.c

@@ -18,6 +18,7 @@ struct BusyOperationData {
   int read_fd;
   int write_fd;
   volatile int cancelled;
+  volatile int executed;
   double duration;  // How long to wait (for testing)
   clock_t start_time;
   clock_t end_time;
@@ -29,6 +30,9 @@ struct BusyOperationData {
 static void* busy_blocking_operation(void *data) {
 	struct BusyOperationData *busy_data = (struct BusyOperationData*)data;
 
+	// Record that the operation actually ran (a cancelled operation may be skipped entirely):
+	busy_data->executed = 1;
+
 	// Use select() to wait for the pipe to become readable
 	fd_set read_fds;
 	struct timeval timeout;
@@ -143,6 +147,7 @@ static VALUE worker_pool_test_busy(int argc, VALUE *argv, VALUE self) {
 		.read_fd = pipe_fds[0],
 		.write_fd = pipe_fds[1],
 		.cancelled = 0,
+		.executed = 0,
 		.duration = duration,
 		.start_time = 0,
 		.end_time = 0,
@@ -165,6 +170,7 @@ static VALUE worker_pool_test_busy(int argc, VALUE *argv, VALUE self) {
 	VALUE result = rb_hash_new();
 	rb_hash_aset(result, ID2SYM(rb_intern("duration")), DBL2NUM(duration));
 	rb_hash_aset(result, ID2SYM(rb_intern("elapsed")), DBL2NUM(elapsed));
+	rb_hash_aset(result, ID2SYM(rb_intern("executed")), busy_data.executed ? Qtrue : Qfalse);
 
 	// Determine result based on operation outcome
 	if (busy_data.exception != Qnil) {

test/io/event/worker_pool.rb

@@ -324,4 +324,69 @@ def block(...)
 			expect(rescued).to be == true
 		end
 	end
+
+	with "scheduler that re-enters the worker pool from block" do
+		# Ruby can run GC finalizers inside the scheduler's `block` hook, before the fiber suspends. If a finalizer performs an offloaded blocking operation (e.g. `File.unlink` from a `Tempfile` finalizer), the same fiber calls `WorkerPool#call` again while its outer wait is still in progress. This scheduler simulates that deterministically by performing one nested blocking operation from inside the first worker pool `block`.
+		class ReentrantBlockScheduler < IO::Event::TestScheduler
+			attr_reader :nested_result
+
+			def block(blocker, timeout = nil)
+				# Simulate a finalizer that performs an offloaded blocking operation, exactly once:
+				if blocker.is_a?(IO::Event::WorkerPool) and !@reentered
+					@reentered = true
+					@nested_result = IO::Event::WorkerPool.busy(duration: 0.01)
+				end
+
+				super
+			end
+		end
+
+		class WaitTimeout < StandardError
+		end
+
+		it "wakes the outer operation and runs the nested operation" do
+			result = nil
+			nested_result = nil
+
+			Thread.new do
+				# A single worker guarantees that the nested operation queues behind the outer one:
+				scheduler = ReentrantBlockScheduler.new(maximum_worker_count: 1)
+				Fiber.set_scheduler(scheduler)
+
+				fiber = Fiber.schedule do
+					result = IO::Event::WorkerPool.busy(duration: 0.1)
+				end
+
+				# Bound the outer wait so a lost wakeup fails the test instead of hanging it:
+				Fiber.schedule do
+					deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 2.0
+
+					while result.nil? and Process.clock_gettime(Process::CLOCK_MONOTONIC) < deadline
+						sleep(0.01)
+					end
+
+					Fiber.scheduler.fiber_interrupt(fiber, WaitTimeout) if result.nil?
+				end
+
+				scheduler.run
+				nested_result = scheduler.nested_result
+			ensure
+				Fiber.set_scheduler(nil)
+				scheduler&.close
+			end.join
+
+			# The nested operation should run to completion, rather than being cancelled when the outer operation's wakeup resumes the nested wait (depending on timing, the cancelled operation is either interrupted or skipped entirely):
+			expect(nested_result).to have_keys(
+				executed: be == true,
+				result: be == :completed
+			)
+
+			# The outer operation should complete without being interrupted by the timeout:
+			expect(result[:exception]).to be_nil
+			expect(result).to have_keys(
+				executed: be == true,
+				result: be == :completed
+			)
+		end
+	end
 end