usockets: accept with AcceptEx on a listening socket that another process holds (Windows)

packages/bun-usockets/src/bsd.c

@@ -58,6 +58,7 @@ static void init_debug_logging() {
 #include <errno.h>
 #else /* _WIN32 */
 #include <mstcpip.h>
+#include <mswsock.h>
 #endif
 
 #if defined(__APPLE__)
@@ -810,22 +811,116 @@ int bsd_addr_get_port(struct bsd_addr_t *addr) {
     return addr->port;
 }
 
+#ifdef _WIN32
+/* accept() for a listening socket that another process holds too. Every process
+ * that polls the socket wakes up for each connection. A non-blocking accept()
+ * returns WSAEWOULDBLOCK only if no connection is pending when it starts. If a
+ * connection is pending and another process takes it first, accept() waits for
+ * the next connection, and the event loop stops. An AcceptEx request that finds
+ * no connection is canceled here instead. libuv accepts with AcceptEx on every
+ * listening socket, shared ones included (win/tcp.c). */
+static SOCKET bsd_accept_shared_socket(SOCKET fd, struct bsd_addr_t *addr) {
+    /* Do not create a socket when no connection is pending. */
+    fd_set readable;
+    FD_ZERO(&readable);
+    FD_SET(fd, &readable);
+    struct timeval no_wait = {0, 0};
+    int ready = select(0, &readable, NULL, NULL, &no_wait);
+    if (ready != 1) {
+        if (ready == 0) {
+            WSASetLastError(WSAEWOULDBLOCK);
+        }
+        return INVALID_SOCKET;
+    }
+
+    GUID acceptex_guid = WSAID_ACCEPTEX;
+    LPFN_ACCEPTEX acceptex = NULL;
+    DWORD bytes = 0;
+    if (WSAIoctl(fd, SIO_GET_EXTENSION_FUNCTION_POINTER, &acceptex_guid, sizeof(acceptex_guid),
+                 &acceptex, sizeof(acceptex), &bytes, NULL, NULL) != 0) {
+        /* A provider without AcceptEx keeps the plain accept(). */
+        return accept(fd, (struct sockaddr *) &addr->mem, &addr->len);
+    }
+
+    struct sockaddr_storage local;
+    int local_len = sizeof(local);
+    if (getsockname(fd, (struct sockaddr *) &local, &local_len) != 0) {
+        return INVALID_SOCKET;
+    }
+    SOCKET accepted = WSASocketW(local.ss_family, SOCK_STREAM, 0, NULL, 0,
+                                 WSA_FLAG_OVERLAPPED | WSA_FLAG_NO_HANDLE_INHERIT);
+    if (accepted == INVALID_SOCKET) {
+        return INVALID_SOCKET;
+    }
+    HANDLE event = CreateEventW(NULL, TRUE, FALSE, NULL);
+    if (event == NULL) {
+        closesocket(accepted);
+        return INVALID_SOCKET;
+    }
+
+    /* AcceptEx writes both addresses here, each with 16 bytes of extra space. */
+    char addresses[2 * (sizeof(struct sockaddr_storage) + 16)];
+    OVERLAPPED overlapped;
+    memset(&overlapped, 0, sizeof(overlapped));
+    /* The low bit keeps the completion off any I/O completion port. */
+    overlapped.hEvent = (HANDLE) ((ULONG_PTR) event | 1);
+    int err = 0;
+    if (!acceptex(fd, accepted, addresses, 0, sizeof(struct sockaddr_storage) + 16,
+                  sizeof(struct sockaddr_storage) + 16, &bytes, &overlapped)) {
+        err = WSAGetLastError();
+        if (err == WSA_IO_PENDING) {
+            /* The request gets a connection that arrives before the cancel, or it is canceled.
+             * Either way it completes and sets the event. Wait for that: the request writes
+             * to overlapped and addresses, which are on this stack. */
+            CancelIoEx((HANDLE) fd, &overlapped);
+            WaitForSingleObject(event, INFINITE);
+            DWORD flags = 0;
+            err = WSAGetOverlappedResult(fd, &overlapped, &bytes, FALSE, &flags) ? 0 : WSAGetLastError();
+            if (err == WSA_OPERATION_ABORTED) {
+                err = WSAEWOULDBLOCK;
+            }
+        }
+    }
+    CloseHandle(event);
+
+    if (err == 0 && setsockopt(accepted, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char *) &fd, sizeof(fd)) != 0) {
+        err = WSAGetLastError();
+    }
+    if (err != 0) {
+        closesocket(accepted);
+        WSASetLastError(err);
+        return INVALID_SOCKET;
+    }
+    /* A connection that the peer already reset has no address. The first read reports the reset. */
+    if (getpeername(accepted, (struct sockaddr *) &addr->mem, &addr->len) != 0) {
+        addr->mem.ss_family = AF_UNSPEC;
+    }
+    return accepted;
+}
+#endif
+
 // called by dispatch_ready_poll
