cluster: port Node's cluster and child_process handle-passing suites (+43 upstream tests; cluster 54 → 85) and implement what they expose — round-robin fd handoff, SCHED_NONE shared handles, UDP clustering, IPC handle passing

.claude/skills/verify/SKILL.md

@@ -37,6 +37,8 @@ For worker/subprocess-shaped changes, spawn a subprocess (still `-e`) so worker
   formula shadows the pinned nightly, and `bun bd` dies with `the option 'Z' is only
   accepted on the nightly compiler`. `bun bd` re-runs cargo on every invocation, so
   this is needed for follow-up runs too, not just the first build.
+- `node:cluster` changes can't be driven with `-e`: `cluster.fork()` re-execs `argv[1]`, so workers need a real file on disk. Write a scratch script and run `./build/debug/bun-debug <file>`.
+- Only one `bun bd` per worktree at a time — a second one blocks on the build lock and looks like a runtime hang. Build once, then drive `./build/debug/bun-debug` directly under `timeout`.
 - `BUN_DEBUG_QUIET_LOGS=1` suppresses debug-build log spam.
 - Debug builds print `[cachefs]`/`[sys]` lines to stdout; filter them before diffing
   output against `node`.

packages/bun-usockets/src/bsd.c

@@ -135,7 +135,8 @@ int bsd_sendmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_sendbuf* sendbuf, int fl
 #endif
 }
 
