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,14 @@
              * while this dispatch is on the stack. */
             if constexpr (IsNodeHttp) {
                 httpResponseData->state &= ~HttpResponseData<SSL>::HTTP_NODE_PIPELINED_DISPATCH;
+                /* Upgrade body (and its post-body remainder) are fully present in
+                 * this same chunk: the remainder was already delivered as the
+                 * 'upgrade' head via req->head, so suppress it from onSocketData
+                 * once the tunnel switch fires below. Cleared at end of onData. */
+                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 +508,13 @@
             }
 
             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)) {
+                    /* Same-chunk post-body remainder of an Upgrade-with-body: already
+                     * delivered as the 'upgrade' head (req->head), not socket data. */
+                    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) {
@@ -550,6 +564,10 @@
 
         /* Mark that we are no longer parsing Http */
         httpContextData->flags.isParsingHttp = false;
+        if constexpr (IsNodeHttp) {
+            /* Scope the same-chunk-head suppression to this parse call only. */
+            httpResponseData->state &= ~HttpResponseData<SSL>::HTTP_NODE_TUNNEL_HEAD_PENDING;
+        }
         /* If we got fullptr that means the parser wants us to close the socket from error (same as calling the errorHandler) */
         if (httpErrorStatusCode) {
             /* node:http compat: parse errors surface as the server's 'clientError'
@@ -780,15 +798,25 @@
             /* 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's UpgradeStream
+             * ends the socket's readable side and then tears the connection down
+             * (the declared body is unsatisfied, so there is no valid tunnel
+             * state to keep half-open). Closing here drives onClose, which
+             * delivers the body abort and releases the pending-request ref. */
+            if (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY) {
+                if (httpResponseData->socketData && httpContextData->onSocketData) {
+                    httpContextData->onSocketData(httpResponseData->socketData, SSL, s, "", 0, true);
+                }
+                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 @@
          * 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,41 @@
             /* 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).
+             * Node's 'upgrade' head is the chunk remainder AFTER the declared
+             * body, so for a body-carrying Upgrade skip the body bytes (which
+             * are delivered through inStream below) and report whether the
+             * body completes in this chunk. */
+            req->bodyCompleteInHead = false;
             req->head = std::span<const char>(data, length);
+            if constexpr (IsNodeHttp) {
+                if (!isConnectRequest && (transferEncoding.has || contentLengthStringLen)
+                        && req->getHeader("upgrade").data()) [[unlikely]] {
+                    if (transferEncoding.has) {
+                        /* Pre-scan the chunked body with local state so the real
+                         * parse below (which fires inStream per chunk) is unchanged. */
+                        uint64_t scanState = STATE_IS_CHUNKED;
+                        uint64_t scanExt = 0;
+                        std::string scanTrailers;
+                        std::string_view scanView(data, length);
+                        for (auto chunk : uWS::ChunkIterator(&scanView, &scanState, false, &scanExt, &scanTrailers, maxBufferedHeaderSize)) {
+                            (void) chunk;
+                        }
+                        if (!isParsingChunkedEncoding(scanState) && !isParsingInvalidChunkedEncoding(scanState)) {
+                            req->head = std::span<const char>(scanView.data(), scanView.length());
+                            req->bodyCompleteInHead = true;
+                        } else {
+                            req->head = std::span<const char>();
+                        }
+                    } else if (remainingStreamingBytes <= (uint64_t) length) {
+                        unsigned int skip = (unsigned int) remainingStreamingBytes;
+                        req->head = std::span<const char>(data + skip, length - skip);
+                        req->bodyCompleteInHead = true;
+                    } else {
+                        req->head = std::span<const char>();
+                    }
+                }
+            }
             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));
 
@@ -272,6 +273,14 @@ function onDataIncomingMessage(this: any, chunk, isLast, aborted: NodeHTTPRespon
     // Like Node's parserOnMessageComplete: any readStop above left the shared
     // socket's flowing=false, which would swallow the next request's 'pause'.
     if (!this.upgrade && socket && !socket._paused && socket.readable) socket.resume();
+    // 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.complete = true;
+      deferred();
+    }
   }
 }
 
@@ -625,6 +634,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,36 @@ 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 = connectHead ? connectHead : kEmptyBuffer;
+          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();
+            }
+          };
           // 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);
+          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, {});