Fix kqueue false connect success from stale EVFILT_WRITE

The kqueue backend registers sockets for EVFILT_WRITE at open() time.
A freshly created socket is writable, so kqueue fires a stale event
before connect() completes. If the reactor processes this before the
kernel delivers the connect result (e.g. RST for ECONNREFUSED),
getsockopt(SO_ERROR) returns 0 and the connect falsely reports success.

Fix by adding a getpeername() check in connect perform_io() to verify
the connection is actually established when SO_ERROR is 0, returning
EAGAIN to re-park the op if not. Add EAGAIN handling for connect ops
in descriptor_state::operator()() to match the existing read/write
pattern.

Author: mvandeberg

include/boost/corosio/native/detail/kqueue/kqueue_op.hpp

@@ -282,6 +282,21 @@ struct kqueue_connect_op final : kqueue_op
         socklen_t len = sizeof(err);
         if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
             err = errno;
+        // Guard against stale EVFILT_WRITE events from socket creation.
+        // kqueue registers EVFILT_WRITE at open() time; a freshly created
+        // socket is "writable" so the filter fires immediately. If the
+        // reactor processes this event after connect() returns EINPROGRESS
+        // but before the kernel delivers the connect result (e.g. RST for
+        // ECONNREFUSED), SO_ERROR is still 0. Use getpeername() to verify
+        // the connection is actually established.
+        if (err == 0)
+        {
+            sockaddr_storage peer{};
+            socklen_t peer_len = sizeof(peer);
+            if (::getpeername(
+                    fd, reinterpret_cast<sockaddr*>(&peer), &peer_len) < 0)
+                err = EAGAIN;
+        }
         complete(err, 0);
     }
 

include/boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp

@@ -604,8 +604,16 @@ descriptor_state::operator()()
             cn->complete(err, 0);
         else
             cn->perform_io();
-        local_ops.push(cn);
-        cn = nullptr;
+
+        if (cn->errn == EAGAIN || cn->errn == EWOULDBLOCK)
+        {
+            cn->errn = 0;
+        }
+        else
+        {
+            local_ops.push(cn);
+            cn = nullptr;
+        }
     }
 
     if (wr)
@@ -630,7 +638,7 @@ descriptor_state::operator()()
     // have set read_ready/write_ready while we held the op (no read_op
     // was registered, so it cached the edge event). Check the flags
     // under the same lock as re-registration so no edge is lost.
-    while (rd || wr)
+    while (rd || wr || cn)
     {
         bool retry = false;
         {
@@ -661,6 +669,19 @@ descriptor_state::operator()()
                     wr       = nullptr;
                 }
             }
+            if (cn)
+            {
+                if (write_ready)
+                {
+                    write_ready = false;
+                    retry       = true;
+                }
+                else
+                {
+                    connect_op = cn;
+                    cn         = nullptr;
+                }
+            }
         }
 
         if (!retry)
@@ -688,6 +709,17 @@ descriptor_state::operator()()
                 wr = nullptr;
             }
         }
+        if (cn)
+        {
+            cn->perform_io();
+            if (cn->errn == EAGAIN || cn->errn == EWOULDBLOCK)
+                cn->errn = 0;
+            else
+            {
+                local_ops.push(cn);
+                cn = nullptr;
+            }
+        }
     }
 
     // Execute first handler inline — the scheduler's work_cleanup