repro: add TCPSocket.open ENOTCONN stress test with io_wait instrumentation

Add DEBUG_IO_WAIT logging to io_wait_transfer so every io_uring poll
completion is emitted to stderr; lines with flags outside the requested
mask are tagged UNEXPECTED.

Add test/stress/tcp_connect.rb: hammers TCPSocket.open from N concurrent
fibers under the URing backend and reports any ENOTCONN errors, exiting 1
if any are observed.

See bug.md for the full investigation context.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: samuel-williams-shopify

ext/io/event/selector/uring.c

@@ -22,6 +22,9 @@ enum {
 	DEBUG = 0,
 	DEBUG_COMPLETION = 0,
 	DEBUG_CQE = 0,
+	// Log all io_wait poll completions during the TCPSocket.open ENOTCONN repro.
+	// Emits a line per completion so `grep UNEXPECTED` isolates surprising flags.
+	DEBUG_IO_WAIT = 1,
 };
 
 enum {URING_ENTRIES = 64};
@@ -627,7 +630,23 @@ VALUE io_wait_transfer(VALUE _arguments) {
 	} else if (result > 0) {
 		// We explicitly filter the resulting events based on the requested events.
 		// In some cases, poll will report events we didn't ask for.
-		return RB_INT2NUM(events_from_poll_flags(arguments->waiting->result & arguments->flags));
+		int translated = events_from_poll_flags(result);
+		int returned   = events_from_poll_flags(result & arguments->flags);
+		
+		if (DEBUG_IO_WAIT) {
+			short unexpected = result & ~arguments->flags;
+			// Log all completions; mark ones that carry flags outside the requested mask.
+			fprintf(stderr,
+				"io_wait: result=%#06x flags=%#06x translated=%#x returned=%#x%s\n",
+				(unsigned)result,
+				(unsigned)arguments->flags,
+				(unsigned)translated,
+				(unsigned)returned,
+				unexpected ? " UNEXPECTED" : ""
+			);
+		}
+		
+		return RB_INT2NUM(returned);
 	} else {
 		return Qfalse;
 	}

test/stress/tcp_connect.rb

@@ -0,0 +1,117 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+# Stress test for Errno::ENOTCONN on TCPSocket.open with the io_uring backend.
+#
+# When built with DEBUG_IO_WAIT=1 in uring.c, each io_wait poll completion is
+# logged to stderr. Grep for "UNEXPECTED" to find completions whose raw flags
+# contained bits outside the originally-requested poll mask.
+#
+# Usage (on Linux with liburing):
+#   ruby test/stress/tcp_connect.rb [CONCURRENCY [TOTAL]]
+#
+# Exits 0 on clean run, 1 if any ENOTCONN was observed.
+
+require "io/event"
+require "io/event/test_scheduler"
+require "socket"
+
+unless IO::Event::Selector.const_defined?(:URing)
+	abort "ERROR: URing backend not available. Run this on Linux with liburing installed."
+end
+
+CONCURRENCY = (ARGV[0] || 100).to_i
+TOTAL       = (ARGV[1] || 5_000).to_i
+PER_FIBER   = (TOTAL.to_f / CONCURRENCY).ceil
+
+$stderr.puts "io-event stress test: TCPSocket.open ENOTCONN repro"
+$stderr.puts "  Backend   : #{IO::Event::Selector::URing}"
+$stderr.puts "  Concurrency: #{CONCURRENCY} fibers"
+$stderr.puts "  Total      : #{TOTAL} connections (#{PER_FIBER} per fiber)"
+$stderr.puts
+
+# ── acceptor ────────────────────────────────────────────────────────────────
+
+server = TCPServer.new("127.0.0.1", 0)
+port   = server.addr[1]
+$stderr.puts "Server listening on 127.0.0.1:#{port}"
+
+# Accept connections in a background thread so the scheduler fibers are never
+# blocked waiting for an accept. Each accepted socket is closed immediately so
+# the connect side sees a clean EOF rather than a hung connection.
+acceptor_thread = Thread.new do
+	loop do
+		begin
+			conn = server.accept_nonblock
+			conn.close
+		rescue IO::WaitReadable
+			IO.select([server])
+			retry
+		rescue IOError, Errno::EBADF
+			break  # server was closed
+		rescue => e
+			$stderr.puts "[acceptor] #{e.class}: #{e.message}"
+		end
+	end
+end
+
+# ── scheduler ───────────────────────────────────────────────────────────────
+
+counters = Hash.new(0)
+mutex    = Mutex.new
+
+selector  = IO::Event::Selector::URing.new(Fiber.current)
+scheduler = IO::Event::TestScheduler.new(selector: selector)
+
+Fiber.set_scheduler(scheduler)
+
+# ── worker fibers ────────────────────────────────────────────────────────────
+
+CONCURRENCY.times do
+	Fiber.schedule do
+		PER_FIBER.times do
+			begin
+				sock = TCPSocket.new("127.0.0.1", port)
+				sock.close
+				mutex.synchronize { counters[:ok] += 1 }
+			rescue Errno::ENOTCONN => e
+				mutex.synchronize { counters[:ENOTCONN] += 1 }
+				$stderr.puts "[ENOTCONN] #{e.message}"
+			rescue Errno::ECONNREFUSED
+				mutex.synchronize { counters[:ECONNREFUSED] += 1 }
+			rescue => e
+				mutex.synchronize { counters[e.class] += 1 }
+				$stderr.puts "[ERROR] #{e.class}: #{e.message}"
+			end
+		end
+	end
+end
+
+scheduler.run
+Fiber.set_scheduler(nil)
+
+# ── tear-down ────────────────────────────────────────────────────────────────
+
+server.close
+acceptor_thread.join(2)
+
+# ── report ───────────────────────────────────────────────────────────────────
+
+$stderr.puts
+$stderr.puts "=== Results ==="
+$stderr.puts "  Successful : #{counters[:ok]}"
+$stderr.puts "  ENOTCONN   : #{counters[:ENOTCONN]}"
+$stderr.puts "  Other      : #{(counters.reject { |k, _| k == :ok || k == :ENOTCONN }).inspect}" unless counters.size <= 2
+
+if counters[:ENOTCONN] > 0
+	$stderr.puts
+	$stderr.puts "REPRODUCED: #{counters[:ENOTCONN]} ENOTCONN error(s) observed."
+	exit 1
+else
+	$stderr.puts
+	$stderr.puts "No ENOTCONN errors observed in this run."
+	exit 0
+end