From bd8cd55614f164127cb977ea38af3f198cdc23bd Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:19:11 +0000 Subject: [PATCH 1/8] node:http: keep receiving the request body after the response has ended A request whose response is ended before its body has been received used to be completed immediately: the native side dropped the body read state in res.end() (and in the dispatch tail), and IncomingMessage._dump()/_read() then fabricated EOF, so req.complete, readableEnded and destroyed flipped to true and 'end'/'close' fired while most of the body was still in flight. A consumer attached before (or in the same tick as) a synchronous res.end() never saw the body either. Like Node, the body now keeps flowing into the IncomingMessage after the response: native re-arms the connection's body data handler after uws's end() dropped it (while the request's reader is still armed and the body is still arriving), pause/resume and arming keep working in that window, and the request is released when the body's fin arrives or the connection goes away. JS stops clearing the native callback in _dump() and stops treating a dumped request as ended in _read(); the dump decision itself moves to the response's 'finish' like Node's resOnFinish. The never-set hasCustomOnData flag, which only existed to gate those discards, is removed. --- src/js/node/_http_incoming.ts | 17 +- src/js/node/_http_server.ts | 22 +- src/runtime/server/NodeHTTPResponse.rs | 198 ++++++++---- src/runtime/server/mod.rs | 3 +- src/runtime/server/server.classes.ts | 4 - ...ode-http-ondata-reregister-leak.fixture.js | 2 +- .../node-http-server-abort-events.test.ts | 300 +++++++++++++++++- 7 files changed, 457 insertions(+), 89 deletions(-) diff --git a/src/js/node/_http_incoming.ts b/src/js/node/_http_incoming.ts index da2f8c66f846..a70df7984a8f 100644 --- a/src/js/node/_http_incoming.ts +++ b/src/js/node/_http_incoming.ts @@ -389,11 +389,10 @@ IncomingMessage.prototype._read = function _read(_n) { const bodyReadState = handle.hasBody; - if ( - (bodyReadState & NodeHTTPBodyReadState.done) !== 0 || - bodyReadState === NodeHTTPBodyReadState.none || - this._dumped - ) { + // A dumped request (res.end() before the body was read) is not complete + // here: like Node, it only ends once the rest of its body has actually + // arrived, which onDataIncomingMessage reports with isLast. + if ((bodyReadState & NodeHTTPBodyReadState.done) !== 0 || bodyReadState === NodeHTTPBodyReadState.none) { emitEOFIncomingMessage(this); } @@ -406,7 +405,6 @@ IncomingMessage.prototype._read = function _read(_n) { if (!handle.ondata) { handle.ondata = onDataIncomingMessage.bind(this); - handle.hasCustomOnData = false; } }; @@ -759,10 +757,9 @@ IncomingMessage.prototype._dump = function _dump() { // If there is buffered data, it may trigger 'data' events. // Remove 'data' event listeners explicitly. this.removeAllListeners("data"); - const handle = this[kHandle]; - if (handle) { - handle.ondata = undefined; - } + // The native ondata callback stays armed: onDataIncomingMessage drops the + // chunks of a dumped request and still delivers the body's fin, which is + // what completes the request (see _read). this.resume(); } }; diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index 45fb072718c6..1a1e9971de87 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -824,10 +824,12 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort handle.onabort = socket[kBoundOnAbort] ??= onServerRequestEvent.bind(socket); // 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. + // the socket (which emits 'pause' on it) once the buffer fills. The + // callback stays armed until the body's fin chunk even if the response + // is ended first (native keeps delivering), so req 'end'/'close' and + // req.complete track the body actually being received, like Node. if (hasBody) { handle.ondata = onDataIncomingMessage.bind(http_req); - handle.hasCustomOnData = false; } drainMicrotasks(); @@ -2387,10 +2389,17 @@ function stopServerResponsePerf(this: any) { // arm keep-alive) runs first because onResponseFinishHandleSocket's guards // read pre-detach state, then detach the socket and advance the pipeline. function emitResponseFinish() { + const req = this.req; + // If the user never called req.read(), and didn't pipe() or + // .resume() or .on('data'), then we call req._dump() so that the + // bytes will be pulled off the wire. + if (req && !req._consuming && !req._readableState?.resumeScheduled) { + req._dump(); + } // req.socket is nulled by the stream destroyer (pipeline/compose cleanup); // the response's own socket (set by assignSocket, cleared only by // detachSocket) still references the connection then. - const socket = this.req?.socket ?? this.socket; + const socket = req?.socket ?? this.socket; onResponseFinishHandleSocket(socket?.server, socket, this); // The dispatcher detached a synchronously-finished response itself; // advancing the pipeline again here would skip a queued response. @@ -3172,10 +3181,9 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) { } } this._header = " "; - const req = this.req; - if (!req._consuming && !req?._readableState?.resumeScheduled) { - req._dump(); - } + // The unread-body dump decision is made on 'finish' (emitResponseFinish), + // like Node.js's resOnFinish: a consumer attached in the same tick as + // res.end() still gets the body. // The socket is NOT detached here: like Node.js, res.socket stays assigned // until the response 'finish' machinery runs (the dispatcher detaches it // right after a synchronously-finished handler returns, or via its 'finish' diff --git a/src/runtime/server/NodeHTTPResponse.rs b/src/runtime/server/NodeHTTPResponse.rs index 9c5f1edb87a6..d83a0295ac24 100644 --- a/src/runtime/server/NodeHTTPResponse.rs +++ b/src/runtime/server/NodeHTTPResponse.rs @@ -87,7 +87,6 @@ bitflags! { const REQUEST_HAS_COMPLETED = 1 << 1; const ENDED = 1 << 2; const UPGRADED = 1 << 3; - const HAS_CUSTOM_ON_DATA = 1 << 4; const IS_REQUEST_PENDING = 1 << 5; const IS_DATA_BUFFERED_DURING_PAUSE = 1 << 6; /// Did we receive the last chunk of data during pause? @@ -615,32 +614,62 @@ impl NodeHTTPResponse { true } - pub(crate) fn maybe_stop_reading_body(&self, vm: &mut VirtualMachine, this_value: JSValue) { + /// Is a JS `ondata` callback armed on the wrapper that set it (the + /// IncomingMessage is still wired up to receive this request's body)? + fn has_body_reader(&self) -> bool { + let armed = self.armed_this_value.get(); + !armed.is_empty() && js::on_data_get_cached(armed).is_some_and(|cb| cb.is_cell()) + } + + /// The connection's parser is still inside this request's body and its + /// bytes are being routed to this response (uws `inStream` belongs to it). + /// False once the fin chunk arrived, including a fin that was parked in + /// `buffered_request_body_data_during_pause`: from then on the parser may + /// already be on the next pipelined request, whose response owns the + /// shared per-connection callback slots. + fn body_still_arriving(&self) -> bool { + self.body_read_state.get() == BodyReadState::Pending + && !self + .flags + .get() + .contains(Flags::IS_DATA_BUFFERED_DURING_PAUSE_LAST) + } + + /// Called once the handler is done with the response. A body that is still + /// unread is only discarded here when nothing on the JS side is wired to + /// receive it. With an IncomingMessage armed it keeps flowing after + /// `res.end()`, like Node's parser does: consumed by the handler, or + /// discarded by `req._dump()`, with the request completing ('end' / + /// 'close', `req.complete`) only once its fin chunk is actually received. + pub(crate) fn maybe_stop_reading_body(&self, vm: &mut VirtualMachine) { self.upgrade_context.with_mut(|c| c.reset()); // we can discard the upgrade context now let flags = self.flags.get(); - if (flags.contains(Flags::UPGRADED) - || flags.contains(Flags::SOCKET_CLOSED) - || flags.contains(Flags::ENDED)) - && (self.body_read_ref.get().has - || self.body_read_state.get() == BodyReadState::Pending) - && (!flags.contains(Flags::HAS_CUSTOM_ON_DATA) - || js::on_data_get_cached(this_value).is_none()) - { - let had_ref = self.body_read_ref.get().has; - if !flags.contains(Flags::UPGRADED) && !flags.contains(Flags::SOCKET_CLOSED) { - scoped_log!(NodeHTTPResponse, "clearOnData"); - if let Some(raw_response) = self.raw_response.get() { - raw_response.clear_on_data(); - } + // Upgraded / closed: uws delivers nothing further on this response, so + // the read state must be released no matter what JS still has armed. + let transport_gone = + flags.contains(Flags::UPGRADED) || flags.contains(Flags::SOCKET_CLOSED); + let discard = transport_gone || (flags.contains(Flags::ENDED) && !self.has_body_reader()); + if !discard { + return; + } + if !self.body_read_ref.get().has && self.body_read_state.get() != BodyReadState::Pending { + return; + } + + let had_ref = self.body_read_ref.get().has; + if !transport_gone { + scoped_log!(NodeHTTPResponse, "clearOnData"); + if let Some(raw_response) = self.raw_response.get() { + raw_response.clear_on_data(); } + } - self.body_read_ref.with_mut(|r| r.unref(vm)); - self.body_read_state.set(BodyReadState::Done); + self.body_read_ref.with_mut(|r| r.unref(vm)); + self.body_read_state.set(BodyReadState::Done); - if had_ref { - self.mark_request_as_done_if_necessary(); - } + if had_ref { + self.mark_request_as_done_if_necessary(); } } @@ -727,6 +756,10 @@ impl NodeHTTPResponse { .with_mut(|b| b.clear_and_free()); let mut server = self.server; self.poll_ref.with_mut(|r| r.unref(vm)); + // Still held when the connection dies while the body was being received + // after the response had already ended (the only abort path that does + // not go through on_data_or_aborted). + self.body_read_ref.with_mut(|r| r.unref(vm)); self.unregister_auto_flush(); server.on_request_complete(); @@ -1333,19 +1366,19 @@ impl NodeHTTPResponse { let Some(raw) = self.raw_response.get() else { return Ok(JSValue::FALSE); }; - if flags.contains(Flags::REQUEST_HAS_COMPLETED) - || flags.contains(Flags::SOCKET_CLOSED) - || flags.contains(Flags::ENDED) - || flags.contains(Flags::UPGRADED) - { + if flags.contains(Flags::SOCKET_CLOSED) || flags.contains(Flags::UPGRADED) { + return Ok(JSValue::FALSE); + } + let body_still_arriving = self.body_still_arriving(); + // Once the response is over, request-body flow control only has + // something to act on while that body is still coming in. + if flags.is_requested_completed_or_ended() && !body_still_arriving { return Ok(JSValue::FALSE); } // 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) - { + if body_still_arriving { self.update_flags(|f| f.insert(Flags::IS_DATA_BUFFERED_DURING_PAUSE)); raw.on_data(on_buffer_paused_shim, self.as_ctx_ptr()); } @@ -1403,9 +1436,7 @@ impl NodeHTTPResponse { let Some(raw) = self.raw_response.get() else { return Ok(JSValue::FALSE); }; - if flags.contains(Flags::REQUEST_HAS_COMPLETED) - || flags.contains(Flags::SOCKET_CLOSED) - || flags.contains(Flags::ENDED) + if flags.contains(Flags::SOCKET_CLOSED) || flags.contains(Flags::UPGRADED) // A CONNECT tunnel's bytes reach JS via onSocketData; arming inStream // here would deliver them twice (and park them in the body buffer). @@ -1413,16 +1444,28 @@ impl NodeHTTPResponse { { return Ok(JSValue::FALSE); } + let response_over = flags.is_requested_completed_or_ended(); + let body_still_arriving = self.body_still_arriving(); + if response_over && !body_still_arriving { + return Ok(JSValue::FALSE); + } // 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(); + if body_still_arriving { + if !response_over { + 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)); + if response_over { + // The response has been detached from the connection socket, so the + // socket-level resume (NodeHTTPServerSocket#resumeSocket) no longer + // knows which request a drained buffer belongs to; leave it for the + // request's own _read() -> drainRequestBody(), which does. + return Ok(JSValue::TRUE); + } Ok(self .drain_buffered_request_body_from_pause(global_object)? .unwrap_or(JSValue::TRUE)) @@ -1454,7 +1497,7 @@ fn node_http_request_on_resolve(global_object: &JSGlobalObject, callframe: &Call p.deinit(); had }); - this.maybe_stop_reading_body(bun_vm_mut(global_object), arguments[1]); + this.maybe_stop_reading_body(bun_vm_mut(global_object)); let flags = this.flags.get(); if !flags.contains(Flags::REQUEST_HAS_COMPLETED) && !flags.contains(Flags::SOCKET_CLOSED) { @@ -1497,7 +1540,7 @@ fn node_http_request_on_reject(global_object: &JSGlobalObject, callframe: &CallF p.deinit(); had }); - this.maybe_stop_reading_body(bun_vm_mut(global_object), arguments[1]); + this.maybe_stop_reading_body(bun_vm_mut(global_object)); let flags = this.flags.get(); if !flags.contains(Flags::REQUEST_HAS_COMPLETED) @@ -1652,7 +1695,8 @@ impl NodeHTTPResponse { if !on_data_armed && body_was_pending && event == AbortEvent::None { // No reader armed yet: pipelined request whose body arrived in the same parse burst // as its headers, before JS ran _read() to install ondata. Park it where pause parks; - // the reader-arm drain picks it up. (Dumped requests move to Done first, never here.) + // the reader-arm drain picks it up. (A dumped request keeps its reader armed and takes + // the branch below, which discards the chunk in JS and still sees the fin.) self.buffered_request_body_data_during_pause .with_mut(|b| b.append_slice(chunk)); self.update_flags(|f| { @@ -1684,10 +1728,13 @@ impl NodeHTTPResponse { // Deferred tail: if last { - if self.body_read_ref.get().has { - self.body_read_ref.with_mut(|r| r.unref(vm_get())); - self.mark_request_as_done_if_necessary(); - } + self.body_read_ref.with_mut(|r| r.unref(vm_get())); + // Not gated on the ref still being held: when the response had already ended, the + // fin callback above ran the request's 'end' -> autoDestroy -> `ondata = undefined` + // (set_on_data) via the nextTick drain, which released the ref first. The body was + // the only thing keeping such a request pending, so it has to be re-evaluated here; + // a request whose response is still in flight stays pending either way. + self.mark_request_as_done_if_necessary(); } } @@ -2012,12 +2059,15 @@ impl NodeHTTPResponse { self.spill_pending_pinned_write(global_object); if IS_END { - // Discard the body read ref if it's pending and no onData callback is set at this point. - // This is the equivalent of req._dump(). - if self.body_read_ref.get().has + // Like Node, a request body that is still arriving when the response + // goes out keeps being parsed and delivered to the IncomingMessage + // (which consumes it, or discards it once req._dump() ran); the + // request completes at the body's fin, not here. Only a body nobody + // is wired to receive is dropped at this point. + let keep_reading_body = self.body_still_arriving() && self.has_body_reader(); + if !keep_reading_body + && self.body_read_ref.get().has && self.body_read_state.get() == BodyReadState::Pending - && (!self.flags.get().contains(Flags::HAS_CUSTOM_ON_DATA) - || js::on_data_get_cached(this_value).is_none()) { self.body_read_ref.with_mut(|r| r.unref(vm_get())); self.body_read_state.set(BodyReadState::None); @@ -2038,6 +2088,24 @@ impl NodeHTTPResponse { } else { raw_response.end_stream(state.is_http_connection_close()); } + if keep_reading_body { + // uws's end() (markDone) dropped the connection's body data + // handler; put it back, in whichever mode the reader left it. + // A Connection: close response may have closed the socket + // synchronously inside end() (set_closed ran), in which case + // there is nothing left to read from and the close path + // releases the read state. + let flags = self.flags.get(); + if !flags.contains(Flags::SOCKET_CLOSED) { + if let Some(raw_response) = self.raw_response.get() { + if flags.contains(Flags::IS_DATA_BUFFERED_DURING_PAUSE) { + raw_response.on_data(on_buffer_paused_shim, self.as_ctx_ptr()); + } else { + raw_response.on_data(on_data_shim, self.as_ctx_ptr()); + } + } + } + } self.on_request_complete(); Ok(JSValue::js_number_from_uint64(bytes.len() as u64)) @@ -2196,18 +2264,10 @@ impl NodeHTTPResponse { js::on_data_get_cached(this_value).unwrap_or(JSValue::UNDEFINED) } - pub(crate) fn get_has_custom_on_data(&self, _global: &JSGlobalObject) -> JSValue { - JSValue::from(self.flags.get().contains(Flags::HAS_CUSTOM_ON_DATA)) - } - pub(crate) fn get_upgraded(&self, _global: &JSGlobalObject) -> JSValue { JSValue::from(self.flags.get().contains(Flags::UPGRADED)) } - pub(crate) fn set_has_custom_on_data(&self, _global: &JSGlobalObject, value: JSValue) { - self.update_flags(|f| f.set(Flags::HAS_CUSTOM_ON_DATA, value.to_boolean())); - } - fn clear_on_data_callback(&self, this_value: JSValue, global_object: &JSGlobalObject) { scoped_log!(NodeHTTPResponse, "clearOnDataCallback"); // Clear on the wrapper that armed ondata (see on_data): the parameter may @@ -2237,22 +2297,26 @@ impl NodeHTTPResponse { global_object: &JSGlobalObject, value: JSValue, ) { - // Only `.pending` accepts a callback. `.done` means either uSockets delivered last=true or JS - // previously cleared `ondata` (which already called clearOnData()); either way, there is no - // more body to read, so don't re-register with uSockets or churn refs. + // A callback is only accepted while this request's body is still arriving (whether or not + // the response has ended). `.done` / a parked fin means either uSockets delivered last=true + // or JS previously cleared `ondata` (which already called clearOnData()); either way, there + // is no more body to read, so don't re-register with uSockets or churn refs. let flags = self.flags.get(); + let body_still_arriving = self.body_still_arriving(); if value.is_undefined_or_null() - || flags.contains(Flags::ENDED) || flags.contains(Flags::SOCKET_CLOSED) - || self.body_read_state.get() != BodyReadState::Pending - || flags.contains(Flags::IS_DATA_BUFFERED_DURING_PAUSE_LAST) || flags.contains(Flags::UPGRADED) + || !body_still_arriving { js::on_data_set_cached(this_value, global_object, JSValue::UNDEFINED); self.armed_this_value.set(JSValue::ZERO); + let mut stopped_reading_body = false; match self.body_read_state.get() { BodyReadState::Pending | BodyReadState::Done => { - if !flags.contains(Flags::REQUEST_HAS_COMPLETED) + // Once the response has completed, the connection's data handler is only + // still this request's while its body is the one being parsed (write_or_end + // re-armed it); afterwards it may already belong to the next request. + if (body_still_arriving || !flags.contains(Flags::REQUEST_HAS_COMPLETED)) && !flags.contains(Flags::SOCKET_CLOSED) && !flags.contains(Flags::UPGRADED) { @@ -2260,6 +2324,7 @@ impl NodeHTTPResponse { if let Some(raw_response) = self.raw_response.get() { raw_response.clear_on_data(); } + stopped_reading_body = body_still_arriving; } self.body_read_state.set(BodyReadState::Done); } @@ -2269,6 +2334,12 @@ impl NodeHTTPResponse { self.body_read_ref .with_mut(|r| r.unref(bun_vm_mut(global_object))); } + if stopped_reading_body { + // The reader was torn down (req.destroy()) mid-body and uws will not call back + // into this response for it anymore; when the response is over as well, that body + // was all that still kept the request pending. + self.mark_request_as_done_if_necessary(); + } return; } @@ -2278,7 +2349,6 @@ impl NodeHTTPResponse { value.with_async_context_if_needed(global_object), ); self.armed_this_value.set(this_value); - self.update_flags(|f| f.insert(Flags::HAS_CUSTOM_ON_DATA)); if let Some(raw_response) = self.raw_response.get() { raw_response.on_data(on_data_shim, self.as_ctx_ptr()); } diff --git a/src/runtime/server/mod.rs b/src/runtime/server/mod.rs index 5f24285425cf..0c98f1533479 100644 --- a/src/runtime/server/mod.rs +++ b/src/runtime/server/mod.rs @@ -1480,9 +1480,8 @@ impl NewServer { } // If we ended the response without attaching an ondata handler, we discard the body read stream else if !matches!(http_result, HttpResult::Pending) { - let this_value = nhr.get_this_value(); // SAFETY: `vm` is the process-static VirtualMachine. - nhr.maybe_stop_reading_body(unsafe { &mut *vm }, this_value); + nhr.maybe_stop_reading_body(unsafe { &mut *vm }); } } if nhr_flags.contains(NhrFlags::TUNNELED) { diff --git a/src/runtime/server/server.classes.ts b/src/runtime/server/server.classes.ts index a3f910968d9b..0e01de0feb6e 100644 --- a/src/runtime/server/server.classes.ts +++ b/src/runtime/server/server.classes.ts @@ -222,10 +222,6 @@ export default [ setter: "setOnAbort", this: true, }, - hasCustomOnData: { - getter: "getHasCustomOnData", - setter: "setHasCustomOnData", - }, upgraded: { getter: "getUpgraded", }, diff --git a/test/js/node/http/node-http-ondata-reregister-leak.fixture.js b/test/js/node/http/node-http-ondata-reregister-leak.fixture.js index 426b5c459a09..a42e32334a80 100644 --- a/test/js/node/http/node-http-ondata-reregister-leak.fixture.js +++ b/test/js/node/http/node-http-ondata-reregister-leak.fixture.js @@ -17,7 +17,7 @@ const server = http.createServer(async (req, res) => { for (const sym of Object.getOwnPropertySymbols(req)) { if (sym.description !== "handle") continue; const val = req[sym]; - if (val && typeof val === "object" && "hasBody" in val && "hasCustomOnData" in val) { + if (val && typeof val === "object" && "hasBody" in val && "ondata" in val) { handle = val; break; } diff --git a/test/js/node/http/node-http-server-abort-events.test.ts b/test/js/node/http/node-http-server-abort-events.test.ts index 44fcf440345f..aa51f06e2251 100644 --- a/test/js/node/http/node-http-server-abort-events.test.ts +++ b/test/js/node/http/node-http-server-abort-events.test.ts @@ -2,9 +2,11 @@ * This test must also pass in Node.js. */ import { describe, expect, test } from "bun:test"; +import { bunEnv, bunExe } from "harness"; import { once } from "node:events"; +import type { Server } from "node:http"; import { createServer, IncomingMessage, ServerResponse } from "node:http"; -import type { AddressInfo } from "node:net"; +import type { AddressInfo, Socket } from "node:net"; import { connect } from "node:net"; import { duplexPair } from "node:stream"; @@ -299,3 +301,299 @@ describe("res.destroy() defers 'close'", () => { expect(events).toEqual(["destroy()", "destroy() returned (closed: false)", "res.close (closed: true)"]); }); }); + +// A handler that answers before the request body has been received (the usual +// shape of an early 413/401/redirect). Node keeps parsing the rest of the body +// in the background: the IncomingMessage is only complete ('end', then 'close', +// req.complete === true) once the body has actually arrived, a consumer +// attached before or in the same tick as res.end() still gets every byte, and +// a connection that drops mid-body leaves the request as it was. +describe("request body arriving after the response was ended", () => { + function reqState(req: IncomingMessage) { + return { complete: req.complete, readableEnded: req.readableEnded, destroyed: req.destroyed, aborted: req.aborted }; + } + + async function listen(server: Server) { + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + return (server.address() as AddressInfo).port; + } + + // A raw client so the request body can be sent in pieces. + async function rawClient(port: number) { + const socket = connect(port, "127.0.0.1"); + socket.on("error", () => {}); + let received = ""; + let onData: (() => void) | undefined; + socket.on("data", chunk => { + received += chunk; + onData?.(); + }); + await once(socket, "connect"); + return { + socket, + write: (data: string) => new Promise(resolve => socket.write(data, () => resolve())), + // Resolves once the response body `marker` has been received; the + // response is on the wire before the server-side request can complete. + async response(marker: string) { + while (!received.includes(marker)) { + await new Promise(resolve => (onData = resolve)); + } + const out = received; + received = ""; + return out; + }, + }; + } + + function closeServer(server: Server) { + // Resolves only once every request has been released by the server; a + // request that is never accounted as finished hangs this (and the test). + return new Promise((resolve, reject) => server.close(err => (err ? reject(err) : resolve()))); + } + + test("req completes with 'end' then 'close' once the rest of the body arrives, not at res.end()", async () => { + const events: string[] = []; + const { promise: request, resolve: gotRequest } = Promise.withResolvers(); + const { promise: reqClosed, resolve: resolveReqClosed } = Promise.withResolvers(); + const server = createServer((req, res) => { + req.on("aborted", () => events.push("aborted")); + req.on("end", () => events.push("end")); + req.on("close", () => { + events.push("close"); + resolveReqClosed(); + }); + res.end("first"); + gotRequest(req); + }); + try { + const client = await rawClient(await listen(server)); + await client.write("POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 6\r\n\r\nabc"); + const req = await request; + await client.response("first"); + + // Half of the body is still outstanding: the message is not complete. + expect({ ...reqState(req), events: [...events] }).toEqual({ + complete: false, + readableEnded: false, + destroyed: false, + aborted: false, + events: [], + }); + + await client.write("def"); + await reqClosed; + expect({ ...reqState(req), events }).toEqual({ + complete: true, + readableEnded: true, + destroyed: true, + aborted: false, + events: ["end", "close"], + }); + + // The trailing body bytes were consumed as body, so the kept-alive + // connection is still in sync for the next request. + await client.write("GET / HTTP/1.1\r\nHost: x\r\n\r\n"); + expect(await client.response("first")).toStartWith("HTTP/1.1 200 OK"); + + client.socket.destroy(); + await closeServer(server); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + test("a connection dropped mid-body after the response leaves req incomplete and emits nothing", async () => { + const events: string[] = []; + const { promise: request, resolve: gotRequest } = Promise.withResolvers(); + const { promise: serverSocketClosed, resolve: resolveServerSocketClosed } = Promise.withResolvers(); + const server = createServer((req, res) => { + for (const name of ["aborted", "end", "close", "error"]) req.on(name, () => events.push(name)); + (req.socket as Socket).on("close", () => resolveServerSocketClosed()); + res.end("first"); + gotRequest(req); + }); + // The half-sent body makes the peer's close a parse error on the + // connection ('clientError', like Node); the connection is already gone. + server.on("clientError", (_err, socket) => socket.destroy()); + try { + const client = await rawClient(await listen(server)); + await client.write("POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 6\r\n\r\nabc"); + const req = await request; + await client.response("first"); + + client.socket.destroy(); + await serverSocketClosed; + // Like Node's socketOnClose, only requests whose response has not + // finished are aborted; this one is simply left incomplete. + await closeServer(server); + expect({ ...reqState(req), events, socketDestroyed: req.socket.destroyed }).toEqual({ + complete: false, + readableEnded: false, + destroyed: false, + aborted: false, + events: [], + socketDestroyed: true, + }); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + // Issues #4733 and #18613: the body is lost when res.end() runs + // synchronously in the handler, whether it was in the same packet as the + // headers (curl -d) or is still in flight. + test.each([ + ["in the same packet as the headers", "hello", ""], + ["still in flight", "he", "llo"], + ])("a 'data' listener attached before a synchronous res.end() receives a body %s", async (_, first, rest) => { + const { promise: body, resolve: resolveBody } = Promise.withResolvers(); + const server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", chunk => chunks.push(chunk)); + req.on("end", () => resolveBody({ body: Buffer.concat(chunks).toString(), ...reqState(req) })); + res.end("first"); + }); + try { + const client = await rawClient(await listen(server)); + await client.write(`POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\n\r\n${first}`); + await client.response("first"); + if (rest) await client.write(rest); + expect(await body).toEqual({ + body: "hello", + complete: true, + readableEnded: true, + destroyed: false, + aborted: false, + }); + client.socket.destroy(); + await closeServer(server); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + test("a consumer attached in the same tick after res.end() still receives the body", async () => { + // Node decides whether to dump an unread body on the response's 'finish' + // (resOnFinish), so a listener attached right after res.end() counts. + const { promise: body, resolve: resolveBody } = Promise.withResolvers(); + const server = createServer((req, res) => { + res.end("first"); + const chunks: Buffer[] = []; + req.on("data", chunk => chunks.push(chunk)); + req.on("end", () => resolveBody({ body: Buffer.concat(chunks).toString(), complete: req.complete })); + }); + try { + const client = await rawClient(await listen(server)); + await client.write("POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\n\r\nhe"); + await client.response("first"); + await client.write("llo"); + expect(await body).toEqual({ body: "hello", complete: true }); + client.socket.destroy(); + await closeServer(server); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + test("req.pause()/resume() keep working for a body arriving after the response", async () => { + const { promise: request, resolve: gotRequest } = Promise.withResolvers(); + const { promise: firstChunk, resolve: gotFirstChunk } = Promise.withResolvers(); + const { promise: body, resolve: resolveBody } = Promise.withResolvers(); + const server = createServer((req, res) => { + res.end("first"); + const chunks: Buffer[] = []; + req.on("data", chunk => { + chunks.push(chunk); + if (chunks.length === 1) { + req.pause(); + gotFirstChunk(); + } + }); + req.on("end", () => resolveBody({ body: Buffer.concat(chunks).toString(), complete: req.complete })); + gotRequest(req); + }); + try { + const client = await rawClient(await listen(server)); + await client.write("POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 9\r\n\r\n"); + const req = await request; + await client.response("first"); + await client.write("abc"); + await firstChunk; + expect(req.isPaused()).toBe(true); + await client.write("defghi"); + req.resume(); + expect(await body).toEqual({ body: "abcdefghi", complete: true }); + client.socket.destroy(); + await closeServer(server); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + // The body outliving the response must not leave anything holding the event + // loop open: the process has to exit on its own both when the body does + // arrive later (with the connection then reused, so the early request is no + // longer the connection's current one when it closes) and when the client + // goes away mid-body instead. + test.concurrent.each(["complete", "abort"])( + "process exits on its own after an early response (%s)", + async mode => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { createServer } = require("node:http"); + const { connect } = require("node:net"); + const server = createServer((req, res) => { + if (req.url === "/early") { + req.on("close", () => console.log("req close complete=" + req.complete)); + req.socket.on("close", () => console.log("socket close complete=" + req.complete)); + } + res.end("ok:" + req.url); + }); + server.listen(0, "127.0.0.1", () => { + const client = connect(server.address().port, "127.0.0.1"); + let received = ""; + client.on("data", chunk => { + received += chunk; + if (received.endsWith("ok:/early")) { + received = ""; + if (${JSON.stringify(mode)} === "abort") { + client.destroy(); + server.close(() => console.log("server closed")); + } else { + client.write("def"); + client.write("GET /second HTTP/1.1\\r\\nHost: x\\r\\n\\r\\n"); + } + } else if (received.endsWith("ok:/second")) { + client.end(); + server.close(() => console.log("server closed")); + } + }); + client.write("POST /early HTTP/1.1\\r\\nHost: x\\r\\nContent-Length: 6\\r\\n\\r\\nabc"); + }); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout.trim().split("\n").sort()).toEqual( + mode === "abort" + ? ["server closed", "socket close complete=false"] + : ["req close complete=true", "server closed", "socket close complete=true"], + ); + expect(exitCode).toBe(0); + }, + 30_000, + ); +}); From 682d83263f6618a9e57bb9c6614da888315add07 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 13 Aug 2026 20:07:36 +0000 Subject: [PATCH 2/8] node:http: add the consumer-side tests for a body arriving after res.end() Carried over from #35489: pipe()/on('data') before and after res.end(), res.write()+res.end(), res.end() on nextTick, the _dump() decision being made on 'finish', keep-alive reuse across three consumed bodies, a chunked body split across segments and a mid-upload close reaching beforeExit. --- test/js/node/http/node-http.test.ts | 255 ++++++++++++++++++++++++++++ 1 file changed, 255 insertions(+) diff --git a/test/js/node/http/node-http.test.ts b/test/js/node/http/node-http.test.ts index c555aa1b5576..55038932755b 100644 --- a/test/js/node/http/node-http.test.ts +++ b/test/js/node/http/node-http.test.ts @@ -4366,3 +4366,258 @@ describe("response header values are written as latin-1 bytes", () => { ]); }); }); + +// https://github.com/oven-sh/bun/issues/4733 +// https://github.com/oven-sh/bun/issues/18613 +// Ending the response inside the request handler must not drop the request +// body: like Node.js, the body keeps flowing to req's 'data' listeners and +// piped destinations until it has been fully received, and resOnFinish (the +// 'finish' listener) decides whether to _dump() based on the state *at* +// 'finish', so a consumer attached after res.end() in the same tick still +// receives the body. (req.complete / 'end' / 'close' tracking the body, and +// connections the response closes, are covered in +// node-http-server-abort-events.test.ts.) +describe("request body still flows after res.end() was called in the handler", () => { + async function run(handler: (req: IncomingMessage, res: ServerResponse, out: Writable) => void) { + const events: string[] = []; + const chunks: Buffer[] = []; + const { promise: finished, resolve: finish, reject } = Promise.withResolvers(); + + let reqRef: IncomingMessage | undefined; + await using server = createServer((req, res) => { + reqRef = req; + const out = new Writable({ + write(chunk, _enc, cb) { + chunks.push(Buffer.from(chunk)); + cb(); + }, + }); + req.once("end", () => events.push("req-end")); + req.once("close", () => events.push("req-close")); + req.once("error", reject); + out.once("error", reject); + out.once("finish", () => { + events.push("out-finish"); + finish(); + }); + handler(req, res, out); + }); + + await once(server.listen(0, "127.0.0.1"), "listening"); + const { port } = server.address() as AddressInfo; + const resp = await fetch(`http://127.0.0.1:${port}/`, { method: "POST", body: "testing-body" }); + expect(await resp.text()).toBe("ok"); + await finished; + + expect({ + body: Buffer.concat(chunks).toString(), + events, + dumped: reqRef!._dumped, + }).toEqual({ + body: "testing-body", + events: ["req-end", "out-finish", "req-close"], + dumped: false, + }); + } + + it("req.pipe(out) before res.end()", async () => { + await run((req, res, out) => { + req.pipe(out); + res.end("ok"); + }); + }); + + it("req.on('data') before res.end()", async () => { + await run((req, res, out) => { + req.on("data", c => out.write(c)); + req.on("end", () => out.end()); + res.end("ok"); + }); + }); + + it("req.pipe(out) after res.end() in the same tick", async () => { + await run((req, res, out) => { + res.end("ok"); + req.pipe(out); + }); + }); + + it("req.on('data') after res.end() in the same tick", async () => { + await run((req, res, out) => { + res.end("ok"); + req.on("data", c => out.write(c)); + req.on("end", () => out.end()); + }); + }); + + it("req.pipe(out) with res.write() + res.end()", async () => { + await run((req, res, out) => { + req.pipe(out); + res.write("o"); + res.end("k"); + }); + }); + + it("req.pipe(out) with res.end() on nextTick", async () => { + await run((req, res, out) => { + req.pipe(out); + process.nextTick(() => res.end("ok")); + }); + }); + + it("with no consumer, req is _dumped on 'finish' like Node's resOnFinish", async () => { + let dumpedAtFinish: boolean | undefined; + let reqRef: IncomingMessage | undefined; + const { promise: closed, resolve, reject } = Promise.withResolvers(); + await using server = createServer((req, res) => { + reqRef = req; + req.once("error", reject); + req.once("close", resolve); + res.end("ok"); + // emitResponseFinish is registered before the 'request' event, so by the + // time this listener runs req._dump() has already been called. + res.on("finish", () => (dumpedAtFinish = req._dumped)); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const { port } = server.address() as AddressInfo; + const resp = await fetch(`http://127.0.0.1:${port}/`, { method: "POST", body: "testing-body" }); + expect(await resp.text()).toBe("ok"); + await closed; + expect({ dumpedAtFinish, dumped: reqRef!._dumped }).toEqual({ dumpedAtFinish: true, dumped: true }); + }); + + it("req.resume() after res.end() in the same tick prevents _dump()", async () => { + let dumped: boolean | undefined; + const { promise: ended, resolve, reject } = Promise.withResolvers(); + await using server = createServer((req, res) => { + req.once("error", reject); + res.end("ok"); + req.resume(); + req.once("end", () => { + dumped = req._dumped; + resolve(); + }); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const { port } = server.address() as AddressInfo; + const resp = await fetch(`http://127.0.0.1:${port}/`, { method: "POST", body: "testing-body" }); + expect(await resp.text()).toBe("ok"); + await ended; + expect(dumped).toBe(false); + }); + + it("keep-alive connection reused after a consumed body releases the request", async () => { + // The body's fin arriving on the re-armed inStream after res.end() must + // release the pending-request ref so server.close() resolves and the + // process exits. + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const http = require("http"); + const { once } = require("events"); + (async () => { + const bodies = []; + const server = http.createServer((req, res) => { + let body = ""; + req.on("data", c => body += c); + req.once("end", () => bodies.push(body)); + res.end("ok"); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const agent = new http.Agent({ keepAlive: true, maxSockets: 1 }); + const port = server.address().port; + for (let i = 0; i < 3; i++) { + await new Promise((resolve, reject) => { + const r = http.request({ agent, method: "POST", port }, res => { + res.resume(); + res.on("end", resolve); + res.on("error", reject); + }); + r.on("error", reject); + r.end("body" + i); + }); + } + agent.destroy(); + await new Promise(r => server.close(r)); + console.log(JSON.stringify(bodies)); + })();`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout.trim()).toBe('["body0","body1","body2"]'); + expect(exitCode).toBe(0); + }, 20_000); + + it("chunked request body split across writes", async () => { + let body = ""; + const { promise: ended, resolve, reject } = Promise.withResolvers(); + await using server = createServer((req, res) => { + req.on("data", c => (body += c)); + req.once("end", resolve); + req.once("error", reject); + req.once("close", () => reject(new Error("closed before end"))); + res.end("ok"); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const { port } = server.address() as AddressInfo; + + const sock = connect(port, "127.0.0.1"); + await once(sock, "connect"); + sock.write("POST / HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n"); + // A second TCP segment carrying the remainder (via setNoDelay + event-loop + // bounce) so res.end() has already run when it arrives. + sock.setNoDelay(true); + await new Promise(r => setImmediate(r)); + sock.write("6\r\n world\r\n0\r\n\r\n"); + + await ended; + sock.end(); + expect(body).toBe("hello world"); + }); + + it("socket closed mid-upload does not strand the event loop", async () => { + // Covers the case where the body's fin never arrives after the response + // ended: the body-read ref must be released on teardown. + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const http = require("http"); + const net = require("net"); + const { once } = require("events"); + (async () => { + const events = []; + const server = http.createServer((req, res) => { + req.on("data", c => events.push("data(" + c.length + ")")); + req.on("end", () => events.push("end")); + res.end("ok"); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const s = net.connect(server.address().port, "127.0.0.1"); + await once(s, "connect"); + s.write("POST / HTTP/1.1\\r\\nHost: x\\r\\nContent-Length: 100\\r\\n\\r\\nabc"); + s.resume(); + await once(s, "data"); + s.destroy(); + await once(s, "close"); + server.close(); + process.on("beforeExit", () => console.log(JSON.stringify(events))); + })();`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + // Like Node.js: the partial body is delivered, 'end' is not (incomplete), + // and the process reaches beforeExit. + expect(stdout.trim()).toBe('["data(3)"]'); + expect(exitCode).toBe(0); + }, 20_000); +}); From 1576084218d31dc7a6215d0b1ea2cc6a1936fa7d Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 13 Aug 2026 20:07:36 +0000 Subject: [PATCH 3/8] node:http: finish parsing the read before closing a non-keep-alive request's connection For an HTTP/1.0 or Connection: close request the response 'finish' listener ends the server socket. When the handler responded before the body was read, that end() runs while the read carrying the request is still being parsed; the shutdown made the parser stop after the request head, so the body bytes already in the buffer were dropped and the request never completed. Hand the close to uWS's post-parse gate in that case (as is already done when response bytes are still buffered): the body is delivered first and the connection is closed right after, which is what Node does. --- packages/bun-uws/src/HttpResponse.h | 16 +++ .../bindings/node/JSNodeHTTPServerSocket.cpp | 15 ++- .../bindings/node/JSNodeHTTPServerSocket.h | 4 +- .../node-http-server-abort-events.test.ts | 119 ++++++++++++++++++ 4 files changed, 149 insertions(+), 5 deletions(-) diff --git a/packages/bun-uws/src/HttpResponse.h b/packages/bun-uws/src/HttpResponse.h index c556f4dece84..aa347d87d0b9 100644 --- a/packages/bun-uws/src/HttpResponse.h +++ b/packages/bun-uws/src/HttpResponse.h @@ -851,6 +851,22 @@ struct HttpResponse : public AsyncSocket { return !(httpResponseData->state & HttpResponseData::HTTP_RESPONSE_PENDING); } + /* node:http compat: the response is complete, but the read that carried its + * request is still being parsed and a body handler was (re-)armed after the + * response ended, i.e. the request body is still being delivered out of the + * current buffer. A shutdown issued now makes the parser stop right after + * the request head (HttpContext's request hook bails on a shut-down socket) + * and drops that body; onData's post-parse close gate is the place to close + * such a connection, once the buffer has been consumed. */ + bool isDeliveringBodyAfterResponse() { + HttpResponseData *httpResponseData = getHttpResponseData(); + + return hasResponded() + && !httpResponseData->isConnectRequest + && httpResponseData->inStream != nullptr + && HttpContext::fromSocket((us_socket_t *) this)->getSocketContextData()->parsingSocket == (us_socket_t *) this; + } + /* Corks the response if possible. Leaves already corked socket be. */ HttpResponse *cork(MoveOnlyFunction &&handler) { if (!Super::isCorked()) { diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp index 92aba2c148f1..357afe4b1f97 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp @@ -245,12 +245,19 @@ bool JSNodeHTTPServerSocket::isClosed() const template static bool deferShutdownUntilResponseDrains(us_socket_t* socket) { - if (reinterpret_cast*>(socket)->getBufferedAmount() == 0) { + auto* response = reinterpret_cast*>(socket); + /* The second case is the 'finish' listener ending the connection of a + * handler that answered before its request body was read (the body is + * still being parsed out of the read that carried the request); Node parses + * everything it has read before resOnFinish's destroySoon() runs. */ + if (reinterpret_cast*>(socket)->getBufferedAmount() == 0 && !response->isDeliveringBodyAfterResponse()) { return false; } - /* HttpContext::onWritable shuts the socket down once the buffered - * response data has flushed and HTTP_CONNECTION_CLOSE is set, so the FIN - * is sequenced after the response bytes (like Node's destroySoon). */ + /* HttpContext's close gates (after the current parse, or from + * onWritable once the buffered response data has flushed) shut the socket + * down when HTTP_CONNECTION_CLOSE is set, so the FIN is sequenced after the + * response bytes and after the body bytes already read (like Node's + * destroySoon). */ auto* httpResponseData = reinterpret_cast*>(us_socket_ext(socket)); httpResponseData->state |= uWS::HttpResponseData::HTTP_CONNECTION_CLOSE; return true; diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.h b/src/jsc/bindings/node/JSNodeHTTPServerSocket.h index eef438fce6b5..def053332fd6 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.h +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.h @@ -105,7 +105,9 @@ class JSNodeHTTPServerSocket : public JSC::JSDestructibleObject { /* node:http socket.end(): when the in-flight response still has bytes in * uWS's send buffer, a shutdown now would put the FIN ahead of them and - * truncate the response. Returns true after handing the close to uWS. */ + * truncate the response; when the finished response's request body is + * still being parsed out of the current read, it would drop that body. + * Returns true after handing the close to uWS in either case. */ bool shutdownAfterResponseDrains(); /* Switch the connection into CONNECT-style tunnel mode after an accepted diff --git a/test/js/node/http/node-http-server-abort-events.test.ts b/test/js/node/http/node-http-server-abort-events.test.ts index aa51f06e2251..5505820ec831 100644 --- a/test/js/node/http/node-http-server-abort-events.test.ts +++ b/test/js/node/http/node-http-server-abort-events.test.ts @@ -536,6 +536,125 @@ describe("request body arriving after the response was ended", () => { } }); + // A request that forbids connection reuse makes the server close the + // connection right after the response. Node still parses everything it has + // already read first: a body that came in with the headers is delivered (or + // dumped) and completes the request before the socket is closed. + describe("on a connection the response closes", () => { + const requestHeads = [ + ["Connection: close", "POST / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n"], + ["HTTP/1.0", "POST / HTTP/1.0\r\nHost: x\r\n"], + ] as const; + + // Resolves with the request's state as observed when the server-side socket + // closed; the closing is what the early response triggers, so by then the + // request must already be in its final state. + function observeUntilSocketClose(server: Server, onRequest: (req: IncomingMessage) => object) { + const { promise, resolve } = Promise.withResolvers(); + server.on("request", req => { + (req.socket as Socket).once("close", () => resolve(onRequest(req))); + }); + return promise; + } + + test.each(requestHeads)( + "a consumer attached before the synchronous res.end() receives a body sent with the headers (%s)", + async (_, head) => { + const events: string[] = []; + const chunks: Buffer[] = []; + const server = createServer((req, res) => { + req.on("data", chunk => chunks.push(chunk)); + req.on("end", () => events.push("end")); + req.on("close", () => events.push("close")); + req.on("error", err => events.push(`error:${(err as NodeJS.ErrnoException).code}`)); + req.on("aborted", () => events.push("aborted")); + res.end("first"); + }); + const observed = observeUntilSocketClose(server, req => ({ + body: Buffer.concat(chunks).toString(), + events: [...events], + ...reqState(req), + })); + try { + const client = await rawClient(await listen(server)); + await client.write(`${head}Content-Length: 5\r\n\r\nhello`); + expect(await observed).toEqual({ + body: "hello", + events: ["end", "close"], + complete: true, + readableEnded: true, + destroyed: true, + aborted: false, + }); + // The server closed the connection, as the request asked, and the + // response made it out before that. + await once(client.socket, "close"); + expect(await client.response("first")).toStartWith("HTTP/1.1 200 OK"); + await closeServer(server); + } finally { + server.closeAllConnections(); + server.close(); + } + }, + ); + + test("an unread body sent with the headers is dumped and still completes the request", async () => { + const events: string[] = []; + const server = createServer((req, res) => { + req.on("end", () => events.push("end")); + req.on("close", () => events.push("close")); + res.end("first"); + }); + const observed = observeUntilSocketClose(server, req => ({ events: [...events], ...reqState(req) })); + try { + const client = await rawClient(await listen(server)); + await client.write("POST / HTTP/1.1\r\nHost: x\r\nConnection: close\r\nContent-Length: 5\r\n\r\nhello"); + expect(await observed).toEqual({ + events: ["end", "close"], + complete: true, + readableEnded: true, + destroyed: true, + aborted: false, + }); + await once(client.socket, "close"); + await closeServer(server); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + + test("the part of the body that had arrived is delivered and the request is left incomplete", async () => { + const events: string[] = []; + const server = createServer((req, res) => { + req.on("data", chunk => events.push(`data:${chunk}`)); + for (const name of ["end", "close", "aborted", "error"]) req.on(name, () => events.push(name)); + res.end("first"); + }); + const observed = observeUntilSocketClose(server, req => ({ events: [...events], ...reqState(req) })); + try { + const client = await rawClient(await listen(server)); + // The rest of the body never comes; the server closes the connection + // after the response regardless, like Node's destroySoon(). + await client.write("POST / HTTP/1.1\r\nHost: x\r\nConnection: close\r\nContent-Length: 100\r\n\r\nabc"); + expect(await observed).toEqual({ + events: ["data:abc"], + complete: false, + readableEnded: false, + destroyed: false, + aborted: false, + }); + await once(client.socket, "close"); + expect(await client.response("first")).toStartWith("HTTP/1.1 200 OK"); + // The request was released even though its body never completed. + await closeServer(server); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + }); + // The body outliving the response must not leave anything holding the event // loop open: the process has to exit on its own both when the body does // arrive later (with the connection then reused, so the early request is no From 4e0d97099104134b215df7c9cd5865476f9d1ab5 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 13 Aug 2026 20:15:50 +0000 Subject: [PATCH 4/8] node:http: register the client close listener before sending in the connection-close tests Client and server share the test's event loop, so the client's 'close' can fire before the server-side close has been observed; a listener added after waiting on the latter could miss it. --- .../node-http-server-abort-events.test.ts | 37 ++++++++++++------- 1 file changed, 24 insertions(+), 13 deletions(-) diff --git a/test/js/node/http/node-http-server-abort-events.test.ts b/test/js/node/http/node-http-server-abort-events.test.ts index 5505820ec831..1bcaf27610ee 100644 --- a/test/js/node/http/node-http-server-abort-events.test.ts +++ b/test/js/node/http/node-http-server-abort-events.test.ts @@ -557,6 +557,18 @@ describe("request body arriving after the response was ended", () => { return promise; } + // Sends the request in one packet and waits for the server to close the + // connection; resolves with what the client received. The close listener is + // registered before writing: client and server share this event loop, so + // the client's 'close' may fire before the server-side one is observed. + async function requestUntilClosed(server: Server, request: string) { + const client = await rawClient(await listen(server)); + const clientClosed = once(client.socket, "close"); + await client.write(request); + await clientClosed; + return client.response("first"); + } + test.each(requestHeads)( "a consumer attached before the synchronous res.end() receives a body sent with the headers (%s)", async (_, head) => { @@ -576,8 +588,7 @@ describe("request body arriving after the response was ended", () => { ...reqState(req), })); try { - const client = await rawClient(await listen(server)); - await client.write(`${head}Content-Length: 5\r\n\r\nhello`); + const response = await requestUntilClosed(server, `${head}Content-Length: 5\r\n\r\nhello`); expect(await observed).toEqual({ body: "hello", events: ["end", "close"], @@ -586,10 +597,8 @@ describe("request body arriving after the response was ended", () => { destroyed: true, aborted: false, }); - // The server closed the connection, as the request asked, and the - // response made it out before that. - await once(client.socket, "close"); - expect(await client.response("first")).toStartWith("HTTP/1.1 200 OK"); + // The response made it out before the connection was closed. + expect(response).toStartWith("HTTP/1.1 200 OK"); await closeServer(server); } finally { server.closeAllConnections(); @@ -607,8 +616,10 @@ describe("request body arriving after the response was ended", () => { }); const observed = observeUntilSocketClose(server, req => ({ events: [...events], ...reqState(req) })); try { - const client = await rawClient(await listen(server)); - await client.write("POST / HTTP/1.1\r\nHost: x\r\nConnection: close\r\nContent-Length: 5\r\n\r\nhello"); + await requestUntilClosed( + server, + "POST / HTTP/1.1\r\nHost: x\r\nConnection: close\r\nContent-Length: 5\r\n\r\nhello", + ); expect(await observed).toEqual({ events: ["end", "close"], complete: true, @@ -616,7 +627,6 @@ describe("request body arriving after the response was ended", () => { destroyed: true, aborted: false, }); - await once(client.socket, "close"); await closeServer(server); } finally { server.closeAllConnections(); @@ -633,10 +643,12 @@ describe("request body arriving after the response was ended", () => { }); const observed = observeUntilSocketClose(server, req => ({ events: [...events], ...reqState(req) })); try { - const client = await rawClient(await listen(server)); // The rest of the body never comes; the server closes the connection // after the response regardless, like Node's destroySoon(). - await client.write("POST / HTTP/1.1\r\nHost: x\r\nConnection: close\r\nContent-Length: 100\r\n\r\nabc"); + const response = await requestUntilClosed( + server, + "POST / HTTP/1.1\r\nHost: x\r\nConnection: close\r\nContent-Length: 100\r\n\r\nabc", + ); expect(await observed).toEqual({ events: ["data:abc"], complete: false, @@ -644,8 +656,7 @@ describe("request body arriving after the response was ended", () => { destroyed: false, aborted: false, }); - await once(client.socket, "close"); - expect(await client.response("first")).toStartWith("HTTP/1.1 200 OK"); + expect(response).toStartWith("HTTP/1.1 200 OK"); // The request was released even though its body never completed. await closeServer(server); } finally { From a3ebec063610cf8840a73e161fdd7bf69946aa8f Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 14 Aug 2026 01:40:56 +0000 Subject: [PATCH 5/8] node:http: only defer the response-finish end(), and dispatch nothing behind it The deferral is now taken only for the end() that onResponseFinishHandleSocket issues (passed down from _final), so a socket.end() from user code behaves as before. While a deferred close is pending, the parser no longer dispatches a request pipelined behind the current message: dispatching it would start a new response and clear the connection's close mark, leaving the ended connection open and serving. Such a request is reported as HPE_CLOSED_CONNECTION like one behind a Connection: close request, and the parse-error path now runs the close gate too, so the connection is closed even when a 'clientError' listener does not destroy it. --- packages/bun-uws/src/HttpContext.h | 10 +++++ packages/bun-uws/src/HttpParser.h | 14 +++++- src/js/node/_http_server.ts | 13 +++++- .../bindings/node/JSNodeHTTPServerSocket.cpp | 28 +++++++----- .../bindings/node/JSNodeHTTPServerSocket.h | 9 ++-- .../node/JSNodeHTTPServerSocketPrototype.cpp | 9 ++-- .../node-http-server-abort-events.test.ts | 45 +++++++++++++++++++ 7 files changed, 105 insertions(+), 23 deletions(-) diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index c3fbda0d72c7..6a50cf5c4e25 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -655,6 +655,16 @@ struct HttpContext { /* Flush anything the 'clientError' handler wrote (uncorking a * closed socket is a no-op). */ ((AsyncSocket *) s)->uncork(); + /* A connection whose close was already decided and deferred to + * the gates (Connection: close, or socket.end() while its last + * message was still being parsed) must still get closed when the + * parse ends in an error instead of reaching the gate below; the + * 'clientError' listener is not required to destroy it. Closed or + * shut-down sockets, and responses still in flight, are left to + * the listener exactly as before. */ + if (!us_socket_is_closed(s) && !us_socket_is_shut_down(s)) { + ((HttpResponse *) s)->closeIfDoneAndMarked(httpResponseData); + } return s; } if(httpContextData->onClientError) { diff --git a/packages/bun-uws/src/HttpParser.h b/packages/bun-uws/src/HttpParser.h index 7925b43cdf92..6a501071dd71 100644 --- a/packages/bun-uws/src/HttpParser.h +++ b/packages/bun-uws/src/HttpParser.h @@ -484,6 +484,15 @@ struct HttpResponseData; return remainingStreamingBytes != 0; } + /* node:http server compat: the JS layer ended the connection (socket.end()) + * but the close itself is left to the connection-close gates so that the + * message being parsed can still be delivered. Nothing after that message + * may be dispatched, exactly as after a message that forbade keep-alive: + * a further request head is reported as HPE_CLOSED_CONNECTION instead. */ + void nodeHttpStopDispatchingAfterCurrentMessage() { + nodeHttpSawConnectionClose = true; + } + /* Maximum number of trailer fields surfaced to JS (the section size cap * already bounds memory; this matches the regular-header count cap). */ static constexpr unsigned MAX_TRAILER_FIELDS = UWS_HTTP_MAX_HEADERS_COUNT - 1; @@ -604,8 +613,9 @@ struct HttpResponseData; /* This guy really has only 30 bits since we reserve two highest bits to chunked encoding parsing state */ uint64_t remainingStreamingBytes = 0; /* node:http compat: a completed request on this connection forbade keep-alive - * (Connection: close, or HTTP/1.0), so no further message may be dispatched - * (llhttp parses nothing after such a message: HPE_CLOSED_CONNECTION). */ + * (Connection: close, or HTTP/1.0), or the JS layer ended the connection + * (nodeHttpStopDispatchingAfterCurrentMessage), so no further message may be + * dispatched (llhttp parses nothing after such a message: HPE_CLOSED_CONNECTION). */ bool nodeHttpSawConnectionClose = false; const size_t MAX_FALLBACK_SIZE = BUN_DEFAULT_MAX_HTTP_HEADER_SIZE; diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index 1a1e9971de87..28e364759a3f 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -1349,6 +1349,11 @@ const kKeepAliveTimeoutSet = Symbol("keepAliveTimeoutSet"); // the socket timer on every response; onSocketTimeoutTimerExpired reads it to // grant the remaining idle budget when the timer actually fires. const kKeepAliveIdleStart = Symbol("keepAliveIdleStart"); +// Set while onResponseFinishHandleSocket ends a connection that must close +// after its response (Node's resOnFinish -> destroySoon), so _final can tell +// that end() from one issued by user code: for it, native lets a request body +// that is still being parsed out of the current read reach the request first. +const kEndAfterResponse = Symbol("kEndAfterResponse"); // HTTP/1.1 pipelining (responses queued behind an in-flight response): // - on the socket: array of queued ServerResponses, in arrival order // - on a queued response: { ops, bytes, needDrain, ended, isAncient } while it @@ -1487,6 +1492,7 @@ function getNodeHTTPServerSocket() { [kBytesWritten] = 0; [kHandle]; [kUpgradeIncoming] = undefined; + [kEndAfterResponse] = false; server: Server; _httpMessage; _secureEstablished = false; @@ -1767,7 +1773,7 @@ function getNodeHTTPServerSocket() { callback(); return; } - handle.end(); + handle.end(this[kEndAfterResponse]); callback(); } @@ -2414,7 +2420,10 @@ function emitResponseFinish() { // is eventually closed. function onResponseFinishHandleSocket(server, socket, res) { if (res[kMustCloseConnection]) { - socket?.end(); + if (socket != null) { + socket[kEndAfterResponse] = true; + socket.end(); + } return; } if (!socket || socket.destroyed || typeof socket.setTimeout !== "function") { diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp index 357afe4b1f97..955f59869a6a 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp @@ -243,35 +243,39 @@ bool JSNodeHTTPServerSocket::isClosed() const } template -static bool deferShutdownUntilResponseDrains(us_socket_t* socket) -{ - auto* response = reinterpret_cast*>(socket); - /* The second case is the 'finish' listener ending the connection of a - * handler that answered before its request body was read (the body is - * still being parsed out of the read that carried the request); Node parses - * everything it has read before resOnFinish's destroySoon() runs. */ - if (reinterpret_cast*>(socket)->getBufferedAmount() == 0 && !response->isDeliveringBodyAfterResponse()) { +static bool deferShutdownUntilResponseDrains(us_socket_t* socket, bool afterResponseFinished) +{ + /* The 'finish' listener ending the connection of a handler that answered + * before its request body was read: the body is still being parsed out of + * the read that carried the request, and Node parses everything it has + * read before resOnFinish's destroySoon() runs. */ + bool bodyStillParsing = afterResponseFinished && reinterpret_cast*>(socket)->isDeliveringBodyAfterResponse(); + if (!bodyStillParsing && reinterpret_cast*>(socket)->getBufferedAmount() == 0) { return false; } /* HttpContext's close gates (after the current parse, or from * onWritable once the buffered response data has flushed) shut the socket * down when HTTP_CONNECTION_CLOSE is set, so the FIN is sequenced after the * response bytes and after the body bytes already read (like Node's - * destroySoon). */ + * destroySoon). Until then the parser may still run over bytes that were + * read together with the current message; a further request in there must + * not be dispatched (it would start a new response on this ended + * connection, and starting one clears HTTP_CONNECTION_CLOSE again). */ auto* httpResponseData = reinterpret_cast*>(us_socket_ext(socket)); httpResponseData->state |= uWS::HttpResponseData::HTTP_CONNECTION_CLOSE; + httpResponseData->nodeHttpStopDispatchingAfterCurrentMessage(); return true; } -bool JSNodeHTTPServerSocket::shutdownAfterResponseDrains() +bool JSNodeHTTPServerSocket::shutdownAfterResponseDrains(bool afterResponseFinished) { if (!socket || upgraded || us_socket_is_closed(socket) || us_socket_is_shut_down(socket)) { return false; } if (is_ssl) { - return deferShutdownUntilResponseDrains(socket); + return deferShutdownUntilResponseDrains(socket, afterResponseFinished); } - return deferShutdownUntilResponseDrains(socket); + return deferShutdownUntilResponseDrains(socket, afterResponseFinished); } template diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.h b/src/jsc/bindings/node/JSNodeHTTPServerSocket.h index def053332fd6..3cd3cd5d7da2 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.h +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.h @@ -105,10 +105,11 @@ class JSNodeHTTPServerSocket : public JSC::JSDestructibleObject { /* node:http socket.end(): when the in-flight response still has bytes in * uWS's send buffer, a shutdown now would put the FIN ahead of them and - * truncate the response; when the finished response's request body is - * still being parsed out of the current read, it would drop that body. - * Returns true after handing the close to uWS in either case. */ - bool shutdownAfterResponseDrains(); + * truncate the response; for the end() issued because the response + * finished (afterResponseFinished), a shutdown while that response's + * request body is still being parsed out of the current read would drop + * the body. Returns true after handing the close to uWS in either case. */ + bool shutdownAfterResponseDrains(bool afterResponseFinished); /* Switch the connection into CONNECT-style tunnel mode after an accepted * Upgrade: subsequent bytes bypass the HTTP parser and stream to the diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp index bdfc8aca98a0..5a595325e07d 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp @@ -216,9 +216,12 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketEnd, (JSC::JSGlobalObject } thisObject->ended = true; - // The response's buffered body must reach the kernel before the FIN; uWS - // performs the shutdown after its send buffer drains. - if (thisObject->shutdownAfterResponseDrains()) { + // The response's buffered body must reach the kernel before the FIN, and + // for the end() that follows a finished response (argument 0) so must the + // request body still being parsed reach the request; uWS performs the + // shutdown once that is through. + bool afterResponseFinished = callFrame->argument(0).isTrue(); + if (thisObject->shutdownAfterResponseDrains(afterResponseFinished)) { return JSValue::encode(JSC::jsUndefined()); } auto bufferedSize = thisObject->streamBuffer.bufferedSize(); diff --git a/test/js/node/http/node-http-server-abort-events.test.ts b/test/js/node/http/node-http-server-abort-events.test.ts index 1bcaf27610ee..0c9da766b993 100644 --- a/test/js/node/http/node-http-server-abort-events.test.ts +++ b/test/js/node/http/node-http-server-abort-events.test.ts @@ -664,6 +664,51 @@ describe("request body arriving after the response was ended", () => { server.close(); } }); + + // A second request pipelined behind the one that closes the connection, in + // the same packet as its body: the body is still delivered, and the + // connection closes after the first response without answering the second + // request, whether the request or the response asked for the close, and + // whether or not a 'clientError' listener (which sees the rejected second + // request) takes care of destroying the connection itself. + const pipelined = "GET /second HTTP/1.1\r\nHost: x\r\n\r\n"; + test.each([ + ["the request asked to close", "Connection: close\r\n", false, false], + ["the response asked to close", "", true, false], + [ + "the request asked to close and a 'clientError' listener ignores the rest", + "Connection: close\r\n", + false, + true, + ], + ])( + "a request pipelined behind it is not answered (%s)", + async (_, closeHeader, closeFromResponse, ignoreClientErrors) => { + const { promise: body, resolve: resolveBody } = Promise.withResolvers(); + const server = createServer((req, res) => { + if (req.url === "/first") { + const chunks: Buffer[] = []; + req.on("data", chunk => chunks.push(chunk)); + req.on("end", () => resolveBody(Buffer.concat(chunks).toString())); + if (closeFromResponse) res.setHeader("Connection", "close"); + } + res.end("first"); + }); + if (ignoreClientErrors) server.on("clientError", () => {}); + try { + const response = await requestUntilClosed( + server, + `POST /first HTTP/1.1\r\nHost: x\r\n${closeHeader}Content-Length: 5\r\n\r\nhello${pipelined}`, + ); + expect(await body).toBe("hello"); + expect(response.match(/HTTP\/1\.1 200 OK/g)).toHaveLength(1); + await closeServer(server); + } finally { + server.closeAllConnections(); + server.close(); + } + }, + ); }); // The body outliving the response must not leave anything holding the event From 3a2ee08dcacb4eaef170387e0c86276e01b5f829 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 14 Aug 2026 01:54:03 +0000 Subject: [PATCH 6/8] node:http: tighten review-flagged comments --- src/js/node/_http_incoming.ts | 10 +-- src/js/node/_http_server.ts | 22 ++--- .../bindings/node/JSNodeHTTPServerSocket.cpp | 18 ++-- .../bindings/node/JSNodeHTTPServerSocket.h | 9 +- .../node/JSNodeHTTPServerSocketPrototype.cpp | 7 +- src/runtime/server/NodeHTTPResponse.rs | 90 ++++++++----------- 6 files changed, 60 insertions(+), 96 deletions(-) diff --git a/src/js/node/_http_incoming.ts b/src/js/node/_http_incoming.ts index a70df7984a8f..a7f5d28e904d 100644 --- a/src/js/node/_http_incoming.ts +++ b/src/js/node/_http_incoming.ts @@ -389,9 +389,8 @@ IncomingMessage.prototype._read = function _read(_n) { const bodyReadState = handle.hasBody; - // A dumped request (res.end() before the body was read) is not complete - // here: like Node, it only ends once the rest of its body has actually - // arrived, which onDataIncomingMessage reports with isLast. + // A dumped request is not complete yet: like Node, it ends only when the + // rest of its body arrives (onDataIncomingMessage's isLast). if ((bodyReadState & NodeHTTPBodyReadState.done) !== 0 || bodyReadState === NodeHTTPBodyReadState.none) { emitEOFIncomingMessage(this); } @@ -757,9 +756,8 @@ IncomingMessage.prototype._dump = function _dump() { // If there is buffered data, it may trigger 'data' events. // Remove 'data' event listeners explicitly. this.removeAllListeners("data"); - // The native ondata callback stays armed: onDataIncomingMessage drops the - // chunks of a dumped request and still delivers the body's fin, which is - // what completes the request (see _read). + // The native ondata stays armed: it discards the chunks but still + // delivers the fin that completes the request (see _read). this.resume(); } }; diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index 28e364759a3f..a547fd16ebaf 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -824,10 +824,8 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort handle.onabort = socket[kBoundOnAbort] ??= onServerRequestEvent.bind(socket); // 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. The - // callback stays armed until the body's fin chunk even if the response - // is ended first (native keeps delivering), so req 'end'/'close' and - // req.complete track the body actually being received, like Node. + // the socket once the buffer fills. The callback stays armed past + // res.end(), so req.complete tracks the body actually being received. if (hasBody) { handle.ondata = onDataIncomingMessage.bind(http_req); } @@ -1349,10 +1347,9 @@ const kKeepAliveTimeoutSet = Symbol("keepAliveTimeoutSet"); // the socket timer on every response; onSocketTimeoutTimerExpired reads it to // grant the remaining idle budget when the timer actually fires. const kKeepAliveIdleStart = Symbol("keepAliveIdleStart"); -// Set while onResponseFinishHandleSocket ends a connection that must close -// after its response (Node's resOnFinish -> destroySoon), so _final can tell -// that end() from one issued by user code: for it, native lets a request body -// that is still being parsed out of the current read reach the request first. +// Distinguishes the end() from onResponseFinishHandleSocket (Node's +// destroySoon) from a user-issued one: for it, native lets a body still being +// parsed reach the request before the shutdown. const kEndAfterResponse = Symbol("kEndAfterResponse"); // HTTP/1.1 pipelining (responses queued behind an in-flight response): // - on the socket: array of queued ServerResponses, in arrival order @@ -2396,9 +2393,7 @@ function stopServerResponsePerf(this: any) { // read pre-detach state, then detach the socket and advance the pipeline. function emitResponseFinish() { const req = this.req; - // If the user never called req.read(), and didn't pipe() or - // .resume() or .on('data'), then we call req._dump() so that the - // bytes will be pulled off the wire. + // Dump the body if the user never consumed or resumed it (Node's resOnFinish). if (req && !req._consuming && !req._readableState?.resumeScheduled) { req._dump(); } @@ -3190,9 +3185,8 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) { } } this._header = " "; - // The unread-body dump decision is made on 'finish' (emitResponseFinish), - // like Node.js's resOnFinish: a consumer attached in the same tick as - // res.end() still gets the body. + // The unread-body dump happens on 'finish' (emitResponseFinish), like + // Node's resOnFinish, so a same-tick consumer still gets the body. // The socket is NOT detached here: like Node.js, res.socket stays assigned // until the response 'finish' machinery runs (the dispatcher detaches it // right after a synchronously-finished handler returns, or via its 'finish' diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp index 955f59869a6a..b9e62cd3e167 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp @@ -245,22 +245,16 @@ bool JSNodeHTTPServerSocket::isClosed() const template static bool deferShutdownUntilResponseDrains(us_socket_t* socket, bool afterResponseFinished) { - /* The 'finish' listener ending the connection of a handler that answered - * before its request body was read: the body is still being parsed out of - * the read that carried the request, and Node parses everything it has - * read before resOnFinish's destroySoon() runs. */ + /* The 'finish' end() of a handler that answered before reading its body: + * Node parses everything already read before destroySoon() runs. */ bool bodyStillParsing = afterResponseFinished && reinterpret_cast*>(socket)->isDeliveringBodyAfterResponse(); if (!bodyStillParsing && reinterpret_cast*>(socket)->getBufferedAmount() == 0) { return false; } - /* HttpContext's close gates (after the current parse, or from - * onWritable once the buffered response data has flushed) shut the socket - * down when HTTP_CONNECTION_CLOSE is set, so the FIN is sequenced after the - * response bytes and after the body bytes already read (like Node's - * destroySoon). Until then the parser may still run over bytes that were - * read together with the current message; a further request in there must - * not be dispatched (it would start a new response on this ended - * connection, and starting one clears HTTP_CONNECTION_CLOSE again). */ + /* uWS shuts down after the current parse and flush, sequencing the FIN + * after the response and the body bytes already read (Node's destroySoon). + * Dispatching another request meanwhile would start a new response and + * clear HTTP_CONNECTION_CLOSE, so stop dispatching too. */ auto* httpResponseData = reinterpret_cast*>(us_socket_ext(socket)); httpResponseData->state |= uWS::HttpResponseData::HTTP_CONNECTION_CLOSE; httpResponseData->nodeHttpStopDispatchingAfterCurrentMessage(); diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.h b/src/jsc/bindings/node/JSNodeHTTPServerSocket.h index 3cd3cd5d7da2..507e7974ad94 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.h +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.h @@ -103,12 +103,9 @@ class JSNodeHTTPServerSocket : public JSC::JSDestructibleObject { * parser when 'close' is emitted on the socket). */ void stopHTTPParsing(); - /* node:http socket.end(): when the in-flight response still has bytes in - * uWS's send buffer, a shutdown now would put the FIN ahead of them and - * truncate the response; for the end() issued because the response - * finished (afterResponseFinished), a shutdown while that response's - * request body is still being parsed out of the current read would drop - * the body. Returns true after handing the close to uWS in either case. */ + /* node:http socket.end(): defer the shutdown while an immediate FIN would + * truncate the buffered response or (afterResponseFinished) drop a request + * body still being parsed. Returns true after handing the close to uWS. */ bool shutdownAfterResponseDrains(bool afterResponseFinished); /* Switch the connection into CONNECT-style tunnel mode after an accepted diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp index 5a595325e07d..729ce9f4b7fe 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp @@ -216,10 +216,9 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketEnd, (JSC::JSGlobalObject } thisObject->ended = true; - // The response's buffered body must reach the kernel before the FIN, and - // for the end() that follows a finished response (argument 0) so must the - // request body still being parsed reach the request; uWS performs the - // shutdown once that is through. + // uWS performs the shutdown once the buffered response has drained and, + // for the end() after a finished response (argument 0), the body parse is + // done. bool afterResponseFinished = callFrame->argument(0).isTrue(); if (thisObject->shutdownAfterResponseDrains(afterResponseFinished)) { return JSValue::encode(JSC::jsUndefined()); diff --git a/src/runtime/server/NodeHTTPResponse.rs b/src/runtime/server/NodeHTTPResponse.rs index d83a0295ac24..024f1dae60df 100644 --- a/src/runtime/server/NodeHTTPResponse.rs +++ b/src/runtime/server/NodeHTTPResponse.rs @@ -614,19 +614,16 @@ impl NodeHTTPResponse { true } - /// Is a JS `ondata` callback armed on the wrapper that set it (the - /// IncomingMessage is still wired up to receive this request's body)? + /// An IncomingMessage is still wired up (via `ondata`) to receive this + /// request's body. fn has_body_reader(&self) -> bool { let armed = self.armed_this_value.get(); !armed.is_empty() && js::on_data_get_cached(armed).is_some_and(|cb| cb.is_cell()) } - /// The connection's parser is still inside this request's body and its - /// bytes are being routed to this response (uws `inStream` belongs to it). - /// False once the fin chunk arrived, including a fin that was parked in - /// `buffered_request_body_data_during_pause`: from then on the parser may - /// already be on the next pipelined request, whose response owns the - /// shared per-connection callback slots. + /// The parser is still inside this request's body. False once the fin + /// arrived (even parked in the pause buffer): the parser may then be on + /// the next pipelined request, which owns the shared callback slots. fn body_still_arriving(&self) -> bool { self.body_read_state.get() == BodyReadState::Pending && !self @@ -635,18 +632,16 @@ impl NodeHTTPResponse { .contains(Flags::IS_DATA_BUFFERED_DURING_PAUSE_LAST) } - /// Called once the handler is done with the response. A body that is still - /// unread is only discarded here when nothing on the JS side is wired to - /// receive it. With an IncomingMessage armed it keeps flowing after - /// `res.end()`, like Node's parser does: consumed by the handler, or - /// discarded by `req._dump()`, with the request completing ('end' / - /// 'close', `req.complete`) only once its fin chunk is actually received. + /// Called once the handler is done with the response. An unread body is + /// discarded only when no reader is wired to receive it; otherwise it + /// keeps flowing after `res.end()` (like Node's parser) and the request + /// completes at its fin. pub(crate) fn maybe_stop_reading_body(&self, vm: &mut VirtualMachine) { self.upgrade_context.with_mut(|c| c.reset()); // we can discard the upgrade context now let flags = self.flags.get(); - // Upgraded / closed: uws delivers nothing further on this response, so - // the read state must be released no matter what JS still has armed. + // Upgraded / closed: uws delivers nothing further, so release the + // read state regardless of what JS has armed. let transport_gone = flags.contains(Flags::UPGRADED) || flags.contains(Flags::SOCKET_CLOSED); let discard = transport_gone || (flags.contains(Flags::ENDED) && !self.has_body_reader()); @@ -756,9 +751,8 @@ impl NodeHTTPResponse { .with_mut(|b| b.clear_and_free()); let mut server = self.server; self.poll_ref.with_mut(|r| r.unref(vm)); - // Still held when the connection dies while the body was being received - // after the response had already ended (the only abort path that does - // not go through on_data_or_aborted). + // Still held if the connection died mid-body after the response ended + // (the one abort path that skips on_data_or_aborted). self.body_read_ref.with_mut(|r| r.unref(vm)); self.unregister_auto_flush(); @@ -1370,8 +1364,8 @@ impl NodeHTTPResponse { return Ok(JSValue::FALSE); } let body_still_arriving = self.body_still_arriving(); - // Once the response is over, request-body flow control only has - // something to act on while that body is still coming in. + // After the response, flow control only matters while the body is + // still arriving. if flags.is_requested_completed_or_ended() && !body_still_arriving { return Ok(JSValue::FALSE); } @@ -1460,10 +1454,8 @@ impl NodeHTTPResponse { } self.update_flags(|f| f.remove(Flags::IS_DATA_BUFFERED_DURING_PAUSE)); if response_over { - // The response has been detached from the connection socket, so the - // socket-level resume (NodeHTTPServerSocket#resumeSocket) no longer - // knows which request a drained buffer belongs to; leave it for the - // request's own _read() -> drainRequestBody(), which does. + // The socket-level resume can no longer attribute a drained buffer + // to a request; leave it for the request's own drainRequestBody(). return Ok(JSValue::TRUE); } Ok(self @@ -1694,9 +1686,8 @@ impl NodeHTTPResponse { let on_data_armed = js::on_data_get_cached(this_value).is_some_and(|cb| cb.is_cell()); if !on_data_armed && body_was_pending && event == AbortEvent::None { // No reader armed yet: pipelined request whose body arrived in the same parse burst - // as its headers, before JS ran _read() to install ondata. Park it where pause parks; - // the reader-arm drain picks it up. (A dumped request keeps its reader armed and takes - // the branch below, which discards the chunk in JS and still sees the fin.) + // as its headers. Park it where pause parks; the reader-arm drain picks it up. + // (A dumped request keeps its reader armed and takes the branch below.) self.buffered_request_body_data_during_pause .with_mut(|b| b.append_slice(chunk)); self.update_flags(|f| { @@ -1729,11 +1720,9 @@ impl NodeHTTPResponse { // Deferred tail: if last { self.body_read_ref.with_mut(|r| r.unref(vm_get())); - // Not gated on the ref still being held: when the response had already ended, the - // fin callback above ran the request's 'end' -> autoDestroy -> `ondata = undefined` - // (set_on_data) via the nextTick drain, which released the ref first. The body was - // the only thing keeping such a request pending, so it has to be re-evaluated here; - // a request whose response is still in flight stays pending either way. + // Not gated on the ref: the fin callback above may have already released it + // ('end' -> autoDestroy -> set_on_data(undefined)), and the body was the only + // thing keeping such a request pending. self.mark_request_as_done_if_necessary(); } } @@ -2059,11 +2048,9 @@ impl NodeHTTPResponse { self.spill_pending_pinned_write(global_object); if IS_END { - // Like Node, a request body that is still arriving when the response - // goes out keeps being parsed and delivered to the IncomingMessage - // (which consumes it, or discards it once req._dump() ran); the - // request completes at the body's fin, not here. Only a body nobody - // is wired to receive is dropped at this point. + // Like Node, a body with a reader wired keeps being delivered after + // the response ends and the request completes at its fin; only a + // body nobody receives is dropped here. let keep_reading_body = self.body_still_arriving() && self.has_body_reader(); if !keep_reading_body && self.body_read_ref.get().has @@ -2089,12 +2076,9 @@ impl NodeHTTPResponse { raw_response.end_stream(state.is_http_connection_close()); } if keep_reading_body { - // uws's end() (markDone) dropped the connection's body data - // handler; put it back, in whichever mode the reader left it. - // A Connection: close response may have closed the socket - // synchronously inside end() (set_closed ran), in which case - // there is nothing left to read from and the close path - // releases the read state. + // uws's end() (markDone) dropped the body data handler; re-arm + // it in whichever mode the reader left it, unless end() closed + // the socket itself (the close path releases the read state). let flags = self.flags.get(); if !flags.contains(Flags::SOCKET_CLOSED) { if let Some(raw_response) = self.raw_response.get() { @@ -2297,10 +2281,9 @@ impl NodeHTTPResponse { global_object: &JSGlobalObject, value: JSValue, ) { - // A callback is only accepted while this request's body is still arriving (whether or not - // the response has ended). `.done` / a parked fin means either uSockets delivered last=true - // or JS previously cleared `ondata` (which already called clearOnData()); either way, there - // is no more body to read, so don't re-register with uSockets or churn refs. + // A callback is only accepted while the body is still arriving (whether or not the + // response has ended); past that there is no more body to read, so don't re-register + // with uSockets or churn refs. let flags = self.flags.get(); let body_still_arriving = self.body_still_arriving(); if value.is_undefined_or_null() @@ -2313,9 +2296,9 @@ impl NodeHTTPResponse { let mut stopped_reading_body = false; match self.body_read_state.get() { BodyReadState::Pending | BodyReadState::Done => { - // Once the response has completed, the connection's data handler is only - // still this request's while its body is the one being parsed (write_or_end - // re-armed it); afterwards it may already belong to the next request. + // After the response completed, the connection's data handler is only this + // request's while its body is still being parsed; afterwards it may belong + // to the next pipelined request. if (body_still_arriving || !flags.contains(Flags::REQUEST_HAS_COMPLETED)) && !flags.contains(Flags::SOCKET_CLOSED) && !flags.contains(Flags::UPGRADED) @@ -2335,9 +2318,8 @@ impl NodeHTTPResponse { .with_mut(|r| r.unref(bun_vm_mut(global_object))); } if stopped_reading_body { - // The reader was torn down (req.destroy()) mid-body and uws will not call back - // into this response for it anymore; when the response is over as well, that body - // was all that still kept the request pending. + // Reader torn down (req.destroy()) mid-body: uws won't call back for it, + // and the body may have been all that kept the request pending. self.mark_request_as_done_if_necessary(); } return; From c15ea51cb0031ed1bcda0eafc931e629472305cc Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 14 Aug 2026 01:54:03 +0000 Subject: [PATCH 7/8] node:http: move the subprocess exit test next to the other spawn tests; bind the proxy test to 127.0.0.1 The abort-events file's scenarios are verified against Node.js, so keep it free of Bun-only helpers. The proxy test's listen("localhost") can bind ::1 while the client connects to 127.0.0.1; bind the IPv4 loopback explicitly. --- .../node-http-server-abort-events.test.ts | 62 ------------------- test/js/node/http/node-http.test.ts | 61 ++++++++++++++++++ 2 files changed, 61 insertions(+), 62 deletions(-) diff --git a/test/js/node/http/node-http-server-abort-events.test.ts b/test/js/node/http/node-http-server-abort-events.test.ts index 0c9da766b993..067d64b7c2e3 100644 --- a/test/js/node/http/node-http-server-abort-events.test.ts +++ b/test/js/node/http/node-http-server-abort-events.test.ts @@ -2,7 +2,6 @@ * This test must also pass in Node.js. */ import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe } from "harness"; import { once } from "node:events"; import type { Server } from "node:http"; import { createServer, IncomingMessage, ServerResponse } from "node:http"; @@ -710,65 +709,4 @@ describe("request body arriving after the response was ended", () => { }, ); }); - - // The body outliving the response must not leave anything holding the event - // loop open: the process has to exit on its own both when the body does - // arrive later (with the connection then reused, so the early request is no - // longer the connection's current one when it closes) and when the client - // goes away mid-body instead. - test.concurrent.each(["complete", "abort"])( - "process exits on its own after an early response (%s)", - async mode => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - const { createServer } = require("node:http"); - const { connect } = require("node:net"); - const server = createServer((req, res) => { - if (req.url === "/early") { - req.on("close", () => console.log("req close complete=" + req.complete)); - req.socket.on("close", () => console.log("socket close complete=" + req.complete)); - } - res.end("ok:" + req.url); - }); - server.listen(0, "127.0.0.1", () => { - const client = connect(server.address().port, "127.0.0.1"); - let received = ""; - client.on("data", chunk => { - received += chunk; - if (received.endsWith("ok:/early")) { - received = ""; - if (${JSON.stringify(mode)} === "abort") { - client.destroy(); - server.close(() => console.log("server closed")); - } else { - client.write("def"); - client.write("GET /second HTTP/1.1\\r\\nHost: x\\r\\n\\r\\n"); - } - } else if (received.endsWith("ok:/second")) { - client.end(); - server.close(() => console.log("server closed")); - } - }); - client.write("POST /early HTTP/1.1\\r\\nHost: x\\r\\nContent-Length: 6\\r\\n\\r\\nabc"); - }); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout.trim().split("\n").sort()).toEqual( - mode === "abort" - ? ["server closed", "socket close complete=false"] - : ["req close complete=true", "server closed", "socket close complete=true"], - ); - expect(exitCode).toBe(0); - }, - 30_000, - ); }); diff --git a/test/js/node/http/node-http.test.ts b/test/js/node/http/node-http.test.ts index 55038932755b..fb936f327637 100644 --- a/test/js/node/http/node-http.test.ts +++ b/test/js/node/http/node-http.test.ts @@ -4620,4 +4620,65 @@ describe("request body still flows after res.end() was called in the handler", ( expect(stdout.trim()).toBe('["data(3)"]'); expect(exitCode).toBe(0); }, 20_000); + + // The body outliving the response must not leave anything holding the event + // loop open: the process has to exit on its own both when the body does + // arrive later (with the connection then reused, so the early request is no + // longer the connection's current one when it closes) and when the client + // goes away mid-body instead. + it.each(["complete", "abort"])( + "process exits on its own after an early response (%s)", + async mode => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { createServer } = require("node:http"); + const { connect } = require("node:net"); + const server = createServer((req, res) => { + if (req.url === "/early") { + req.on("close", () => console.log("req close complete=" + req.complete)); + req.socket.on("close", () => console.log("socket close complete=" + req.complete)); + } + res.end("ok:" + req.url); + }); + server.listen(0, "127.0.0.1", () => { + const client = connect(server.address().port, "127.0.0.1"); + let received = ""; + client.on("data", chunk => { + received += chunk; + if (received.endsWith("ok:/early")) { + received = ""; + if (${JSON.stringify(mode)} === "abort") { + client.destroy(); + server.close(() => console.log("server closed")); + } else { + client.write("def"); + client.write("GET /second HTTP/1.1\\r\\nHost: x\\r\\n\\r\\n"); + } + } else if (received.endsWith("ok:/second")) { + client.end(); + server.close(() => console.log("server closed")); + } + }); + client.write("POST /early HTTP/1.1\\r\\nHost: x\\r\\nContent-Length: 6\\r\\n\\r\\nabc"); + }); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout.trim().split("\n").sort()).toEqual( + mode === "abort" + ? ["server closed", "socket close complete=false"] + : ["req close complete=true", "server closed", "socket close complete=true"], + ); + expect(exitCode).toBe(0); + }, + 30_000, + ); }); From 3e88b4f98647053f4b233b631ddcf4d3bc27e272 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 14 Aug 2026 08:42:33 +0000 Subject: [PATCH 8/8] ci: retrigger