node:net: route throws from 'data'/'connection' listeners to uncaughtException and keep the socket alive

src/js/node/net.ts

@@ -49,6 +49,9 @@
 const ArrayPrototypePush = Array.prototype.push;
 const MathMax = Math.max;
 const MathMin = Math.min;
+// Captured at module load so user code clobbering globalThis.reportError cannot
+// defeat the uncaughtException routing below.
+const reportError = globalThis.reportError;
 
 const { UV_ECANCELED, UV_ENOBUFS, UV_ETIMEDOUT } = process.binding("uv");
 const isWindows = process.platform === "win32";
@@ -329,6 +332,24 @@
   return new ConnResetException("socket hang up");
 }
 
+// Readable.push() synchronously runs user 'data' listeners. A throw escaping
+// this handler is caught by the native socket dispatch and routed to the
+// handler table's `error` entry, which is for transport failures (ECONNRESET
+// etc.), so a programming error in a 'data' listener would be reported as a
+// socket 'error' and the connection torn down. Node surfaces the throw as
+// uncaughtException and leaves the socket reading; the next chunk is still
+// delivered. Catching here keeps the socket alive and matches that.
+function pushDataToSocket(self, socket, buffer) {
+  let full;
+  try {
+    full = self.push(buffer) === false;
+  } catch (e) {
+    reportError(e);
+    return;
+  }
+  if (full) socket.pause();
+}
+
 const SocketHandlers: SocketHandler = {
   close(socket, err) {
     const self = socket.data;
@@ -345,9 +366,7 @@
 
     self._unrefTimer();
     self.bytesRead += buffer.length;
-    if (!self.push(buffer)) {
-      socket.pause();
-    }
+    pushDataToSocket(self, socket, buffer);
   },
   drain(socket) {
     const self = socket.data;
@@ -706,9 +725,7 @@
 
     self._unrefTimer();
     self.bytesRead += buffer.length;
-    if (!self.push(buffer)) {
-      socket.pause();
-    }
+    pushDataToSocket(self, socket, buffer);
   },
   keylog(socket, line) {
     const { data: self } = socket;
@@ -1183,7 +1200,15 @@
   }
   if (isTLS) initAcceptedTLSSocket(self, _socket);
 
-  self.emit("connection", _socket);
+  // A 'connection' listener throw that reached the native open dispatch would
+  // be treated as an open failure and the accepted socket closed. Node reports
+  // it as uncaughtException and the connection stays established; match that
+  // and fall through so reading is still started.
+  try {
+    self.emit("connection", _socket);
+  } catch (e) {
+    reportError(e);
+  }
   if (!pauseOnConnect && !isTLS) {
     _socket.read(0);
   }
@@ -1224,7 +1249,7 @@
     const { self } = socket.data;
     self._unrefTimer();
     self.bytesRead += buffer.length;
-    if (!self.push(buffer)) socket.pause();
+    pushDataToSocket(self, socket, buffer);
   },
   drain(socket) {
     $debug("Bun.Socket drain");

src/uws_sys/libuwsockets.cpp

@@ -1363,15 +1363,18 @@ 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; a header or CRLF here would corrupt the body. */
+      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);
       }
@@ -1383,15 +1386,17 @@ extern "C"
     {
       uWS::HttpResponse<false> *uwsRes = (uWS::HttpResponse<false> *)res;
       auto *data = uwsRes->getHttpResponseData();
+      /* See the SSL branch above. */
+      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

@@ -4054,3 +4054,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);
 });
+
+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<string>();
+  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: x\r\nConnection: keep-alive\r\n\r\n"));
+  client.on("data", chunk => {
+    chunks.push(chunk);
+    // Destroy mid-stream only once the chunk frame is 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).toString("latin1")));
+  const raw = await promise;
+  const headerEnd = raw.indexOf("\r\n\r\n");
+  expect(headerEnd).toBeGreaterThan(0);
+  // The body must be exactly the one chunk frame that was written; a
+  // Connection: close header or stray CRLF written by the abort path would
+  // land here and corrupt the chunked framing.
+  expect(raw.slice(headerEnd + 4)).toBe(chunkFrame);
+});

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

@@ -1794,3 +1794,122 @@ it.skipIf(isWindows)("connect({ localPort }) succeeds when the local port has TI
     target.close();
   }
 });
