From 92b4ba7f6e4608387bfc8ea22cf6c278b07eb112 Mon Sep 17 00:00:00 2001
From: Shizuo Fujita <fujita@clear-code.com>
Date: Wed, 21 Oct 2026 17:03:51 +0900
Subject: [PATCH] Fix listen failing with EISCONN on Windows after a Loop with
 a StatWatcher is freed

Since 1.10.0, a Loop with a StatWatcher on Windows creates libev's
evpipe, a pair of loopback TCP sockets. ev_loop_destroy closed them
with the CRT close(), which only calls CloseHandle() and leaves Winsock
state behind. A socket later imported from a WSAPROTOCOL_INFO with the
same handle value inherits that state, and listen() fails with
WSAEISCONN. Fluentd hits this through the socket manager of
ServerEngine. Register the evpipe sockets with Ruby and close them with
rb_w32_close(), which calls closesocket().

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 ext/cool.io/ev_wrap.h               |  7 +++
 spec/fixtures/windows_stat_native.c | 71 +++++++++++++++++++++++++++++
 spec/windows_stat_native_spec.rb    | 19 ++++++++
 3 files changed, 97 insertions(+)

diff --git a/ext/cool.io/ev_wrap.h b/ext/cool.io/ev_wrap.h
index 2894697..0cc0910 100644
--- a/ext/cool.io/ev_wrap.h
+++ b/ext/cool.io/ev_wrap.h
@@ -1,10 +1,17 @@
 #define EV_STANDALONE              /* keeps ev from requiring config.h */
 
 #ifdef _WIN32
+#include "ruby.h"
 #define EV_SELECT_IS_WINSOCKET 1   /* configure libev for windows select */
 #define EV_USE_MONOTONIC 0
 #define EV_USE_REALTIME 0
 #define EV_WIN32_STAT_CODEPAGE CP_UTF8 /* Ruby paths are UTF-8. */
+/* Close the evpipe sockets with closesocket() through Ruby. The CRT close()
+ * only calls CloseHandle(), which leaves Winsock state behind. A socket later
+ * imported from a WSAPROTOCOL_INFO with the same handle value inherits it, and
+ * listen() fails with WSAEISCONN. */
+#define EV_WIN32_HANDLE_TO_FD(handle) rb_w32_wrap_io_handle((HANDLE)(handle), 0)
+#define EV_WIN32_CLOSE_FD(fd) rb_w32_close(fd)
 #endif
 
 #include "../libev/ev.h"
diff --git a/spec/fixtures/windows_stat_native.c b/spec/fixtures/windows_stat_native.c
index f8b8a3d..a1352a3 100644
--- a/spec/fixtures/windows_stat_native.c
+++ b/spec/fixtures/windows_stat_native.c
@@ -1284,6 +1284,76 @@ alias_recovery (VALUE self, VALUE path)
     retried ? Qtrue : Qfalse, cached ? Qtrue : Qfalse);
 }
 
+static VALUE
+imported_listen_failures (VALUE self, VALUE tries)
+{
+  enum { limit = 64 };
+  WSADATA data;
+  WSAPROTOCOL_INFOW info;
+  struct sockaddr_in addr;
+  SOCKET sockets[limit * 2], base, imported;
+  const char *failure = 0;
+  int i, error = 0, opened = 0, failures = 0, count = NUM2INT (tries);
+  if (count < 0 || count > limit)
+    {
+      rb_raise (rb_eArgError, "tries must be between 0 and %d", limit);
+    }
+  if (WSAStartup (MAKEWORD (2, 2), &data))
+    {
+      rb_raise (rb_eRuntimeError, "WSAStartup failed");
+    }
+  memset (&addr, 0, sizeof (addr));
+  addr.sin_family = AF_INET;
+  addr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
+  for (i = 0; i < count; ++i)
+    {
+      base = WSASocketW (AF_INET, SOCK_STREAM, IPPROTO_TCP, 0, 0, WSA_FLAG_OVERLAPPED);
+      if (base == INVALID_SOCKET)
+        {
+          failure = "WSASocketW failed";
+          error = WSAGetLastError ();
+          break;
+        }
+      sockets[opened++] = base;
+      if (bind (base, (struct sockaddr *)&addr, sizeof (addr))
+          || WSADuplicateSocketW (base, GetCurrentProcessId (), &info))
+        {
+          failure = "could not prepare a socket to import";
+          error = WSAGetLastError ();
+          break;
+        }
+      imported = WSASocketW (FROM_PROTOCOL_INFO, FROM_PROTOCOL_INFO, FROM_PROTOCOL_INFO,
+                             &info, 0, WSA_FLAG_OVERLAPPED);
+      if (imported == INVALID_SOCKET)
+        {
+          failure = "WSASocketW could not import the socket";
+          error = WSAGetLastError ();
+          break;
+        }
+      sockets[opened++] = imported;
+      if (listen (imported, 1) == SOCKET_ERROR)
+        {
+          error = WSAGetLastError ();
+          if (error != WSAEISCONN)
+            {
+              failure = "listen failed";
+              break;
+            }
+          ++failures;
+        }
+    }
+  while (opened)
+    {
+      closesocket (sockets[--opened]);
+    }
+  WSACleanup ();
+  if (failure)
+    {
+      rb_raise (rb_eRuntimeError, "%s: %d", failure, error);
+    }
+  return INT2NUM (failures);
+}
+
 RUBY_FUNC_EXPORTED void
 Init_coolio_stat_native (void)
 {
@@ -1302,4 +1372,5 @@ Init_coolio_stat_native (void)
   rb_define_singleton_method (module, "children_cwd", children_cwd, 3);
   rb_define_singleton_method (module, "replaced_dir", replaced_dir, 2);
   rb_define_singleton_method (module, "alias_recovery", alias_recovery, 1);
+  rb_define_singleton_method (module, "imported_listen_failures", imported_listen_failures, 1);
 }
diff --git a/spec/windows_stat_native_spec.rb b/spec/windows_stat_native_spec.rb
index 37b24d5..da76428 100644
--- a/spec/windows_stat_native_spec.rb
+++ b/spec/windows_stat_native_spec.rb
@@ -177,6 +177,25 @@ def replaced_dir(mode)
     expect(result).to eq([true, true, true, true])
   end
 
+  it 'closes the notification pipe of a destroyed loop so that imported sockets can listen' do
+    lib = File.expand_path('../lib', __dir__)
+    result = native_result(<<~'RUBY', lib, @build)
+      $LOAD_PATH.unshift(ARGV[0])
+      require 'cool.io'
+      path = File.join(ARGV[1], 'evpipe.txt')
+      File.write(path, 'a')
+      20.times do
+        reactor = Coolio::Loop.new
+        watcher = Coolio::StatWatcher.new(path, 60)
+        watcher.attach(reactor)
+        watcher.detach
+      end
+      3.times { GC.start }
+      $stdout.write(Marshal.dump(CoolioStatNativeSpec.imported_listen_failures(30)))
+    RUBY
+    expect(result).to eq(0)
+  end
+
   it 'uses effective notification flags and respects EVFLAG_NOENV' do
     path = File.join(@build, 'flags.txt')
     # EVFLAG_NOINOTIFY = 0x00100000; EVFLAG_NOENV = 0x01000000.