diff --git a/src/js/node/net.ts b/src/js/node/net.ts index ebed05641f93..6ca474b12aad 100644 --- a/src/js/node/net.ts +++ b/src/js/node/net.ts @@ -49,6 +49,9 @@ const ArrayPrototypeJoin = Array.prototype.join; 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 @@ function tlsHandshakeError(verifyError) { 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 @@ const SocketHandlers: SocketHandler = { 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 @@ const ServerHandlers: SocketHandler = { 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 @@ function onconnection(err, clientHandle) { } 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 SocketHandlers2: SocketHandler *uwsRes = (uWS::HttpResponse *)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::HTTP_WRITE_CALLED; if (close_connection) { - if (!(data->state & uWS::HttpResponseData::HTTP_CONNECTION_CLOSE)) + if (!bodyStarted && !(data->state & uWS::HttpResponseData::HTTP_CONNECTION_CLOSE)) { uwsRes->writeHeader("Connection", "close"); } data->state |= uWS::HttpResponseData::HTTP_CONNECTION_CLOSE; } - if (!(data->state & uWS::HttpResponseData::HTTP_END_CALLED)) + if (!bodyStarted && !(data->state & uWS::HttpResponseData::HTTP_END_CALLED)) { uwsRes->AsyncSocket::write("\r\n", 2); } @@ -1383,15 +1386,17 @@ extern "C" { uWS::HttpResponse *uwsRes = (uWS::HttpResponse *)res; auto *data = uwsRes->getHttpResponseData(); + /* See the SSL branch above. */ + bool bodyStarted = data->state & uWS::HttpResponseData::HTTP_WRITE_CALLED; if (close_connection) { - if (!(data->state & uWS::HttpResponseData::HTTP_CONNECTION_CLOSE)) + if (!bodyStarted && !(data->state & uWS::HttpResponseData::HTTP_CONNECTION_CLOSE)) { uwsRes->writeHeader("Connection", "close"); } data->state |= uWS::HttpResponseData::HTTP_CONNECTION_CLOSE; } - if (!(data->state & uWS::HttpResponseData::HTTP_END_CALLED)) + if (!bodyStarted && !(data->state & uWS::HttpResponseData::HTTP_END_CALLED)) { // Some HTTP clients require the complete "
\r\n\r\n" to be sent. // If not, they may throw a ConnectionError. diff --git a/test/js/node/http/node-http.test.ts b/test/js/node/http/node-http.test.ts index a204e6d37259..b83ee845b141 100644 --- a/test/js/node/http/node-http.test.ts +++ b/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(); + let serverRes: InstanceType | 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); +}); diff --git a/test/js/node/net/node-net.test.ts b/test/js/node/net/node-net.test.ts index 6a972d3848da..673fd479190f 100644 --- a/test/js/node/net/node-net.test.ts +++ b/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); + }); +});