usockets: keep connect()'s answer; a peer's early close is not a failed connect

packages/bun-usockets/src/bsd.c

@@ -1891,14 +1891,18 @@ int bsd_disconnect_udp_socket(LIBUS_SOCKET_DESCRIPTOR fd) {
     }
 }
 
-static int bsd_do_connect_raw(LIBUS_SOCKET_DESCRIPTOR fd, struct sockaddr *addr, size_t namelen)
+/* Returns 0 for a connect that is established or in progress. *connected
+ * tells the two apart: it is 1 only when connect() itself returned 0. */
+static int bsd_do_connect_raw(LIBUS_SOCKET_DESCRIPTOR fd, struct sockaddr *addr, size_t namelen, int *connected)
 {
     ssize_t injected = 0; int unused = 0;
+    *connected = 0;
     if (US_FAULT_CHECK(US_FAULT_CONNECT, fd, injected, unused)) return errno;
     (void)injected; (void)unused;
 #ifdef _WIN32
     while (1) {
         if (connect(fd, (struct sockaddr *)addr, namelen) == 0) {
+            *connected = 1;
             return 0;
         }
 
@@ -1936,6 +1940,7 @@ static int bsd_do_connect_raw(LIBUS_SOCKET_DESCRIPTOR fd, struct sockaddr *addr,
         return errno;
     }
 
+    *connected = r == 0;
     return 0;
 #endif
 }
@@ -2040,7 +2045,9 @@ LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket(struct sockaddr_storage *addr,
     }
 
 #endif
-    int rc = bsd_do_connect_raw(fd, (struct sockaddr*) addr, addr->ss_family == AF_INET ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6));
+    /* Unused: only bsd_create_connect_socket_unix reports it. */
+    int connected;
+    int rc = bsd_do_connect_raw(fd, (struct sockaddr*) addr, addr->ss_family == AF_INET ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6), &connected);
 
     if (rc != 0) {
         bsd_close_socket(fd);
@@ -2055,7 +2062,7 @@ LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket(struct sockaddr_storage *addr,
     return fd;
 }
 
-static LIBUS_SOCKET_DESCRIPTOR internal_bsd_create_connect_socket_unix(const char *server_path, size_t len, int options, struct sockaddr_un* server_address, const size_t addrlen) {
+static LIBUS_SOCKET_DESCRIPTOR internal_bsd_create_connect_socket_unix(const char *server_path, size_t len, int options, struct sockaddr_un* server_address, const size_t addrlen, int *connected) {
     LIBUS_SOCKET_DESCRIPTOR fd = bsd_create_socket(AF_UNIX, SOCK_STREAM, 0, NULL);
 
     if (fd == LIBUS_SOCKET_ERROR) {
@@ -2064,7 +2071,7 @@ static LIBUS_SOCKET_DESCRIPTOR internal_bsd_create_connect_socket_unix(const cha
 
     win32_set_nonblocking(fd);
 
-    int rc = bsd_do_connect_raw(fd, (struct sockaddr *)server_address, addrlen);
+    int rc = bsd_do_connect_raw(fd, (struct sockaddr *)server_address, addrlen, connected);
     if (rc != 0) {
         bsd_close_socket(fd);
 #ifdef _WIN32
@@ -2076,7 +2083,7 @@ static LIBUS_SOCKET_DESCRIPTOR internal_bsd_create_connect_socket_unix(const cha
     return fd;
 }
 
-LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket_unix(const char *server_path, size_t len, int options) {
+LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket_unix(const char *server_path, size_t len, int options, int *connected) {
     struct sockaddr_un server_address;
     size_t addrlen = 0;
     int dirfd_workaround_for_unix_path_len = -1;
@@ -2094,7 +2101,7 @@ LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket_unix(const char *server_path,
     }
 #endif
 
-    LIBUS_SOCKET_DESCRIPTOR fd = internal_bsd_create_connect_socket_unix(server_path, len, options, &server_address, addrlen);
+    LIBUS_SOCKET_DESCRIPTOR fd = internal_bsd_create_connect_socket_unix(server_path, len, options, &server_address, addrlen, connected);
 
 #if defined(__APPLE__)
     if (dirfd_workaround_for_unix_path_len != -1) {

packages/bun-usockets/src/context.c

@@ -347,6 +347,7 @@ static void us_internal_init_listen_socket(struct us_listen_socket_t *ls,
     s->unclassified_send_failures = 0;
     s->read_eof = 0;
     s->hangup_closes_unsent = 0;
+    s->connect_returned_zero = 0;
     s->next = 0;
     s->prev = 0;
     s->connect_state = NULL;
@@ -527,6 +528,7 @@ static inline void us_internal_init_connect_socket(struct us_socket_t *s,
     s->unclassified_send_failures = 0;
     s->read_eof = 0;
     s->hangup_closes_unsent = 0;
+    s->connect_returned_zero = 0;
     s->connect_state = NULL;
     s->connect_next = NULL;
 }
@@ -653,7 +655,8 @@ void *us_socket_group_connect(struct us_socket_group_t *group, unsigned char kin
 struct us_socket_t *us_socket_group_connect_unix(struct us_socket_group_t *group,
         unsigned char kind, struct ssl_ctx_st *ssl_ctx,
         const char *server_path, size_t pathlen, int options, int socket_ext_size) {
-    LIBUS_SOCKET_DESCRIPTOR connect_socket_fd = bsd_create_connect_socket_unix(server_path, pathlen, options);
+    int connected = 0;
+    LIBUS_SOCKET_DESCRIPTOR connect_socket_fd = bsd_create_connect_socket_unix(server_path, pathlen, options, &connected);
     if (connect_socket_fd == LIBUS_SOCKET_ERROR) {
         return 0;
     }
@@ -670,6 +673,7 @@ struct us_socket_t *us_socket_group_connect_unix(struct us_socket_group_t *group
 
     struct us_socket_t *connect_socket = (struct us_socket_t *) p;
     us_internal_init_connect_socket(connect_socket, group, kind, options);
+    connect_socket->connect_returned_zero = connected;
 
     if (ssl_ctx) {
         us_internal_ssl_attach(connect_socket, ssl_ctx, 1, NULL, NULL);

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

@@ -370,6 +370,8 @@ struct us_socket_t {
   /* A hangup that leaves bytes unsent closes the socket even while it is paused (loop.c defers it otherwise).
    * For an owner whose pause can wait for those bytes to drain: node:http's pipelining. */
   unsigned char hangup_closes_unsent : 1;
+  /* connect() itself returned 0 (an AF_UNIX connect): the connection was established before the first poll event. */
+  unsigned char connect_returned_zero : 1;
   /* The close code passed to the deferred close (e.g. a reset requested from
    * inside a handshake callback must still RST, not FIN, when it is finally
    * performed). */

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

@@ -242,7 +242,9 @@ int bsd_disconnect_udp_socket(LIBUS_SOCKET_DESCRIPTOR fd);
 
 LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket(struct sockaddr_storage *addr, struct sockaddr_storage *local_addr, int options);
 
-LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket_unix(const char *server_path, size_t pathlen, int options);
+/* *connected becomes 1 when connect() itself returned 0: the connection is
+ * then established before this returns. */
+LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket_unix(const char *server_path, size_t pathlen, int options, int *connected);
 
 int bsd_socket_export_size(void);
 int bsd_socket_export(LIBUS_SOCKET_DESCRIPTOR fd, unsigned int target_pid, void *info_out);

packages/bun-usockets/src/loop.c

@@ -498,16 +498,18 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
              * rather than exact equality so the connect-complete is still
              * recognized; listen sockets only ever poll READABLE. */
             if (us_poll_events(p) & LIBUS_SOCKET_WRITABLE) {
-                /* The connecting fd became writable with an error/HUP flag also
-                 * set: the handshake may have completed and then been reset
-                 * before we collected the event. Report the kernel's actual
-                 * SO_ERROR (ECONNRESET for that race) instead of the literal
-                 * boolean, which downstream would misreport as ECONNREFUSED.
-                 * libuv does the same getsockopt in uv__stream_connect. */
+                /* An error or hangup flag on the first event: SO_ERROR has the
+                 * verdict, as in libuv's uv__stream_connect. SO_ERROR 0 has two
+                 * meanings. If connect() already returned 0, the connection is
+                 * established and its peer closed before this event: open it,
+                 * and the socket arm below reads what the peer sent, then the
+                 * end. If connect() was in progress, something took the error
+                 * first (a send() on this fd, another SO_ERROR read): the
+                 * connect failed, and ECONNRESET stands in for the lost code. */
                 int connect_error = 0;
                 if (error || eof) {
                     connect_error = us_socket_get_error((struct us_socket_t *) p);
-                    if (connect_error == 0) {
+                    if (connect_error == 0 && !((struct us_socket_t *) p)->connect_returned_zero) {
                         connect_error = LIBUS_ECONNRESET;
                     }
                 }
@@ -556,6 +558,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                         s->unclassified_send_failures = 0;
                         s->read_eof = 0;
                         s->hangup_closes_unsent = 0;
+                        s->connect_returned_zero = 0;
 
                         /* We always use nodelay */
                         bsd_socket_nodelay(client_fd, 1);

packages/bun-usockets/src/socket.c

@@ -445,6 +445,7 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned
     s->unclassified_send_failures = 0;
     s->read_eof = 0;
     s->hangup_closes_unsent = 0;
+    s->connect_returned_zero = 0;
     s->connect_state = NULL;
 
     /* We always use nodelay */

test/js/bun/net/unix-socket-peer-closes-first.test.ts

@@ -0,0 +1,224 @@
+import { SQL } from "bun";
+import { afterAll, beforeAll, describe, expect, it } from "bun:test";
+import { isLinux, isWindows, tempDir } from "harness";
+import { once } from "node:events";
+import net from "node:net";
+import { join } from "node:path";
+import { Worker } from "node:worker_threads";
+import { mysqlRawPacket } from "../../sql/wire-frames";
+
+// The peer runs on a worker thread, so it can act while this thread is blocked and cannot poll.
+// The first event this thread then sees for the connection is the peer's hangup.
+// Not concurrent: the tests share one peer and its state.
+describe.skipIf(isWindows)("a unix socket peer that closes before the connecting thread polls", () => {
+  // state[0]: the peer sets it when its part is done (1: as asked, 2: not as asked).
+  // state[1]: this thread sets it when its connect call has returned.
+  const state = new Int32Array(new SharedArrayBuffer(8));
+  let dir: ReturnType<typeof tempDir>;
+  let worker: Worker;
+  let sockets = 0;
+
+  beforeAll(() => {
+    dir = tempDir("uds-peer-closes-first", {
+      "peer.mjs": `
+        import { parentPort, workerData as state } from "node:worker_threads";
+        let listener;
+        parentPort.on("message", ({ listen, answer }) => {
+          listener?.stop(true);
+          Atomics.store(state, 0, 0);
+          Atomics.store(state, 1, 0);
+          let accepted = 0;
+          listener = Bun.listen({
+            ...listen,
+            socket: {
+              open(socket) {
+                accepted++;
+                if (!answer) return;
+                const wrote = socket.write(answer);
+                // close(2) at once for a unix socket. A FIN for TCP, where terminate() is a reset.
+                if (listen.unix) socket.terminate();
+                else socket.end();
+                Atomics.store(state, 0, wrote === answer.byteLength ? 1 : 2);
+                Atomics.notify(state, 0);
+              },
+              data() {},
+            },
+          });
+          parentPort.postMessage(listener.port ?? 0);
+          if (answer) return;
+          // No answer: close the listener with the connection still in its accept queue.
+          // This thread stays blocked, so it cannot accept, until the client has connected.
+          Atomics.wait(state, 1, 0, 10_000);
+          listener.stop(true);
+          listener = undefined;
+          Atomics.store(state, 0, accepted === 0 ? 1 : 2);
+          Atomics.notify(state, 0);
+        });
+      `,
+    });
+    worker = new Worker(join(String(dir), "peer.mjs"), { workerData: state });
+  });
+
+  afterAll(async () => {
+    await worker?.terminate();
+    dir?.[Symbol.dispose]();
+  });
+
+  // With an answer, the peer sends it as soon as it accepts, and closes.
+  // With no answer, the peer closes its listener and never accepts.
+  async function peer(answer?: Uint8Array, transport: "unix" | "tcp" = "unix") {
+    const unix = join(String(dir), `${sockets++}.sock`);
+    const listening = once(worker, "message");
+    worker.postMessage({ listen: transport === "tcp" ? { hostname: "127.0.0.1", port: 0 } : { unix }, answer });
+    const [port] = await listening;
+    return { unix, port: port as number };
+  }
+
+  // This thread must not poll between its connect call and the peer's last step:
+  // call this right after the connect call, with no await between the two.
+  function blockUntilPeerIsDone() {
+    Atomics.store(state, 1, 1);
+    Atomics.notify(state, 1);
+    if (Atomics.wait(state, 0, 0, 10_000) === "timed-out") return "timed out";
+    return Atomics.load(state, 0) === 1 ? "done" : "not as asked";
+  }
+
+  async function bunConnect(unix: string) {
+    const events: string[] = [];
+    const chunks: Buffer[] = [];
+    const closed = Promise.withResolvers<void>();
+    const connecting = Bun.connect({
+      unix,
+      socket: {
+        open() {
+          events.push("open");
+        },
+        data(_socket, chunk) {
+          chunks.push(Buffer.from(chunk));
+        },
+        end() {
+          events.push("end");
+        },
+        close() {
+          events.push("close");
+          closed.resolve();
+        },
+        connectError(_socket, error) {
+          events.push(`connectError ${(error as NodeJS.ErrnoException).code}`);
+          closed.resolve();
+        },
+      },
+    });
+    const peer = blockUntilPeerIsDone();
+    const connect = await connecting.then(
+      () => "fulfilled",
+      () => "rejected",
+    );
+    await closed.promise;
+    return { peer, connect, events, received: Buffer.concat(chunks) };
+  }
+
+  async function netConnect(unix: string) {
+    const events: string[] = [];
+    const chunks: Buffer[] = [];
+    const closed = Promise.withResolvers<void>();
+    const socket = net.connect(unix);
+    socket.on("connect", () => events.push("connect"));
+    socket.on("data", chunk => chunks.push(chunk));
+    socket.on("end", () => events.push("end"));
+    socket.on("error", (error: NodeJS.ErrnoException) => events.push(`error ${error.syscall} ${error.code}`));
+    socket.on("close", () => {
+      events.push("close");
+      closed.resolve();
+    });
+    const peer = blockUntilPeerIsDone();
+    await closed.promise;
+    return { peer, events, received: Buffer.concat(chunks) };
+  }
+
+  // The peer's one write has to fit in the socket buffer, because the client does not read yet.
+  // The default buffer of a unix socket is 8 KiB on macOS. The large answer is more than the
+  // 64 KiB that a node:net stream takes before it pauses its handle.
+  for (const bytes of [0, 5, isLinux ? 126892 : 4096]) {
+    const answer = Buffer.alloc(bytes, "x");
+
+    it(`Bun.connect opens and reads an answer of ${bytes} bytes`, async () => {
+      const { unix } = await peer(answer);
+      const { received, ...rest } = await bunConnect(unix);
+      expect({ ...rest, received: received.length }).toEqual({
+        peer: "done",
+        connect: "fulfilled",
+        events: ["open", "end", "close"],
+        received: bytes,
+      });
+      expect(received.equals(answer)).toBe(true);
+    });
+
+    it(`net.connect emits 'connect' and reads an answer of ${bytes} bytes`, async () => {
+      const { unix } = await peer(answer);
+      const { received, ...rest } = await netConnect(unix);
+      expect({ ...rest, received: received.length }).toEqual({
+        peer: "done",
+        events: ["connect", "end", "close"],
+        received: bytes,
+      });
+      expect(received.equals(answer)).toBe(true);
+    });
+  }
+
+  it("Bun.SQL (mysql) reports a full server through its socket file as it does over TCP", async () => {
+    // A MySQL server at max_connections sends error 1040 as soon as it accepts, and closes.
+    const tooManyConnections = mysqlRawPacket(
+      0,
+      Buffer.concat([Buffer.from([0xff, 1040 & 0xff, 1040 >> 8]), Buffer.from("Too many connections")]),
+    );
+    async function connectError(transport: "unix" | "tcp") {
+      const { unix, port } = await peer(tooManyConnections, transport);
+      const sql = new SQL({
+        adapter: "mysql",
+        username: "u",
+        password: "p",
+        database: "d",
+        max: 1,
+        ...(transport === "unix" ? { path: unix } : { hostname: "127.0.0.1", port }),
+      });
+      try {
+        const connecting = sql.connect();
+        const peer = blockUntilPeerIsDone();
+        const error = await connecting.then(
+          () => null,
+          error => error,
+        );
+        return { peer, code: error?.code };
+      } finally {
+        await sql.close({ timeout: 0 });
+      }
+    }
+    // Over TCP the peer's close is a FIN, which is not a hangup: the client always reads the packet.
+    const overTcp = await connectError("tcp");
+    expect(overTcp.peer).toBe("done");
+    expect(await connectError("unix")).toEqual(overTcp);
+  });
+
+  // Linux resets a connection that its listener never accepted: SO_ERROR is ECONNRESET.
+  it.skipIf(!isLinux)("Bun.connect fails for a connection that the listener never accepted", async () => {
+    const { unix } = await peer();
+    const { received, ...rest } = await bunConnect(unix);
+    expect({ ...rest, received: received.length }).toEqual({
+      peer: "done",
+      connect: "rejected",
+      events: ["connectError ECONNRESET"],
+      received: 0,
+    });
+  });
+
+  it.skipIf(!isLinux)("net.connect fails for a connection that the listener never accepted", async () => {
+    const { unix } = await peer();
+    const { received, ...rest } = await netConnect(unix);
+    expect({ ...rest, received: received.length }).toEqual({
+      peer: "done",
+      events: ["error connect ECONNRESET", "close"],
+      received: 0,
+    });
+  });
+});