diff --git a/src/js/internal/http.ts b/src/js/internal/http.ts index f8e26db96fad..089ae5b8d2c3 100644 --- a/src/js/internal/http.ts +++ b/src/js/internal/http.ts @@ -242,6 +242,36 @@ function emitEOFIncomingMessage(self) { process.nextTick(emitEOFIncomingMessageOuter, self); } +function onDataIncomingMessage(this: any, chunk, isLast, aborted: NodeHTTPResponseAbortEvent) { + if (aborted === NodeHTTPResponseAbortEvent.abort) { + this.destroy(); + return; + } + + // Incoming request-body bytes are socket activity: push the connection's + // inactivity timeout (socket.setTimeout / server.timeout) further out, like + // Node.js does for reads on the socket. + const socket = this.socket; + socket?._unrefTimer?.(); + + if (chunk && !this._dumped) { + if (!this.push(chunk)) { + // Like Node's parserOnBody: pause the connection once the buffer fills. + // Upgrade-with-body routes through its own handle so the socket's flow + // state stays with the upgrade listener; _read() balances it. + if (this.upgrade) this[kHandle]?.pause(); + else if (socket && !socket.writableEnded) socket.pause(); + } + } + + if (isLast) { + emitEOFIncomingMessage(this); + // Like Node's parserOnMessageComplete: any readStop above left the shared + // socket's flowing=false, which would swallow the next request's 'pause'. + if (!this.upgrade && socket && !socket._paused && socket.readable) socket.resume(); + } +} + function validateMsecs(numberlike: any, field: string) { if (typeof numberlike !== "number" || numberlike < 0) { throw $ERR_INVALID_ARG_TYPE(field, "number", numberlike); @@ -623,6 +653,7 @@ export { kUseDefaultPort, kWaitForProxyTunnel, noBodySymbol, + onDataIncomingMessage, optionsSymbol, parseProxyConfigFromEnv, parseProxyUrl, diff --git a/src/js/node/_http_incoming.ts b/src/js/node/_http_incoming.ts index ce445fef1c0f..da2f8c66f846 100644 --- a/src/js/node/_http_incoming.ts +++ b/src/js/node/_http_incoming.ts @@ -18,7 +18,7 @@ const { emitErrorNextTickIfErrorListenerNT, NodeHTTPBodyReadState, emitEOFIncomingMessage, - NodeHTTPResponseAbortEvent, + onDataIncomingMessage, kAbortController, } = require("internal/http"); @@ -410,29 +410,6 @@ IncomingMessage.prototype._read = function _read(_n) { } }; -function onDataIncomingMessage( - this: import("node:http").IncomingMessage, - chunk, - isLast, - aborted: NodeHTTPResponseAbortEvent, -) { - if (aborted === NodeHTTPResponseAbortEvent.abort) { - this.destroy(); - return; - } - - // Incoming request-body bytes are socket activity: push the connection's - // inactivity timeout (socket.setTimeout / server.timeout) further out, like - // Node.js does for reads on the socket. - this.socket?._unrefTimer?.(); - - if (chunk && !this._dumped) this.push(chunk); - - if (isLast) { - emitEOFIncomingMessage(this); - } -} - // It's possible that the socket will be destroyed, and removed from // any messages, before ever calling this. In that case, just skip // it, since something else is destroying this connection anyway. diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index 96fb896e9790..91892b00e869 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -64,6 +64,7 @@ const { fakeSocketSymbol, noBodySymbol, kOutHeaders, + onDataIncomingMessage, validateMsecs, } = require("internal/http"); const { FakeSocket } = require("internal/http/FakeSocket"); @@ -865,9 +866,12 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort setIsNextIncomingMessageHTTPS(prevIsNextIncomingMessageHTTPS); handle.onabort = socket[kBoundOnAbort] ??= onServerRequestEvent.bind(socket); - // start buffering data if any, the user will need to resume() or .on("data") to read it + // Like Node's connectionListener -> parserOnBody: body bytes flow into + // the IncomingMessage as they arrive, and the push callback readStop()s + // the socket (which emits 'pause' on it) once the buffer fills. if (hasBody) { - handle.pause(); + handle.ondata = onDataIncomingMessage.bind(http_req); + handle.hasCustomOnData = false; } drainMicrotasks(); diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp index bd6f5a1d15de..51f1e22aa0c3 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp @@ -12,6 +12,7 @@ extern "C" EncodedJSValue us_socket_buffered_js_write(void* socket, bool is_ssl, extern "C" uint64_t uws_res_get_remote_address_info(void* res, const char** dest, int* port, bool* is_ipv6); extern "C" uint64_t uws_res_get_local_address_info(void* res, const char** dest, int* port, bool* is_ipv6); extern "C" void us_socket_resume(us_socket_t*); +extern "C" void us_socket_pause(us_socket_t*); namespace Bun { @@ -220,10 +221,16 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketEnd, (JSC::JSGlobalObject } auto bufferedSize = thisObject->streamBuffer.bufferedSize(); if (bufferedSize == 0) { + // onNodeHTTPRequest no longer pauses at dispatch; pause here so the + // shutdown+resume below still cycles kqueue's EVFILT_READ (delete then + // re-add), without which macOS 26 does not deliver the peer's close. + if (thisObject->socket && !thisObject->upgraded) { + us_socket_pause(thisObject->socket); + } auto result = us_socket_buffered_js_write(thisObject->socket, thisObject->is_ssl, thisObject->ended, &thisObject->streamBuffer, globalObject, JSValue::encode(JSC::jsUndefined()), JSValue::encode(JSC::jsUndefined())); - // Undo onNodeHTTPRequest's body-buffering pause after the shutdown so - // the unread body drains and kqueue's one-shot EVFILT_WRITE (which - // delivers EV_EOF on SHUT_WR) is not deleted by a W -> R|W -> R step. + // Undo the pause above after the shutdown so the unread body drains + // and kqueue's one-shot EVFILT_WRITE (which delivers EV_EOF on + // SHUT_WR) is not deleted by a W -> R|W -> R step. if (thisObject->socket && !thisObject->upgraded) { us_socket_resume(thisObject->socket); } diff --git a/src/runtime/server/NodeHTTPResponse.rs b/src/runtime/server/NodeHTTPResponse.rs index de1405f5f857..01bead8d6221 100644 --- a/src/runtime/server/NodeHTTPResponse.rs +++ b/src/runtime/server/NodeHTTPResponse.rs @@ -1354,8 +1354,15 @@ impl NodeHTTPResponse { { return Ok(JSValue::FALSE); } - self.update_flags(|f| f.insert(Flags::IS_DATA_BUFFERED_DURING_PAUSE)); - raw.on_data(on_buffer_paused_shim, self.as_ctx_ptr()); + // Body already delivered: nothing to buffer, and re-arming onData would + // overwrite a pipelined request's userData on the shared HttpResponseData. + // pause_socket() still runs so pausePipelineReads can gate the fd. + if self.body_read_state.get() == BodyReadState::Pending + && !flags.contains(Flags::IS_DATA_BUFFERED_DURING_PAUSE_LAST) + { + self.update_flags(|f| f.insert(Flags::IS_DATA_BUFFERED_DURING_PAUSE)); + raw.on_data(on_buffer_paused_shim, self.as_ctx_ptr()); + } // TODO: figure out why windows is not emitting EOF with UV_DISCONNECT #[cfg(not(windows))] @@ -1419,8 +1426,15 @@ impl NodeHTTPResponse { { return JSValue::FALSE; } - self.set_on_aborted_handler(); - raw.on_data(on_data_shim, self.as_ctx_ptr()); + // Body already delivered: re-arming onData/onTimeout would overwrite a + // pipelined request's userData on the shared HttpResponseData. The drain + // below still runs so a body buffered-while-paused reaches its own caller. + if self.body_read_state.get() == BodyReadState::Pending + && !flags.contains(Flags::IS_DATA_BUFFERED_DURING_PAUSE_LAST) + { + self.set_on_aborted_handler(); + raw.on_data(on_data_shim, self.as_ctx_ptr()); + } self.update_flags(|f| f.remove(Flags::IS_DATA_BUFFERED_DURING_PAUSE)); let mut result: JSValue = JSValue::TRUE; diff --git a/test/js/node/http/node-http-req-socket-pause.test.ts b/test/js/node/http/node-http-req-socket-pause.test.ts new file mode 100644 index 000000000000..e5261a6729ae --- /dev/null +++ b/test/js/node/http/node-http-req-socket-pause.test.ts @@ -0,0 +1,114 @@ +/** + * All tests in this file should also run in Node.js. + */ +import { expect, it } from "bun:test"; +import { once } from "node:events"; +import { Agent, createServer, request } from "node:http"; +import type { AddressInfo } from "node:net"; + +it("req.socket emits 'pause' once an unread request body fills the IncomingMessage buffer", async () => { + // Node's test-http-no-read-no-dump: a handler that never reads the body sees + // 'pause' on req.connection once the IncomingMessage push() backpressures. + const { promise: paused, resolve: onPause, reject } = Promise.withResolvers(); + const server = createServer((req, res) => { + req.connection!.on("pause", () => { + onPause((req as any).readableLength); + res.end("ok"); + }); + res.writeHead(200); + res.flushHeaders(); + }); + try { + await once(server.listen(0), "listening"); + const port = (server.address() as AddressInfo).port; + const post = request({ method: "POST", port }); + post.on("error", reject); + post.flushHeaders(); + // One body chunk at the default highWaterMark: the first parserOnBody push + // returns false and Node's readStop pauses the socket. + post.write(Buffer.alloc(64 * 1024, "X")); + const buffered = await paused; + expect(buffered).toBeGreaterThan(0); + post.destroy(); + } finally { + server.closeAllConnections(); + server.close(); + } +}); + +it("body reading from 'pause' still delivers every byte and 'end'", async () => { + // A handler that keys its slow-reader flow off the socket's 'pause' event + // (the pattern from Node's test-http-no-read-no-dump) must still be able to + // drain the full body once it attaches a 'data' listener. + const { promise: done, resolve, reject } = Promise.withResolvers<{ pauses: number; received: number }>(); + const server = createServer((req, res) => { + let pauses = 0; + let received = 0; + req.connection!.on("pause", () => { + pauses++; + if (pauses > 1) return; + req.on("data", chunk => (received += chunk.length)); + req.on("end", () => { + res.end("ok"); + resolve({ pauses, received }); + }); + }); + res.writeHead(200); + res.flushHeaders(); + }); + try { + await once(server.listen(0), "listening"); + const port = (server.address() as AddressInfo).port; + const payload = 256 * 1024; + const post = request({ method: "POST", port }); + post.on("error", reject); + post.flushHeaders(); + await once(post, "response"); + post.end(Buffer.alloc(payload, "X")); + const { pauses, received } = await done; + expect(received).toBe(payload); + expect(pauses).toBeGreaterThanOrEqual(1); + } finally { + server.closeAllConnections(); + server.close(); + } +}); + +it("req.socket emits 'pause' on every body-bearing keep-alive request, not just the first", async () => { + const pauses: string[] = []; + const sockets: unknown[] = []; + const ended = Promise.withResolvers(); + const server = createServer((req, res) => { + sockets.push(req.socket); + req.connection!.once("pause", () => { + pauses.push(req.url!); + res.end("ok"); + if (req.url === "/b") ended.resolve(); + }); + res.writeHead(200); + res.flushHeaders(); + }); + try { + await once(server.listen(0), "listening"); + const port = (server.address() as AddressInfo).port; + const agent = new Agent({ keepAlive: true, maxSockets: 1 }); + for (const path of ["/a", "/b"]) { + await new Promise((resolve, reject) => { + const post = request({ method: "POST", port, path, agent }, res => { + res.resume(); + res.on("end", resolve); + }); + post.on("error", reject); + post.end(Buffer.alloc(128 * 1024, "X")); + }); + } + await ended.promise; + agent.destroy(); + expect(sockets.length).toBe(2); + expect(sockets[0]).toBe(sockets[1]); + expect(pauses).toEqual(["/a", "/b"]); + } finally { + server.closeAllConnections(); + server.close(); + } +}); diff --git a/test/js/node/test/parallel/test-http-no-read-no-dump.js b/test/js/node/test/parallel/test-http-no-read-no-dump.js new file mode 100644 index 000000000000..b31e25e42ae6 --- /dev/null +++ b/test/js/node/test/parallel/test-http-no-read-no-dump.js @@ -0,0 +1,56 @@ +'use strict'; +const common = require('../common'); +const http = require('http'); + +let onPause = null; + +const server = http.createServer(common.mustCallAtLeast((req, res) => { + if (req.method === 'GET') + return res.end(); + + res.writeHead(200); + res.flushHeaders(); + + req.on('close', common.mustCall(() => { + req.on('end', common.mustNotCall()); + })); + + req.connection.on('pause', () => { + res.end(); + onPause(); + }); +})).listen(0, common.mustCall(() => { + const agent = new http.Agent({ + maxSockets: 1, + keepAlive: true + }); + + const port = server.address().port; + + const post = http.request({ + agent, + method: 'POST', + port, + }, common.mustCall((res) => { + res.resume(); + + post.write(Buffer.alloc(64 * 1024).fill('X')); + onPause = () => { + post.end('something'); + }; + })); + + // What happens here is that the server `end`s the response before we send + // `something`, and the client thought that this is a green light for sending + // next GET request + post.write('initial'); + + http.request({ + agent, + method: 'GET', + port, + }, common.mustCall((res) => { + server.close(); + res.connection.end(); + })).end(); +}));