node:http: keep receiving the request body after the response has ended

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<SSL> *) 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<SSL> *) s)->closeIfDoneAndMarked(httpResponseData);
+                }
                 return s;
             }
             if(httpContextData->onClientError) {

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;

packages/bun-uws/src/HttpResponse.h

@@ -851,6 +851,22 @@ struct HttpResponse : public AsyncSocket<SSL> {
         return !(httpResponseData->state & HttpResponseData<SSL>::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<SSL> *httpResponseData = getHttpResponseData();
+
+        return hasResponded()
+            && !httpResponseData->isConnectRequest
+            && httpResponseData->inStream != nullptr
+            && HttpContext<SSL>::fromSocket((us_socket_t *) this)->getSocketContextData()->parsingSocket == (us_socket_t *) this;
+    }
+
      /* Corks the response if possible. Leaves already corked socket be. */
     HttpResponse *cork(MoveOnlyFunction<void()> &&handler) {
         if (!Super::isCorked()) {

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();
   }
 };

src/js/node/_http_server.ts

@@ -824,10 +824,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();
 
@@ -1347,6 +1347,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
@@ -1485,6 +1489,7 @@ function getNodeHTTPServerSocket() {
     [kBytesWritten] = 0;
     [kHandle];
     [kUpgradeIncoming] = undefined;
+    [kEndAfterResponse] = false;
     server: Server;
     _httpMessage;
     _secureEstablished = false;
@@ -1765,7 +1770,7 @@ function getNodeHTTPServerSocket() {
         callback();
         return;
       }
-      handle.end();
+      handle.end(this[kEndAfterResponse]);
       callback();
     }
 
@@ -2387,10 +2392,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.
@@ -2405,7 +2415,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") {
@@ -3172,10 +3185,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'

src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp

@@ -243,28 +243,33 @@ bool JSNodeHTTPServerSocket::isClosed() const
 }
 
 template<bool SSL>
-static bool deferShutdownUntilResponseDrains(us_socket_t* socket)
+static bool deferShutdownUntilResponseDrains(us_socket_t* socket, bool afterResponseFinished)
 {
-    if (reinterpret_cast<uWS::AsyncSocket<SSL>*>(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<uWS::HttpResponse<SSL>*>(socket)->isDeliveringBodyAfterResponse();
+    if (!bodyStillParsing && reinterpret_cast<uWS::AsyncSocket<SSL>*>(socket)->getBufferedAmount() == 0) {
         return false;
     }
-    /* HttpContext<SSL>::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<uWS::HttpResponseData<SSL>*>(us_socket_ext(socket));
     httpResponseData->state |= uWS::HttpResponseData<SSL>::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<true>(socket);
+        return deferShutdownUntilResponseDrains<true>(socket, afterResponseFinished);
     }
-    return deferShutdownUntilResponseDrains<false>(socket);
+    return deferShutdownUntilResponseDrains<false>(socket, afterResponseFinished);
 }
 
 template<bool SSL>

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

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();