-LIBUS_SOCKET_DESCRIPTOR bsd_accept_socket(LIBUS_SOCKET_DESCRIPTOR fd, struct bsd_addr_t *addr) {
+LIBUS_SOCKET_DESCRIPTOR bsd_accept_socket(LIBUS_SOCKET_DESCRIPTOR fd, struct bsd_addr_t *addr, int shared) {
     LIBUS_SOCKET_DESCRIPTOR accepted_fd;
 
     ssize_t injected = 0; int unused = 0;
     if (US_FAULT_CHECK(US_FAULT_ACCEPT, fd, injected, unused)) return LIBUS_SOCKET_ERROR;
     (void)injected; (void)unused;
+#ifndef _WIN32
+    /* A POSIX accept() honors O_NONBLOCK, also on a socket that another process holds. */
+    (void)shared;
+#endif
 
     while (1) {
         addr->len = sizeof(addr->mem);
 
 #if defined(SOCK_CLOEXEC) && defined(SOCK_NONBLOCK)
     // Linux, FreeBSD
     accepted_fd = accept4(fd, (struct sockaddr *) addr, &addr->len, SOCK_CLOEXEC | SOCK_NONBLOCK);
+#elif defined(_WIN32)
+    accepted_fd = shared ? bsd_accept_shared_socket(fd, addr) : accept(fd, (struct sockaddr *) addr, &addr->len);
 #else
-    // Windows, OS X
+    // OS X
     accepted_fd = accept(fd, (struct sockaddr *) addr, &addr->len);
 #endif
 

packages/bun-usockets/src/context.c

@@ -370,6 +370,7 @@ static void us_internal_init_listen_socket(struct us_listen_socket_t *ls,
     ls->socket_ext_size = socket_ext_size;
     ls->deferred_accept = 0;
     ls->accept_paused = (options & LIBUS_SOCKET_OPEN_PAUSED) && !ssl_ctx;
+    ls->shared = 0;
 
     /* Link into the group so close_all() / test-isolation can find it. */
     ls->next = group->head_listen_sockets;
@@ -435,6 +436,8 @@ struct us_listen_socket_t *us_socket_group_listen_fd(struct us_socket_group_t *g
 
     struct us_listen_socket_t *ls = (struct us_listen_socket_t *) p;
     us_internal_init_listen_socket(ls, group, kind, ssl_ctx, options, socket_ext_size);
+    /* The process that handed over the descriptor can still hold it. */
+    ls->shared = 1;
 
     if (options & LIBUS_LISTEN_DEFER_ACCEPT) {
         ls->deferred_accept = bsd_set_defer_accept(fd);
@@ -517,6 +520,10 @@ struct us_socket_group_t *us_listen_socket_group(struct us_listen_socket_t *ls)
     return ls->accept_group;
 }
 
+void us_listen_socket_set_shared(struct us_listen_socket_t *ls) {
+    ls->shared = 1;
+}
+
 /* ── Connect ────────────────────────────────────────────────────────────── */
 
 static inline void us_internal_init_connect_socket(struct us_socket_t *s,

packages/bun-usockets/src/internal/internal.h

@@ -464,6 +464,9 @@ struct us_listen_socket_t {
   unsigned char deferred_accept;
   /* LIBUS_SOCKET_OPEN_PAUSED: accepted sockets start without read interest. */
   unsigned char accept_paused;
+  /* Another process can accept on this socket too: it was sent over IPC or
+   * adopted from a descriptor. See bsd_accept_socket. */
+  unsigned char shared;
 };
 
 void us_internal_socket_group_link_connecting_socket(us_socket_group_r group, struct us_connecting_socket_t *c);

packages/bun-usockets/src/internal/networking/bsd.h

@@ -210,8 +210,8 @@ int bsd_addr_get_ip_length(struct bsd_addr_t *addr);
 
 int bsd_addr_get_port(struct bsd_addr_t *addr);
 
-// called by dispatch_ready_poll
-LIBUS_SOCKET_DESCRIPTOR bsd_accept_socket(LIBUS_SOCKET_DESCRIPTOR fd, struct bsd_addr_t *addr);
+// called by dispatch_ready_poll. shared: another process can accept on fd too.
+LIBUS_SOCKET_DESCRIPTOR bsd_accept_socket(LIBUS_SOCKET_DESCRIPTOR fd, struct bsd_addr_t *addr, int shared);
 
 ssize_t bsd_recv(LIBUS_SOCKET_DESCRIPTOR fd, void *buf, int length, int flags);
 #if !defined(_WIN32)

packages/bun-usockets/src/libusockets.h

@@ -699,6 +699,8 @@ void *us_listen_socket_ext(struct us_listen_socket_t *ls) nonnull_fn_decl;
 LIBUS_SOCKET_DESCRIPTOR us_listen_socket_get_fd(struct us_listen_socket_t *ls) nonnull_fn_decl;
 int us_listen_socket_port(struct us_listen_socket_t *ls) nonnull_fn_decl;
 struct us_socket_group_t *us_listen_socket_group(struct us_listen_socket_t *ls) nonnull_fn_decl;
+/* Call before another process gets a duplicate of the listening socket. */
+void us_listen_socket_set_shared(struct us_listen_socket_t *ls) nonnull_fn_decl;
 /* Walk a group's live listeners. The list is the source of truth — anything
  * that caches us_listen_socket_t* across event-loop ticks (e.g. a std::vector
  * in C++) is a UAF once a listener is closed and freed in loop_post. */

packages/bun-usockets/src/loop.c

@@ -474,7 +474,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                 struct us_loop_t *loop = accept_group->loop;
                 struct bsd_addr_t addr;
 
-                LIBUS_SOCKET_DESCRIPTOR client_fd = bsd_accept_socket(us_poll_fd(p), &addr);
+                LIBUS_SOCKET_DESCRIPTOR client_fd = bsd_accept_socket(us_poll_fd(p), &addr, listen_socket->shared);
                 if (client_fd == LIBUS_SOCKET_ERROR) {
                     /* Todo: start timer here */
 
@@ -540,7 +540,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                             break;
                         }
 
-                    } while ((client_fd = bsd_accept_socket(us_poll_fd(p), &addr)) != LIBUS_SOCKET_ERROR);
+                    } while ((client_fd = bsd_accept_socket(us_poll_fd(p), &addr, listen_socket->shared)) != LIBUS_SOCKET_ERROR);
                 }
             }
         break;

src/runtime/ipc_host.rs

@@ -176,10 +176,10 @@ pub(crate) fn do_send(
             match unsafe { (*listener).listener.get() } {
                 crate::socket::listener::ListenerType::Uws(socket_uws) => {
                     // may need to handle ssl case
-                    // SAFETY: `socket_uws` is a live non-null `*mut ListenSocket`
-                    // owned by uSockets; `get_socket` only reinterpret-casts to
-                    // `&mut us_socket_t` and `get_fd` is a read-only FFI call.
-                    let fd = unsafe { &mut *socket_uws }.get_socket().get_fd();
+                    // SAFETY: `socket_uws` is a live non-null `*mut ListenSocket` owned by uSockets.
+                    let listen_socket = unsafe { &mut *socket_uws };
+                    listen_socket.set_shared();
+                    let fd = listen_socket.get_socket().get_fd();
                     #[cfg(not(windows))]
                     match Handle::init_dup(fd, handle, false) {
                         Ok(h) => zig_handle = Some(h),

src/uws_sys/ListenSocket.rs

@@ -36,6 +36,11 @@ impl ListenSocket {
         ))
     }
 
+    /// Call before another process gets a duplicate of this socket (see `bsd_accept_socket`).
+    pub fn set_shared(&mut self) {
+        us_listen_socket_set_shared(self)
+    }
+
     /// Group accepted sockets are linked into.
     pub fn group(&mut self) -> &mut SocketGroup {
         // SAFETY: self is a valid listen socket; C returns a non-null group.
@@ -102,6 +107,7 @@ unsafe extern "C" {
     safe fn us_listen_socket_close(ls: &mut ListenSocket);
     safe fn us_listen_socket_group(ls: &mut ListenSocket) -> *mut SocketGroup;
     safe fn us_listen_socket_get_fd(ls: &mut ListenSocket) -> LIBUS_SOCKET_DESCRIPTOR;
+    safe fn us_listen_socket_set_shared(ls: &mut ListenSocket);
     fn us_listen_socket_add_server_name(
         ls: *mut ListenSocket,
         hostname: *const c_char,

test/js/node/child_process/child_process_ipc_handle.test.ts

@@ -781,3 +781,74 @@ const server = net.createServer().listen(0, '127.0.0.1', () => {
     expect(exitCode).toBe(0);
   });
 });
+
+// On Windows, the child gets a duplicate of the parent's listening socket, and both processes wake
+// up for each connection. The process that did not get the connection called a non-blocking accept()
+// that blocked until the next connection, so its event loop stopped. Here a third process makes the
+// connections, one at a time, so both processes that accept are idle when a connection arrives.
+// After each connection, the parent must handle a message from the client process and the acceptor
+// must answer a ping: a process that is blocked in accept() does neither. Both processes poll the
+// socket, so each one accepts about half of the connections.
+test.skipIf(!isWindows)(
+  "a net.Server sent to a child: both processes keep handling events while they accept",
+  async () => {
+    using dir = tempDir("ipc-handle-shared-accept", {
+      "parent.js": `
+const { fork } = require('node:child_process');
+const net = require('node:net');
+const N = 100;
+const accepted = { parent: 0, child: 0 };
+const server = net.createServer(socket => { accepted.parent++; socket.destroy(); });
+const acceptor = fork('child.js', ['acceptor']);
+const client = fork('child.js', ['client']);
+const connect = i => client.send({ port: server.address().port, i });
+client.on('message', ({ closed }) => acceptor.send({ ping: closed }));
+acceptor.on('message', m => {
+  if (m === 'ready') return connect(0);
+  accepted.child = m.accepted;
+  if (m.pong + 1 < N) return connect(m.pong + 1);
+  server.close();
+  client.disconnect();
+  acceptor.disconnect();
+});
+acceptor.on('exit', code => console.log(JSON.stringify({ ...accepted, code })));
+server.listen(0, '127.0.0.1', () => acceptor.send('server', server));
+`,
+      "child.js": `
+const net = require('node:net');
+if (process.argv[2] === 'acceptor') {
+  let accepted = 0;
+  process.on('message', (m, server) => {
+    if (m !== 'server') return process.send({ pong: m.ping, accepted });
+    server.on('connection', socket => { accepted++; socket.destroy(); });
+    process.on('disconnect', () => server.close());
+    process.send('ready');
+  });
+} else {
+  process.on('message', ({ port, i }) => {
+    const socket = net.connect(port, '127.0.0.1');
+    socket.on('error', () => {});
+    socket.on('close', () => process.send({ closed: i }));
+  });
+}
+`,
+    });
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "parent.js"],
+      env: bunEnv,
+      cwd: String(dir),
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    const { parent, child, code } = JSON.parse(stdout.trim());
+    expect({ accepted: parent + child, parentAccepted: parent > 0, childAccepted: child > 0, code, stderr }).toEqual({
+      accepted: 100,
+      parentAccepted: true,
+      childAccepted: true,
+      code: 0,
+      stderr: "",
+    });
+    expect(exitCode).toBe(0);
+  },
+);