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

src/js/node/net.ts

@@ -256,7 +256,41 @@ function tlsHandshakeError(verifyError) {
   return new ConnResetException("socket hang up");
 }
 
-const SocketHandlers: SocketHandler = {
+function rethrowUncaught(err) {
+  throw err;
+}
+
+// An exception escaping a native socket callback (a throwing user listener,
+// e.g. 'data') must surface as an uncaughtException like in Node. The native
+// dispatch would instead route it to this table's `error` handler, which is
+// the documented behavior for the public Bun.listen/Bun.connect API but makes
+// node:net deliver it to the socket's 'error' listeners as if it were a
+// socket error. The socket is torn down afterwards: the callback was
+// interrupted mid-dispatch, so its stream state is unreliable (and this
+// matches the teardown the previous error routing performed).
+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 +492,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 +631,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;
@@ -863,7 +897,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 +1023,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 +1255,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,7 +1494,7 @@ 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();
@@ -1469,7 +1503,7 @@ function Socket(options?) {
         } catch (e) {
           self.emit("error", e);
         }
-      },
+      }),
     };
   }
   if (signal) {

src/uws_sys/libuwsockets.cpp

@@ -1338,15 +1338,22 @@ extern "C"
     {
       uWS::HttpResponse<true> *uwsRes = (uWS::HttpResponse<true> *)res;
       auto *data = uwsRes->getHttpResponseData();
+      // Once write() ran, the header section is already terminated and body
+      // bytes are on the wire, so header bytes here would land in the middle
+      // of the body and corrupt the stream - only update the state flags in
+      // that case. (HTTP_WROTE_CONTENT_LENGTH_HEADER does not imply the
+      // section is terminated: the file-route HEAD path sets it for a plain
+      // header line and relies on this function for the terminating 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 +1365,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,35 @@ 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 (unmasked by routing listener
+// throws to uncaughtException): aborting a chunked response used to inject
+// "Connection: close\r\n\r\n" into the middle of the body.
+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,113 @@ it.skipIf(isWindows)("connect({ localPort }) succeeds when the local port has TI
     target.close();
   }
 });
+
+// https://github.com/oven-sh/bun/issues/34064
+describe("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);
+  });
+});