more child-process (#18688)

Co-authored-by: pfgithub <6010774+pfgithub@users.noreply.github.com>

Author: pfgithub

package.json

@@ -32,6 +32,7 @@
   "scripts": {
     "build": "bun run build:debug",
     "watch": "zig build check --watch -fincremental --prominent-compile-errors --global-cache-dir build/debug/zig-check-cache --zig-lib-dir vendor/zig/lib",
+    "watch-windows": "zig build check-windows --watch -fincremental --prominent-compile-errors --global-cache-dir build/debug/zig-check-cache --zig-lib-dir vendor/zig/lib",
     "bd": "(bun run --silent build:debug &> /tmp/bun.debug.build.log || (cat /tmp/bun.debug.build.log && rm -rf /tmp/bun.debug.build.log && exit 1)) && rm -f /tmp/bun.debug.build.log && ./build/debug/bun-debug",
     "build:debug": "bun ./scripts/build.mjs -GNinja -DCMAKE_BUILD_TYPE=Debug -B build/debug",
     "build:valgrind": "bun ./scripts/build.mjs -GNinja -DCMAKE_BUILD_TYPE=Debug -DENABLE_BASELINE=ON -ENABLE_VALGRIND=ON -B build/debug-valgrind",

packages/bun-types/bun.d.ts

@@ -6613,9 +6613,10 @@ declare module "bun" {
       ipc?(
         message: any,
         /**
-         * The {@link Subprocess} that sent the message
+         * The {@link Subprocess} that received the message
          */
         subprocess: Subprocess<In, Out, Err>,
+        handle?: unknown,
       ): void;
 
       /**

packages/bun-usockets/src/bsd.c

@@ -725,6 +725,20 @@ ssize_t bsd_recv(LIBUS_SOCKET_DESCRIPTOR fd, void *buf, int length, int flags) {
     }
 }
 
+#if !defined(_WIN32)
+ssize_t bsd_recvmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct msghdr *msg, int flags) {
+    while (1) {
+        ssize_t ret = recvmsg(fd, msg, flags);
+
+        if (UNLIKELY(IS_EINTR(ret))) {
+            continue;
+        }
+
+        return ret;
+    }
+}
+#endif
+
 #if !defined(_WIN32)
 #include <sys/uio.h>
 
@@ -783,6 +797,20 @@ ssize_t bsd_send(LIBUS_SOCKET_DESCRIPTOR fd, const char *buf, int length, int ms
     }
 }
 
+#if !defined(_WIN32)
+ssize_t bsd_sendmsg(LIBUS_SOCKET_DESCRIPTOR fd, const struct msghdr *msg, int flags) {
+    while (1) {
+        ssize_t rc = sendmsg(fd, msg, flags);
+
+        if (UNLIKELY(IS_EINTR(rc))) {
+            continue;
+        }
+
+        return rc;
+    }
+}
+#endif
+
 int bsd_would_block() {
 #ifdef _WIN32
     return WSAGetLastError() == WSAEWOULDBLOCK;

packages/bun-usockets/src/context.c

@@ -279,14 +279,15 @@ struct us_socket_context_t *us_create_socket_context(int ssl, struct us_loop_t *
     return context;
 }
 
-struct us_socket_context_t *us_create_bun_socket_context(int ssl, struct us_loop_t *loop, int context_ext_size, struct us_bun_socket_context_options_t options, enum create_bun_socket_error_t *err) {
+struct us_socket_context_t *us_create_bun_ssl_socket_context(struct us_loop_t *loop, int context_ext_size, struct us_bun_socket_context_options_t options, enum create_bun_socket_error_t *err) {
 #ifndef LIBUS_NO_SSL
-    if (ssl) {
-        /* This function will call us, again, with SSL = false and a bigger ext_size */
-        return (struct us_socket_context_t *) us_internal_bun_create_ssl_socket_context(loop, context_ext_size, options, err);
-    }
+    /* This function will call us, again, with SSL = false and a bigger ext_size */
+    return (struct us_socket_context_t *) us_internal_bun_create_ssl_socket_context(loop, context_ext_size, options, err);
 #endif
+    return us_create_bun_nossl_socket_context(loop, context_ext_size);
+}
 
+struct us_socket_context_t *us_create_bun_nossl_socket_context(struct us_loop_t *loop, int context_ext_size) {
     /* This path is taken once either way - always BEFORE whatever SSL may do LATER.
      * context_ext_size will however be modified larger in case of SSL, to hold SSL extensions */
 
@@ -370,8 +371,8 @@ struct us_listen_socket_t *us_socket_context_listen(int ssl, struct us_socket_co
     ls->s.timeout = 255;
     ls->s.long_timeout = 255;
     ls->s.flags.low_prio_state = 0;
-        ls->s.flags.is_paused = 0;
-
+    ls->s.flags.is_paused = 0;
+    ls->s.flags.is_ipc = 0;
     ls->s.next = 0;
     ls->s.flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN);
     us_internal_socket_context_link_listen_socket(context, ls);
@@ -406,6 +407,7 @@ struct us_listen_socket_t *us_socket_context_listen_unix(int ssl, struct us_sock
     ls->s.flags.low_prio_state = 0;
     ls->s.flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN);
     ls->s.flags.is_paused = 0;
+    ls->s.flags.is_ipc = 0;
     ls->s.next = 0;
     us_internal_socket_context_link_listen_socket(context, ls);
 
@@ -414,7 +416,6 @@ struct us_listen_socket_t *us_socket_context_listen_unix(int ssl, struct us_sock
     return ls;
 }
 
-
 struct us_socket_t* us_socket_context_connect_resolved_dns(struct us_socket_context_t *context, struct sockaddr_storage* addr, int options, int socket_ext_size) {
     LIBUS_SOCKET_DESCRIPTOR connect_socket_fd = bsd_create_connect_socket(addr, options);
     if (connect_socket_fd == LIBUS_SOCKET_ERROR) {
@@ -437,6 +438,7 @@ struct us_socket_t* us_socket_context_connect_resolved_dns(struct us_socket_cont
     socket->flags.low_prio_state = 0;
     socket->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN);
     socket->flags.is_paused = 0;
+    socket->flags.is_ipc = 0;
     socket->connect_state = NULL;
 
 
@@ -563,6 +565,7 @@ int start_connections(struct us_connecting_socket_t *c, int count) {
         s->flags.low_prio_state = 0;
         s->flags.allow_half_open = (c->options & LIBUS_SOCKET_ALLOW_HALF_OPEN);
         s->flags.is_paused = 0;
+        s->flags.is_ipc = 0;
         /* Link it into context so that timeout fires properly */
         us_internal_socket_context_link_socket(s->context, s);
 
@@ -739,6 +742,7 @@ struct us_socket_t *us_socket_context_connect_unix(int ssl, struct us_socket_con
     connect_socket->flags.low_prio_state = 0;
     connect_socket->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN);
     connect_socket->flags.is_paused = 0;
+    connect_socket->flags.is_ipc = 0;
     connect_socket->connect_state = NULL;
     connect_socket->connect_next = NULL;
     us_internal_socket_context_link_socket(context, connect_socket);
@@ -843,6 +847,14 @@ void us_socket_context_on_data(int ssl, struct us_socket_context_t *context, str
     context->on_data = on_data;
 }
 
+void us_socket_context_on_fd(int ssl, struct us_socket_context_t *context, struct us_socket_t *(*on_fd)(struct us_socket_t *s, int fd)) {
+#ifndef LIBUS_NO_SSL
+    if (ssl) return;
+#endif
+
+    context->on_fd = on_fd;
+}
+
 void us_socket_context_on_writable(int ssl, struct us_socket_context_t *context, struct us_socket_t *(*on_writable)(struct us_socket_t *s)) {
 #ifndef LIBUS_NO_SSL
     if (ssl) {

packages/bun-usockets/src/crypto/openssl.c

@@ -1535,10 +1535,9 @@ us_internal_bun_create_ssl_socket_context(
   /* Otherwise ee continue by creating a non-SSL context, but with larger ext to
    * hold our SSL stuff */
   struct us_internal_ssl_socket_context_t *context =
-      (struct us_internal_ssl_socket_context_t *)us_create_bun_socket_context(
-          0, loop,
-          sizeof(struct us_internal_ssl_socket_context_t) + context_ext_size,
-          options, err);
+      (struct us_internal_ssl_socket_context_t *)us_create_bun_nossl_socket_context(
+          loop,
+          sizeof(struct us_internal_ssl_socket_context_t) + context_ext_size);
 
   /* I guess this is the only optional callback */
   context->on_server_name = NULL;
@@ -2080,8 +2079,8 @@ struct us_internal_ssl_socket_t *us_internal_ssl_socket_wrap_with_tls(
   us_socket_context_ref(0,old_context);
 
   enum create_bun_socket_error_t err = CREATE_BUN_SOCKET_ERROR_NONE;
-  struct us_socket_context_t *context = us_create_bun_socket_context(
-      1, old_context->loop, sizeof(struct us_wrapped_socket_context_t),
+  struct us_socket_context_t *context = us_create_bun_ssl_socket_context(
+      old_context->loop, sizeof(struct us_wrapped_socket_context_t),
       options, &err);
 
   // Handle SSL context creation failure

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

@@ -170,6 +170,8 @@ struct us_socket_flags {
     bool allow_half_open: 1; 
     /* 0 = not in low-prio queue, 1 = is in low-prio queue, 2 = was in low-prio queue in this iteration */
     unsigned char low_prio_state: 2; 
+    /* If true, the socket should be read using readmsg to support receiving file descriptors */
+    bool is_ipc: 1;
 
 } __attribute__((packed));
 
@@ -287,6 +289,7 @@ struct us_socket_context_t {
   struct us_socket_t *(*on_open)(struct us_socket_t *, int is_client, char *ip,
                                  int ip_length);
   struct us_socket_t *(*on_data)(struct us_socket_t *, char *data, int length);
+  struct us_socket_t *(*on_fd)(struct us_socket_t *, int fd);
   struct us_socket_t *(*on_writable)(struct us_socket_t *);
   struct us_socket_t *(*on_close)(struct us_socket_t *, int code, void *reason);
   // void (*on_timeout)(struct us_socket_context *);

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

@@ -207,7 +207,13 @@ int bsd_addr_get_port(struct bsd_addr_t *addr);
 LIBUS_SOCKET_DESCRIPTOR bsd_accept_socket(LIBUS_SOCKET_DESCRIPTOR fd, struct bsd_addr_t *addr);
 
 ssize_t bsd_recv(LIBUS_SOCKET_DESCRIPTOR fd, void *buf, int length, int flags);
+#if !defined(_WIN32)
+ssize_t bsd_recvmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct msghdr *msg, int flags);
+#endif
 ssize_t bsd_send(LIBUS_SOCKET_DESCRIPTOR fd, const char *buf, int length, int msg_more);
+#if !defined(_WIN32)
+ssize_t bsd_sendmsg(LIBUS_SOCKET_DESCRIPTOR fd, const struct msghdr *msg, int flags);
+#endif
 ssize_t bsd_write2(LIBUS_SOCKET_DESCRIPTOR fd, const char *header, int header_length, const char *payload, int payload_length);
 int bsd_would_block();
 

packages/bun-usockets/src/libusockets.h

@@ -264,8 +264,10 @@ enum create_bun_socket_error_t {
   CREATE_BUN_SOCKET_ERROR_INVALID_CA,
 };
 
-struct us_socket_context_t *us_create_bun_socket_context(int ssl, struct us_loop_t *loop,
+struct us_socket_context_t *us_create_bun_ssl_socket_context(struct us_loop_t *loop,
     int ext_size, struct us_bun_socket_context_options_t options, enum create_bun_socket_error_t *err);
+struct us_socket_context_t *us_create_bun_nossl_socket_context(struct us_loop_t *loop,
+    int ext_size);
 
 /* Delete resources allocated at creation time (will call unref now and only free when ref count == 0). */
 void us_socket_context_free(int ssl, us_socket_context_r context) nonnull_fn_decl;
@@ -280,6 +282,8 @@ void us_socket_context_on_close(int ssl, us_socket_context_r context,
     struct us_socket_t *(*on_close)(us_socket_r s, int code, void *reason));
 void us_socket_context_on_data(int ssl, us_socket_context_r context,
     struct us_socket_t *(*on_data)(us_socket_r s, char *data, int length));
+void us_socket_context_on_fd(int ssl, us_socket_context_r context,
+    struct us_socket_t *(*on_fd)(us_socket_r s, int fd));
 void us_socket_context_on_writable(int ssl, us_socket_context_r context,
     struct us_socket_t *(*on_writable)(us_socket_r s));
 void us_socket_context_on_timeout(int ssl, us_socket_context_r context,
@@ -465,7 +469,7 @@ void us_socket_local_address(int ssl, us_socket_r s, char *nonnull_arg buf, int
 
 /* Bun extras */
 struct us_socket_t *us_socket_pair(struct us_socket_context_t *ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR* fds);
-struct us_socket_t *us_socket_from_fd(struct us_socket_context_t *ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR fd);
+struct us_socket_t *us_socket_from_fd(struct us_socket_context_t *ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR fd, int ipc);
 struct us_socket_t *us_socket_wrap_with_tls(int ssl, us_socket_r s, struct us_bun_socket_context_options_t options, struct us_socket_events_t events, int socket_ext_size);
 int us_socket_raw_write(int ssl, us_socket_r s, const char *data, int length, int msg_more);
 struct us_socket_t* us_socket_open(int ssl, struct us_socket_t * s, int is_client, char* ip, int ip_length);

packages/bun-usockets/src/loop.c

@@ -313,6 +313,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                         s->flags.low_prio_state = 0;
                         s->flags.allow_half_open = listen_socket->s.flags.allow_half_open;
                         s->flags.is_paused = 0;
+                        s->flags.is_ipc = 0;
 
                         /* We always use nodelay */
                         bsd_socket_nodelay(client_fd, 1);
@@ -391,7 +392,43 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                       const int recv_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
                     #endif
 
-                    int length = bsd_recv(us_poll_fd(&s->p), loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, LIBUS_RECV_BUFFER_LENGTH, recv_flags);
+                    int length;
+                    #if !defined(_WIN32)
+                    if(s->flags.is_ipc) {
+                        struct msghdr msg = {0};
+                        struct iovec iov = {0};
+                        char cmsg_buf[CMSG_SPACE(sizeof(int))];
+                        
+                        iov.iov_base = loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING;
+                        iov.iov_len = LIBUS_RECV_BUFFER_LENGTH;
+
+                        msg.msg_flags = 0;                        
+                        msg.msg_iov = &iov;
+                        msg.msg_iovlen = 1;
+                        msg.msg_name = NULL;
+                        msg.msg_namelen = 0;
+                        msg.msg_controllen = CMSG_LEN(sizeof(int));
+                        msg.msg_control = cmsg_buf;
+                        
+                        length = bsd_recvmsg(us_poll_fd(&s->p), &msg, recv_flags);
+                        
+                        // Extract file descriptor if present
+                        if (length > 0 && msg.msg_controllen > 0) {
+                            struct cmsghdr *cmsg_ptr = CMSG_FIRSTHDR(&msg);
+                            if (cmsg_ptr && cmsg_ptr->cmsg_level == SOL_SOCKET && cmsg_ptr->cmsg_type == SCM_RIGHTS) {
+                                int fd = *(int *)CMSG_DATA(cmsg_ptr);
+                                s = s->context->on_fd(s, fd);
+                                if(us_socket_is_closed(0, s)) {
+                                    break;
+                                }
+                            }
+                        }
+                    }else{
+                    #endif
+                        length = bsd_recv(us_poll_fd(&s->p), loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, LIBUS_RECV_BUFFER_LENGTH, recv_flags);
+                    #if !defined(_WIN32)
+                    }
+                    #endif
 
                     if (length > 0) {
                         s = s->context->on_data(s, loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, length);

packages/bun-usockets/src/socket.c

@@ -284,7 +284,7 @@ struct us_socket_t *us_socket_pair(struct us_socket_context_t *ctx, int socket_e
         return 0;
     }
 
-    return us_socket_from_fd(ctx, socket_ext_size, fds[0]);
+    return us_socket_from_fd(ctx, socket_ext_size, fds[0], 0);
 #endif
 }
 
@@ -302,7 +302,7 @@ int us_socket_write2(int ssl, struct us_socket_t *s, const char *header, int hea
     return written < 0 ? 0 : written;
 }
 
-struct us_socket_t *us_socket_from_fd(struct us_socket_context_t *ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR fd) {
+struct us_socket_t *us_socket_from_fd(struct us_socket_context_t *ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR fd, int ipc) {
 #if defined(LIBUS_USE_LIBUV) || defined(WIN32)
     return 0;
 #else
@@ -321,6 +321,8 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_context_t *ctx, int socke
     s->flags.low_prio_state = 0;
     s->flags.allow_half_open = 0;
     s->flags.is_paused = 0;
+    s->flags.is_ipc = 0;
+    s->flags.is_ipc = ipc;
     s->connect_state = NULL;
 
     /* We always use nodelay */
@@ -374,6 +376,44 @@ int us_socket_write(int ssl, struct us_socket_t *s, const char *data, int length
     return written < 0 ? 0 : written;
 }
 
+#if !defined(_WIN32)
+/* Send a message with data and an attached file descriptor, for use in IPC. Returns the number of bytes written. If that
+    number is less than the length, the file descriptor was not sent. */
+int us_socket_ipc_write_fd(struct us_socket_t *s, const char* data, int length, int fd) {
+    if (us_socket_is_closed(0, s) || us_socket_is_shut_down(0, s)) {
+        return 0;
+    }
+
+    struct msghdr msg = {0};
+    struct iovec iov = {0};
+    char cmsgbuf[CMSG_SPACE(sizeof(int))];
+    
+    iov.iov_base = (void*)data;
+    iov.iov_len = length;
+    
+    msg.msg_iov = &iov;
+    msg.msg_iovlen = 1;
+    msg.msg_control = cmsgbuf;
+    msg.msg_controllen = CMSG_SPACE(sizeof(int));
+    
+    struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg);
+    cmsg->cmsg_level = SOL_SOCKET;
+    cmsg->cmsg_type = SCM_RIGHTS;
+    cmsg->cmsg_len = CMSG_LEN(sizeof(int));
+    
+    *(int *)CMSG_DATA(cmsg) = fd;
+    
+    int sent = bsd_sendmsg(us_poll_fd(&s->p), &msg, 0);
+    
+    if (sent != length) {
+        s->context->loop->data.last_write_failed = 1;
+        us_poll_change(&s->p, s->context->loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE);
+    }
+    
+    return sent < 0 ? 0 : sent;
+}
+#endif
+
 void *us_socket_ext(int ssl, struct us_socket_t *s) {
 #ifndef LIBUS_NO_SSL
     if (ssl) {