+
+// A throw from a user listener invoked synchronously from a native socket
+// dispatch must reach process.on('uncaughtException') the way Node reports it,
+// not be routed to the socket's 'error' event or silently dropped, and the
+// connection must stay alive so subsequent bytes are still delivered.
+describe.concurrent("uncaughtException from socket listeners", () => {
+  async function runFixture(src: string) {
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", src],
+      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("server-side 'data' listener throw reaches uncaughtException and the socket keeps reading", async () => {
+    const { stdout, stderr, exitCode } = await runFixture(`
+      const net = require("node:net");
+      const ev = [];
+      const done = () => { console.log(JSON.stringify(ev)); process.exit(0); };
+      process.on("uncaughtException", e => ev.push("uncaught:" + e.message));
+      const srv = net.createServer(s => {
+        ev.push("connection");
+        s.on("error", e => ev.push("socket-error:" + e.message));
+        s.on("data", d => {
+          ev.push("data:" + d);
+          // Queued before the throw so the client can pace the next write on
+          // the ack instead of time.
+          s.write(".");
+          if (String(d) === "A") throw new Error("data-boom");
+        });
+        s.on("close", had => { ev.push("close:" + had); srv.close(done); });
+      });
+      srv.listen(0, "127.0.0.1", () => {
+        const c = net.connect(srv.address().port, "127.0.0.1", () => c.write("A"));
+        let acks = 0;
+        c.on("data", () => { ++acks === 1 ? c.write("B") : c.end(); });
+        c.on("error", () => {});
+      });
+      setTimeout(done, 5000).unref();
+    `);
+    expect({ stdout: stdout.trim(), exitCode, ...(exitCode === 0 ? {} : { stderr }) }).toEqual({
+      stdout: JSON.stringify(["connection", "data:A", "uncaught:data-boom", "data:B", "close:false"]),
+      exitCode: 0,
+    });
+  });
+
+  it("client-side 'data' listener throw reaches uncaughtException and the socket keeps reading", async () => {
+    const { stdout, stderr, exitCode } = await runFixture(`
+      const net = require("node:net");
+      const ev = [];
+      const done = () => { console.log(JSON.stringify(ev)); process.exit(0); };
+      process.on("uncaughtException", e => ev.push("uncaught:" + e.message));
+      const srv = net.createServer(s => {
+        s.write("A");
+        s.once("data", () => s.end("B"));
+      });
+      srv.listen(0, "127.0.0.1", () => {
+        const c = net.connect(srv.address().port, "127.0.0.1");
+        c.on("error", e => ev.push("socket-error:" + e.message));
+        c.on("data", d => {
+          ev.push("data:" + d);
+          if (String(d) === "A") { c.write("."); throw new Error("data-boom"); }
+        });
+        c.on("close", had => { ev.push("close:" + had); srv.close(done); });
+      });
+      setTimeout(done, 5000).unref();
+    `);
+    expect({ stdout: stdout.trim(), exitCode, ...(exitCode === 0 ? {} : { stderr }) }).toEqual({
+      stdout: JSON.stringify(["data:A", "uncaught:data-boom", "data:B", "close:false"]),
+      exitCode: 0,
+    });
+  });
+
+  it("'connection' listener throw reaches uncaughtException and the accepted socket keeps reading", async () => {
+    const { stdout, stderr, exitCode } = await runFixture(`
+      const net = require("node:net");
+      const ev = [];
+      const done = () => { console.log(JSON.stringify(ev)); process.exit(0); };
+      process.on("uncaughtException", e => ev.push("uncaught:" + e.message));
+      const srv = net.createServer(s => {
+        ev.push("connection");
+        s.on("error", e => ev.push("socket-error:" + e.message));
+        s.on("data", d => { ev.push("data:" + d); s.write("."); });
+        s.on("close", had => { ev.push("close:" + had); srv.close(done); });
+        throw new Error("conn-boom");
+      });
+      srv.on("error", e => ev.push("server-error:" + e.message));
+      srv.listen(0, "127.0.0.1", () => {
+        const c = net.connect(srv.address().port, "127.0.0.1", () => c.write("A"));
+        let acks = 0;
+        c.on("data", () => { ++acks === 1 ? c.write("B") : c.end(); });
+        c.on("error", () => {});
+      });
+      setTimeout(done, 5000).unref();
+    `);
+    expect({ stdout: stdout.trim(), exitCode, ...(exitCode === 0 ? {} : { stderr }) }).toEqual({
+      stdout: JSON.stringify(["connection", "uncaught:conn-boom", "data:A", "data:B", "close:false"]),
+      exitCode: 0,
+    });
+  });
+
+  it("without an uncaughtException handler a throwing 'data' listener crashes the process", async () => {
+    const { stdout, stderr, exitCode } = await runFixture(`
+      const net = require("node:net");
+      const srv = net.createServer(s => s.end("x"));
+      srv.listen(0, "127.0.0.1", () => {
+        const c = net.connect(srv.address().port, "127.0.0.1");
+        c.on("error", e => { console.log("socket-error:" + e.message); process.exit(7); });
+        c.on("data", () => { throw new Error("fatal-boom"); });
+      });
+    `);
+    expect(stdout).not.toContain("socket-error:");
+    expect(stderr).toContain("fatal-boom");
+    expect(exitCode).toBe(1);
+  });
+});