node:net: route exceptions thrown by socket event listeners to uncaughtException

src/js/node/net.ts

@@ -256,7 +256,36 @@ function tlsHandshakeError(verifyError) {
   return new ConnResetException("socket hang up");
 }
 
-const SocketHandlers: SocketHandler = {
+function rethrowUncaught(err) {
+  throw err;
+}
+
+// Reroute exceptions escaping a handler to uncaughtException like in Node,
+// instead of the socket's 'error' event (Bun.listen/Bun.connect's documented
+// behavior), then tear the socket down: it was interrupted mid-dispatch.
+function protectHandler(fn) {
+  return function (socket, a, b) {
+    try {
+      return fn.$call(this, socket, a, b);
+    } catch (err) {
+      process.nextTick(rethrowUncaught, err);
+      // Optional call: serverName receives the owning tls.Server as its
+      // first argument rather than a native socket handle.
+      socket?.terminate?.();
+    }
+  };
+}
+
+function protectHandlers<T extends object>(handlers: T): T {
+  const protectedHandlers = {} as T;
+  for (const key in handlers) {
+    const value = handlers[key];
+    protectedHandlers[key] = typeof value === "function" ? protectHandler(value) : value;
+  }
+  return protectedHandlers;
+}
+
+const SocketHandlers: SocketHandler = protectHandlers({
   close(socket, err) {
     const self = socket.data;
     if (!self || self[kclosed]) return;
@@ -458,7 +487,7 @@ const SocketHandlers: SocketHandler = {
     self.emit("timeout", self);
   },
   binaryType: "buffer",
-} as const;
+} as const);
 
 function SocketEmitEndNT(self, _err?) {
   // A read error delivered with the close (e.g. a received RST surfacing as
@@ -597,7 +626,7 @@ function onSNIResolution(state, err, context) {
   }
 }
 
-const ServerHandlers: SocketHandler<NetSocket> = {
+const ServerHandlers: SocketHandler<NetSocket> = protectHandlers({
   data(socket, buffer) {
     const { data: self } = socket;
     if (!self) return;
@@ -796,8 +825,10 @@ const ServerHandlers: SocketHandler<NetSocket> = {
       if (verifyError) {
         self.authorized = false;
         self.authorizationError = verifyError.code || verifyError.message;
-        server?.emit("tlsClientError", verifyError, self);
         if (self._rejectUnauthorized) {
+          // Only a rejected connection reports tlsClientError; an
+          // unauthorized-but-admitted one proceeds silently like in Node.
+          server?.emit("tlsClientError", verifyError, self);
           // if we reject we still need to emit secure
           self.emit("secure", self);
           // No error argument: the socket has no 'error' listener yet, so destroy(err)
@@ -863,7 +894,7 @@ const ServerHandlers: SocketHandler<NetSocket> = {
     SocketHandlers.drain(socket);
   },
   binaryType: "buffer",
-} as const;
+} as const);
 
 // Node.js-compatible onconnection: assigned to server._handle.onconnection in
 // kRealListen and invoked from ServerHandlers.open with `this` bound to the
@@ -989,7 +1020,7 @@ function onconnection(err, clientHandle) {
 }
 
 // TODO: SocketHandlers2 is a bad name but its temporary. reworking the Server in a followup PR
-const SocketHandlers2: SocketHandler<NonNullable<import("node:net").Socket["_handle"]>["data"]> = {
+const SocketHandlers2: SocketHandler<NonNullable<import("node:net").Socket["_handle"]>["data"]> = protectHandlers({
   open(socket) {
     $debug("Bun.Socket open");
     let { self, req } = socket.data;
@@ -1221,7 +1252,7 @@ const SocketHandlers2: SocketHandler<NonNullable<import("node:net").Socket["_han
     }
     req!.oncomplete(error.errno, self._handle, req, true, true);
   },
-};
+});
 
 // The same table minus the per-connection callback members: a listener whose
 // config has neither handler never registers the native SNI/ALPN dispatches,
@@ -1460,16 +1491,14 @@ function Socket(options?) {
     // when the onread option is specified we use a different handlers object
     this[khandlers] = {
       ...SocketHandlers2,
-      data(socket, buffer) {
+      data: protectHandler(function data(socket, buffer) {
         const { self } = socket.data;
         if (!self) return;
         self._unrefTimer();
-        try {
-          onread.callback(buffer.length, buffer);
-        } catch (e) {
-          self.emit("error", e);
-        }
-      },
+        // A throwing callback reaches uncaughtException via protectHandler,
+        // matching Node (onStreamRead does not catch it).
+        onread.callback(buffer.length, buffer);
+      }),
     };
   }
   if (signal) {

src/uws_sys/libuwsockets.cpp

@@ -1338,15 +1338,19 @@ extern "C"
     {
       uWS::HttpResponse<true> *uwsRes = (uWS::HttpResponse<true> *)res;
       auto *data = uwsRes->getHttpResponseData();
+      // Once write() ran the header section is terminated and body bytes are
+      // on the wire; header bytes here would corrupt the body. Not gated on
+      // WROTE_CONTENT_LENGTH: the file-route HEAD path still needs the CRLF.
+      bool bodyStarted = data->state & uWS::HttpResponseData<true>::HTTP_WRITE_CALLED;
       if (close_connection)
       {
-        if (!(data->state & uWS::HttpResponseData<true>::HTTP_CONNECTION_CLOSE))
+        if (!bodyStarted && !(data->state & uWS::HttpResponseData<true>::HTTP_CONNECTION_CLOSE))
         {
           uwsRes->writeHeader("Connection", "close");
         }
         data->state |= uWS::HttpResponseData<true>::HTTP_CONNECTION_CLOSE;
       }
-      if (!(data->state & uWS::HttpResponseData<true>::HTTP_END_CALLED))
+      if (!bodyStarted && !(data->state & uWS::HttpResponseData<true>::HTTP_END_CALLED))
       {
         uwsRes->AsyncSocket<true>::write("\r\n", 2);
       }
@@ -1358,15 +1362,17 @@ extern "C"
     {
       uWS::HttpResponse<false> *uwsRes = (uWS::HttpResponse<false> *)res;
       auto *data = uwsRes->getHttpResponseData();
+      // See the SSL branch: never write header bytes mid-body.
+      bool bodyStarted = data->state & uWS::HttpResponseData<false>::HTTP_WRITE_CALLED;
       if (close_connection)
       {
-        if (!(data->state & uWS::HttpResponseData<false>::HTTP_CONNECTION_CLOSE))
+        if (!bodyStarted && !(data->state & uWS::HttpResponseData<false>::HTTP_CONNECTION_CLOSE))
         {
           uwsRes->writeHeader("Connection", "close");
         }
         data->state |= uWS::HttpResponseData<false>::HTTP_CONNECTION_CLOSE;
       }
-      if (!(data->state & uWS::HttpResponseData<false>::HTTP_END_CALLED))
+      if (!bodyStarted && !(data->state & uWS::HttpResponseData<false>::HTTP_END_CALLED))
       {
         // Some HTTP clients require the complete "<header>\r\n\r\n" to be sent.
         // If not, they may throw a ConnectionError.

test/js/node/http/node-http.test.ts

@@ -3749,3 +3749,33 @@ it("OutgoingMessage outputData is per-instance and _flushOutput is defined", ()
   c.outputData.push({ data: "y", encoding: "utf8", callback: null });
   expect(d.outputData.length).toBe(0);
 });
+
+// https://github.com/oven-sh/bun/issues/34064
+it("destroying a chunked response mid-stream writes no header bytes into the body", async () => {
+  const chunkFrame = "f\r\nPart of my res.\r\n";
+  const { promise, resolve, reject } = Promise.withResolvers<Buffer>();
+  let serverRes: InstanceType<typeof http.ServerResponse> | undefined;
+  await using server = http.createServer((req, res) => {
+    res.write("Part of my res.");
+    serverRes = res;
+  });
+  server.listen(0, "127.0.0.1");
+  await once(server, "listening");
+  const client = connect((server.address() as AddressInfo).port, "127.0.0.1");
+  const chunks: Buffer[] = [];
+  client.on("connect", () => client.write("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n"));
+  client.on("data", chunk => {
+    chunks.push(chunk);
+    // Destroy mid-stream only once the chunk frame is provably on the wire;
+    // anything the abort appends after it arrives before 'close'.
+    if (Buffer.concat(chunks).includes(chunkFrame)) serverRes!.destroy();
+  });
+  client.on("error", reject);
+  client.on("close", () => resolve(Buffer.concat(chunks)));
+  const raw = (await promise).toString();
+  const headerEnd = raw.indexOf("\r\n\r\n");
+  expect(headerEnd).toBeGreaterThan(0);
+  // The body must be exactly the one chunk frame that was written, nothing
+  // appended by the abort.
+  expect(raw.slice(headerEnd + 4)).toBe(chunkFrame);
+});

test/js/node/net/node-net.test.ts

@@ -1062,3 +1062,142 @@ it.skipIf(isWindows)("connect({ localPort }) succeeds when the local port has TI
     target.close();
   }
 });
+
+// https://github.com/oven-sh/bun/issues/34064
+describe.concurrent("exceptions thrown from socket event listeners", () => {
+  async function run(fixture: string) {
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", fixture],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    return { stdout, stderr, exitCode };
+  }
+
+  it("a throwing 'data' listener reaches uncaughtException, not the socket 'error' listener", async () => {
+    const { stdout, stderr, exitCode } = await run(`
+      const net = require("node:net");
+      process.on("uncaughtException", (e) => {
+        console.log("uncaughtException: " + e.message);
+        process.exit(3);
+      });
+      const server = net.createServer((s) => s.end("x"));
+      server.listen(0, "127.0.0.1", () => {
+        const c = net.connect(server.address().port, "127.0.0.1");
+        c.on("error", (e) => {
+          console.log("socket error listener: " + e.message);
+          process.exit(7);
+        });
+        c.on("data", () => {
+          throw new Error("boom from data listener");
+        });
+      });
+    `);
+    expect(stdout.trim()).toBe("uncaughtException: boom from data listener");
+    expect(exitCode).toBe(3);
+  });
+
+  it("a throwing 'data' listener on an accepted server socket reaches uncaughtException", async () => {
+    const { stdout, stderr, exitCode } = await run(`
+      const net = require("node:net");
+      process.on("uncaughtException", (e) => {
+        console.log("uncaughtException: " + e.message);
+        process.exit(3);
+      });
+      const server = net.createServer((s) => {
+        s.on("error", (e) => {
+          console.log("socket error listener: " + e.message);
+          process.exit(7);
+        });
+        s.on("data", () => {
+          throw new Error("boom from server data listener");
+        });
+      });
+      server.listen(0, "127.0.0.1", () => {
+        const c = net.connect(server.address().port, "127.0.0.1");
+        c.on("error", () => {});
+        c.on("connect", () => c.write("hello"));
+      });
+    `);
+    expect(stdout.trim()).toBe("uncaughtException: boom from server data listener");
+    expect(exitCode).toBe(3);
+  });
+
+  it("without an uncaughtException handler, a throwing 'data' listener crashes the process", async () => {
+    const { stdout, stderr, exitCode } = await run(`
+      const net = require("node:net");
+      const server = net.createServer((s) => s.end("x"));
+      server.listen(0, "127.0.0.1", () => {
+        const c = net.connect(server.address().port, "127.0.0.1");
+        c.on("error", (e) => {
+          console.log("socket error listener: " + e.message);
+          process.exit(7);
+        });
+        c.on("data", () => {
+          throw new Error("fatal boom");
+        });
+      });
+    `);
+    expect(stdout).not.toContain("socket error listener");
+    expect(stderr).toContain("fatal boom");
+    expect(exitCode).toBe(1);
+  });
+
+  it("emitting an unhandled 'error' on another emitter from a 'data' listener reaches uncaughtException", async () => {
+    // The pg pool shape from the issue: the data listener re-emits 'error' on
+    // an emitter with no listeners, so emit() rethrows the error.
+    const { stdout, stderr, exitCode } = await run(`
+      const net = require("node:net");
+      const { EventEmitter } = require("node:events");
+      process.on("uncaughtException", (e) => {
+        console.log("uncaughtException: " + e.message);
+        process.exit(3);
+      });
+      const pool = new EventEmitter();
+      const server = net.createServer((s) => s.end("x"));
+      server.listen(0, "127.0.0.1", () => {
+        const c = net.connect(server.address().port, "127.0.0.1");
+        c.on("error", (e) => {
+          console.log("socket error listener: " + e.message);
+          process.exit(7);
+        });
+        c.on("data", () => {
+          pool.emit("error", new Error("pool error"));
+        });
+      });
+    `);
+    expect(stdout.trim()).toBe("uncaughtException: pool error");
+    expect(exitCode).toBe(3);
+  });
+
+  it("a throwing onread callback reaches uncaughtException, not the socket 'error' listener", async () => {
+    const { stdout, stderr, exitCode } = await run(`
+      const net = require("node:net");
+      process.on("uncaughtException", (e) => {
+        console.log("uncaughtException: " + e.message);
+        process.exit(3);
+      });
+      const server = net.createServer((s) => s.end("x"));
+      server.listen(0, "127.0.0.1", () => {
+        const c = net.connect({
+          port: server.address().port,
+          host: "127.0.0.1",
+          onread: {
+            buffer: Buffer.alloc(4096),
+            callback: () => {
+              throw new Error("boom from onread callback");
+            },
+          },
+        });
+        c.on("error", (e) => {
+          console.log("socket error listener: " + e.message);
+          process.exit(7);
+        });
+      });
+    `);
+    expect(stdout.trim()).toBe("uncaughtException: boom from onread callback");
+    expect(exitCode).toBe(3);
+  });
+});

test/js/node/tls/node-tls-cert.test.ts

@@ -337,6 +337,10 @@ it("explicit rejectUnauthorized: false still admits an unverified client certifi
     },
     socket => onHandledSocket(socket),
   );
+  // An unauthorized-but-admitted connection is not a client error; Node only
+  // reports tlsClientError when the connection is torn down.
+  const tlsClientErrors: Error[] = [];
+  server.on("tlsClientError", err => tlsClientErrors.push(err));
   await once(server.listen(0, "127.0.0.1"), "listening");
   const port = (server.address() as AddressInfo).port;
 
@@ -355,6 +359,7 @@ it("explicit rejectUnauthorized: false still admits an unverified client certifi
     const [serverSocket] = await Promise.all([handledSocket, once(client, "secureConnect")]);
     expect(serverSocket.authorized).toBe(false);
     expect(serverSocket.authorizationError).toBe("UNABLE_TO_VERIFY_LEAF_SIGNATURE");
+    expect(tlsClientErrors).toEqual([]);
   } finally {
     client.end();
     server.close();