diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index 6f7cf7441cca..06c4d6059f0f 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -604,6 +604,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 1e880df048e9..8c7ba2956f44 100644 --- a/packages/bun-uws/src/HttpParser.h +++ b/packages/bun-uws/src/HttpParser.h @@ -496,6 +496,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; @@ -616,8 +625,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/packages/bun-uws/src/HttpResponse.h b/packages/bun-uws/src/HttpResponse.h index ac140fa59a73..f274cb92b682 100644 --- a/packages/bun-uws/src/HttpResponse.h +++ b/packages/bun-uws/src/HttpResponse.h @@ -864,6 +864,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/js/node/_http_incoming.ts b/src/js/node/_http_incoming.ts index da2f8c66f846..a7f5d28e904d 100644 --- a/src/js/node/_http_incoming.ts +++ b/src/js/node/_http_incoming.ts @@ -389,11 +389,9 @@ IncomingMessage.prototype._read = function _read(_n) { const bodyReadState = handle.hasBody; - if ( - (bodyReadState & NodeHTTPBodyReadState.done) !== 0 || - bodyReadState === NodeHTTPBodyReadState.none || - this._dumped - ) { + // 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); } @@ -406,7 +404,6 @@ IncomingMessage.prototype._read = function _read(_n) { if (!handle.ondata) { handle.ondata = onDataIncomingMessage.bind(this); - handle.hasCustomOnData = false; } }; @@ -759,10 +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"); - const handle = this[kHandle]; - if (handle) { - handle.ondata = undefined; - } + // 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 89b6db00bacd..d6febf79425b 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -898,10 +898,10 @@ 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 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); - handle.hasCustomOnData = false; } drainMicrotasks(); @@ -1421,6 +1421,10 @@ 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"); +// 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 // - on a queued response: { ops, bytes, needDrain, ended, isAncient } while it @@ -1554,6 +1558,7 @@ const NodeHTTPServerSocket = class Socket extends NetSocket { [kBytesWritten] = 0; [kHandle]; [kUpgradeIncoming] = undefined; + [kEndAfterResponse] = false; server: Server; _httpMessage; _secureEstablished = false; @@ -1834,7 +1839,7 @@ const NodeHTTPServerSocket = class Socket extends NetSocket { callback(); return; } - handle.end(); + handle.end(this[kEndAfterResponse]); callback(); } @@ -2463,10 +2468,15 @@ 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; + // Dump the body if the user never consumed or resumed it (Node's resOnFinish). + 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. @@ -2481,7 +2491,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") { @@ -3248,10 +3261,8 @@ 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 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 441e75d4fbe4..bec3fcc126a5 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp @@ -242,28 +242,33 @@ bool JSNodeHTTPServerSocket::isClosed() const } template -static bool deferShutdownUntilResponseDrains(us_socket_t* socket) +static bool deferShutdownUntilResponseDrains(us_socket_t* socket, bool afterResponseFinished) { - if (reinterpret_cast*>(socket)->getBufferedAmount() == 0) { + /* 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::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). */ + /* 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(); 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 42795b49c421..e401b335bb4a 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.h +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.h @@ -103,10 +103,10 @@ 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. Returns true after handing the close to uWS. */ - bool shutdownAfterResponseDrains(); + /* 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 * 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 59e3be6835e9..5d2cf816e4e0 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp @@ -216,9 +216,11 @@ 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()) { + // 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()); } auto bufferedSize = thisObject->streamBuffer.bufferedSize(); diff --git a/src/runtime/server/NodeHTTPResponse.rs b/src/runtime/server/NodeHTTPResponse.rs index 59caf52d354b..b75565b68859 100644 --- a/src/runtime/server/NodeHTTPResponse.rs +++ b/src/runtime/server/NodeHTTPResponse.rs @@ -92,7 +92,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? @@ -640,32 +639,57 @@ impl NodeHTTPResponse { true } - pub(crate) fn maybe_stop_reading_body(&self, vm: &mut VirtualMachine, this_value: JSValue) { + /// 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 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 + .flags + .get() + .contains(Flags::IS_DATA_BUFFERED_DURING_PAUSE_LAST) + } + + /// 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(); - 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, 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()); + 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(); } } @@ -747,6 +771,9 @@ 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 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(); server.on_request_complete(); @@ -1386,19 +1413,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(); + // 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); } // 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()); } @@ -1456,9 +1483,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). @@ -1466,16 +1491,26 @@ 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 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 .drain_buffered_request_body_from_pause(global_object)? .unwrap_or(JSValue::TRUE)) @@ -1508,7 +1543,7 @@ fn node_http_request_on_resolve(global_object: &JSGlobalObject, callframe: &Call had }); // defer this.deref(); — moved to tail. - 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) { @@ -1551,7 +1586,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)); // defer this.deref(); — moved to tail. @@ -1708,8 +1743,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. (Dumped requests move to Done first, never here.) + // 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| { @@ -1741,10 +1776,11 @@ 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: 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(); self.deref(); } } @@ -2079,12 +2115,13 @@ 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 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 && 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); @@ -2105,6 +2142,21 @@ impl NodeHTTPResponse { } else { raw_response.end_stream(state.is_http_connection_close()); } + if keep_reading_body { + // 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() { + 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)) @@ -2263,18 +2315,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 @@ -2304,23 +2348,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 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() - || 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; // defer { if body_read_ref.has { unref } } — moved to tail of this branch. match self.body_read_state.get() { BodyReadState::Pending | BodyReadState::Done => { - if !flags.contains(Flags::REQUEST_HAS_COMPLETED) + // 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) { @@ -2328,6 +2375,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); } @@ -2337,6 +2385,11 @@ impl NodeHTTPResponse { self.body_read_ref .with_mut(|r| r.unref(bun_vm_mut(global_object))); } + if stopped_reading_body { + // 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; } @@ -2346,7 +2399,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 01077a8455a2..dfcf1142a0a6 100644 --- a/src/runtime/server/mod.rs +++ b/src/runtime/server/mod.rs @@ -1486,9 +1486,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 473778e699c1..fc3c8ab583d5 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-proxy.js b/test/js/node/http/node-http-proxy.js index 8b82678ae9e4..76694392f885 100644 --- a/test/js/node/http/node-http-proxy.js +++ b/test/js/node/http/node-http-proxy.js @@ -32,7 +32,9 @@ export async function run() { req.pipe(proxyRequest); // Use pipe instead of manual data handling }); - proxyServer.listen(0, "localhost", async () => { + // "localhost" can resolve to ::1 for the listen while the client connects to + // 127.0.0.1; bind the IPv4 loopback explicitly so they always match. + proxyServer.listen(0, "127.0.0.1", async () => { const address = proxyServer.address(); const options = { 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 ac148f126e44..05ef9bbf3b1a 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 @@ -1,10 +1,11 @@ /** * This test must also pass in Node.js. */ -import { expect, test } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { once } from "node:events"; +import type { IncomingMessage, Server } from "node:http"; import { createServer } from "node:http"; -import type { AddressInfo } from "node:net"; +import type { AddressInfo, Socket } from "node:net"; import { connect } from "node:net"; test("aborted request body emits 'error' ECONNRESET and res 'close' before req 'close'", async () => { @@ -47,3 +48,413 @@ test("aborted request body emits 'error' ECONNRESET and res 'close' before req ' server.close(); } }); + +// 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(); + } + }); + + // 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; + } + + // 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) => { + 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 response = await requestUntilClosed(server, `${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 response made it out before the connection was closed. + expect(response).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 { + 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, + readableEnded: true, + destroyed: true, + aborted: false, + }); + 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 { + // The rest of the body never comes; the server closes the connection + // after the response regardless, like Node's destroySoon(). + 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, + readableEnded: false, + destroyed: false, + aborted: false, + }); + expect(response).toStartWith("HTTP/1.1 200 OK"); + // The request was released even though its body never completed. + await closeServer(server); + } finally { + server.closeAllConnections(); + 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(); + } + }, + ); + }); +}); diff --git a/test/js/node/http/node-http.test.ts b/test/js/node/http/node-http.test.ts index b9edb1f12756..92dda78cdc5f 100644 --- a/test/js/node/http/node-http.test.ts +++ b/test/js/node/http/node-http.test.ts @@ -4139,3 +4139,319 @@ it("connectionListener hands off Upgrade and CONNECT like Node", async () => { expect(serverSide.destroyed).toBe(true); } }); + +// 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); + + // 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, + ); +});