Fix file descriptor leak in ipc (#8165)

* Fix file descriptor leak in ipc

* Do not assume sequential file descriptors

---------

Co-authored-by: Jarred Sumner <709451+Jarred-Sumner@users.noreply.github.com>

packages/bun-usockets/src/socket.c

@@ -22,6 +22,10 @@
 #include <string.h>
 #include <stdint.h>
 
+#ifndef WIN32
+#include <fcntl.h>
+#endif
+
 /* Shared with SSL */
 
 int us_socket_local_port(int ssl, struct us_socket_t *s) {
@@ -247,6 +251,12 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_context_t *ctx, int socke
     /* We always use nodelay */
     bsd_socket_nodelay(fd, 1);
 
+    int flags = fcntl(fd, F_GETFL, 0);
+    if (flags != -1) {
+        flags |= O_NONBLOCK;
+        fcntl(fd, F_SETFL, flags);
+    }
+
     us_internal_socket_context_link_socket(ctx, s);
 
     return s;

src/bun.js/api/bun/subprocess.zig

@@ -1387,6 +1387,13 @@ pub const Subprocess = struct {
         var ipc_mode = IPCMode.none;
         var ipc_callback: JSValue = .zero;
         var stdio_pipes: std.ArrayListUnmanaged(Stdio.PipeExtra) = .{};
+        var pipes_to_close: std.ArrayListUnmanaged(bun.FileDescriptor) = .{};
+        defer {
+            for (pipes_to_close.items) |pipe_fd| {
+                _ = bun.sys.close(pipe_fd);
+            }
+            pipes_to_close.clearAndFree(bun.default_allocator);
+        }
 
         var windows_hide: if (Environment.isWindows) u1 else u0 = 0;
 
@@ -1863,6 +1870,7 @@ pub const Subprocess = struct {
                     .PROTONOSUPPORT => break :blk error.ProtocolNotSupported,
                     else => |err| break :blk std.os.unexpectedErrno(err),
                 }
+                pipes_to_close.append(bun.default_allocator, bun.toFD(fds[1])) catch |err| break :blk err;
                 actions.dup2(bun.toFD(fds[1]), bun.toFD(item.fileno)) catch |err| break :blk err;
                 actions.close(bun.toFD(fds[1])) catch |err| break :blk err;
                 item.fd = fds[0];
@@ -1915,7 +1923,14 @@ pub const Subprocess = struct {
                 });
                 return .zero;
             };
+            pipes_to_close.append(bun.default_allocator, bun.toFD(fds[1])) catch |err| return globalThis.handleError(err, "in posix_spawn");
             actions.dup2(bun.toFD(fds[1]), bun.toFD(3)) catch |err| return globalThis.handleError(err, "in posix_spawn");
+            actions.close(bun.toFD(fds[1])) catch |err| return globalThis.handleError(err, "in posix_spawn");
+            // enable non-block
+            const before = std.c.fcntl(fds[0], os.F.GETFL);
+            _ = std.c.fcntl(fds[0], os.F.SETFL, before | os.O.NONBLOCK);
+            // enable SOCK_CLOXEC
+            _ = std.c.fcntl(fds[0], os.FD_CLOEXEC);
         }
 
         env_array.append(allocator, null) catch {
@@ -1935,9 +1950,12 @@ pub const Subprocess = struct {
                 if (stdio[2].isPiped()) {
                     _ = bun.sys.close(bun.toFD(stderr_pipe[1]));
                 }
-                for (stdio_pipes.items) |item| {
-                    _ = bun.sys.close(bun.toFD(item.fd + 1));
+
+                // we always close these, but we want to close these earlier
+                for (pipes_to_close.items) |pipe_fd| {
+                    _ = bun.sys.close(pipe_fd);
                 }
+                pipes_to_close.clearAndFree(bun.default_allocator);
             }
 
             break :brk switch (PosixSpawn.spawnZ(argv.items[0].?, actions, attr, @as([*:null]?[*:0]const u8, @ptrCast(argv.items[0..].ptr)), env)) {

test/integration/next/default-pages-dir/test/dev-server.test.ts

@@ -4,6 +4,7 @@ import { bunEnv, bunExe } from "../../../../harness";
 import { Subprocess } from "bun";
 import { copyFileSync, rmSync } from "fs";
 import { join } from "path";
+import { StringDecoder } from "string_decoder";
 
 const root = join(import.meta.dir, "../");
 let dev_server: undefined | Subprocess<"ignore", "pipe", "inherit">;
@@ -25,9 +26,11 @@ test("the dev server can start", async () => {
   dev_server.exited.then(() => {
     dev_server = undefined;
   });
+
+  var string_decoder = new StringDecoder("utf-8");
   for await (const chunk of dev_server.stdout) {
-    console.error({ chunk: new TextDecoder().decode(chunk) });
-    const str = new TextDecoder().decode(chunk);
+    const str = string_decoder.write(chunk);
+    console.error(str);
     let match = str.match(/http:\/\/localhost:\d+/);
     if (match) {
       baseUrl = match[0];
@@ -66,7 +69,7 @@ test("ssr works for 100 requests", async () => {
   for (const y of x) {
     expect(y.status).toBe("fulfilled");
   }
-}, 10000);
+}, 100000);
 
 test("hot reloading works on the client (+ tailwind hmr)", async () => {
   expect(dev_server).not.toBeUndefined();
@@ -81,5 +84,6 @@ test("hot reloading works on the client (+ tailwind hmr)", async () => {
 }, 30000);
 
 afterAll(() => {
-  Bun.spawnSync(["pkill", "-P", dev_server!.pid.toString()]);
+  const pid = dev_server?.pid?.toString?.()!;
+  if (pid) Bun.spawnSync(["pkill", "-P", pid]);
 });