node:http: slice 'upgrade' head to the post-body remainder and close on unsatisfied-body FIN

packages/bun-uws/src/HttpContext.h

@@ -460,6 +460,12 @@ struct HttpContext {
              * while this dispatch is on the stack. */
             if constexpr (IsNodeHttp) {
                 httpResponseData->state &= ~HttpResponseData<SSL>::HTTP_NODE_PIPELINED_DISPATCH;
+                /* Same-chunk Upgrade body remainder was already delivered as
+                 * req->head: suppress the one onSocketData call for it below. */
+                if (httpRequest->bodyCompleteInHead
+                        && (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY)) {
+                    httpResponseData->state |= HttpResponseData<SSL>::HTTP_NODE_TUNNEL_HEAD_PENDING;
+                }
             }
 
             /* Returning from a request handler without responding or attaching an onAborted handler is ill-use */
@@ -500,7 +506,11 @@ struct HttpContext {
             }
 
             if (httpResponseData->isConnectRequest && httpResponseData->socketData && httpContextData->onSocketData) {
-                httpContextData->onSocketData(httpResponseData->socketData, SSL, (struct us_socket_t *) user, data.data(), data.length(), fin);
+                if (IsNodeHttp && (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_HEAD_PENDING)) {
+                    httpResponseData->state &= ~HttpResponseData<SSL>::HTTP_NODE_TUNNEL_HEAD_PENDING;
+                } else {
+                    httpContextData->onSocketData(httpResponseData->socketData, SSL, (struct us_socket_t *) user, data.data(), data.length(), fin);
+                }
             }
 
             if (switchToTunnelAfterThisChunk) {
@@ -597,6 +607,8 @@ struct HttpContext {
                     nodeHttpResponseData->lastMessageStartMs = nodeCompatMonotonicMs();
                     nodeHttpResponseData->headersCompleted = false;
                 }
+                /* Scope the same-chunk-head suppression to this parse call only. */
+                httpResponseData->state &= ~HttpResponseData<SSL>::HTTP_NODE_TUNNEL_HEAD_PENDING;
             }
 
             /* Timeout on uncork failure */
@@ -780,15 +792,20 @@ struct HttpContext {
             /* CONNECT/Upgrade tunnels allow half-open: the peer finishing its
              * writable side ends the JS socket's readable side ('end' event) but
              * the server can keep writing until it ends the socket itself, like
-             * Node's http server (allowHalfOpen: true). This includes an accepted
-             * Upgrade whose body never completed (HTTP_NODE_TUNNEL_AFTER_BODY): the
-             * EOF ends the upgrade socket, exactly like Node's UpgradeStream. */
-            if (httpResponseData->isConnectRequest || (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY)) {
+             * Node's http server (allowHalfOpen: true). */
+            if (httpResponseData->isConnectRequest) {
                 if (httpResponseData->socketData && httpContextData->onSocketData) {
                     httpContextData->onSocketData(httpResponseData->socketData, SSL, s, "", 0, true);
                 }
                 return s;
             }
+            /* Accepted Upgrade whose body never completed: Node tears the
+             * connection down (unsatisfied body, no valid tunnel state to keep
+             * half-open). onClose delivers the fin and releases the request. */
+            if (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY) {
+                asyncSocket->close();
+                return s;
+            }
 
             if (httpContextData->onClientError && !(httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_PARSING_STOPPED)
                 && (httpResponseData->hasBufferedPartialRequestHeaders()

packages/bun-uws/src/HttpParser.h

@@ -209,6 +209,10 @@ struct HttpResponseData;
          * WARNING: This points to data in the receive buffer and may be stack-allocated.
          * Must be cloned before the request handler returns. */
         std::span<const char> head;
+        /* For an Upgrade request with a body, whether that body completes
+         * within this same parse chunk (so head above is the post-body
+         * remainder, and req.complete can be set at 'upgrade' time). */
+        bool bodyCompleteInHead = false;
 
         bool isAncient()
         {
@@ -1248,8 +1252,23 @@ struct HttpResponseData;
             /* If returned socket is not what we put in we need
              * to break here as we either have upgraded to
              * WebSockets or otherwise closed the socket. */
-            /* Store any remaining data as head for Node.js compat (connect/upgrade events) */
+            /* Store any remaining data as head for Node.js compat (connect/upgrade events).
+             * For an Upgrade with a Content-Length body that completes in this
+             * chunk, slice head to the post-body remainder like Node. */
+            req->bodyCompleteInHead = false;
             req->head = std::span<const char>(data, length);
+            if constexpr (IsNodeHttp && !ConsumeMinimally) {
+                if (!isConnectRequest && contentLengthStringLen && !transferEncoding.has
+                        && remainingStreamingBytes <= (uint64_t) length
+                        && req->getHeader("upgrade").data()) [[unlikely]] {
+                    std::string_view m = req->getCaseSensitiveMethod();
+                    if (m != "HEAD" && m != "TRACE") {
+                        unsigned int skip = (unsigned int) remainingStreamingBytes;
+                        req->head = std::span<const char>(data + skip, length - skip);
+                        req->bodyCompleteInHead = true;
+                    }
+                }
+            }
             void *returnedUser = requestHandler(user, req);
             if (returnedUser != user) {
                 /* We are upgraded to WebSocket or otherwise broken */

packages/bun-uws/src/HttpResponseData.h

@@ -131,6 +131,11 @@ struct HttpResponseData : AsyncSocketData<SSL>, HttpParser {
          * and everything after the end of the message is opaque data for the
          * 'upgrade' listener's socket. */
         HTTP_NODE_TUNNEL_AFTER_BODY = 1 << 14,
+        /* The tunnel-after-body switch just happened during this parse call.
+         * The post-body remainder of the same chunk is the 'upgrade' head
+         * argument (already handed to JS via req->head), not raw socket data,
+         * so the next isConnectRequest onSocketData is suppressed once. */
+        HTTP_NODE_TUNNEL_HEAD_PENDING = 1 << 17,
         /* The peer half-closed (FIN) while there was still something to flush
          * before teardown: buffered/pinned response bytes, or (with
          * httpAllowHalfOpen) an in-flight or queued pipelined response. The

src/js/internal/http.ts

@@ -98,6 +98,7 @@ const serverSymbol = Symbol.for("::bunternal::");
 const kPendingCallbacks = Symbol("pendingCallbacks");
 const kRequest = Symbol("request");
 const kCloseCallback = Symbol("closeCallback");
+const kDeferredUpgradeEmit = Symbol("deferredUpgradeEmit");
 
 const kEmptyObject = Object.freeze(Object.create(null));
 
@@ -268,6 +269,16 @@ function onDataIncomingMessage(this: any, chunk, isLast, aborted: NodeHTTPRespon
   }
 
   if (isLast) {
+    // Deferred 'upgrade' emit: body was in the same chunk as the headers, so
+    // fire now that it is buffered on req (Node's parser.execute() order).
+    const deferred = this[kDeferredUpgradeEmit];
+    if (deferred !== undefined) {
+      this[kDeferredUpgradeEmit] = undefined;
+      this[eofInProgress] = true;
+      emitEOFIncomingMessageOuter(this);
+      deferred();
+      return;
+    }
     emitEOFIncomingMessage(this);
     // Like Node's parserOnMessageComplete: any readStop above left the shared
     // socket's flowing=false, which would swallow the next request's 'pause'.
@@ -625,6 +636,7 @@ export {
   kBodyChunks,
   kClearTimeout,
   kCloseCallback,
+  kDeferredUpgradeEmit,
   kDeprecatedReplySymbol,
   kEmitState,
   kEmptyObject,

src/js/node/_http_server.ts

@@ -41,6 +41,7 @@ const {
   kPendingCallbacks,
   kRequest,
   kCloseCallback,
+  kDeferredUpgradeEmit,
   NodeHTTPResponseFlags,
   emitErrorNextTickIfErrorListenerNT,
   getIsNextIncomingMessageHTTPS,
@@ -717,6 +718,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
         isAncientHTTP: boolean,
         connectHead?: Buffer,
         isPipelinedDispatch?: boolean,
+        bodyCompleteInHead?: boolean,
       ) {
         const prevIsNextIncomingMessageHTTPS = getIsNextIncomingMessageHTTPS();
         setIsNextIncomingMessageHTTPS(isHTTPS);
@@ -992,13 +994,10 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
           }
         } else if (is_upgrade) {
           // Hand the raw socket over to the 'upgrade' listener, like Node.js.
-          // Without a body the message is already complete: bytes that arrived
-          // after the request head become the upgradeHead and the connection
-          // switches into CONNECT-style tunnel mode immediately. With a body
-          // (Node 26 semantics) the body keeps being parsed and delivered
-          // through req; the connection only switches to tunnel mode once the
-          // message completes, so the upgradeHead is empty and everything after
-          // the end of the message reaches the socket as raw data.
+          // With a body (Node 26 semantics) the body keeps being parsed and
+          // delivered through req; the connection only switches to tunnel mode
+          // once the message completes, and head is the post-body remainder of
+          // the same chunk (the parser already computed it in connectHead).
           socketHandle.upgradeToTunnel(hasBody);
           socket[kEnableStreaming](true);
           detachSocketListenersForHandoff(socket);
@@ -1008,28 +1007,38 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
             socket[kUpgradeIncoming] = http_req;
             http_req.once("end", clearUpgradeIncoming.bind(undefined, socket));
           }
-          const upgradeHead = !hasBody && connectHead ? connectHead : kEmptyBuffer;
-          let upgradeHandled;
-          try {
-            upgradeHandled = server.emit("upgrade", http_req, socket, upgradeHead);
-          } catch (err) {
-            // A throwing 'upgrade' listener surfaces as an uncaught
-            // exception, like Node.js (the emit happens outside any JS try
-            // frame there).
-            process.nextTick(rethrowUncaught, err);
-            upgradeHandled = true;
-          }
-          if (!upgradeHandled) {
-            // shouldUpgradeCallback accepted the upgrade but no 'upgrade'
-            // listener is installed: Node.js destroys the socket.
-            socket.destroy();
-            return;
-          }
+          const upgradeHead = (bodyCompleteInHead || !hasBody) && connectHead ? connectHead : kEmptyBuffer;
           // Like CONNECT: the connection is detached from the HTTP request
           // machinery; hold the native callback open until the raw socket
           // closes.
           const { promise: upgradePromise, resolve: resolveUpgrade } = $newPromiseCapability(Promise);
-          socket.once("close", resolveUpgrade);
+          const emitUpgrade = () => {
+            let handled;
+            try {
+              handled = server.emit("upgrade", http_req, socket, upgradeHead);
+            } catch (err) {
+              // A throwing 'upgrade' listener surfaces as an uncaught
+              // exception, like Node.js (the emit happens outside any JS try
+              // frame there).
+              process.nextTick(rethrowUncaught, err);
+              handled = true;
+            }
+            if (!handled) {
+              // shouldUpgradeCallback accepted the upgrade but no 'upgrade'
+              // listener is installed: Node.js destroys the socket.
+              socket.destroy();
+            }
+            // Attach the internal close listener after the user's handler ran:
+            // Node.js hands the socket over with no listeners (same as CONNECT).
+            socket.once("close", resolveUpgrade);
+          };
+          if (hasBody && bodyCompleteInHead) {
+            // Body is in this chunk: defer to onDataIncomingMessage's fin so
+            // the listener observes req.complete and the buffered body.
+            http_req[kDeferredUpgradeEmit] = emitUpgrade;
+          } else {
+            emitUpgrade();
+          }
           return upgradePromise;
         } else if (
           server.requireHostHeader &&

src/jsc/bindings/NodeHTTP.cpp

@@ -719,6 +719,7 @@ static EncodedJSValue NodeHTTPServer__onRequest(
     }
 
     args.append(jsBoolean(isPipelinedDispatch));
+    args.append(jsBoolean(request->bodyCompleteInHead));
 
     JSValue returnValue = AsyncContextFrame::profiledCall(globalObject, callbackObject, jsUndefined(), args);
     RETURN_IF_EXCEPTION(scope, {});