-int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_recvbuf *recvbuf, int flags) {
+int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_recvbuf *recvbuf, int flags, int max_packets) {
+    if (max_packets > LIBUS_UDP_RECV_COUNT) max_packets = LIBUS_UDP_RECV_COUNT;
 #if defined(_WIN32)
     for (int i = 0; i < LIBUS_UDP_RECV_COUNT; i++) {
         while (1) {
@@ -162,12 +163,12 @@ int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_recvbuf *recvbuf, int fl
 #elif defined(__APPLE__)
     if (Bun__doesMacOSVersionSupportSendRecvMsgX()) {
         while (1) {
-            int ret = recvmsg_x(fd, recvbuf->msgvec, LIBUS_UDP_RECV_COUNT, flags);
+            int ret = recvmsg_x(fd, recvbuf->msgvec, max_packets, flags);
             if (ret >= 0 || errno != EINTR) return ret;
         }
     }
 
-    for (int i = 0; i < LIBUS_UDP_RECV_COUNT; ++i) {
+    for (int i = 0; i < max_packets; ++i) {
         while (1) {
             ssize_t ret = recvmsg(fd, &recvbuf->msgvec[i].msg_hdr, flags);
             if (ret < 0) {
@@ -179,10 +180,10 @@ int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_recvbuf *recvbuf, int fl
             break;
         }
     }
-    return LIBUS_UDP_RECV_COUNT;
+    return max_packets;
 #else
     while (1) {
-        int ret = recvmmsg(fd, (struct mmsghdr *)&recvbuf->msgvec, LIBUS_UDP_RECV_COUNT, flags, 0);
+        int ret = recvmmsg(fd, (struct mmsghdr *)&recvbuf->msgvec, max_packets, flags, 0);
         if (ret >= 0 || errno != EINTR) return ret;
     }
 #endif
@@ -1216,6 +1217,140 @@ int bsd_set_defer_accept(LIBUS_SOCKET_DESCRIPTOR listenFd) {
 
 // return LIBUS_SOCKET_ERROR or the fd that represents listen socket
 // listen both on ipv6 and ipv4
+int bsd_socket_export_size(void) {
+#ifdef _WIN32
+    return (int) sizeof(WSAPROTOCOL_INFOW);
+#else
+    return 0;
+#endif
+}
+
+int bsd_socket_export(LIBUS_SOCKET_DESCRIPTOR fd, unsigned int target_pid, void *info_out) {
+#ifdef _WIN32
+    if (WSADuplicateSocketW(fd, (DWORD) target_pid, (WSAPROTOCOL_INFOW *) info_out) != 0) {
+        return WSAGetLastError();
+    }
+    return 0;
+#else
+    (void) fd; (void) target_pid; (void) info_out;
+    return ENOTSUP;
+#endif
+}
+
+LIBUS_SOCKET_DESCRIPTOR bsd_socket_import(void *info, int *err) {
+#ifdef _WIN32
+    SOCKET s = WSASocketW(FROM_PROTOCOL_INFO, FROM_PROTOCOL_INFO, FROM_PROTOCOL_INFO,
+                          (WSAPROTOCOL_INFOW *) info, 0, WSA_FLAG_OVERLAPPED);
+    if (s == INVALID_SOCKET) {
+        *err = WSAGetLastError();
+        return LIBUS_SOCKET_ERROR;
+    }
+    return s;
+#else
+    (void) info;
+    *err = ENOTSUP;
+    return LIBUS_SOCKET_ERROR;
+#endif
+}
+
+/* Windows rejects listen() on a duplicate of an already-listening socket, where POSIX
+ * no-ops it; cluster workers each listen on their own dup of one shared fd. libuv
+ * sidesteps this by listening before the xfer (UV_HANDLE_SHARED_TCP_SOCKET, win/tcp.c). */
+int bsd_socket_listen_error_is_benign(LIBUS_SOCKET_DESCRIPTOR fd) {
+#ifdef _WIN32
+    int listening = 0;
+    int optlen = (int) sizeof(listening);
+    if (getsockopt(fd, SOL_SOCKET, SO_ACCEPTCONN, (char *) &listening, &optlen) != 0) {
+        return 0;
+    }
+    return listening != 0;
+#else
+    (void) fd;
+    return 0;
+#endif
+}
+
+LIBUS_SOCKET_DESCRIPTOR bsd_create_bound_socket(const char *host, int port, int options, int *out_port, int *error) {
+    struct addrinfo hints, *result;
+    memset(&hints, 0, sizeof(struct addrinfo));
+    hints.ai_flags = AI_PASSIVE;
+    hints.ai_family = AF_UNSPEC;
+    hints.ai_socktype = SOCK_STREAM;
+
+    char port_string[16];
+    snprintf(port_string, 16, "%d", port);
+
+    int gai = getaddrinfo(host, port_string, &hints, &result);
+    if (gai != 0) {
+#ifdef _WIN32
+        *error = gai;
+#else
+        *error = EINVAL;
+#endif
+        return LIBUS_SOCKET_ERROR;
+    }
+
+    LIBUS_SOCKET_DESCRIPTOR fd = LIBUS_SOCKET_ERROR;
+    for (int family = AF_INET6; fd == LIBUS_SOCKET_ERROR && family >= AF_INET; family -= (AF_INET6 - AF_INET)) {
+        for (struct addrinfo *a = result; a != NULL; a = a->ai_next) {
+            if (a->ai_family != family) {
+                continue;
+            }
+            fd = bsd_create_socket(a->ai_family, a->ai_socktype, a->ai_protocol, NULL);
+            if (fd == LIBUS_SOCKET_ERROR) {
+                *error = LIBUS_ERR;
+                continue;
+            }
+#if defined(SO_REUSEADDR) && !defined(_WIN32)
+            int one = 1;
+            setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
+#endif
+#ifdef IPV6_V6ONLY
+            if (a->ai_family == AF_INET6) {
+                int enabled = (options & LIBUS_SOCKET_IPV6_ONLY) != 0;
+                setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, (void *) &enabled, sizeof(enabled));
+            }
+#endif
+            int rc;
+            do
+                rc = bind(fd, a->ai_addr, (socklen_t) a->ai_addrlen);
+            while (IS_EINTR(rc));
+            if (rc != 0) {
+                *error = LIBUS_ERR;
+                bsd_close_socket(fd);
+                fd = LIBUS_SOCKET_ERROR;
+                continue;
+            }
+            break;
+        }
+    }
+    freeaddrinfo(result);
+    if (fd == LIBUS_SOCKET_ERROR) {
+        return LIBUS_SOCKET_ERROR;
+    }
+#ifdef _WIN32
+    /* Windows rejects listen() on a duplicate of an already-listening socket
+     * only after another duplicate has listened, so two workers racing on the
+     * same shared fd can observe listen() -> WSAEINVAL while SO_ACCEPTCONN
+     * still reads 0 (the benign check in us_socket_group_listen_fd then
+     * fails). libuv's approach (UV_HANDLE_SHARED_TCP_SOCKET, win/tcp.c) is to
+     * listen in the primary before WSADuplicateSocket; every worker's
+     * duplicate is then already listening and SO_ACCEPTCONN is reliably set. */
+    if (listen(fd, 511) != 0) {
+        *error = LIBUS_ERR;
+        bsd_close_socket(fd);
+        return LIBUS_SOCKET_ERROR;
+    }
+#endif
+    struct bsd_addr_t tmp;
+    if (bsd_local_addr(fd, &tmp) == 0) {
+        *out_port = bsd_addr_get_port(&tmp);
+    } else {
+        *out_port = port;
+    }
+    return fd;
+}
+
 LIBUS_SOCKET_DESCRIPTOR bsd_create_listen_socket(const char *host, int port, int options, int* error) {
     struct addrinfo hints, *result;
     memset(&hints, 0, sizeof(struct addrinfo));

packages/bun-usockets/src/context.c

@@ -418,6 +418,40 @@ struct us_listen_socket_t *us_socket_group_listen(struct us_socket_group_t *grou
     return ls;
 }
 
+struct us_listen_socket_t *us_socket_group_listen_fd(struct us_socket_group_t *group,
+        unsigned char kind, struct ssl_ctx_st *ssl_ctx,
+        LIBUS_SOCKET_DESCRIPTOR fd, int backlog, int options, int socket_ext_size, int *error) {
+    /* Validate with listen(2) before touching the descriptor's flags: on failure the caller keeps
+     * the fd (it may be its stdio), and a non-socket must come back untouched. */
+    if (listen(fd, backlog > 0 ? backlog : 512)) {
+        int listen_err = LIBUS_ERR;
+        if (!bsd_socket_listen_error_is_benign(fd)) {
+            *error = listen_err;
+            return 0;
+        }
+    }
+    apple_no_sigpipe(fd);
+    bsd_set_nonblocking(fd);
+
+    struct us_poll_t *p = us_create_poll(group->loop, 0, sizeof(struct us_listen_socket_t));
+    us_poll_init(p, fd, POLL_TYPE_SEMI_SOCKET);
+    if (us_poll_start_rc(p, group->loop, LIBUS_SOCKET_READABLE) != 0) {
+        int saved_errno = LIBUS_ERR;
+        us_poll_free(p, group->loop);
+        *error = saved_errno;
+        return 0;
+    }
+
+    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);
+
+    if (options & LIBUS_LISTEN_DEFER_ACCEPT) {
+        ls->deferred_accept = bsd_set_defer_accept(fd);
+    }
+
+    return ls;
+}
+
 struct us_listen_socket_t *us_socket_group_listen_unix(struct us_socket_group_t *group,
         unsigned char kind, struct ssl_ctx_st *ssl_ctx,
         const char *path, size_t pathlen, int options, int socket_ext_size, int *error) {

packages/bun-usockets/src/eventing/libuv.c

@@ -574,4 +574,4 @@ int us_socket_get_error(struct us_socket_t *s) {
   return error;
 }
 
-#endif
+#endif

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

@@ -390,6 +390,7 @@ struct us_udp_socket_t {
     uint16_t port;
     uint16_t closed : 1;
     uint16_t connected : 1;
+    uint16_t shared_fd : 1;
     struct us_udp_socket_t *next;
 };
 

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

@@ -160,7 +160,7 @@ struct udp_sendbuf {
 };
 
 int bsd_sendmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_sendbuf* sendbuf, int flags);
-int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_recvbuf *recvbuf, int flags);
+int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct udp_recvbuf *recvbuf, int flags, int max_packets);
 void bsd_udp_setup_recvbuf(struct udp_recvbuf *recvbuf, void *databuf, size_t databuflen);
 int bsd_udp_setup_sendbuf(struct udp_sendbuf *buf, size_t bufsize, void** payloads, size_t* lengths, void** addresses, int num);
 int bsd_udp_packet_buffer_payload_length(struct udp_recvbuf *msgvec, int index);
@@ -243,6 +243,13 @@ LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket(struct sockaddr_storage *addr,
 
 LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket_unix(const char *server_path, size_t pathlen, int options);
 
+int bsd_socket_export_size(void);
+int bsd_socket_export(LIBUS_SOCKET_DESCRIPTOR fd, unsigned int target_pid, void *info_out);
+LIBUS_SOCKET_DESCRIPTOR bsd_socket_import(void *info, int *err);
+int bsd_socket_listen_error_is_benign(LIBUS_SOCKET_DESCRIPTOR fd);
+
+LIBUS_SOCKET_DESCRIPTOR bsd_create_bound_socket(const char *host, int port, int options, int *out_port, int *error);
+
 #ifndef MSG_DONTWAIT
 #define MSG_DONTWAIT 0
 #endif

packages/bun-usockets/src/libusockets.h

@@ -227,6 +227,7 @@ struct us_udp_packet_buffer_t *us_create_udp_packet_buffer();
 
 struct us_udp_socket_t *us_create_udp_socket(us_loop_r loop, void (*data_cb)(struct us_udp_socket_t *, void *, int), void (*drain_cb)(struct us_udp_socket_t *), void (*close_cb)(struct us_udp_socket_t *), void (*recv_error_cb)(struct us_udp_socket_t *, int, int), const char *host, unsigned short port, int flags, int *err, void *user);
 
+
 void us_udp_socket_close(struct us_udp_socket_t *s);
 
 int us_udp_socket_set_broadcast(struct us_udp_socket_t *s, int enabled);
@@ -242,7 +243,7 @@ LIBUS_SOCKET_DESCRIPTOR us_udp_socket_fd(struct us_udp_socket_t *s);
 /* Adopts an already created (and usually already bound) UDP socket descriptor
  * instead of creating a new one. The fd is made non-blocking and the standard
  * receive-path options are applied. Returns null with *err set on failure. */
-struct us_udp_socket_t *us_create_udp_socket_from_fd(us_loop_r loop, void (*data_cb)(struct us_udp_socket_t *, void *, int), void (*drain_cb)(struct us_udp_socket_t *), void (*close_cb)(struct us_udp_socket_t *), void (*recv_error_cb)(struct us_udp_socket_t *, int, int), LIBUS_SOCKET_DESCRIPTOR fd, int *err, void *user);
+struct us_udp_socket_t *us_create_udp_socket_from_fd(us_loop_r loop, void (*data_cb)(struct us_udp_socket_t *, void *, int), void (*drain_cb)(struct us_udp_socket_t *), void (*close_cb)(struct us_udp_socket_t *), void (*recv_error_cb)(struct us_udp_socket_t *, int, int), LIBUS_SOCKET_DESCRIPTOR fd, int shared, int *err, void *user);
 
 /* This one is ugly, should be ext! not user */
 void *us_udp_socket_user(struct us_udp_socket_t *s);
@@ -395,6 +396,10 @@ struct us_listen_socket_t *us_socket_group_listen_unix(us_socket_group_r group,
     unsigned char kind, struct ssl_ctx_st *ssl_ctx,
     const char *path, size_t pathlen, int options, int socket_ext_size, int *error)
     __attribute__((nonnull(1, 4, 8)));  /* ssl_ctx nullable */
+struct us_listen_socket_t *us_socket_group_listen_fd(us_socket_group_r group,
+    unsigned char kind, struct ssl_ctx_st *ssl_ctx,
+    LIBUS_SOCKET_DESCRIPTOR fd, int backlog, int options, int socket_ext_size, int *error)
+    __attribute__((nonnull(1, 8)));  /* ssl_ctx nullable */
 void us_listen_socket_close(struct us_listen_socket_t *ls) nonnull_fn_decl;
 
 /* SNI: tree hangs off the listen socket. ssl_ctx is up_ref'd; user is opaque
@@ -706,7 +711,7 @@ LIBUS_SOCKET_DESCRIPTOR us_socket_get_fd(us_socket_r s) nonnull_fn_decl;
 
 /* Bun extras */
 struct us_socket_t *us_socket_pair(us_socket_group_r group, unsigned char kind, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR *fds) nonnull_fn_decl;
-struct us_socket_t *us_socket_from_fd(us_socket_group_r group, unsigned char kind, struct ssl_ctx_st *ssl_ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR fd, int ipc)
+struct us_socket_t *us_socket_from_fd(us_socket_group_r group, unsigned char kind, struct ssl_ctx_st *ssl_ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR fd, int options, int ipc)
     __attribute__((nonnull(1)));  /* ssl_ctx nullable */
 struct us_socket_t *us_socket_open(struct us_socket_t *s, int is_client, char *ip, int ip_length);
 int us_raw_root_certs(struct us_cert_string_t **out);

packages/bun-usockets/src/loop.c

@@ -961,7 +961,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                 do {
                     struct udp_recvbuf recvbuf;
                     bsd_udp_setup_recvbuf(&recvbuf, u->loop->data.recv_buf, LIBUS_RECV_BUFFER_LENGTH);
-                    int npackets = bsd_recvmmsg(us_poll_fd(p), &recvbuf, MSG_DONTWAIT);
+                    int npackets = bsd_recvmmsg(us_poll_fd(p), &recvbuf, MSG_DONTWAIT, u->shared_fd ? 1 : LIBUS_UDP_RECV_COUNT);
                     if (npackets > 0) {
                         u->on_data(u, &recvbuf, npackets);
                     } else {

packages/bun-usockets/src/socket.c

@@ -405,7 +405,7 @@ struct us_socket_t *us_socket_pair(struct us_socket_group_t *group, unsigned cha
         return 0;
     }
 
-    return us_socket_from_fd(group, kind, NULL, socket_ext_size, fds[0], 0);
+    return us_socket_from_fd(group, kind, NULL, socket_ext_size, fds[0], 0, 0);
 #endif
 }
 
@@ -445,10 +445,7 @@ int us_socket_write2(struct us_socket_t *s, const char *header, int header_lengt
     return written < 0 ? 0 : written;
 }
 
-struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned char kind, struct ssl_ctx_st *ssl_ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR fd, int ipc) {
-#if defined(LIBUS_USE_LIBUV) || defined(WIN32)
-    return 0;
-#else
+struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned char kind, struct ssl_ctx_st *ssl_ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR fd, int options, int ipc) {
     struct us_poll_t *p1 = us_create_poll(group->loop, 0, sizeof(struct us_socket_t) + socket_ext_size);
     us_poll_init(p1, fd, POLL_TYPE_SOCKET);
     int rc = us_poll_start_rc(p1, group->loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE);
@@ -464,7 +461,7 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned
     s->timeout = 255;
     s->long_timeout = 255;
     s->flags.low_prio_state = 0;
-    s->flags.allow_half_open = 0;
+    s->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN) != 0;
     s->flags.is_paused = 0;
     s->flags.is_ipc = ipc;
     s->flags.is_closed = 0;
@@ -490,7 +487,6 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned
     }
 
     return s;
-#endif
 }
 
 void *us_socket_get_native_handle(struct us_socket_t *s) {
@@ -690,12 +686,21 @@ int us_socket_ipc_write_fd(struct us_socket_t *s, const char *data, int length,
 
     int sent = bsd_sendmsg(us_poll_fd(&s->p), &msg, 0);
 
+    if (sent < 0) {
+        if (errno == EAGAIN || errno == EWOULDBLOCK || errno == ENOBUFS) {
+            s->flags.last_write_failed = 1;
+            us_poll_change(&s->p, s->group->loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE);
+            return 0;
+        }
+        return -1;
+    }
+
     if (sent != length) {
         s->flags.last_write_failed = 1;
         us_internal_rearm_writable(s);
     }
 
-    return sent < 0 ? 0 : sent;
+    return sent;
 }
 #endif